diff --git a/src/backend/executor/execMain.c b/src/backend/executor/execMain.c index 33b172b..0cac472 100644 --- a/src/backend/executor/execMain.c +++ b/src/backend/executor/execMain.c @@ -42,6 +42,7 @@ #include "access/transam.h" #include "access/xact.h" #include "catalog/namespace.h" +#include "catalog/pg_collation.h" #include "commands/matview.h" #include "commands/trigger.h" #include "executor/execdebug.h" @@ -2702,3 +2703,317 @@ EvalPlanQualEnd(EPQState *epqstate) epqstate->planstate = NULL; epqstate->origslot = NULL; } + +/* + * ExecFindPartition - find a partition to insert a tuple (in slot) + * + * XXX - if not found, returns itself. Currently only good for the + * range strategy (including multi-column range strategy). + */ +Relation +ExecFindPartition(ResultRelInfo *resultRelInfo, TupleTableSlot *slot) +{ + Relation rel = resultRelInfo->ri_RelationDesc; + Relation partition; + + /* partition key and bound info */ + int partnatts = rel->rd_partnatts; + AttrNumber *partattrs = rel->rd_partattrs; + FmgrInfo *partattcmpfn = rel->rd_partattcmpfn; + PartitionInfo *pinfo = rel->rd_partitioninfo; + + FunctionCallInfoData fcinfo; + + /* input tuple to find a home for */ + Datum *values = slot->tts_values; + bool *isnull = slot->tts_isnull; + + AttrNumber attno; + AttrNumber finalattno; + int attidx; + int finalattidx; + + int least, highest; + int lobound, hibound; + int newlo, newhi; + int probe; + int cmpval; + bool bsearch_aborted; + + int j; + + /* let's get this out of the way */ + for(attidx = 0; attidx < partnatts; attidx++) + { + AttrNumber attno = partattrs[attidx]; + + if(isnull[attno - 1]) + ereport(ERROR, + (errcode(ERRCODE_NOT_NULL_VIOLATION), + errmsg("value in partition key column \"%s\" cannot be null", + NameStr(rel->rd_att->attrs[attno - 1]->attname)))); + } + + /* + * Let the binary search begin! + * The invariant is kind of complicated because of incorporating + * the logic for multi-column range strategy. + */ + attidx = 0; + finalattno = partattrs[0]; + finalattidx = attidx; + + least = lobound = 0; + highest = hibound = pinfo->numpartitions - 1; + + newlo = -1; + newhi = pinfo->numpartitions; + + bsearch_aborted = false; + while(lobound < hibound) + { + bool isless; + probe = (lobound + hibound) / 2; + + attno = partattrs[attidx]; + InitFunctionCallInfoData(fcinfo, &(partattcmpfn[attidx]), 2, + DEFAULT_COLLATION_OID, NULL, NULL); + /* tuple(attno) < rangemax(attidx)? */ + fcinfo.arg[0] = values[attno - 1]; + fcinfo.arg[1] = pinfo->rangemaxs[attidx][probe]; + cmpval = DatumGetInt32(FunctionCallInvoke(&fcinfo)); + + if(!cmpval) + { + /* + * tuple(attno) == rangemax(attidx). + */ + + /* + * The last attribute matching a rangemax means we are + * pretty close to the partition we're looking for - namely + * probe+1 + */ + if(attidx == partnatts - 1) + { + /* + * memorize to use after the binary search ends + */ + probe = probe + 1; + + if(probe > highest) + probe = highest; + + finalattidx = attidx; + finalattno = attno; + bsearch_aborted = true; + break; + } + + /* + * Before moving on to the next attribute, restrict the binary + * search space the sub-range where rangemax(attidx) is the + * same. The sub-range consists of partitions newlo..newhi + */ + + /* look left of probe */ + j = probe; + do + { + --j; + + if(j < 0) + break; + + fcinfo.arg[0] = values[attno - 1]; + fcinfo.arg[1] = pinfo->rangemaxs[attidx][j]; + cmpval = DatumGetInt32(FunctionCallInvoke(&fcinfo)); + } while(!cmpval); + + /* possibly new lobound */ + newlo = ++j; + + /* look right of probe */ + j = probe; + do + { + ++j; + + if(j > pinfo->numpartitions - 1) + break; + + fcinfo.arg[0] = values[attno - 1]; + fcinfo.arg[1] = pinfo->rangemaxs[attidx][j]; + cmpval = DatumGetInt32(FunctionCallInvoke(&fcinfo)); + } while(!cmpval); + + /* possibly new hibound */ + newhi = --j; + + /* + * We indeed are moving to our next binary search + */ + if(newlo != newhi) + { + attidx++; + lobound = newlo; + hibound = newhi; + } + } + else + { + /* + * tuple(attno) != rangemax(attidx). + * + * Move a level down in the current binary search + */ + isless = (cmpval < 0); + + if (isless) + hibound = probe; + else + lobound = probe + 1; + + /* + * Remember newlo, newhi are the sub-range bounds covering the + * range of values rangemax[attidx] for which rangemax[attidx-1] + * is the same. As determined when attidx was set to its current + * value. + * + * It is however possible that the desired partition may be the + * one just after the last one of the partitions newlo..newhi, + * or the one just before. + */ + if(lobound >= newhi) + { + /* + * Before concluding so, check if this is not the one + */ + probe = (lobound + hibound) / 2; + InitFunctionCallInfoData(fcinfo, &(partattcmpfn[attidx]), 2, + DEFAULT_COLLATION_OID, NULL, NULL); + fcinfo.arg[0] = values[attno - 1]; + fcinfo.arg[1] = pinfo->rangemaxs[attidx][probe]; + cmpval = DatumGetInt32(FunctionCallInvoke(&fcinfo)); + + if(cmpval < 0) + { + InitFunctionCallInfoData(fcinfo, + &(partattcmpfn[finalattidx]), + 2, DEFAULT_COLLATION_OID, NULL, NULL); + + fcinfo.arg[0] = values[attno - 1]; + fcinfo.arg[1] = pinfo->rangemins[attidx][probe]; + + cmpval = DatumGetInt32(FunctionCallInvoke(&fcinfo)); + + if(cmpval >= 0) + { + /* Yes, this is it, return. */ + partition = heap_open(pinfo->oids[probe], RowExclusiveLock); + return partition; + } + } + + /* + * We are about to abort the bsearch. + * Reset attidx. + */ + attidx = 0; + attno = partattrs[attidx]; + probe = newhi + 1; + + if(probe > highest) + probe = highest; + + /* + * memorize to use after the binary search ends + */ + finalattidx = attidx; + finalattno = attno; + bsearch_aborted = true; + break; + } + else if(hibound <= newlo) + { + /* + * Before concluding so, check if this is not the one + */ + probe = (lobound + hibound) / 2; + InitFunctionCallInfoData(fcinfo, &(partattcmpfn[finalattidx]), + 2, DEFAULT_COLLATION_OID, NULL, NULL); + + fcinfo.arg[0] = values[attno - 1]; + fcinfo.arg[1] = pinfo->rangemins[attidx][probe]; + + cmpval = DatumGetInt32(FunctionCallInvoke(&fcinfo)); + + if(cmpval >= 0) + { + /* Yes, this is it, return. */ + partition = heap_open(pinfo->oids[probe], RowExclusiveLock); + return partition; + } + + /* + * We're about to abort the bsearch. + * Reset attidx. + */ + attidx = 0; + attno = partattrs[attidx]; + probe = newlo - 1; + + if(probe < least) + probe = least; + + /* + * memorize to use after the binary search ends + */ + finalattidx = attidx; + finalattno = attno; + bsearch_aborted = true; + break; + } + } + + /* + * memorize to use after the binary search ends + */ + finalattidx = attidx; + finalattno = attno; + } + + if(!bsearch_aborted) + probe = (lobound + hibound) / 2; + + /* tuple < rangemaxs(probe)? */ + InitFunctionCallInfoData(fcinfo, &(partattcmpfn[finalattidx]), 2, + DEFAULT_COLLATION_OID, NULL, NULL); + + fcinfo.arg[0] = values[finalattno - 1]; + fcinfo.arg[1] = pinfo->rangemaxs[finalattidx][probe]; + cmpval = DatumGetInt32(FunctionCallInvoke(&fcinfo)); + + /* + * So, tuple < rangemaxs(probe). + * Is tuple >= rangemins(probe)? + */ + if(cmpval < 0) + { + InitFunctionCallInfoData(fcinfo, &(partattcmpfn[finalattidx]), 2, + DEFAULT_COLLATION_OID, NULL, NULL); + + fcinfo.arg[0] = values[finalattno - 1]; + fcinfo.arg[1] = pinfo->rangemins[finalattidx][probe]; + + cmpval = DatumGetInt32(FunctionCallInvoke(&fcinfo)); + + if(cmpval >= 0) + { + partition = heap_open(pinfo->oids[probe], RowExclusiveLock); + return partition; + } + } + + return rel; +} diff --git a/src/backend/executor/nodeModifyTable.c b/src/backend/executor/nodeModifyTable.c index f96fb24..5cd24f5 100644 --- a/src/backend/executor/nodeModifyTable.c +++ b/src/backend/executor/nodeModifyTable.c @@ -168,7 +168,9 @@ ExecInsert(TupleTableSlot *slot, { HeapTuple tuple; ResultRelInfo *resultRelInfo; + ResultRelInfo *saved_resultRelInfo; Relation resultRelationDesc; + Relation saved_resultRelationDesc; Oid newId; List *recheckIndexes = NIL; @@ -259,6 +261,33 @@ ExecInsert(TupleTableSlot *slot, ExecConstraints(resultRelInfo, slot, estate); /* + * a hack/idea to route tuples to a valid partition + * if none found, resultRelationDesc remains unchanged + * + * XXX - should this be before ExecConstraints()? + */ + saved_resultRelationDesc = resultRelationDesc; + saved_resultRelInfo = resultRelInfo; + if(resultRelationDesc->rd_rel->relispartitioned) + { + resultRelationDesc = ExecFindPartition(resultRelInfo, slot); + + /* for ExecInsertIndexTuples() */ + if(resultRelationDesc != saved_resultRelationDesc) + { + resultRelInfo = makeNode(ResultRelInfo); + InitResultRelInfo(resultRelInfo, + resultRelationDesc, + 1, + 0); + + ExecOpenIndices(resultRelInfo); + + estate->es_result_relation_info = resultRelInfo; + } + } + + /* * insert the tuple * * Note: heap_insert returns the tid (location) of the new tuple in @@ -273,6 +302,17 @@ ExecInsert(TupleTableSlot *slot, if (resultRelInfo->ri_NumIndices > 0) recheckIndexes = ExecInsertIndexTuples(slot, &(tuple->t_self), estate); + + /* close the partition heap and reset estate */ + if(resultRelationDesc != saved_resultRelationDesc) + { + pfree(resultRelInfo); + heap_close(resultRelationDesc, RowExclusiveLock); + estate->es_result_relation_info = saved_resultRelInfo; + resultRelInfo = saved_resultRelInfo; + } + + resultRelationDesc = saved_resultRelationDesc; } if (canSetTag)