#include "access/nbtree.h"
#include "fmgr.h"
#include "utils/lsyscache.h"
+#include "utils/rel.h"
#include "utils/sortsupport.h"
}
/*
- * Fill in SortSupport given an ordering operator (btree "<" or ">" operator).
- *
- * Caller must previously have zeroed the SortSupportData structure and then
- * filled in ssup_cxt, ssup_collation, and ssup_nulls_first. This will fill
- * in ssup_reverse as well as the comparator function pointer.
+ * Look up and call sortsupport function to setup SortSupport comparator;
+ * or if no such function exists or it declines to set up the appropriate
+ * state, prepare a suitable shim.
*/
-void
-PrepareSortSupportFromOrderingOp(Oid orderingOp, SortSupport ssup)
+static void
+FinishSortSupportFunction(Oid opfamily, Oid opcintype, SortSupport ssup)
{
Oid sortSupportFunction;
- Oid opfamily;
- Oid opcintype;
- int16 strategy;
-
- Assert(ssup->comparator == NULL);
-
- /* Find the operator in pg_amop */
- if (!get_ordering_op_properties(orderingOp, &opfamily, &opcintype,
- &strategy))
- elog(ERROR, "operator %u is not a valid ordering operator",
- orderingOp);
- ssup->ssup_reverse = (strategy == BTGreaterStrategyNumber);
/* Look for a sort support function */
sortSupportFunction = get_opfamily_proc(opfamily, opcintype, opcintype,
PrepareSortSupportComparisonShim(sortFunction, ssup);
}
}
+
+/*
+ * Fill in SortSupport given an ordering operator (btree "<" or ">" operator).
+ *
+ * Caller must previously have zeroed the SortSupportData structure and then
+ * filled in ssup_cxt, ssup_collation, and ssup_nulls_first. This will fill
+ * in ssup_reverse as well as the comparator function pointer.
+ */
+void
+PrepareSortSupportFromOrderingOp(Oid orderingOp, SortSupport ssup)
+{
+ Oid opfamily;
+ Oid opcintype;
+ int16 strategy;
+
+ Assert(ssup->comparator == NULL);
+
+ /* Find the operator in pg_amop */
+ if (!get_ordering_op_properties(orderingOp, &opfamily, &opcintype,
+ &strategy))
+ elog(ERROR, "operator %u is not a valid ordering operator",
+ orderingOp);
+ ssup->ssup_reverse = (strategy == BTGreaterStrategyNumber);
+
+ FinishSortSupportFunction(opfamily, opcintype, ssup);
+}
+
+/*
+ * Fill in SortSupport given an index relation, attribute, and strategy.
+ *
+ * Caller must previously have zeroed the SortSupportData structure and then
+ * filled in ssup_cxt, ssup_attno, ssup_collation, and ssup_nulls_first. This
+ * will fill in ssup_reverse (based on the supplied strategy), as well as the
+ * comparator function pointer.
+ */
+void
+PrepareSortSupportFromIndexRel(Relation indexRel, int16 strategy,
+ SortSupport ssup)
+{
+ Oid opfamily = indexRel->rd_opfamily[ssup->ssup_attno - 1];
+ Oid opcintype = indexRel->rd_opcintype[ssup->ssup_attno - 1];
+
+ Assert(ssup->comparator == NULL);
+
+ /* Find the operator in pg_amop */
+ if (indexRel->rd_rel->relam != BTREE_AM_OID)
+ elog(ERROR, "unexpected non-btree AM: %u", indexRel->rd_rel->relam);
+ if (strategy != BTGreaterStrategyNumber &&
+ strategy != BTLessStrategyNumber)
+ elog(ERROR, "unexpected sort support strategy: %d", strategy);
+ ssup->ssup_reverse = (strategy == BTGreaterStrategyNumber);
+
+ FinishSortSupportFunction(opfamily, opcintype, ssup);
+}
void (*readtup) (Tuplesortstate *state, SortTuple *stup,
int tapenum, unsigned int len);
- /*
- * Function to reverse the sort direction from its current state. (We
- * could dispense with this if we wanted to enforce that all variants
- * represent the sort key information alike.)
- */
- void (*reversedirection) (Tuplesortstate *state);
-
/*
* This array holds the tuples now in sort memory. If we are in state
* INITIAL, the tuples are in no particular order; if we are in state
bool markpos_eof; /* saved "eof_reached" */
/*
- * These variables are specific to the MinimalTuple case; they are set by
- * tuplesort_begin_heap and used only by the MinimalTuple routines.
+ * The sortKeys variable is used by every case other than the hash index
+ * case; it is set by tuplesort_begin_xxx. tupDesc is only used by the
+ * MinimalTuple and CLUSTER routines, though.
*/
TupleDesc tupDesc;
SortSupport sortKeys; /* array of length nKeys */
/*
* These variables are specific to the CLUSTER case; they are set by
- * tuplesort_begin_cluster. Note CLUSTER also uses tupDesc and
- * indexScanKey.
+ * tuplesort_begin_cluster.
*/
IndexInfo *indexInfo; /* info about index being used for reference */
EState *estate; /* for evaluating index expressions */
Relation indexRel; /* index being built */
/* These are specific to the index_btree subcase: */
- ScanKey indexScanKey;
bool enforceUnique; /* complain if we find duplicate tuples */
/* These are specific to the index_hash subcase: */
#define COPYTUP(state,stup,tup) ((*(state)->copytup) (state, stup, tup))
#define WRITETUP(state,tape,stup) ((*(state)->writetup) (state, tape, stup))
#define READTUP(state,stup,tape,len) ((*(state)->readtup) (state, stup, tape, len))
-#define REVERSEDIRECTION(state) ((*(state)->reversedirection) (state))
#define LACKMEM(state) ((state)->availMem < 0)
#define USEMEM(state,amt) ((state)->availMem -= (amt))
#define FREEMEM(state,amt) ((state)->availMem += (amt))
static void tuplesort_heap_insert(Tuplesortstate *state, SortTuple *tuple,
int tupleindex, bool checkIndex);
static void tuplesort_heap_siftup(Tuplesortstate *state, bool checkIndex);
+static void reversedirection(Tuplesortstate *state);
static unsigned int getlen(Tuplesortstate *state, int tapenum, bool eofOK);
static void markrunend(Tuplesortstate *state, int tapenum);
static int comparetup_heap(const SortTuple *a, const SortTuple *b,
SortTuple *stup);
static void readtup_heap(Tuplesortstate *state, SortTuple *stup,
int tapenum, unsigned int len);
-static void reversedirection_heap(Tuplesortstate *state);
static int comparetup_cluster(const SortTuple *a, const SortTuple *b,
Tuplesortstate *state);
static void copytup_cluster(Tuplesortstate *state, SortTuple *stup, void *tup);
SortTuple *stup);
static void readtup_index(Tuplesortstate *state, SortTuple *stup,
int tapenum, unsigned int len);
-static void reversedirection_index_btree(Tuplesortstate *state);
-static void reversedirection_index_hash(Tuplesortstate *state);
static int comparetup_datum(const SortTuple *a, const SortTuple *b,
Tuplesortstate *state);
static void copytup_datum(Tuplesortstate *state, SortTuple *stup, void *tup);
SortTuple *stup);
static void readtup_datum(Tuplesortstate *state, SortTuple *stup,
int tapenum, unsigned int len);
-static void reversedirection_datum(Tuplesortstate *state);
static void free_sort_tuple(Tuplesortstate *state, SortTuple *stup);
/*
state->copytup = copytup_heap;
state->writetup = writetup_heap;
state->readtup = readtup_heap;
- state->reversedirection = reversedirection_heap;
state->tupDesc = tupDesc; /* assume we need not copy tupDesc */
int workMem, bool randomAccess)
{
Tuplesortstate *state = tuplesort_begin_common(workMem, randomAccess);
+ ScanKey indexScanKey;
MemoryContext oldcontext;
+ int i;
Assert(indexRel->rd_rel->relam == BTREE_AM_OID);
state->copytup = copytup_cluster;
state->writetup = writetup_cluster;
state->readtup = readtup_cluster;
- state->reversedirection = reversedirection_index_btree;
state->indexInfo = BuildIndexInfo(indexRel);
- state->indexScanKey = _bt_mkscankey_nodata(indexRel);
state->tupDesc = tupDesc; /* assume we need not copy tupDesc */
+ indexScanKey = _bt_mkscankey_nodata(indexRel);
+
if (state->indexInfo->ii_Expressions != NULL)
{
TupleTableSlot *slot;
econtext->ecxt_scantuple = slot;
}
+ /* Prepare SortSupport data for each column */
+ state->sortKeys = (SortSupport) palloc0(state->nKeys *
+ sizeof(SortSupportData));
+
+ for (i = 0; i < state->nKeys; i++)
+ {
+ SortSupport sortKey = state->sortKeys + i;
+ ScanKey scanKey = indexScanKey + i;
+ int16 strategy;
+
+ sortKey->ssup_cxt = CurrentMemoryContext;
+ sortKey->ssup_collation = scanKey->sk_collation;
+ sortKey->ssup_nulls_first =
+ (scanKey->sk_flags & SK_BT_NULLS_FIRST) != 0;
+ sortKey->ssup_attno = scanKey->sk_attno;
+
+ AssertState(sortKey->ssup_attno != 0);
+
+ strategy = (scanKey->sk_flags & SK_BT_DESC) != 0 ?
+ BTGreaterStrategyNumber : BTLessStrategyNumber;
+
+ PrepareSortSupportFromIndexRel(indexRel, strategy, sortKey);
+ }
+
+ _bt_freeskey(indexScanKey);
+
MemoryContextSwitchTo(oldcontext);
return state;
int workMem, bool randomAccess)
{
Tuplesortstate *state = tuplesort_begin_common(workMem, randomAccess);
+ ScanKey indexScanKey;
MemoryContext oldcontext;
+ int i;
oldcontext = MemoryContextSwitchTo(state->sortcontext);
state->copytup = copytup_index;
state->writetup = writetup_index;
state->readtup = readtup_index;
- state->reversedirection = reversedirection_index_btree;
state->heapRel = heapRel;
state->indexRel = indexRel;
- state->indexScanKey = _bt_mkscankey_nodata(indexRel);
state->enforceUnique = enforceUnique;
+ indexScanKey = _bt_mkscankey_nodata(indexRel);
+ state->nKeys = RelationGetNumberOfAttributes(indexRel);
+
+ /* Prepare SortSupport data for each column */
+ state->sortKeys = (SortSupport) palloc0(state->nKeys *
+ sizeof(SortSupportData));
+
+ for (i = 0; i < state->nKeys; i++)
+ {
+ SortSupport sortKey = state->sortKeys + i;
+ ScanKey scanKey = indexScanKey + i;
+ int16 strategy;
+
+ sortKey->ssup_cxt = CurrentMemoryContext;
+ sortKey->ssup_collation = scanKey->sk_collation;
+ sortKey->ssup_nulls_first =
+ (scanKey->sk_flags & SK_BT_NULLS_FIRST) != 0;
+ sortKey->ssup_attno = scanKey->sk_attno;
+
+ AssertState(sortKey->ssup_attno != 0);
+
+ strategy = (scanKey->sk_flags & SK_BT_DESC) != 0 ?
+ BTGreaterStrategyNumber : BTLessStrategyNumber;
+
+ PrepareSortSupportFromIndexRel(indexRel, strategy, sortKey);
+ }
+
+ _bt_freeskey(indexScanKey);
+
MemoryContextSwitchTo(oldcontext);
return state;
state->copytup = copytup_index;
state->writetup = writetup_index;
state->readtup = readtup_index;
- state->reversedirection = reversedirection_index_hash;
state->heapRel = heapRel;
state->indexRel = indexRel;
state->copytup = copytup_datum;
state->writetup = writetup_datum;
state->readtup = readtup_datum;
- state->reversedirection = reversedirection_datum;
state->datumType = datumType;
Assert(tupcount >= state->bound);
/* Reverse sort direction so largest entry will be at root */
- REVERSEDIRECTION(state);
+ reversedirection(state);
state->memtupcount = 0; /* make the heap empty */
for (i = 0; i < tupcount; i++)
* Reverse sort direction back to the original state. This is not
* actually necessary but seems like a good idea for tidiness.
*/
- REVERSEDIRECTION(state);
+ reversedirection(state);
state->status = TSS_SORTEDINMEM;
state->boundUsed = true;
memtuples[i] = *tuple;
}
+/*
+ * Function to reverse the sort direction from its current state
+ *
+ * It is not safe to call this when performing hash tuplesorts
+ */
+static void
+reversedirection(Tuplesortstate *state)
+{
+ SortSupport sortKey = state->sortKeys;
+ int nkey;
+
+ for (nkey = 0; nkey < state->nKeys; nkey++, sortKey++)
+ {
+ sortKey->ssup_reverse = !sortKey->ssup_reverse;
+ sortKey->ssup_nulls_first = !sortKey->ssup_nulls_first;
+ }
+}
+
/*
* Tape interface routines
}
-/*
- * Inline-able copy of FunctionCall2Coll() to save some cycles in sorting.
- */
-static inline Datum
-myFunctionCall2Coll(FmgrInfo *flinfo, Oid collation, Datum arg1, Datum arg2)
-{
- FunctionCallInfoData fcinfo;
- Datum result;
-
- InitFunctionCallInfoData(fcinfo, flinfo, 2, collation, NULL, NULL);
-
- fcinfo.arg[0] = arg1;
- fcinfo.arg[1] = arg2;
- fcinfo.argnull[0] = false;
- fcinfo.argnull[1] = false;
-
- result = FunctionCallInvoke(&fcinfo);
-
- /* Check for null result, since caller is clearly not expecting one */
- if (fcinfo.isnull)
- elog(ERROR, "function %u returned NULL", fcinfo.flinfo->fn_oid);
-
- return result;
-}
-
-/*
- * Apply a sort function (by now converted to fmgr lookup form)
- * and return a 3-way comparison result. This takes care of handling
- * reverse-sort and NULLs-ordering properly. We assume that DESC and
- * NULLS_FIRST options are encoded in sk_flags the same way btree does it.
- */
-static inline int32
-inlineApplySortFunction(FmgrInfo *sortFunction, int sk_flags, Oid collation,
- Datum datum1, bool isNull1,
- Datum datum2, bool isNull2)
-{
- int32 compare;
-
- if (isNull1)
- {
- if (isNull2)
- compare = 0; /* NULL "=" NULL */
- else if (sk_flags & SK_BT_NULLS_FIRST)
- compare = -1; /* NULL "<" NOT_NULL */
- else
- compare = 1; /* NULL ">" NOT_NULL */
- }
- else if (isNull2)
- {
- if (sk_flags & SK_BT_NULLS_FIRST)
- compare = 1; /* NOT_NULL ">" NULL */
- else
- compare = -1; /* NOT_NULL "<" NULL */
- }
- else
- {
- compare = DatumGetInt32(myFunctionCall2Coll(sortFunction, collation,
- datum1, datum2));
-
- if (sk_flags & SK_BT_DESC)
- compare = -compare;
- }
-
- return compare;
-}
-
-
/*
* Routines specialized for HeapTuple (actually MinimalTuple) case
*/
&stup->isnull1);
}
-static void
-reversedirection_heap(Tuplesortstate *state)
-{
- SortSupport sortKey = state->sortKeys;
- int nkey;
-
- for (nkey = 0; nkey < state->nKeys; nkey++, sortKey++)
- {
- sortKey->ssup_reverse = !sortKey->ssup_reverse;
- sortKey->ssup_nulls_first = !sortKey->ssup_nulls_first;
- }
-}
-
-
/*
* Routines specialized for the CLUSTER case (HeapTuple data, with
* comparisons per a btree index definition)
comparetup_cluster(const SortTuple *a, const SortTuple *b,
Tuplesortstate *state)
{
- ScanKey scanKey = state->indexScanKey;
+ SortSupport sortKey = state->sortKeys;
HeapTuple ltup;
HeapTuple rtup;
TupleDesc tupDesc;
/* Compare the leading sort key, if it's simple */
if (state->indexInfo->ii_KeyAttrNumbers[0] != 0)
{
- compare = inlineApplySortFunction(&scanKey->sk_func, scanKey->sk_flags,
- scanKey->sk_collation,
- a->datum1, a->isnull1,
- b->datum1, b->isnull1);
+ compare = ApplySortComparator(a->datum1, a->isnull1,
+ b->datum1, b->isnull1,
+ sortKey);
if (compare != 0 || state->nKeys == 1)
return compare;
/* Compare additional columns the hard way */
- scanKey++;
+ sortKey++;
nkey = 1;
}
else
/* If not expression index, just compare the proper heap attrs */
tupDesc = state->tupDesc;
- for (; nkey < state->nKeys; nkey++, scanKey++)
+ for (; nkey < state->nKeys; nkey++, sortKey++)
{
AttrNumber attno = state->indexInfo->ii_KeyAttrNumbers[nkey];
Datum datum1,
datum1 = heap_getattr(ltup, attno, tupDesc, &isnull1);
datum2 = heap_getattr(rtup, attno, tupDesc, &isnull2);
- compare = inlineApplySortFunction(&scanKey->sk_func,
- scanKey->sk_flags,
- scanKey->sk_collation,
- datum1, isnull1,
- datum2, isnull2);
+ compare = ApplySortComparator(datum1, isnull1,
+ datum2, isnull2,
+ sortKey);
if (compare != 0)
return compare;
}
FormIndexDatum(state->indexInfo, ecxt_scantuple, state->estate,
r_index_values, r_index_isnull);
- for (; nkey < state->nKeys; nkey++, scanKey++)
+ for (; nkey < state->nKeys; nkey++, sortKey++)
{
- compare = inlineApplySortFunction(&scanKey->sk_func,
- scanKey->sk_flags,
- scanKey->sk_collation,
- l_index_values[nkey],
- l_index_isnull[nkey],
- r_index_values[nkey],
- r_index_isnull[nkey]);
+ compare = ApplySortComparator(l_index_values[nkey],
+ l_index_isnull[nkey],
+ r_index_values[nkey],
+ r_index_isnull[nkey],
+ sortKey);
if (compare != 0)
return compare;
}
* is also special handling for enforcing uniqueness, and special treatment
* for equal keys at the end.
*/
- ScanKey scanKey = state->indexScanKey;
+ SortSupport sortKey = state->sortKeys;
IndexTuple tuple1;
IndexTuple tuple2;
int keysz;
int32 compare;
/* Compare the leading sort key */
- compare = inlineApplySortFunction(&scanKey->sk_func, scanKey->sk_flags,
- scanKey->sk_collation,
- a->datum1, a->isnull1,
- b->datum1, b->isnull1);
+ compare = ApplySortComparator(a->datum1, a->isnull1,
+ b->datum1, b->isnull1,
+ sortKey);
if (compare != 0)
return compare;
tuple2 = (IndexTuple) b->tuple;
keysz = state->nKeys;
tupDes = RelationGetDescr(state->indexRel);
- scanKey++;
- for (nkey = 2; nkey <= keysz; nkey++, scanKey++)
+ sortKey++;
+ for (nkey = 2; nkey <= keysz; nkey++, sortKey++)
{
Datum datum1,
datum2;
datum1 = index_getattr(tuple1, nkey, tupDes, &isnull1);
datum2 = index_getattr(tuple2, nkey, tupDes, &isnull2);
- compare = inlineApplySortFunction(&scanKey->sk_func, scanKey->sk_flags,
- scanKey->sk_collation,
- datum1, isnull1,
- datum2, isnull2);
+ compare = ApplySortComparator(datum1, isnull1,
+ datum2, isnull2,
+ sortKey);
if (compare != 0)
return compare; /* done when we find unequal attributes */
&stup->isnull1);
}
-static void
-reversedirection_index_btree(Tuplesortstate *state)
-{
- ScanKey scanKey = state->indexScanKey;
- int nkey;
-
- for (nkey = 0; nkey < state->nKeys; nkey++, scanKey++)
- {
- scanKey->sk_flags ^= (SK_BT_DESC | SK_BT_NULLS_FIRST);
- }
-}
-
-static void
-reversedirection_index_hash(Tuplesortstate *state)
-{
- /* We don't support reversing direction in a hash index sort */
- elog(ERROR, "reversedirection_index_hash is not implemented");
-}
-
-
/*
* Routines specialized for DatumTuple case
*/
&tuplen, sizeof(tuplen));
}
-static void
-reversedirection_datum(Tuplesortstate *state)
-{
- state->onlyKey->ssup_reverse = !state->onlyKey->ssup_reverse;
- state->onlyKey->ssup_nulls_first = !state->onlyKey->ssup_nulls_first;
-}
-
/*
* Convenience routine to free a tuple previously loaded into sort memory
*/
projects / postgresql.git / commitdiff