From 0d7d2c90ca2a61dd679bbc98ecf69d865c8e8055 Mon Sep 17 00:00:00 2001 From: amit Date: Wed, 10 Feb 2016 19:42:54 +0900 Subject: [PATCH 09/10] Tuple routing functionality. Newly introduced get_partition_for_tuple takes a tuple (TupleTableSlot) and returns OID of the leaf partition to which the tuple maps to. Teach CopyFrom() and ExecInsert() to utilize this new interface. Newly introduced PartitionKeyExecInfo structure can be used to derive partition key from a tuple being routed. It mirrors the PartitionKey but contains some executor state as well. New field ri_PartitionKeyInfo is a list of these structures. ResultRelInfo has one more new field ri_PartitionDesc which is the partition descriptor of the root table and initialized in InitResultRelInfo along with ri_PartitionKeyInfo. Have to find a way to be able to somehow cache relation descriptors of internal partition in EState because they need to be computed (well copied from relcache) for each tuple routed. ExecUpdate and ExecDelete require more information viz. OID of the partition that the modified or to-be-deleted row came from. Extend JunkFilter mechanism to help with that. Also, during rewrite, if the UPDATE/DELETE target is a partitioned table, add junk TLE for tableoid column so that the above works. Of course, UPDATE and DELETE cases don't work yet because there is no way yet to scan a partitioned table which will be fixed shortly. Note that INSERT's ON CONFLICT clause is not supported. --- src/backend/catalog/partition.c | 338 ++++++++++++++++++++++++++++++++ src/backend/commands/copy.c | 106 ++++++++++- src/backend/executor/execMain.c | 26 +-- src/backend/executor/nodeModifyTable.c | 299 ++++++++++++++++++++++++++-- src/backend/parser/parse_clause.c | 9 + src/backend/rewrite/rewriteHandler.c | 25 +++ src/include/catalog/partition.h | 9 + src/include/executor/executor.h | 15 ++ src/include/nodes/execnodes.h | 25 +++ src/include/nodes/nodes.h | 1 + 10 files changed, 815 insertions(+), 38 deletions(-) diff --git a/src/backend/catalog/partition.c b/src/backend/catalog/partition.c index cf48bf4..9baa0a0 100644 --- a/src/backend/catalog/partition.c +++ b/src/backend/catalog/partition.c @@ -76,6 +76,31 @@ static int32 rangemaxs_cmp_datum(const Datum *vals, int partnatts, FmgrInfo *partsupfunc); +static void FormPartitionKeyDatum(PartitionKeyExecInfo *pkinfo, + TupleTableSlot *slot, + EState *estate, + Datum *values, + bool *isnull); +static Oid get_partition_for_tuple_recurse(Relation rel, + ListCell *key_cell, + ListCell *pkinfo_cell, + PartitionDesc pdesc, + TupleTableSlot *slot, + EState *estate); +static Oid list_partition_for_tuple(PartitionKey key, PartitionDesc pdesc, + Datum value); +static Oid range_partition_for_tuple(PartitionKey key, PartitionDesc pdesc, + Datum *values); +typedef int32 (*bsearch_compare_fn) (const Datum *vals, + const Datum *rangemaxs, + int partnatts, + FmgrInfo *partsupfunc); +static int range_minmax_bsearch(PartitionDesc pdesc, + FmgrInfo *partsupfunc, + int partnatts, + const Datum *key, + bsearch_compare_fn cmp_fn); + /* * StorePartitionKey * Store the partition keys of rel into pg_partitioned_rel catalog @@ -1447,3 +1472,316 @@ rangemaxs_cmp_datum(const Datum *vals, const Datum *rangemaxs, return 0; } + +/* ---------------- + * BuildPartitionKeyExecInfo + * Construct a list of PartitionKeyExecInfo records for an open + * relation + * + * PartitionKeyExecInfo stores the information about the partition key + * that's needed when inserting tuples into a partitioned relations; + * especially partition expression state. Normally we build a + * PartitionKeyExecInfo for a partitioned relation just once per + * command, and then use it for (potentially) many tuples. + * ---------------- + */ +List * +BuildPartitionKeyExecInfo(Relation rel) +{ + List *result = NIL; + ListCell *cell; + int16 level; + + level = 1; + foreach (cell, rel->rd_partkeys) + { + int i; + PartitionKey key = (PartitionKey) lfirst(cell); + PartitionKeyExecInfo *pi = makeNode(PartitionKeyExecInfo); + int numKeys = key->partnatts; + + if (numKeys < 1 || numKeys > PARTITION_MAX_KEYS) + elog(ERROR, "invalid partnatts %d for level %d key of partitioned" + " relation %u", numKeys, level, RelationGetRelid(rel)); + + pi->pi_NumKeyAttrs = numKeys; + for (i = 0; i < numKeys; i++) + pi->pi_KeyAttrNumbers[i] = key->partattrs[i]; + + /* Fetch any partition key expressions */ + pi->pi_Expressions = copyObject(key->partexprs); + pi->pi_ExpressionsState = NIL; + + result = lappend(result, pi); + level++; + } + + return result; +} + +/* + * FormPartitionKeyDatum + * Construct values[] and isnull[] arrays for partition key columns + * from the tuple contained in slot + */ +static void +FormPartitionKeyDatum(PartitionKeyExecInfo *pkinfo, + TupleTableSlot *slot, + EState *estate, + Datum *values, + bool *isnull) +{ + ListCell *partexpr_item; + int i; + + if (pkinfo->pi_Expressions != NIL && + pkinfo->pi_ExpressionsState == NIL) + { + /* First time through, set up expression evaluation state */ + pkinfo->pi_ExpressionsState = (List *) + ExecPrepareExpr((Expr *) pkinfo->pi_Expressions, + estate); + /* Check caller has set up context correctly */ + Assert(GetPerTupleExprContext(estate)->ecxt_scantuple == slot); + } + partexpr_item = list_head(pkinfo->pi_ExpressionsState); + + for (i = 0; i < pkinfo->pi_NumKeyAttrs; i++) + { + AttrNumber keycol = pkinfo->pi_KeyAttrNumbers[i]; + Datum pkDatum; + bool isNull; + + if (keycol != 0) + { + /* Plain column; get the value directly from the heap tuple */ + pkDatum = slot_getattr(slot, keycol, &isNull); + } + else + { + /* Expression; need to evaluate it */ + if (partexpr_item == NULL) + elog(ERROR, "wrong number of partition key expressions"); + pkDatum = ExecEvalExprSwitchContext((ExprState *) lfirst(partexpr_item), + GetPerTupleExprContext(estate), + &isNull, + NULL); + partexpr_item = lnext(partexpr_item); + } + values[i] = pkDatum; + isnull[i] = isNull; + } + + if (partexpr_item != NULL) + elog(ERROR, "wrong number of partition key expressions"); +} + +/* + * get_partition_for_tuple + * Find the partition for tuple (in slot) + */ +Oid +get_partition_for_tuple(Relation rel, + List *pkinfo, + PartitionDesc pdesc, + TupleTableSlot *slot, + EState *estate) +{ + if (pdesc->numparts == 0) + return InvalidOid; + + return get_partition_for_tuple_recurse(rel, + list_head(rel->rd_partkeys), + list_head(pkinfo), + pdesc, slot, estate); +} + +static Oid +get_partition_for_tuple_recurse(Relation rel, + ListCell *key_cell, + ListCell *pkinfo_cell, + PartitionDesc pdesc, + TupleTableSlot *slot, + EState *estate) +{ + PartitionKey key = lfirst(key_cell); + PartitionKeyExecInfo *pkinfo = lfirst(pkinfo_cell); + Relation targetRel; + PartitionKey targetKey; + PartitionDesc target_pdesc; + Oid targetOid, + relid = RelationGetRelid(rel); + Datum values[PARTITION_MAX_KEYS]; + bool isnull[PARTITION_MAX_KEYS]; + int i; + + if (pdesc->numparts == 0) + return InvalidOid; + + /* Extract partition key from tuple */ + Assert(GetPerTupleExprContext(estate)->ecxt_scantuple == slot); + FormPartitionKeyDatum(pkinfo, + slot, + estate, + values, + isnull); + + /* Disallow nulls in partition key */ + for (i = 0; i < pkinfo->pi_NumKeyAttrs; i++) + if (isnull[i]) + ereport(ERROR, + (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED), + errmsg("null value for partition key column \"%s\" not allowed", + get_attname(relid, pkinfo->pi_KeyAttrNumbers[i])))); + + + switch (key->strategy) + { + case PARTITION_STRAT_LIST: + /* Remember, list partition key always consists of single column */ + targetOid = list_partition_for_tuple(key, pdesc, values[0]); + break; + + case PARTITION_STRAT_RANGE: + targetOid = range_partition_for_tuple(key, pdesc, values); + break; + } + + /* this is how recursion ends (or doesn't occur at all) */ + if (lnext(key_cell) == NULL || !OidIsValid(targetOid)) + return targetOid; + + targetRel = heap_open(targetOid, AccessShareLock); + targetKey = lfirst(lnext(key_cell)); + target_pdesc = RelationGetPartitionDesc(targetRel, targetKey, true); + targetOid = get_partition_for_tuple_recurse(rel, lnext(key_cell), + lnext(pkinfo_cell), + target_pdesc, slot, estate); + heap_close(targetRel, AccessShareLock); + + return targetOid; +} + +/* + * list_partition_for_tuple + * Find the list partition for a tuple + * + * There is single datum to compare because the partition key in case of + * list partitioning consists of only one column. + */ +static Oid +list_partition_for_tuple(PartitionKey key, PartitionDesc pdesc, Datum value) +{ + int i, + j; + int32 result; + FmgrInfo cmpfn = key->partsupfunc[0]; + + Assert(pdesc->numparts > 0); + + for (i = 0; i < pdesc->numparts; i++) + { + for (j = 0; j < pdesc->listnvalues[i]; j++) + { + result = compare_using_func(cmpfn, + pdesc->listvalues[i][j], + value); + + if (!result) + return pdesc->oids[i]; + } + } + + return InvalidOid; +} + +/* + * range_partition_for_tuple + * Find the range partition for a tuple + * + * Range partitioning supports multi-column key, so the key may contain + * more than one datum. Comparison function rangemaxs_cmp_datum can + * deal with that. + */ +static Oid +range_partition_for_tuple(PartitionKey key, PartitionDesc pdesc, Datum *values) +{ + int partidx; + + Assert(pdesc->numparts > 0); + + partidx = range_minmax_bsearch(pdesc, + key->partsupfunc, + key->partnatts, + values, + rangemaxs_cmp_datum); + + if (partidx >= 0) + return pdesc->oids[partidx]; + + return InvalidOid; +} + +/* + * range_minmax_bsearch + * Search for partition such that left.rangemax <= key < rangemax + * + * Returns index of the so found partition or -1 if key is found to be + * greater than the rangemax of the last partition. + */ +static int +range_minmax_bsearch(PartitionDesc pdesc, + FmgrInfo *partsupfunc, + int partnatts, + const Datum *key, + bsearch_compare_fn cmp_fn) +{ + int32 cmpval; + int low, high, idx, i, result; + Datum *rangemaxs; + + rangemaxs = (Datum *) palloc0(partnatts * sizeof(Datum)); + + /* Good ol' bsearch */ + low = 0; + high = pdesc->numparts - 1; + result = -1; + while (low <= high) + { + idx = (low + high) / 2; + + for (i = 0; i < partnatts; i++) + rangemaxs[i] = pdesc->rangemaxs[i][idx]; + + cmpval = cmp_fn(key, rangemaxs, partnatts, partsupfunc); + if (cmpval < 0) + { + if (idx == 0) + { + /* Nothing on left */ + result = idx; + break; + } + else + { + for (i = 0; i < partnatts; i++) + rangemaxs[i] = pdesc->rangemaxs[i][idx-1]; + + cmpval = cmp_fn(key, rangemaxs, partnatts, partsupfunc); + + if (cmpval >= 0) + { + result = idx; + break; + } + else + high = idx - 1; + } + } + else + low = idx + 1; + } + + pfree(rangemaxs); + return result; +} diff --git a/src/backend/commands/copy.c b/src/backend/commands/copy.c index 62b650f..e54a61e 100644 --- a/src/backend/commands/copy.c +++ b/src/backend/commands/copy.c @@ -36,6 +36,7 @@ #include "miscadmin.h" #include "optimizer/clauses.h" #include "optimizer/planner.h" +#include "parser/parsetree.h" #include "nodes/makefuncs.h" #include "rewrite/rewriteHandler.h" #include "storage/fd.h" @@ -43,6 +44,7 @@ #include "utils/builtins.h" #include "utils/lsyscache.h" #include "utils/memutils.h" +#include "catalog/partition.h" #include "utils/portal.h" #include "utils/rel.h" #include "utils/rls.h" @@ -2223,6 +2225,7 @@ CopyFrom(CopyState cstate) Datum *values; bool *nulls; ResultRelInfo *resultRelInfo; + ResultRelInfo *saved_resultRelInfo = NULL; /* for partitioned table case */ EState *estate = CreateExecutorState(); /* for ExecConstraints() */ ExprContext *econtext; TupleTableSlot *myslot; @@ -2231,7 +2234,7 @@ CopyFrom(CopyState cstate) ErrorContextCallback errcallback; CommandId mycid = GetCurrentCommandId(true); int hi_options = 0; /* start with default heap_insert options */ - BulkInsertState bistate; + BulkInsertState bistate = NULL; uint64 processed = 0; bool useHeapMultiInsert; int nBufferedTuples = 0; @@ -2240,6 +2243,8 @@ CopyFrom(CopyState cstate) HeapTuple *bufferedTuples = NULL; /* initialize to silence warning */ Size bufferedTuplesSize = 0; int firstBufferedLineNo = 0; + Oid prev_partOid = InvalidOid; + ResultRelInfo *prev_partrri = NULL; Assert(cstate->rel); @@ -2381,10 +2386,14 @@ CopyFrom(CopyState cstate) * expressions. Such triggers or expressions might query the table we're * inserting to, and act differently if the tuples that have already been * processed and prepared for insertion are not there. + * + * Also, Do not use multi- mode if inserting into a partitioned table + * because we need to determine a partition for each row individually. */ if ((resultRelInfo->ri_TrigDesc != NULL && (resultRelInfo->ri_TrigDesc->trig_insert_before_row || resultRelInfo->ri_TrigDesc->trig_insert_instead_row)) || + resultRelInfo->ri_PartitionDesc != NULL || cstate->volatile_defexprs) { useHeapMultiInsert = false; @@ -2409,7 +2418,9 @@ CopyFrom(CopyState cstate) values = (Datum *) palloc(tupDesc->natts * sizeof(Datum)); nulls = (bool *) palloc(tupDesc->natts * sizeof(bool)); - bistate = GetBulkInsertState(); + if (resultRelInfo->ri_PartitionDesc == NULL) + bistate = GetBulkInsertState(); + econtext = GetPerTupleExprContext(estate); /* Set up callback to identify error line number */ @@ -2481,6 +2492,76 @@ CopyFrom(CopyState cstate) if (cstate->rel->rd_att->constr) ExecConstraints(resultRelInfo, slot, estate); + /* Find the partition for this tuple */ + if (resultRelInfo->ri_PartitionDesc) + { + Oid partOid; + + econtext->ecxt_scantuple = slot; /* for partition key expressions */ + partOid = get_partition_for_tuple(cstate->rel, + resultRelInfo->ri_PartitionKeyInfo, + resultRelInfo->ri_PartitionDesc, + slot, + estate); + if (OidIsValid(partOid)) + { + saved_resultRelInfo = resultRelInfo; + + /* check if new tuple mapped to the same partition as the last one */ + if (partOid != prev_partOid) + { + Relation partRelDesc; + + /* close the last partition */ + if (prev_partrri) + { + heap_close(prev_partrri->ri_RelationDesc, RowExclusiveLock); + ExecCloseIndices(prev_partrri); + pfree(prev_partrri); + } + + /* open the new one */ + partRelDesc = heap_open(partOid, RowExclusiveLock); + resultRelInfo = makeNode(ResultRelInfo); + InitResultRelInfo(resultRelInfo, + partRelDesc, + 1, /* dummy */ + 0); + ExecOpenIndices(resultRelInfo, false); + prev_partOid = partOid; + } + else + { + /* keep using the last one */ + resultRelInfo = prev_partrri; + } + + estate->es_result_relation_info = resultRelInfo; + } + else + { + char *val_desc; + Bitmapset *modifiedCols; + Bitmapset *insertedCols; + Bitmapset *updatedCols; + + insertedCols = GetInsertedColumns(resultRelInfo, estate); + updatedCols = GetUpdatedColumns(resultRelInfo, estate); + modifiedCols = bms_union(insertedCols, updatedCols); + val_desc = ExecBuildSlotValueDescription(RelationGetRelid(cstate->rel), + slot, + RelationGetDescr(cstate->rel), + modifiedCols, + 64); + + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot find partition of relation \"%s\" for the new row", + RelationGetRelationName(cstate->rel)), + val_desc ? errdetail("Failing row contains %s.", val_desc) : 0)); + } + } + if (useHeapMultiInsert) { /* Add this tuple to the tuple buffer */ @@ -2511,7 +2592,8 @@ CopyFrom(CopyState cstate) List *recheckIndexes = NIL; /* OK, store the tuple and create index entries for it */ - heap_insert(cstate->rel, tuple, mycid, hi_options, bistate); + heap_insert(resultRelInfo->ri_RelationDesc, + tuple, mycid, hi_options, bistate); if (resultRelInfo->ri_NumIndices > 0) recheckIndexes = ExecInsertIndexTuples(slot, &(tuple->t_self), @@ -2532,6 +2614,14 @@ CopyFrom(CopyState cstate) */ processed++; } + + /* Restore the parent resultRelInfo */ + if (saved_resultRelInfo) + { + prev_partrri = resultRelInfo; + resultRelInfo = saved_resultRelInfo; + estate->es_result_relation_info = resultRelInfo; + } } /* Flush any remaining buffered tuples */ @@ -2544,7 +2634,15 @@ CopyFrom(CopyState cstate) /* Done, clean up */ error_context_stack = errcallback.previous; - FreeBulkInsertState(bistate); + if (bistate) + FreeBulkInsertState(bistate); + + if (prev_partrri) + { + heap_close(prev_partrri->ri_RelationDesc, RowExclusiveLock); + ExecCloseIndices(prev_partrri); + pfree(prev_partrri); + } MemoryContextSwitchTo(oldcontext); diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index 0f11922..13b5815 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -87,25 +87,9 @@ static bool ExecCheckRTEPermsModified(Oid relOid, Oid userid, Bitmapset *modifiedCols, AclMode requiredPerms); static void ExecCheckXactReadOnly(PlannedStmt *plannedstmt); -static char *ExecBuildSlotValueDescription(Oid reloid, - TupleTableSlot *slot, - TupleDesc tupdesc, - Bitmapset *modifiedCols, - int maxfieldlen); static void EvalPlanQualStart(EPQState *epqstate, EState *parentestate, Plan *planTree); -/* - * Note that GetUpdatedColumns() also exists in commands/trigger.c. There does - * not appear to be any good header to put it into, given the structures that - * it uses, so we let them be duplicated. Be sure to update both if one needs - * to be changed, however. - */ -#define GetInsertedColumns(relinfo, estate) \ - (rt_fetch((relinfo)->ri_RangeTableIndex, (estate)->es_range_table)->insertedCols) -#define GetUpdatedColumns(relinfo, estate) \ - (rt_fetch((relinfo)->ri_RangeTableIndex, (estate)->es_range_table)->updatedCols) - /* end of local decls */ @@ -1177,6 +1161,7 @@ CheckValidRowMarkRel(Relation rel, RowMarkType markType) switch (rel->rd_rel->relkind) { case RELKIND_RELATION: + case RELKIND_PARTITIONED_REL: /* OK */ break; case RELKIND_SEQUENCE: @@ -1273,6 +1258,13 @@ InitResultRelInfo(ResultRelInfo *resultRelInfo, resultRelInfo->ri_ConstraintExprs = NULL; resultRelInfo->ri_junkFilter = NULL; resultRelInfo->ri_projectReturning = NULL; + if (resultRelationDesc->rd_rel->relkind == RELKIND_PARTITIONED_REL) + { + resultRelInfo->ri_PartitionKeyInfo = BuildPartitionKeyExecInfo(resultRelationDesc); + resultRelInfo->ri_PartitionDesc = RelationGetPartitionDesc(resultRelationDesc, + linitial(resultRelationDesc->rd_partkeys), + true); + } } /* @@ -1914,7 +1906,7 @@ ExecWithCheckOptions(WCOKind kind, ResultRelInfo *resultRelInfo, * column involved, that subset will be returned with a key identifying which * columns they are. */ -static char * +char * ExecBuildSlotValueDescription(Oid reloid, TupleTableSlot *slot, TupleDesc tupdesc, diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index 27051e8..492a9d0 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -39,11 +39,13 @@ #include "access/htup_details.h" #include "access/xact.h" +#include "catalog/partition.h" #include "commands/trigger.h" #include "executor/executor.h" #include "executor/nodeModifyTable.h" #include "foreign/fdwapi.h" #include "miscadmin.h" +#include "parser/parsetree.h" #include "nodes/nodeFuncs.h" #include "storage/bufmgr.h" #include "storage/lmgr.h" @@ -226,6 +228,8 @@ ExecInsert(ModifyTableState *mtstate, { HeapTuple tuple; ResultRelInfo *resultRelInfo; + ResultRelInfo *saved_resultRelInfo = NULL; /* for partitioned table case */ + ExprContext *econtext; Relation resultRelationDesc; Oid newId; List *recheckIndexes = NIL; @@ -340,6 +344,62 @@ ExecInsert(ModifyTableState *mtstate, if (resultRelationDesc->rd_att->constr) ExecConstraints(resultRelInfo, slot, estate); + /* + * Switch resultRelInfo to one corresponding to the partition + * that this tuple maps to. + */ + if (resultRelInfo->ri_PartitionDesc) + { + Oid partOid; + + econtext = GetPerTupleExprContext(estate); + econtext->ecxt_scantuple = slot; + partOid = get_partition_for_tuple(resultRelationDesc, + resultRelInfo->ri_PartitionKeyInfo, + resultRelInfo->ri_PartitionDesc, + slot, + estate); + if (partOid != InvalidOid) + { + Relation partRelDesc; + + partRelDesc = heap_open(partOid, RowExclusiveLock); + + saved_resultRelInfo = resultRelInfo; + resultRelInfo = makeNode(ResultRelInfo); + InitResultRelInfo(resultRelInfo, + partRelDesc, + 1, /* dummy */ + 0); + + ExecOpenIndices(resultRelInfo, false); + estate->es_result_relation_info = resultRelInfo; + resultRelationDesc = resultRelInfo->ri_RelationDesc; + } + else + { + char *val_desc; + Bitmapset *modifiedCols; + Bitmapset *insertedCols; + Bitmapset *updatedCols; + + insertedCols = GetInsertedColumns(resultRelInfo, estate); + updatedCols = GetUpdatedColumns(resultRelInfo, estate); + modifiedCols = bms_union(insertedCols, updatedCols); + val_desc = ExecBuildSlotValueDescription(RelationGetRelid(resultRelationDesc), + slot, + RelationGetDescr(resultRelationDesc), + modifiedCols, + 64); + + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot find partition of relation \"%s\" for the new row", + RelationGetRelationName(resultRelationDesc)), + val_desc ? errdetail("Failing row contains %s.", val_desc) : 0)); + } + } + if (onconflict != ONCONFLICT_NONE && resultRelInfo->ri_NumIndices > 0) { /* Perform a speculative insertion. */ @@ -477,6 +537,17 @@ ExecInsert(ModifyTableState *mtstate, /* AFTER ROW INSERT Triggers */ ExecARInsertTriggers(estate, resultRelInfo, tuple, recheckIndexes); + /* Restore the resultRelInfo, if had been switched. */ + if (saved_resultRelInfo) + { + heap_close(resultRelInfo->ri_RelationDesc, RowExclusiveLock); + ExecCloseIndices(resultRelInfo); + pfree(resultRelInfo); + resultRelInfo = saved_resultRelInfo; + estate->es_result_relation_info = resultRelInfo; + saved_resultRelInfo = NULL; /* aka, paranoia! */ + } + list_free(recheckIndexes); /* @@ -502,6 +573,29 @@ ExecInsert(ModifyTableState *mtstate, return NULL; } +/* + * find_partition_in_es_rt + * A temporary band-aid solution to get partition's RT index in executor + */ +static int +find_partition_in_es_rt(EState *estate, Oid oid) +{ + ListCell *lc; + int index = 1; + + foreach(lc, estate->es_range_table) + { + RangeTblEntry *rte = (RangeTblEntry *) lfirst(lc); + + if (rte->relid == oid) + return index; + + index++; + } + + return 0; +} + /* ---------------------------------------------------------------- * ExecDelete * @@ -522,6 +616,7 @@ ExecInsert(ModifyTableState *mtstate, */ static TupleTableSlot * ExecDelete(ItemPointer tupleid, + Oid partitionOid, HeapTuple oldtuple, TupleTableSlot *planSlot, EPQState *epqstate, @@ -533,12 +628,14 @@ ExecDelete(ItemPointer tupleid, HTSU_Result result; HeapUpdateFailureData hufd; TupleTableSlot *slot = NULL; + Index resultRelationRTindex; /* * get information on the (current) result relation */ resultRelInfo = estate->es_result_relation_info; resultRelationDesc = resultRelInfo->ri_RelationDesc; + resultRelationRTindex = resultRelInfo->ri_RangeTableIndex; /* BEFORE ROW DELETE Triggers */ if (resultRelInfo->ri_TrigDesc && @@ -599,6 +696,13 @@ ExecDelete(ItemPointer tupleid, } else { + if (partitionOid != InvalidOid) + { + resultRelationRTindex = find_partition_in_es_rt(estate, + partitionOid); + resultRelationDesc = heap_open(partitionOid, RowExclusiveLock); + } + /* * delete the tuple * @@ -649,7 +753,7 @@ ldelete:; errhint("Consider using an AFTER trigger instead of a BEFORE trigger to propagate changes to other rows."))); /* Else, already deleted by self; nothing to do */ - return NULL; + goto cleanup; case HeapTupleMayBeUpdated: break; @@ -666,7 +770,7 @@ ldelete:; epqslot = EvalPlanQual(estate, epqstate, resultRelationDesc, - resultRelInfo->ri_RangeTableIndex, + resultRelationRTindex, LockTupleExclusive, &hufd.ctid, hufd.xmax); @@ -677,11 +781,11 @@ ldelete:; } } /* tuple already deleted; nothing to do */ - return NULL; + goto cleanup; default: elog(ERROR, "unrecognized heap_delete status: %u", result); - return NULL; + goto cleanup; } /* @@ -751,10 +855,15 @@ ldelete:; if (BufferIsValid(delbuffer)) ReleaseBuffer(delbuffer); - return rslot; + slot = rslot; } - return NULL; +cleanup: + /* close any partition relation, if we opened */ + if (resultRelationDesc != resultRelInfo->ri_RelationDesc) + heap_close(resultRelationDesc, RowExclusiveLock); + + return slot; } /* ---------------------------------------------------------------- @@ -781,6 +890,8 @@ ldelete:; */ static TupleTableSlot * ExecUpdate(ItemPointer tupleid, + Oid srcPartitionOid, + bool is_partkey_update, HeapTuple oldtuple, TupleTableSlot *slot, TupleTableSlot *planSlot, @@ -790,6 +901,8 @@ ExecUpdate(ItemPointer tupleid, { HeapTuple tuple; ResultRelInfo *resultRelInfo; + ResultRelInfo *saved_resultRelInfo = NULL; /* for partitioned table case */ + ExprContext *econtext; Relation resultRelationDesc; HTSU_Result result; HeapUpdateFailureData hufd; @@ -896,6 +1009,72 @@ lreplace:; ExecConstraints(resultRelInfo, slot, estate); /* + * Check whether the new row changes partition + */ + if (resultRelInfo->ri_PartitionDesc) + { + Oid partOid; + + if (is_partkey_update) + { + econtext = GetPerTupleExprContext(estate); + econtext->ecxt_scantuple = slot; + partOid = get_partition_for_tuple(resultRelationDesc, + resultRelInfo->ri_PartitionKeyInfo, + resultRelInfo->ri_PartitionDesc, + slot, + estate); + + if (partOid != srcPartitionOid) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot perform update that causes a row to change partition"))); + } + else + partOid = srcPartitionOid; + + if (OidIsValid(partOid)) + { + Relation partRelDesc; + + partRelDesc = heap_open(partOid, RowExclusiveLock); + + saved_resultRelInfo = resultRelInfo; + resultRelInfo = makeNode(ResultRelInfo); + InitResultRelInfo(resultRelInfo, + partRelDesc, + find_partition_in_es_rt(estate, partOid), + 0); + + ExecOpenIndices(resultRelInfo, false); + estate->es_result_relation_info = resultRelInfo; + resultRelationDesc = resultRelInfo->ri_RelationDesc; + } + else + { + char *val_desc; + Bitmapset *modifiedCols; + Bitmapset *insertedCols; + Bitmapset *updatedCols; + + insertedCols = GetInsertedColumns(resultRelInfo, estate); + updatedCols = GetUpdatedColumns(resultRelInfo, estate); + modifiedCols = bms_union(insertedCols, updatedCols); + val_desc = ExecBuildSlotValueDescription(RelationGetRelid(resultRelationDesc), + slot, + RelationGetDescr(resultRelationDesc), + modifiedCols, + 64); + + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("cannot find partition of relation \"%s\" for the new row", + RelationGetRelationName(resultRelationDesc)), + val_desc ? errdetail("Failing row contains %s.", val_desc) : 0)); + } + } + + /* * replace the heap tuple * * Note: if es_crosscheck_snapshot isn't InvalidSnapshot, we check @@ -943,7 +1122,7 @@ lreplace:; errhint("Consider using an AFTER trigger instead of a BEFORE trigger to propagate changes to other rows."))); /* Else, already updated by self; nothing to do */ - return NULL; + goto cleanup; case HeapTupleMayBeUpdated: break; @@ -973,11 +1152,11 @@ lreplace:; } } /* tuple already deleted; nothing to do */ - return NULL; + goto cleanup; default: elog(ERROR, "unrecognized heap_update status: %u", result); - return NULL; + goto cleanup; } /* @@ -1008,6 +1187,17 @@ lreplace:; ExecARUpdateTriggers(estate, resultRelInfo, tupleid, oldtuple, tuple, recheckIndexes); + /* Restore the resultRelInfo, if had been switched. */ + if (saved_resultRelInfo) + { + heap_close(resultRelationDesc, RowExclusiveLock); + ExecCloseIndices(resultRelInfo); + pfree(resultRelInfo); + resultRelInfo = saved_resultRelInfo; + estate->es_result_relation_info = resultRelInfo; + saved_resultRelInfo = NULL; + } + list_free(recheckIndexes); /* @@ -1027,6 +1217,15 @@ lreplace:; return ExecProcessReturning(resultRelInfo->ri_projectReturning, slot, planSlot); +cleanup: + if (saved_resultRelInfo) + { + heap_close(resultRelationDesc, RowExclusiveLock); + ExecCloseIndices(resultRelInfo); + pfree(resultRelInfo); + estate->es_result_relation_info = saved_resultRelInfo; + } + return NULL; } @@ -1214,8 +1413,8 @@ ExecOnConflictUpdate(ModifyTableState *mtstate, */ /* Execute UPDATE with projection */ - *returning = ExecUpdate(&tuple.t_self, NULL, - mtstate->mt_conflproj, planSlot, + *returning = ExecUpdate(&tuple.t_self, InvalidOid, false, + NULL, mtstate->mt_conflproj, planSlot, &mtstate->mt_epqstate, mtstate->ps.state, canSetTag); @@ -1299,6 +1498,7 @@ ExecModifyTable(ModifyTableState *node) ItemPointerData tuple_ctid; HeapTupleData oldtupdata; HeapTuple oldtuple; + Oid partitionOid = InvalidOid; /* * This should NOT get called during EvalPlanQual; we should have passed a @@ -1396,7 +1596,9 @@ ExecModifyTable(ModifyTableState *node) bool isNull; relkind = resultRelInfo->ri_RelationDesc->rd_rel->relkind; - if (relkind == RELKIND_RELATION || relkind == RELKIND_MATVIEW) + if (relkind == RELKIND_RELATION || + relkind == RELKIND_PARTITIONED_REL || + relkind == RELKIND_MATVIEW) { datum = ExecGetJunkAttribute(slot, junkfilter->jf_junkAttNo, @@ -1447,6 +1649,22 @@ ExecModifyTable(ModifyTableState *node) } else Assert(relkind == RELKIND_FOREIGN_TABLE); + + /* + * If this is a partition key update, get the source + * partition OID. + */ + if (relkind == RELKIND_PARTITIONED_REL) + { + datum = ExecGetJunkAttribute(slot, + junkfilter->jf_junkExtraAttNo, + &isNull); + /* shouldn't ever get a null result... */ + if (isNull) + elog(ERROR, "tableoid is NULL"); + + partitionOid = DatumGetObjectId(datum); + } } /* @@ -1464,11 +1682,12 @@ ExecModifyTable(ModifyTableState *node) estate, node->canSetTag); break; case CMD_UPDATE: - slot = ExecUpdate(tupleid, oldtuple, slot, planSlot, - &node->mt_epqstate, estate, node->canSetTag); + slot = ExecUpdate(tupleid, partitionOid, node->mt_partkeyupd, + oldtuple, slot, planSlot, &node->mt_epqstate, + estate, node->canSetTag); break; case CMD_DELETE: - slot = ExecDelete(tupleid, oldtuple, planSlot, + slot = ExecDelete(tupleid, partitionOid, oldtuple, planSlot, &node->mt_epqstate, estate, node->canSetTag); break; default: @@ -1500,6 +1719,35 @@ ExecModifyTable(ModifyTableState *node) return NULL; } +/* + * IsPartitionKeyUpdate + * Checks if updated columns include any of the partitin key columns + * + * If there is a partition expression, simply return 'false'. We could maybe + * check if any updated column is involved in the expression hence causing + * partition key to be changed in a given row, but later. + */ +static bool +IsPartitionKeyUpdate(ResultRelInfo *resultRelInfo, EState *estate) +{ + Relation rel = resultRelInfo->ri_RelationDesc; + ListCell *cell; + Bitmapset *updatedCols = GetUpdatedColumns(resultRelInfo, estate); + + foreach (cell, rel->rd_partkeys) + { + PartitionKey key = lfirst(cell); + int i; + + for (i = 0; i < key->partnatts; i++) + if (key->partattrs[i] == 0 || + bms_is_member((int ) key->partattrs[i], updatedCols)) + return true; + } + + return false; +} + /* ---------------------------------------------------------------- * ExecInitModifyTable * ---------------------------------------------------------------- @@ -1516,6 +1764,7 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) Plan *subplan; ListCell *l; int i; + char relkind; /* check for unsupported flags */ Assert(!(eflags & (EXEC_FLAG_BACKWARD | EXEC_FLAG_MARK))); @@ -1828,10 +2077,9 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) if (operation == CMD_UPDATE || operation == CMD_DELETE) { /* For UPDATE/DELETE, find the appropriate junk attr now */ - char relkind; - relkind = resultRelInfo->ri_RelationDesc->rd_rel->relkind; if (relkind == RELKIND_RELATION || + relkind == RELKIND_PARTITIONED_REL || relkind == RELKIND_MATVIEW) { j->jf_junkAttNo = ExecFindJunkAttribute(j, "ctid"); @@ -1852,6 +2100,17 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) if (!AttributeNumberIsValid(j->jf_junkAttNo)) elog(ERROR, "could not find junk wholerow column"); } + + /* + * For partitioned table, find the extra junk attribute, + * ie, "tableoid". + */ + if (relkind == RELKIND_PARTITIONED_REL) + { + j->jf_junkExtraAttNo = ExecFindJunkAttribute(j, "tableoid"); + if (!AttributeNumberIsValid(j->jf_junkExtraAttNo)) + elog(ERROR, "could not find junk tableoid column"); + } } resultRelInfo->ri_junkFilter = j; @@ -1866,6 +2125,12 @@ ExecInitModifyTable(ModifyTable *node, EState *estate, int eflags) } } + relkind = mtstate->resultRelInfo->ri_RelationDesc->rd_rel->relkind; + if (relkind == RELKIND_PARTITIONED_REL && + operation == CMD_UPDATE && + IsPartitionKeyUpdate(mtstate->resultRelInfo, estate)) + mtstate->mt_partkeyupd = true; + /* * Set up a tuple table slot for use for trigger output tuples. In a plan * containing multiple ModifyTable nodes, all can share one such slot, so diff --git a/src/backend/parser/parse_clause.c b/src/backend/parser/parse_clause.c index 04fa127..499da4c 100644 --- a/src/backend/parser/parse_clause.c +++ b/src/backend/parser/parse_clause.c @@ -2881,6 +2881,15 @@ transformOnConflictArbiter(ParseState *pstate, parser_errposition(pstate, exprLocation((Node *) onConflictClause)))); + /* ON CONFLICT on partitioned tables currently disallowed */ + if (pstate->p_target_relation->rd_rel->relkind == RELKIND_PARTITIONED_REL) + ereport(ERROR, + (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), + errmsg("ON CONFLICT is not supported with partitioned tables"), + parser_errposition(pstate, + exprLocation((Node *) onConflictClause)))); + + /* ON CONFLICT DO NOTHING does not require an inference clause */ if (infer) { diff --git a/src/backend/rewrite/rewriteHandler.c b/src/backend/rewrite/rewriteHandler.c index a22a11e..fa28b4d 100644 --- a/src/backend/rewrite/rewriteHandler.c +++ b/src/backend/rewrite/rewriteHandler.c @@ -1222,6 +1222,7 @@ rewriteTargetListUD(Query *parsetree, RangeTblEntry *target_rte, TargetEntry *tle; if (target_relation->rd_rel->relkind == RELKIND_RELATION || + target_relation->rd_rel->relkind == RELKIND_PARTITIONED_REL || target_relation->rd_rel->relkind == RELKIND_MATVIEW) { /* @@ -1293,6 +1294,30 @@ rewriteTargetListUD(Query *parsetree, RangeTblEntry *target_rte, parsetree->targetList = lappend(parsetree->targetList, tle); } + + /* + * Also emit tableoid so that ExecModifyTable knows which partition + * a given row came from. ExecUpdate will use it to throw error if + * the new row does not map to the same partition. + */ + if (target_relation->rd_rel->relkind == RELKIND_PARTITIONED_REL) + { + var = makeVar(parsetree->resultRelation, + TableOidAttributeNumber, + OIDOID, + -1, + InvalidOid, + 0); + + attrname = "tableoid"; + + tle = makeTargetEntry((Expr *) var, + list_length(parsetree->targetList) + 1, + pstrdup(attrname), + true); + + parsetree->targetList = lappend(parsetree->targetList, tle); + } } diff --git a/src/include/catalog/partition.h b/src/include/catalog/partition.h index c54fc1b..81d8232 100644 --- a/src/include/catalog/partition.h +++ b/src/include/catalog/partition.h @@ -14,6 +14,8 @@ #define PARTITION_H #include "fmgr.h" +#include "executor/tuptable.h" +#include "nodes/execnodes.h" #include "utils/relcache.h" /* Type information for partition key columns */ @@ -106,4 +108,11 @@ extern bool list_partition_overlaps(PartitionDesc pdesc, extern bool range_partition_overlaps(PartitionDesc pdesc, PartitionKey key, Datum *rangemaxs); + +extern List *BuildPartitionKeyExecInfo(Relation rel); +extern Oid get_partition_for_tuple(Relation rel, + List *pkinfo, + PartitionDesc pdesc, + TupleTableSlot *slot, + EState *estate); #endif /* PARTITION_H */ diff --git a/src/include/executor/executor.h b/src/include/executor/executor.h index 1a44085..a769126 100644 --- a/src/include/executor/executor.h +++ b/src/include/executor/executor.h @@ -219,6 +219,21 @@ extern void EvalPlanQualBegin(EPQState *epqstate, EState *parentestate); extern void EvalPlanQualEnd(EPQState *epqstate); /* + * Following used to be in execMain.c, brought here to be useful for copy.c, + * trigger.c, among others probably. + */ +#define GetInsertedColumns(relinfo, estate) \ + (rt_fetch((relinfo)->ri_RangeTableIndex, (estate)->es_range_table)->insertedCols) +#define GetUpdatedColumns(relinfo, estate) \ + (rt_fetch((relinfo)->ri_RangeTableIndex, (estate)->es_range_table)->updatedCols) + +extern char *ExecBuildSlotValueDescription(Oid reloid, + TupleTableSlot *slot, + TupleDesc tupdesc, + Bitmapset *modifiedCols, + int maxfieldlen); + +/* * prototypes from functions in execProcnode.c */ extern PlanState *ExecInitNode(Plan *node, EState *estate, int eflags); diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h index 064a050..dec6200 100644 --- a/src/include/nodes/execnodes.h +++ b/src/include/nodes/execnodes.h @@ -76,6 +76,27 @@ typedef struct IndexInfo bool ii_BrokenHotChain; } IndexInfo; +/* + * PartitionKeyExecInfo + * + * This struct holds the information needed to extract partition + * column values from a heap tuple. + * + * NumKeyAttrs number of columns in partition key + * KeyAttrNumbers underlying-rel attribute numbers used as keys + * (zeroes indicate expressions) + * Expressions expr trees for expression entries, or NIL if none + * ExpressionsState exec state for expressions, or NIL if none + */ +typedef struct PartitionKeyExecInfo +{ + NodeTag type; + int pi_NumKeyAttrs; + AttrNumber pi_KeyAttrNumbers[PARTITION_MAX_KEYS]; + List *pi_Expressions; /* list of Expr */ + List *pi_ExpressionsState; /* list of ExprState */ +} PartitionKeyExecInfo; + /* ---------------- * ExprContext_CB * @@ -290,6 +311,7 @@ typedef struct JunkFilter AttrNumber *jf_cleanMap; TupleTableSlot *jf_resultSlot; AttrNumber jf_junkAttNo; + AttrNumber jf_junkExtraAttNo; } JunkFilter; /* ---------------- @@ -341,6 +363,8 @@ typedef struct ResultRelInfo ProjectionInfo *ri_projectReturning; ProjectionInfo *ri_onConflictSetProj; List *ri_onConflictSetWhere; + List *ri_PartitionKeyInfo; + struct PartitionDescData *ri_PartitionDesc; } ResultRelInfo; /* ---------------- @@ -1134,6 +1158,7 @@ typedef struct ModifyTableState * tlist */ TupleTableSlot *mt_conflproj; /* CONFLICT ... SET ... projection * target */ + bool mt_partkeyupd; /* Does UPDATE change partition key cols */ } ModifyTableState; /* ---------------- diff --git a/src/include/nodes/nodes.h b/src/include/nodes/nodes.h index 8032f0d..666bc1e 100644 --- a/src/include/nodes/nodes.h +++ b/src/include/nodes/nodes.h @@ -37,6 +37,7 @@ typedef enum NodeTag T_ResultRelInfo, T_EState, T_TupleTableSlot, + T_PartitionKeyExecInfo, /* * TAGS FOR PLAN NODES (plannodes.h) -- 1.7.1