Improve handling of NULLs in KNN-GiST and KNN-SP-GiST

This commit improves subject in two ways:

 * It removes ugliness of 02f90879e7, which stores distance values and null
   flags in two separate arrays after GISTSearchItem struct.  Instead we pack
   both distance value and null flag in IndexOrderByDistance struct.  Alignment
   overhead should be negligible, because we typically deal with at most few
   "col op const" expressions in ORDER BY clause.
 * It fixes handling of "col op NULL" expression in KNN-SP-GiST.  Now, these
   expression are not passed to support functions, which can't deal with them.
   Instead, NULL result is implicitly assumed.  It future we may decide to
   teach support functions to deal with NULL arguments, but current solution is
   bugfix suitable for backpatch.

Reported-by: Nikita Glukhov
Discussion: https://postgr.es/m/826f57ee-afc7-8977-c44c-6111d18b02ec%40postgrespro.ru
Author: Nikita Glukhov
Reviewed-by: Alexander Korotkov
Backpatch-through: 9.4

1 files changed, 62 insertions, 12 deletions

diff --git a/src/backend/access/spgist/spgscan.c b/src/backend/access/spgist/spgscan.c
index 2bd4037af9e..cfba47067cc 100644
--- a/src/backend/access/spgist/spgscan.c
+++ b/src/backend/access/spgist/spgscan.c
@@ -107,13 +107,13 @@ spgAllocSearchItem(SpGistScanOpaque so, bool isnull, double *distances)
 {
 	/* allocate distance array only for non-NULL items */
 	SpGistSearchItem *item =
-	palloc(SizeOfSpGistSearchItem(isnull ? 0 : so->numberOfOrderBys));
+	palloc(SizeOfSpGistSearchItem(isnull ? 0 : so->numberOfNonNullOrderBys));
 
 	item->isNull = isnull;
 
-	if (!isnull && so->numberOfOrderBys > 0)
+	if (!isnull && so->numberOfNonNullOrderBys > 0)
 		memcpy(item->distances, distances,
-			   so->numberOfOrderBys * sizeof(double));
+			   sizeof(item->distances[0]) * so->numberOfNonNullOrderBys);
 
 	return item;
 }
@@ -208,6 +208,34 @@ spgPrepareScanKeys(IndexScanDesc scan)
 	so->numberOfOrderBys = scan->numberOfOrderBys;
 	so->orderByData = scan->orderByData;
 
+	if (so->numberOfOrderBys <= 0)
+		so->numberOfNonNullOrderBys = 0;
+	else
+	{
+		int			j = 0;
+
+		/*
+		 * Remove all NULL keys, but remember their offsets in the original
+		 * array.
+		 */
+		for (i = 0; i < scan->numberOfOrderBys; i++)
+		{
+			ScanKey		skey = &so->orderByData[i];
+
+			if (skey->sk_flags & SK_ISNULL)
+				so->nonNullOrderByOffsets[i] = -1;
+			else
+			{
+				if (i != j)
+					so->orderByData[j] = *skey;
+
+				so->nonNullOrderByOffsets[i] = j++;
+			}
+		}
+
+		so->numberOfNonNullOrderBys = j;
+	}
+
 	if (scan->numberOfKeys <= 0)
 	{
 		/* If no quals, whole-index scan is required */
@@ -295,6 +323,8 @@ spgbeginscan(Relation rel, int keysz, int orderbysz)
 		/* This will be filled in spgrescan, but allocate the space here */
 		so->orderByTypes = (Oid *)
 			palloc(sizeof(Oid) * scan->numberOfOrderBys);
+		so->nonNullOrderByOffsets = (int *)
+			palloc(sizeof(int) * scan->numberOfOrderBys);
 
 		/* These arrays have constant contents, so we can fill them now */
 		so->zeroDistances = (double *)
@@ -394,6 +424,7 @@ spgendscan(IndexScanDesc scan)
 	if (scan->numberOfOrderBys > 0)
 	{
 		pfree(so->orderByTypes);
+		pfree(so->nonNullOrderByOffsets);
 		pfree(so->zeroDistances);
 		pfree(so->infDistances);
 		pfree(scan->xs_orderbyvals);
@@ -465,7 +496,7 @@ spgLeafTest(SpGistScanOpaque so, SpGistSearchItem *item,
 		in.scankeys = so->keyData;
 		in.nkeys = so->numberOfKeys;
 		in.orderbys = so->orderByData;
-		in.norderbys = so->numberOfOrderBys;
+		in.norderbys = so->numberOfNonNullOrderBys;
 		in.reconstructedValue = item->value;
 		in.traversalValue = item->traversalValue;
 		in.level = item->level;
@@ -492,7 +523,7 @@ spgLeafTest(SpGistScanOpaque so, SpGistSearchItem *item,
 	if (result)
 	{
 		/* item passes the scankeys */
-		if (so->numberOfOrderBys > 0)
+		if (so->numberOfNonNullOrderBys > 0)
 		{
 			/* the scan is ordered -> add the item to the queue */
 			MemoryContext oldCxt = MemoryContextSwitchTo(so->traversalCxt);
@@ -531,7 +562,7 @@ spgInitInnerConsistentIn(spgInnerConsistentIn *in,
 	in->scankeys = so->keyData;
 	in->orderbys = so->orderByData;
 	in->nkeys = so->numberOfKeys;
-	in->norderbys = so->numberOfOrderBys;
+	in->norderbys = so->numberOfNonNullOrderBys;
 	in->reconstructedValue = item->value;
 	in->traversalMemoryContext = so->traversalCxt;
 	in->traversalValue = item->traversalValue;
@@ -751,7 +782,7 @@ redirect:
 		if (item->isLeaf)
 		{
 			/* We store heap items in the queue only in case of ordered search */
-			Assert(so->numberOfOrderBys > 0);
+			Assert(so->numberOfNonNullOrderBys > 0);
 			storeRes(so, &item->heapPtr, item->value, item->isNull,
 					 item->recheck, item->recheckDistances, item->distances);
 			reportedSome = true;
@@ -874,7 +905,7 @@ spggetbitmap(IndexScanDesc scan, TIDBitmap *tbm)
 static void
 storeGettuple(SpGistScanOpaque so, ItemPointer heapPtr,
 			  Datum leafValue, bool isnull, bool recheck, bool recheckDistances,
-			  double *distances)
+			  double *nonNullDistances)
 {
 	Assert(so->nPtrs < MaxIndexTuplesPerPage);
 	so->heapPtrs[so->nPtrs] = *heapPtr;
@@ -883,13 +914,33 @@ storeGettuple(SpGistScanOpaque so, ItemPointer heapPtr,
 
 	if (so->numberOfOrderBys > 0)
 	{
-		if (isnull)
+		if (isnull || so->numberOfNonNullOrderBys <= 0)
 			so->distances[so->nPtrs] = NULL;
 		else
 		{
-			Size		size = sizeof(double) * so->numberOfOrderBys;
+			IndexOrderByDistance *distances =
+			palloc(sizeof(distances[0]) * so->numberOfOrderBys);
+			int			i;
+
+			for (i = 0; i < so->numberOfOrderBys; i++)
+			{
+				int			offset = so->nonNullOrderByOffsets[i];
+
+				if (offset >= 0)
+				{
+					/* Copy non-NULL distance value */
+					distances[i].value = nonNullDistances[offset];
+					distances[i].isnull = false;
+				}
+				else
+				{
+					/* Set distance's NULL flag. */
+					distances[i].value = 0.0;
+					distances[i].isnull = true;
+				}
+			}
 
-			so->distances[so->nPtrs] = memcpy(palloc(size), distances, size);
+			so->distances[so->nPtrs] = distances;
 		}
 	}
 
@@ -929,7 +980,6 @@ spggettuple(IndexScanDesc scan, ScanDirection dir)
 			if (so->numberOfOrderBys > 0)
 				index_store_float8_orderby_distances(scan, so->orderByTypes,
 													 so->distances[so->iPtr],
-													 NULL,
 													 so->recheckDistances[so->iPtr]);
 			so->iPtr++;
 			return true;