diff --git a/doc/src/sgml/spgist.sgml b/doc/src/sgml/spgist.sgml
index dcc3cc2d7331247fe3b3cb998188ed08083ff98a..0202dbcdd5a260bb66e18c2b7baab85de7bad44c 100644
--- a/doc/src/sgml/spgist.sgml
+++ b/doc/src/sgml/spgist.sgml
@@ -439,8 +439,8 @@ CREATE FUNCTION my_inner_consistent(internal, internal) RETURNS void ...
<programlisting>
typedef struct spgInnerConsistentIn
{
- StrategyNumber strategy; /* operator strategy number */
- Datum query; /* operator's RHS value */
+ ScanKey scankeys; /* array of operators and comparison values */
+ int nkeys; /* length of array */
Datum reconstructedValue; /* value reconstructed at parent */
int level; /* current level (counting from zero) */
@@ -463,8 +463,17 @@ typedef struct spgInnerConsistentOut
} spgInnerConsistentOut;
</programlisting>
- <structfield>strategy</> and
- <structfield>query</> describe the index search condition.
+ The array <structfield>scankeys</>, of length <structfield>nkeys</>,
+ describes the index search condition(s). These conditions are
+ combined with AND — only index entries that satisfy all of
+ them are interesting. (Note that <structfield>nkeys</> = 0 implies
+ that all index entries satisfy the query.) Usually the consistent
+ function only cares about the <structfield>sk_strategy</> and
+ <structfield>sk_argument</> fields of each array entry, which
+ respectively give the indexable operator and comparison value.
+ In particular it is not necessary to check <structfield>sk_flags</> to
+ see if the comparison value is NULL, because the SP-GiST core code
+ will filter out such conditions.
<structfield>reconstructedValue</> is the value reconstructed for the
parent tuple; it is <literal>(Datum) 0</> at the root level or if the
<function>inner_consistent</> function did not provide a value at the
@@ -527,8 +536,8 @@ CREATE FUNCTION my_leaf_consistent(internal, internal) RETURNS bool ...
<programlisting>
typedef struct spgLeafConsistentIn
{
- StrategyNumber strategy; /* operator strategy number */
- Datum query; /* operator's RHS value */
+ ScanKey scankeys; /* array of operators and comparison values */
+ int nkeys; /* length of array */
Datum reconstructedValue; /* value reconstructed at parent */
int level; /* current level (counting from zero) */
@@ -544,8 +553,17 @@ typedef struct spgLeafConsistentOut
} spgLeafConsistentOut;
</programlisting>
- <structfield>strategy</> and
- <structfield>query</> define the index search condition.
+ The array <structfield>scankeys</>, of length <structfield>nkeys</>,
+ describes the index search condition(s). These conditions are
+ combined with AND — only index entries that satisfy all of
+ them satisfy the query. (Note that <structfield>nkeys</> = 0 implies
+ that all index entries satisfy the query.) Usually the consistent
+ function only cares about the <structfield>sk_strategy</> and
+ <structfield>sk_argument</> fields of each array entry, which
+ respectively give the indexable operator and comparison value.
+ In particular it is not necessary to check <structfield>sk_flags</> to
+ see if the comparison value is NULL, because the SP-GiST core code
+ will filter out such conditions.
<structfield>reconstructedValue</> is the value reconstructed for the
parent tuple; it is <literal>(Datum) 0</> at the root level or if the
<function>inner_consistent</> function did not provide a value at the
@@ -566,8 +584,8 @@ typedef struct spgLeafConsistentOut
<structfield>leafValue</> must be set to the value originally supplied
to be indexed for this leaf tuple. Also,
<structfield>recheck</> may be set to <literal>true</> if the match
- is uncertain and so the operator must be re-applied to the actual heap
- tuple to verify the match.
+ is uncertain and so the operator(s) must be re-applied to the actual
+ heap tuple to verify the match.
</para>
</listitem>
</varlistentry>
diff --git a/src/backend/access/spgist/spgkdtreeproc.c b/src/backend/access/spgist/spgkdtreeproc.c
index eca972a6f0b8ea5395719701222fe5aeec61beaf..adfe287581b8c101af79a461f176222b338dfecf 100644
--- a/src/backend/access/spgist/spgkdtreeproc.c
+++ b/src/backend/access/spgist/spgkdtreeproc.c
@@ -159,11 +159,10 @@ spg_kd_inner_consistent(PG_FUNCTION_ARGS)
{
spgInnerConsistentIn *in = (spgInnerConsistentIn *) PG_GETARG_POINTER(0);
spgInnerConsistentOut *out = (spgInnerConsistentOut *) PG_GETARG_POINTER(1);
- Point *query;
- BOX *boxQuery;
double coord;
+ int which;
+ int i;
- query = DatumGetPointP(in->query);
Assert(in->hasPrefix);
coord = DatumGetFloat8(in->prefixDatum);
@@ -171,124 +170,97 @@ spg_kd_inner_consistent(PG_FUNCTION_ARGS)
elog(ERROR, "allTheSame should not occur for k-d trees");
Assert(in->nNodes == 2);
- out->nodeNumbers = (int *) palloc(sizeof(int) * 2);
- out->levelAdds = (int *) palloc(sizeof(int) * 2);
- out->levelAdds[0] = 1;
- out->levelAdds[1] = 1;
- out->nNodes = 0;
- switch (in->strategy)
+ /* "which" is a bitmask of children that satisfy all constraints */
+ which = (1 << 1) | (1 << 2);
+
+ for (i = 0; i < in->nkeys; i++)
{
- case RTLeftStrategyNumber:
- out->nNodes = 1;
- out->nodeNumbers[0] = 0;
-
- if ((in->level % 2) == 0 || FPge(query->x, coord))
- {
- out->nodeNumbers[1] = 1;
- out->nNodes++;
- }
- break;
- case RTRightStrategyNumber:
- out->nNodes = 1;
- out->nodeNumbers[0] = 1;
-
- if ((in->level % 2) == 0 || FPle(query->x, coord))
- {
- out->nodeNumbers[1] = 0;
- out->nNodes++;
- }
- break;
- case RTSameStrategyNumber:
- if (in->level % 2)
- {
- if (FPle(query->x, coord))
- {
- out->nodeNumbers[out->nNodes] = 0;
- out->nNodes++;
- }
- if (FPge(query->x, coord))
- {
- out->nodeNumbers[out->nNodes] = 1;
- out->nNodes++;
- }
- }
- else
- {
- if (FPle(query->y, coord))
+ Point *query = DatumGetPointP(in->scankeys[i].sk_argument);
+ BOX *boxQuery;
+
+ switch (in->scankeys[i].sk_strategy)
+ {
+ case RTLeftStrategyNumber:
+ if ((in->level % 2) != 0 && FPlt(query->x, coord))
+ which &= (1 << 1);
+ break;
+ case RTRightStrategyNumber:
+ if ((in->level % 2) != 0 && FPgt(query->x, coord))
+ which &= (1 << 2);
+ break;
+ case RTSameStrategyNumber:
+ if ((in->level % 2) != 0)
{
- out->nodeNumbers[out->nNodes] = 0;
- out->nNodes++;
+ if (FPlt(query->x, coord))
+ which &= (1 << 1);
+ else if (FPgt(query->x, coord))
+ which &= (1 << 2);
}
- if (FPge(query->y, coord))
+ else
{
- out->nodeNumbers[out->nNodes] = 1;
- out->nNodes++;
+ if (FPlt(query->y, coord))
+ which &= (1 << 1);
+ else if (FPgt(query->y, coord))
+ which &= (1 << 2);
}
- }
- break;
- case RTBelowStrategyNumber:
- out->nNodes = 1;
- out->nodeNumbers[0] = 0;
-
- if ((in->level % 2) == 1 || FPge(query->y, coord))
- {
- out->nodeNumbers[1] = 1;
- out->nNodes++;
- }
- break;
- case RTAboveStrategyNumber:
- out->nNodes = 1;
- out->nodeNumbers[0] = 1;
-
- if ((in->level % 2) == 1 || FPle(query->y, coord))
- {
- out->nodeNumbers[1] = 0;
- out->nNodes++;
- }
- break;
- case RTContainedByStrategyNumber:
-
- /*
- * For this operator, the query is a box not a point. We cheat to
- * the extent of assuming that DatumGetPointP won't do anything
- * that would be bad for a pointer-to-box.
- */
- boxQuery = DatumGetBoxP(in->query);
-
- out->nNodes = 1;
- if (in->level % 2)
- {
- if (FPlt(boxQuery->high.x, coord))
- out->nodeNumbers[0] = 0;
- else if (FPgt(boxQuery->low.x, coord))
- out->nodeNumbers[0] = 1;
- else
+ break;
+ case RTBelowStrategyNumber:
+ if ((in->level % 2) == 0 && FPlt(query->y, coord))
+ which &= (1 << 1);
+ break;
+ case RTAboveStrategyNumber:
+ if ((in->level % 2) == 0 && FPgt(query->y, coord))
+ which &= (1 << 2);
+ break;
+ case RTContainedByStrategyNumber:
+
+ /*
+ * For this operator, the query is a box not a point. We
+ * cheat to the extent of assuming that DatumGetPointP won't
+ * do anything that would be bad for a pointer-to-box.
+ */
+ boxQuery = DatumGetBoxP(in->scankeys[i].sk_argument);
+
+ if ((in->level % 2) != 0)
{
- out->nodeNumbers[0] = 0;
- out->nodeNumbers[1] = 1;
- out->nNodes = 2;
+ if (FPlt(boxQuery->high.x, coord))
+ which &= (1 << 1);
+ else if (FPgt(boxQuery->low.x, coord))
+ which &= (1 << 2);
}
- }
- else
- {
- if (FPlt(boxQuery->high.y, coord))
- out->nodeNumbers[0] = 0;
- else if (FPgt(boxQuery->low.y, coord))
- out->nodeNumbers[0] = 1;
else
{
- out->nodeNumbers[0] = 0;
- out->nodeNumbers[1] = 1;
- out->nNodes = 2;
+ if (FPlt(boxQuery->high.y, coord))
+ which &= (1 << 1);
+ else if (FPgt(boxQuery->low.y, coord))
+ which &= (1 << 2);
}
- }
- break;
- default:
- elog(ERROR, "unrecognized strategy number: %d", in->strategy);
- break;
+ break;
+ default:
+ elog(ERROR, "unrecognized strategy number: %d",
+ in->scankeys[i].sk_strategy);
+ break;
+ }
+
+ if (which == 0)
+ break; /* no need to consider remaining conditions */
}
+ /* We must descend into the children identified by which */
+ out->nodeNumbers = (int *) palloc(sizeof(int) * 2);
+ out->nNodes = 0;
+ for (i = 1; i <= 2; i++)
+ {
+ if (which & (1 << i))
+ out->nodeNumbers[out->nNodes++] = i - 1;
+ }
+
+ /* Set up level increments, too */
+ out->levelAdds = (int *) palloc(sizeof(int) * 2);
+ out->levelAdds[0] = 1;
+ out->levelAdds[1] = 1;
+
PG_RETURN_VOID();
}
diff --git a/src/backend/access/spgist/spgquadtreeproc.c b/src/backend/access/spgist/spgquadtreeproc.c
index 231749e992dba9686cdba2af48908622611d3794..10fafe58643b249ae6a32fee229c71b394e0e4eb 100644
--- a/src/backend/access/spgist/spgquadtreeproc.c
+++ b/src/backend/access/spgist/spgquadtreeproc.c
@@ -190,45 +190,21 @@ spg_quad_picksplit(PG_FUNCTION_ARGS)
}
-/* Subroutine to fill out->nodeNumbers[] for spg_quad_inner_consistent */
-static void
-setNodes(spgInnerConsistentOut *out, bool isAll, int first, int second)
-{
- if (isAll)
- {
- out->nNodes = 4;
- out->nodeNumbers[0] = 0;
- out->nodeNumbers[1] = 1;
- out->nodeNumbers[2] = 2;
- out->nodeNumbers[3] = 3;
- }
- else
- {
- out->nNodes = 2;
- out->nodeNumbers[0] = first - 1;
- out->nodeNumbers[1] = second - 1;
- }
-}
-
-
Datum
spg_quad_inner_consistent(PG_FUNCTION_ARGS)
{
spgInnerConsistentIn *in = (spgInnerConsistentIn *) PG_GETARG_POINTER(0);
spgInnerConsistentOut *out = (spgInnerConsistentOut *) PG_GETARG_POINTER(1);
- Point *query,
- *centroid;
- BOX *boxQuery;
+ Point *centroid;
+ int which;
+ int i;
- query = DatumGetPointP(in->query);
Assert(in->hasPrefix);
centroid = DatumGetPointP(in->prefixDatum);
if (in->allTheSame)
{
/* Report that all nodes should be visited */
- int i;
-
out->nNodes = in->nNodes;
out->nodeNumbers = (int *) palloc(sizeof(int) * in->nNodes);
for (i = 0; i < in->nNodes; i++)
@@ -237,76 +213,86 @@ spg_quad_inner_consistent(PG_FUNCTION_ARGS)
}
Assert(in->nNodes == 4);
- out->nodeNumbers = (int *) palloc(sizeof(int) * 4);
- switch (in->strategy)
+ /* "which" is a bitmask of quadrants that satisfy all constraints */
+ which = (1 << 1) | (1 << 2) | (1 << 3) | (1 << 4);
+
+ for (i = 0; i < in->nkeys; i++)
{
- case RTLeftStrategyNumber:
- setNodes(out, SPTEST(point_left, centroid, query), 3, 4);
- break;
- case RTRightStrategyNumber:
- setNodes(out, SPTEST(point_right, centroid, query), 1, 2);
- break;
- case RTSameStrategyNumber:
- out->nNodes = 1;
- out->nodeNumbers[0] = getQuadrant(centroid, query) - 1;
- break;
- case RTBelowStrategyNumber:
- setNodes(out, SPTEST(point_below, centroid, query), 2, 3);
- break;
- case RTAboveStrategyNumber:
- setNodes(out, SPTEST(point_above, centroid, query), 1, 4);
- break;
- case RTContainedByStrategyNumber:
-
- /*
- * For this operator, the query is a box not a point. We cheat to
- * the extent of assuming that DatumGetPointP won't do anything
- * that would be bad for a pointer-to-box.
- */
- boxQuery = DatumGetBoxP(in->query);
-
- if (DatumGetBool(DirectFunctionCall2(box_contain_pt,
- PointerGetDatum(boxQuery),
- PointerGetDatum(centroid))))
- {
- /* centroid is in box, so descend to all quadrants */
- setNodes(out, true, 0, 0);
- }
- else
- {
- /* identify quadrant(s) containing all corners of box */
- Point p;
- int i,
- r = 0;
-
- p = boxQuery->low;
- r |= 1 << (getQuadrant(centroid, &p) - 1);
-
- p.y = boxQuery->high.y;
- r |= 1 << (getQuadrant(centroid, &p) - 1);
-
- p = boxQuery->high;
- r |= 1 << (getQuadrant(centroid, &p) - 1);
-
- p.x = boxQuery->low.x;
- r |= 1 << (getQuadrant(centroid, &p) - 1);
-
- /* we must descend into those quadrant(s) */
- out->nNodes = 0;
- for (i = 0; i < 4; i++)
+ Point *query = DatumGetPointP(in->scankeys[i].sk_argument);
+ BOX *boxQuery;
+
+ switch (in->scankeys[i].sk_strategy)
+ {
+ case RTLeftStrategyNumber:
+ if (SPTEST(point_right, centroid, query))
+ which &= (1 << 3) | (1 << 4);
+ break;
+ case RTRightStrategyNumber:
+ if (SPTEST(point_left, centroid, query))
+ which &= (1 << 1) | (1 << 2);
+ break;
+ case RTSameStrategyNumber:
+ which &= (1 << getQuadrant(centroid, query));
+ break;
+ case RTBelowStrategyNumber:
+ if (SPTEST(point_above, centroid, query))
+ which &= (1 << 2) | (1 << 3);
+ break;
+ case RTAboveStrategyNumber:
+ if (SPTEST(point_below, centroid, query))
+ which &= (1 << 1) | (1 << 4);
+ break;
+ case RTContainedByStrategyNumber:
+
+ /*
+ * For this operator, the query is a box not a point. We
+ * cheat to the extent of assuming that DatumGetPointP won't
+ * do anything that would be bad for a pointer-to-box.
+ */
+ boxQuery = DatumGetBoxP(in->scankeys[i].sk_argument);
+
+ if (DatumGetBool(DirectFunctionCall2(box_contain_pt,
+ PointerGetDatum(boxQuery),
+ PointerGetDatum(centroid))))
{
- if (r & (1 << i))
- {
- out->nodeNumbers[out->nNodes] = i;
- out->nNodes++;
- }
+ /* centroid is in box, so all quadrants are OK */
}
- }
- break;
- default:
- elog(ERROR, "unrecognized strategy number: %d", in->strategy);
- break;
+ else
+ {
+ /* identify quadrant(s) containing all corners of box */
+ Point p;
+ int r = 0;
+
+ p = boxQuery->low;
+ r |= 1 << getQuadrant(centroid, &p);
+ p.y = boxQuery->high.y;
+ r |= 1 << getQuadrant(centroid, &p);
+ p = boxQuery->high;
+ r |= 1 << getQuadrant(centroid, &p);
+ p.x = boxQuery->low.x;
+ r |= 1 << getQuadrant(centroid, &p);
+
+ which &= r;
+ }
+ break;
+ default:
+ elog(ERROR, "unrecognized strategy number: %d",
+ in->scankeys[i].sk_strategy);
+ break;
+ }
+
+ if (which == 0)
+ break; /* no need to consider remaining conditions */
+ }
+
+ /* We must descend into the quadrant(s) identified by which */
+ out->nodeNumbers = (int *) palloc(sizeof(int) * 4);
+ out->nNodes = 0;
+ for (i = 1; i <= 4; i++)
+ {
+ if (which & (1 << i))
+ out->nodeNumbers[out->nNodes++] = i - 1;
}
PG_RETURN_VOID();
@@ -318,9 +304,9 @@ spg_quad_leaf_consistent(PG_FUNCTION_ARGS)
{
spgLeafConsistentIn *in = (spgLeafConsistentIn *) PG_GETARG_POINTER(0);
spgLeafConsistentOut *out = (spgLeafConsistentOut *) PG_GETARG_POINTER(1);
- Point *query = DatumGetPointP(in->query);
Point *datum = DatumGetPointP(in->leafDatum);
bool res;
+ int i;
/* all tests are exact */
out->recheck = false;
@@ -328,35 +314,45 @@ spg_quad_leaf_consistent(PG_FUNCTION_ARGS)
/* leafDatum is what it is... */
out->leafValue = in->leafDatum;
- switch (in->strategy)
+ /* Perform the required comparison(s) */
+ res = true;
+ for (i = 0; i < in->nkeys; i++)
{
- case RTLeftStrategyNumber:
- res = SPTEST(point_left, datum, query);
- break;
- case RTRightStrategyNumber:
- res = SPTEST(point_right, datum, query);
- break;
- case RTSameStrategyNumber:
- res = SPTEST(point_eq, datum, query);
- break;
- case RTBelowStrategyNumber:
- res = SPTEST(point_below, datum, query);
- break;
- case RTAboveStrategyNumber:
- res = SPTEST(point_above, datum, query);
- break;
- case RTContainedByStrategyNumber:
-
- /*
- * For this operator, the query is a box not a point. We cheat to
- * the extent of assuming that DatumGetPointP won't do anything
- * that would be bad for a pointer-to-box.
- */
- res = SPTEST(box_contain_pt, query, datum);
- break;
- default:
- elog(ERROR, "unrecognized strategy number: %d", in->strategy);
- res = false;
+ Point *query = DatumGetPointP(in->scankeys[i].sk_argument);
+
+ switch (in->scankeys[i].sk_strategy)
+ {
+ case RTLeftStrategyNumber:
+ res = SPTEST(point_left, datum, query);
+ break;
+ case RTRightStrategyNumber:
+ res = SPTEST(point_right, datum, query);
+ break;
+ case RTSameStrategyNumber:
+ res = SPTEST(point_eq, datum, query);
+ break;
+ case RTBelowStrategyNumber:
+ res = SPTEST(point_below, datum, query);
+ break;
+ case RTAboveStrategyNumber:
+ res = SPTEST(point_above, datum, query);
+ break;
+ case RTContainedByStrategyNumber:
+
+ /*
+ * For this operator, the query is a box not a point. We
+ * cheat to the extent of assuming that DatumGetPointP won't
+ * do anything that would be bad for a pointer-to-box.
+ */
+ res = SPTEST(box_contain_pt, query, datum);
+ break;
+ default:
+ elog(ERROR, "unrecognized strategy number: %d",
+ in->scankeys[i].sk_strategy);
+ break;
+ }
+
+ if (!res)
break;
}
diff --git a/src/backend/access/spgist/spgscan.c b/src/backend/access/spgist/spgscan.c
index 22cfcc8792933d3a557af017c79a271531b68ba5..99b0852611fbc7454d3e1ef04d7446d89f12b380 100644
--- a/src/backend/access/spgist/spgscan.c
+++ b/src/backend/access/spgist/spgscan.c
@@ -56,18 +56,25 @@ freeScanStack(SpGistScanOpaque so)
}
/*
- * Initialize scanStack with a single entry for the root page, resetting
+ * Initialize scanStack to search the root page, resetting
* any previously active scan
*/
static void
resetSpGistScanOpaque(SpGistScanOpaque so)
{
- ScanStackEntry *startEntry = palloc0(sizeof(ScanStackEntry));
-
- ItemPointerSet(&startEntry->ptr, SPGIST_HEAD_BLKNO, FirstOffsetNumber);
+ ScanStackEntry *startEntry;
freeScanStack(so);
- so->scanStack = list_make1(startEntry);
+
+ Assert(!so->searchNulls); /* XXX fixme */
+
+ if (so->searchNonNulls)
+ {
+ /* Stack a work item to scan the non-null index entries */
+ startEntry = (ScanStackEntry *) palloc0(sizeof(ScanStackEntry));
+ ItemPointerSet(&startEntry->ptr, SPGIST_HEAD_BLKNO, FirstOffsetNumber);
+ so->scanStack = list_make1(startEntry);
+ }
if (so->want_itup)
{
@@ -80,6 +87,82 @@ resetSpGistScanOpaque(SpGistScanOpaque so)
so->iPtr = so->nPtrs = 0;
}
+/*
+ * Prepare scan keys in SpGistScanOpaque from caller-given scan keys
+ *
+ * Sets searchNulls, searchNonNulls, numberOfKeys, keyData fields of *so.
+ *
+ * The point here is to eliminate null-related considerations from what the
+ * opclass consistent functions need to deal with. We assume all SPGiST-
+ * indexable operators are strict, so any null RHS value makes the scan
+ * condition unsatisfiable. We also pull out any IS NULL/IS NOT NULL
+ * conditions; their effect is reflected into searchNulls/searchNonNulls.
+ */
+static void
+spgPrepareScanKeys(IndexScanDesc scan)
+{
+ SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;
+ bool qual_ok;
+ bool haveIsNull;
+ bool haveNotNull;
+ int nkeys;
+ int i;
+
+ if (scan->numberOfKeys <= 0)
+ {
+ /* If no quals, whole-index scan is required */
+ so->searchNulls = true;
+ so->searchNonNulls = true;
+ so->numberOfKeys = 0;
+ return;
+ }
+
+ /* Examine the given quals */
+ qual_ok = true;
+ haveIsNull = haveNotNull = false;
+ nkeys = 0;
+ for (i = 0; i < scan->numberOfKeys; i++)
+ {
+ ScanKey skey = &scan->keyData[i];
+
+ if (skey->sk_flags & SK_SEARCHNULL)
+ haveIsNull = true;
+ else if (skey->sk_flags & SK_SEARCHNOTNULL)
+ haveNotNull = true;
+ else if (skey->sk_flags & SK_ISNULL)
+ {
+ /* ordinary qual with null argument - unsatisfiable */
+ qual_ok = false;
+ break;
+ }
+ else
+ {
+ /* ordinary qual, propagate into so->keyData */
+ so->keyData[nkeys++] = *skey;
+ /* this effectively creates a not-null requirement */
+ haveNotNull = true;
+ }
+ }
+
+ /* IS NULL in combination with something else is unsatisfiable */
+ if (haveIsNull && haveNotNull)
+ qual_ok = false;
+
+ /* Emit results */
+ if (qual_ok)
+ {
+ so->searchNulls = haveIsNull;
+ so->searchNonNulls = haveNotNull;
+ so->numberOfKeys = nkeys;
+ }
+ else
+ {
+ so->searchNulls = false;
+ so->searchNonNulls = false;
+ so->numberOfKeys = 0;
+ }
+}
+
Datum
spgbeginscan(PG_FUNCTION_ARGS)
{
@@ -92,13 +175,16 @@ spgbeginscan(PG_FUNCTION_ARGS)
scan = RelationGetIndexScan(rel, keysz, 0);
so = (SpGistScanOpaque) palloc0(sizeof(SpGistScanOpaqueData));
+ if (keysz > 0)
+ so->keyData = (ScanKey) palloc(sizeof(ScanKeyData) * keysz);
+ else
+ so->keyData = NULL;
initSpGistState(&so->state, scan->indexRelation);
so->tempCxt = AllocSetContextCreate(CurrentMemoryContext,
"SP-GiST search temporary context",
ALLOCSET_DEFAULT_MINSIZE,
ALLOCSET_DEFAULT_INITSIZE,
ALLOCSET_DEFAULT_MAXSIZE);
- resetSpGistScanOpaque(so);
/* Set up indexTupDesc and xs_itupdesc in case it's an index-only scan */
so->indexTupDesc = scan->xs_itupdesc = RelationGetDescr(rel);
@@ -115,12 +201,17 @@ spgrescan(PG_FUNCTION_ARGS)
SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;
ScanKey scankey = (ScanKey) PG_GETARG_POINTER(1);
+ /* copy scankeys into local storage */
if (scankey && scan->numberOfKeys > 0)
{
memmove(scan->keyData, scankey,
scan->numberOfKeys * sizeof(ScanKeyData));
}
+ /* preprocess scankeys, set up the representation in *so */
+ spgPrepareScanKeys(scan);
+
+ /* set up starting stack entries */
resetSpGistScanOpaque(so);
PG_RETURN_VOID();
@@ -162,53 +253,34 @@ spgLeafTest(Relation index, SpGistScanOpaque so, Datum leafDatum,
int level, Datum reconstructedValue,
Datum *leafValue, bool *recheck)
{
- bool result = true;
+ bool result;
spgLeafConsistentIn in;
spgLeafConsistentOut out;
FmgrInfo *procinfo;
MemoryContext oldCtx;
- int i;
- *leafValue = (Datum) 0;
- *recheck = false;
+ /* use temp context for calling leaf_consistent */
+ oldCtx = MemoryContextSwitchTo(so->tempCxt);
- /* set up values that are the same for all quals */
+ in.scankeys = so->keyData;
+ in.nkeys = so->numberOfKeys;
in.reconstructedValue = reconstructedValue;
in.level = level;
in.returnData = so->want_itup;
in.leafDatum = leafDatum;
- /* Apply each leaf consistency check, working in the temp context */
- oldCtx = MemoryContextSwitchTo(so->tempCxt);
+ out.leafValue = (Datum) 0;
+ out.recheck = false;
procinfo = index_getprocinfo(index, 1, SPGIST_LEAF_CONSISTENT_PROC);
+ result = DatumGetBool(FunctionCall2Coll(procinfo,
+ index->rd_indcollation[0],
+ PointerGetDatum(&in),
+ PointerGetDatum(&out)));
- for (i = 0; i < so->numberOfKeys; i++)
- {
- ScanKey skey = &so->keyData[i];
-
- /* Assume SPGiST-indexable operators are strict */
- if (skey->sk_flags & SK_ISNULL)
- {
- result = false;
- break;
- }
+ *leafValue = out.leafValue;
+ *recheck = out.recheck;
- in.strategy = skey->sk_strategy;
- in.query = skey->sk_argument;
-
- out.leafValue = (Datum) 0;
- out.recheck = false;
-
- result = DatumGetBool(FunctionCall2Coll(procinfo,
- skey->sk_collation,
- PointerGetDatum(&in),
- PointerGetDatum(&out)));
- *leafValue = out.leafValue;
- *recheck |= out.recheck;
- if (!result)
- break;
- }
MemoryContextSwitchTo(oldCtx);
return result;
@@ -349,8 +421,13 @@ redirect:
else /* page is inner */
{
SpGistInnerTuple innerTuple;
+ spgInnerConsistentIn in;
+ spgInnerConsistentOut out;
+ FmgrInfo *procinfo;
+ SpGistNodeTuple *nodes;
SpGistNodeTuple node;
int i;
+ MemoryContext oldCtx;
innerTuple = (SpGistInnerTuple) PageGetItem(page,
PageGetItemId(page, offset));
@@ -368,144 +445,68 @@ redirect:
innerTuple->tupstate);
}
- if (so->numberOfKeys == 0)
+ /* use temp context for calling inner_consistent */
+ oldCtx = MemoryContextSwitchTo(so->tempCxt);
+
+ in.scankeys = so->keyData;
+ in.nkeys = so->numberOfKeys;
+ in.reconstructedValue = stackEntry->reconstructedValue;
+ in.level = stackEntry->level;
+ in.returnData = so->want_itup;
+ in.allTheSame = innerTuple->allTheSame;
+ in.hasPrefix = (innerTuple->prefixSize > 0);
+ in.prefixDatum = SGITDATUM(innerTuple, &so->state);
+ in.nNodes = innerTuple->nNodes;
+ in.nodeLabels = spgExtractNodeLabels(&so->state, innerTuple);
+
+ /* collect node pointers */
+ nodes = (SpGistNodeTuple *) palloc(sizeof(SpGistNodeTuple) * in.nNodes);
+ SGITITERATE(innerTuple, i, node)
{
- /*
- * This case cannot happen at the moment, because we don't
- * set pg_am.amoptionalkey for SP-GiST. In order for full
- * index scans to produce correct answers, we'd need to
- * index nulls, which we don't.
- */
- Assert(false);
-
-#ifdef NOT_USED
- /*
- * A full index scan could be done approximately like this,
- * but note that reconstruction of indexed values would be
- * impossible unless the API for inner_consistent is changed.
- */
- SGITITERATE(innerTuple, i, node)
- {
- if (ItemPointerIsValid(&node->t_tid))
- {
- ScanStackEntry *newEntry = palloc(sizeof(ScanStackEntry));
-
- newEntry->ptr = node->t_tid;
- newEntry->level = -1;
- newEntry->reconstructedValue = (Datum) 0;
- so->scanStack = lcons(newEntry, so->scanStack);
- }
- }
-#endif
+ nodes[i] = node;
}
- else
- {
- spgInnerConsistentIn in;
- spgInnerConsistentOut out;
- FmgrInfo *procinfo;
- SpGistNodeTuple *nodes;
- int *andMap;
- int *levelAdds;
- Datum *reconstructedValues;
- int j,
- nMatches = 0;
- MemoryContext oldCtx;
-
- /* use temp context for calling inner_consistent */
- oldCtx = MemoryContextSwitchTo(so->tempCxt);
-
- /* set up values that are the same for all scankeys */
- in.reconstructedValue = stackEntry->reconstructedValue;
- in.level = stackEntry->level;
- in.returnData = so->want_itup;
- in.allTheSame = innerTuple->allTheSame;
- in.hasPrefix = (innerTuple->prefixSize > 0);
- in.prefixDatum = SGITDATUM(innerTuple, &so->state);
- in.nNodes = innerTuple->nNodes;
- in.nodeLabels = spgExtractNodeLabels(&so->state, innerTuple);
-
- /* collect node pointers */
- nodes = (SpGistNodeTuple *) palloc(sizeof(SpGistNodeTuple) * in.nNodes);
- SGITITERATE(innerTuple, i, node)
- {
- nodes[i] = node;
- }
- andMap = (int *) palloc0(sizeof(int) * in.nNodes);
- levelAdds = (int *) palloc0(sizeof(int) * in.nNodes);
- reconstructedValues = (Datum *) palloc0(sizeof(Datum) * in.nNodes);
+ memset(&out, 0, sizeof(out));
- procinfo = index_getprocinfo(index, 1, SPGIST_INNER_CONSISTENT_PROC);
+ procinfo = index_getprocinfo(index, 1, SPGIST_INNER_CONSISTENT_PROC);
+ FunctionCall2Coll(procinfo,
+ index->rd_indcollation[0],
+ PointerGetDatum(&in),
+ PointerGetDatum(&out));
- for (j = 0; j < so->numberOfKeys; j++)
- {
- ScanKey skey = &so->keyData[j];
-
- /* Assume SPGiST-indexable operators are strict */
- if (skey->sk_flags & SK_ISNULL)
- {
- nMatches = 0;
- break;
- }
-
- in.strategy = skey->sk_strategy;
- in.query = skey->sk_argument;
+ MemoryContextSwitchTo(oldCtx);
- memset(&out, 0, sizeof(out));
-
- FunctionCall2Coll(procinfo,
- skey->sk_collation,
- PointerGetDatum(&in),
- PointerGetDatum(&out));
-
- /* If allTheSame, they should all or none of 'em match */
- if (innerTuple->allTheSame)
- if (out.nNodes != 0 && out.nNodes != in.nNodes)
- elog(ERROR, "inconsistent inner_consistent results for allTheSame inner tuple");
-
- nMatches = 0;
- for (i = 0; i < out.nNodes; i++)
- {
- int nodeN = out.nodeNumbers[i];
-
- andMap[nodeN]++;
- if (andMap[nodeN] == j + 1)
- nMatches++;
- if (out.levelAdds)
- levelAdds[nodeN] = out.levelAdds[i];
- if (out.reconstructedValues)
- reconstructedValues[nodeN] = out.reconstructedValues[i];
- }
+ /* If allTheSame, they should all or none of 'em match */
+ if (innerTuple->allTheSame)
+ if (out.nNodes != 0 && out.nNodes != in.nNodes)
+ elog(ERROR, "inconsistent inner_consistent results for allTheSame inner tuple");
- /* quit as soon as all nodes have failed some qual */
- if (nMatches == 0)
- break;
- }
-
- MemoryContextSwitchTo(oldCtx);
+ for (i = 0; i < out.nNodes; i++)
+ {
+ int nodeN = out.nodeNumbers[i];
- if (nMatches > 0)
+ Assert(nodeN >= 0 && nodeN < in.nNodes);
+ if (ItemPointerIsValid(&nodes[nodeN]->t_tid))
{
- for (i = 0; i < in.nNodes; i++)
- {
- if (andMap[i] == so->numberOfKeys &&
- ItemPointerIsValid(&nodes[i]->t_tid))
- {
- ScanStackEntry *newEntry;
-
- /* Create new work item for this node */
- newEntry = palloc(sizeof(ScanStackEntry));
- newEntry->ptr = nodes[i]->t_tid;
- newEntry->level = stackEntry->level + levelAdds[i];
- /* Must copy value out of temp context */
- newEntry->reconstructedValue =
- datumCopy(reconstructedValues[i],
- so->state.attType.attbyval,
- so->state.attType.attlen);
-
- so->scanStack = lcons(newEntry, so->scanStack);
- }
- }
+ ScanStackEntry *newEntry;
+
+ /* Create new work item for this node */
+ newEntry = palloc(sizeof(ScanStackEntry));
+ newEntry->ptr = nodes[nodeN]->t_tid;
+ if (out.levelAdds)
+ newEntry->level = stackEntry->level + out.levelAdds[i];
+ else
+ newEntry->level = stackEntry->level;
+ /* Must copy value out of temp context */
+ if (out.reconstructedValues)
+ newEntry->reconstructedValue =
+ datumCopy(out.reconstructedValues[i],
+ so->state.attType.attbyval,
+ so->state.attType.attlen);
+ else
+ newEntry->reconstructedValue = (Datum) 0;
+
+ so->scanStack = lcons(newEntry, so->scanStack);
}
}
}
@@ -536,10 +537,7 @@ spggetbitmap(PG_FUNCTION_ARGS)
TIDBitmap *tbm = (TIDBitmap *) PG_GETARG_POINTER(1);
SpGistScanOpaque so = (SpGistScanOpaque) scan->opaque;
- /* Copy scankey to *so so we don't need to pass it around separately */
- so->numberOfKeys = scan->numberOfKeys;
- so->keyData = scan->keyData;
- /* Ditto for the want_itup flag */
+ /* Copy want_itup to *so so we don't need to pass it around separately */
so->want_itup = false;
so->tbm = tbm;
@@ -583,10 +581,7 @@ spggettuple(PG_FUNCTION_ARGS)
if (dir != ForwardScanDirection)
elog(ERROR, "SP-GiST only supports forward scan direction");
- /* Copy scankey to *so so we don't need to pass it around separately */
- so->numberOfKeys = scan->numberOfKeys;
- so->keyData = scan->keyData;
- /* Ditto for the want_itup flag */
+ /* Copy want_itup to *so so we don't need to pass it around separately */
so->want_itup = scan->xs_want_itup;
for (;;)
diff --git a/src/backend/access/spgist/spgtextproc.c b/src/backend/access/spgist/spgtextproc.c
index b194fc1b13e52621683f5da165c1ba4091d94a26..656015ea7e6598f1fa194c2676747944d3f7bcca 100644
--- a/src/backend/access/spgist/spgtextproc.c
+++ b/src/backend/access/spgist/spgtextproc.c
@@ -362,25 +362,12 @@ spg_text_inner_consistent(PG_FUNCTION_ARGS)
{
spgInnerConsistentIn *in = (spgInnerConsistentIn *) PG_GETARG_POINTER(0);
spgInnerConsistentOut *out = (spgInnerConsistentOut *) PG_GETARG_POINTER(1);
- StrategyNumber strategy = in->strategy;
- text *inText;
- int inSize;
- int i;
+ bool collate_is_c = lc_collate_is_c(PG_GET_COLLATION());
text *reconstrText = NULL;
int maxReconstrLen = 0;
text *prefixText = NULL;
int prefixSize = 0;
-
- /*
- * If it's a collation-aware operator, but the collation is C, we can
- * treat it as non-collation-aware.
- */
- if (strategy > 10 &&
- lc_collate_is_c(PG_GET_COLLATION()))
- strategy -= 10;
-
- inText = DatumGetTextPP(in->query);
- inSize = VARSIZE_ANY_EXHDR(inText);
+ int i;
/*
* Reconstruct values represented at this tuple, including parent data,
@@ -431,8 +418,8 @@ spg_text_inner_consistent(PG_FUNCTION_ARGS)
{
uint8 nodeChar = DatumGetUInt8(in->nodeLabels[i]);
int thisLen;
- int r;
- bool res = false;
+ bool res = true;
+ int j;
/* If nodeChar is zero, don't include it in data */
if (nodeChar == '\0')
@@ -443,38 +430,57 @@ spg_text_inner_consistent(PG_FUNCTION_ARGS)
thisLen = maxReconstrLen;
}
- r = memcmp(VARDATA(reconstrText), VARDATA_ANY(inText),
- Min(inSize, thisLen));
-
- switch (strategy)
+ for (j = 0; j < in->nkeys; j++)
{
- case BTLessStrategyNumber:
- case BTLessEqualStrategyNumber:
- if (r <= 0)
- res = true;
- break;
- case BTEqualStrategyNumber:
- if (r == 0 && inSize >= thisLen)
- res = true;
- break;
- case BTGreaterEqualStrategyNumber:
- case BTGreaterStrategyNumber:
- if (r >= 0)
- res = true;
- break;
- case BTLessStrategyNumber + 10:
- case BTLessEqualStrategyNumber + 10:
- case BTGreaterEqualStrategyNumber + 10:
- case BTGreaterStrategyNumber + 10:
- /*
- * with non-C collation we need to traverse whole tree :-(
- */
- res = true;
- break;
- default:
- elog(ERROR, "unrecognized strategy number: %d",
- in->strategy);
- break;
+ StrategyNumber strategy = in->scankeys[j].sk_strategy;
+ text *inText;
+ int inSize;
+ int r;
+
+ /*
+ * If it's a collation-aware operator, but the collation is C, we
+ * can treat it as non-collation-aware. With non-C collation we
+ * need to traverse whole tree :-( so there's no point in making
+ * any check here.
+ */
+ if (strategy > 10)
+ {
+ if (collate_is_c)
+ strategy -= 10;
+ else
+ continue;
+ }
+
+ inText = DatumGetTextPP(in->scankeys[j].sk_argument);
+ inSize = VARSIZE_ANY_EXHDR(inText);
+
+ r = memcmp(VARDATA(reconstrText), VARDATA_ANY(inText),
+ Min(inSize, thisLen));
+
+ switch (strategy)
+ {
+ case BTLessStrategyNumber:
+ case BTLessEqualStrategyNumber:
+ if (r > 0)
+ res = false;
+ break;
+ case BTEqualStrategyNumber:
+ if (r != 0 || inSize < thisLen)
+ res = false;
+ break;
+ case BTGreaterEqualStrategyNumber:
+ case BTGreaterStrategyNumber:
+ if (r < 0)
+ res = false;
+ break;
+ default:
+ elog(ERROR, "unrecognized strategy number: %d",
+ in->scankeys[j].sk_strategy);
+ break;
+ }
+
+ if (!res)
+ break; /* no need to consider remaining conditions */
}
if (res)
@@ -496,16 +502,13 @@ spg_text_leaf_consistent(PG_FUNCTION_ARGS)
{
spgLeafConsistentIn *in = (spgLeafConsistentIn *) PG_GETARG_POINTER(0);
spgLeafConsistentOut *out = (spgLeafConsistentOut *) PG_GETARG_POINTER(1);
- StrategyNumber strategy = in->strategy;
- text *query = DatumGetTextPP(in->query);
int level = in->level;
text *leafValue,
*reconstrValue = NULL;
char *fullValue;
int fullLen;
- int queryLen;
- int r;
bool res;
+ int j;
/* all tests are exact */
out->recheck = false;
@@ -518,18 +521,8 @@ spg_text_leaf_consistent(PG_FUNCTION_ARGS)
Assert(level == 0 ? reconstrValue == NULL :
VARSIZE_ANY_EXHDR(reconstrValue) == level);
+ /* Reconstruct the full string represented by this leaf tuple */
fullLen = level + VARSIZE_ANY_EXHDR(leafValue);
-
- queryLen = VARSIZE_ANY_EXHDR(query);
-
- /*
- * For an equality check, we needn't reconstruct fullValue if not same
- * length; it can't match
- */
- if (strategy == BTEqualStrategyNumber && queryLen != fullLen)
- PG_RETURN_BOOL(false);
-
- /* Else, reconstruct the full string represented by this leaf tuple */
if (VARSIZE_ANY_EXHDR(leafValue) == 0 && level > 0)
{
fullValue = VARDATA(reconstrValue);
@@ -549,54 +542,67 @@ spg_text_leaf_consistent(PG_FUNCTION_ARGS)
out->leafValue = PointerGetDatum(fullText);
}
- /* Run the appropriate type of comparison */
- if (strategy > 10)
+ /* Perform the required comparison(s) */
+ res = true;
+ for (j = 0; j < in->nkeys; j++)
{
- /* Collation-aware comparison */
- strategy -= 10;
+ StrategyNumber strategy = in->scankeys[j].sk_strategy;
+ text *query = DatumGetTextPP(in->scankeys[j].sk_argument);
+ int queryLen = VARSIZE_ANY_EXHDR(query);
+ int r;
- /* If asserts are enabled, verify encoding of reconstructed string */
- Assert(pg_verifymbstr(fullValue, fullLen, false));
+ if (strategy > 10)
+ {
+ /* Collation-aware comparison */
+ strategy -= 10;
- r = varstr_cmp(fullValue, Min(queryLen, fullLen),
- VARDATA_ANY(query), Min(queryLen, fullLen),
- PG_GET_COLLATION());
- }
- else
- {
- /* Non-collation-aware comparison */
- r = memcmp(fullValue, VARDATA_ANY(query), Min(queryLen, fullLen));
- }
+ /* If asserts enabled, verify encoding of reconstructed string */
+ Assert(pg_verifymbstr(fullValue, fullLen, false));
- if (r == 0)
- {
- if (queryLen > fullLen)
- r = -1;
- else if (queryLen < fullLen)
- r = 1;
- }
+ r = varstr_cmp(fullValue, Min(queryLen, fullLen),
+ VARDATA_ANY(query), Min(queryLen, fullLen),
+ PG_GET_COLLATION());
+ }
+ else
+ {
+ /* Non-collation-aware comparison */
+ r = memcmp(fullValue, VARDATA_ANY(query), Min(queryLen, fullLen));
+ }
- switch (strategy)
- {
- case BTLessStrategyNumber:
- res = (r < 0);
- break;
- case BTLessEqualStrategyNumber:
- res = (r <= 0);
- break;
- case BTEqualStrategyNumber:
- res = (r == 0);
- break;
- case BTGreaterEqualStrategyNumber:
- res = (r >= 0);
- break;
- case BTGreaterStrategyNumber:
- res = (r > 0);
- break;
- default:
- elog(ERROR, "unrecognized strategy number: %d", in->strategy);
- res = false;
- break;
+ if (r == 0)
+ {
+ if (queryLen > fullLen)
+ r = -1;
+ else if (queryLen < fullLen)
+ r = 1;
+ }
+
+ switch (strategy)
+ {
+ case BTLessStrategyNumber:
+ res = (r < 0);
+ break;
+ case BTLessEqualStrategyNumber:
+ res = (r <= 0);
+ break;
+ case BTEqualStrategyNumber:
+ res = (r == 0);
+ break;
+ case BTGreaterEqualStrategyNumber:
+ res = (r >= 0);
+ break;
+ case BTGreaterStrategyNumber:
+ res = (r > 0);
+ break;
+ default:
+ elog(ERROR, "unrecognized strategy number: %d",
+ in->scankeys[j].sk_strategy);
+ res = false;
+ break;
+ }
+
+ if (!res)
+ break; /* no need to consider remaining conditions */
}
PG_RETURN_BOOL(res);
diff --git a/src/include/access/spgist.h b/src/include/access/spgist.h
index cd6de2c98da8616860563977385f8159edec53f2..8d0205e691f2d0335c08738b723fd4504ca3eef7 100644
--- a/src/include/access/spgist.h
+++ b/src/include/access/spgist.h
@@ -128,8 +128,8 @@ typedef struct spgPickSplitOut
*/
typedef struct spgInnerConsistentIn
{
- StrategyNumber strategy; /* operator strategy number */
- Datum query; /* operator's RHS value */
+ ScanKey scankeys; /* array of operators and comparison values */
+ int nkeys; /* length of array */
Datum reconstructedValue; /* value reconstructed at parent */
int level; /* current level (counting from zero) */
@@ -156,8 +156,8 @@ typedef struct spgInnerConsistentOut
*/
typedef struct spgLeafConsistentIn
{
- StrategyNumber strategy; /* operator strategy number */
- Datum query; /* operator's RHS value */
+ ScanKey scankeys; /* array of operators and comparison values */
+ int nkeys; /* length of array */
Datum reconstructedValue; /* value reconstructed at parent */
int level; /* current level (counting from zero) */
diff --git a/src/include/access/spgist_private.h b/src/include/access/spgist_private.h
index fa23acf6cdeaff70b9ffe2e5ec8ba5586c410fa9..76ea5a1578fc8c5e44b8194be8c33f120f365a7d 100644
--- a/src/include/access/spgist_private.h
+++ b/src/include/access/spgist_private.h
@@ -126,7 +126,11 @@ typedef struct SpGistScanOpaqueData
SpGistState state; /* see above */
MemoryContext tempCxt; /* short-lived memory context */
- /* Index quals for scan (copied from IndexScanDesc for convenience) */
+ /* Control flags showing whether to search nulls and/or non-nulls */
+ bool searchNulls; /* scan matches (all) null entries */
+ bool searchNonNulls; /* scan matches (some) non-null entries */
+
+ /* Index quals to be passed to opclass (null-related quals removed) */
int numberOfKeys; /* number of index qualifier conditions */
ScanKey keyData; /* array of index qualifier descriptors */