Added key ordering fields, scanEntryGetKey()

Author: za-arthur

src/rum.h

@@ -592,6 +592,11 @@ typedef struct RumScanKeyData
 	Datum		outerAddInfo;
 	bool		outerAddInfoIsNull;
 
+	/* Key information, used in orderingFn */
+	Datum		curKey;
+	RumNullCategory curKeyCategory;
+	bool		useCurKey;
+
 	/* other data needed for calling consistentFn */
 	Datum		query;
 	/* NB: these three arrays have only nuserentries elements! */
@@ -641,6 +646,11 @@ typedef struct RumScanEntryData
 	/* current ItemPointer to heap */
 	RumItem		curItem;
 
+	/* Used for ordering using distance */
+	Datum		curKey;
+	RumNullCategory curKeyCategory;
+	bool		useCurKey;
+
 	/*
 	 * For a partial-match or full-scan query, we accumulate all TIDs and
 	 * and additional information here
@@ -1007,5 +1017,11 @@ extern Datum FunctionCall10Coll(FmgrInfo *flinfo, Oid collation,
 				   Datum arg3, Datum arg4, Datum arg5,
 				   Datum arg6, Datum arg7, Datum arg8,
 				   Datum arg9, Datum arg10);
+extern Datum FunctionCall12Coll(FmgrInfo *flinfo, Oid collation,
+				   Datum arg1, Datum arg2,
+				   Datum arg3, Datum arg4, Datum arg5,
+				   Datum arg6, Datum arg7, Datum arg8,
+				   Datum arg9, Datum arg10,
+				   Datum arg11, Datum arg12);
 
 #endif   /* __RUM_H__ */

src/rum_ts_utils.c

@@ -1411,18 +1411,20 @@ rum_tsquery_distance(PG_FUNCTION_ARGS)
 {
 	bool	   *check = (bool *) PG_GETARG_POINTER(0);
 
-	TSQuery		query = PG_GETARG_TSQUERY(2);
-	int			nkeys = PG_GETARG_INT32(3);
-	Pointer	   *extra_data = (Pointer *) PG_GETARG_POINTER(4);
-	Datum	   *addInfo = (Datum *) PG_GETARG_POINTER(8);
-	bool	   *addInfoIsNull = (bool *) PG_GETARG_POINTER(9);
+	/* RumNullCategory keyCategory = PG_GETARG_CHAR(1); */
+	/* Datum	   key = PG_GETARG_DATUM(2); */
+	TSQuery		query = PG_GETARG_TSQUERY(4);
+	int			nkeys = PG_GETARG_INT32(5);
+	Pointer	   *extra_data = (Pointer *) PG_GETARG_POINTER(6);
+	Datum	   *addInfo = (Datum *) PG_GETARG_POINTER(10);
+	bool	   *addInfoIsNull = (bool *) PG_GETARG_POINTER(11);
 	float8		res;
 	int		   *map_item_operand = (int *) (extra_data[0]);
 
 	res = calc_score_addinfo(weights, check, query, map_item_operand,
 							 addInfo, addInfoIsNull, nkeys);
 
-	PG_FREE_IF_COPY(query, 2);
+	PG_FREE_IF_COPY(query, 4);
 	if (res == 0)
 		PG_RETURN_FLOAT8(get_float8_infinity());
 	else

src/rumget.c

@@ -151,6 +151,23 @@ callConsistentFn(RumState * rumstate, RumScanKey key)
 	return res && callAddInfoConsistentFn(rumstate, key);
 }
 
+/*
+ * Extract key value for ordering
+ */
+static void
+scanEntryGetKey(RumScanEntry entry, RumState *rumstate, IndexTuple itup)
+{
+	if (entry->useCurKey == false)
+		return;
+
+	/*
+	 * XXX FIXME only pass-by-value!!! Value should be copied to
+	 * long-lived memory context and, somehow, freeed. Seems, the
+	 * last is real problem.
+	 */
+	entry->curKey = rumtuple_get_key(rumstate, itup, &entry->curKeyCategory);
+}
+
 /*
  * Goes to the next page if current offset is outside of bounds
  */
@@ -641,6 +658,8 @@ startScanEntry(RumState * rumstate, RumScanEntry entry)
 		if (entry->queryCategory == RUM_CAT_EMPTY_QUERY &&
 			entry->scanWithAddInfo)
 			entry->stack = stackEntry;
+
+		scanEntryGetKey(entry, rumstate, itup);
 	}
 
 	if (needUnlock)
@@ -1047,6 +1066,8 @@ entryGetNextItemList(RumState * rumstate, RumScanEntry entry)
 	entry->curItem = entry->list[entry->offset];
 	entry->offset += entry->scanDirection;
 
+	scanEntryGetKey(entry, rumstate, itup);
+
 	/*
 	 * Done with this entry, go to the next for the future.
 	 */
@@ -1207,6 +1228,7 @@ entryGetItem(RumState * rumstate, RumScanEntry entry, bool *nextEntryList)
 			entry->isFinished = TRUE;
 		}
 	}
+	/* Get next item from posting tree */
 	else
 	{
 		do
@@ -2093,9 +2115,11 @@ keyGetOrdering(RumState * rumstate, MemoryContext tempCtx, RumScanKey key,
 		}
 	}
 
-	return DatumGetFloat8(FunctionCall10Coll(&rumstate->orderingFn[key->attnum - 1],
+	return DatumGetFloat8(FunctionCall12Coll(&rumstate->orderingFn[key->attnum - 1],
 								 rumstate->supportCollation[key->attnum - 1],
 											 PointerGetDatum(key->entryRes),
+											 DatumGetChar(key->curKeyCategory),
+											 key->curKey,
 											 UInt16GetDatum(key->strategy),
 											 key->query,
 										   UInt32GetDatum(key->nuserentries),
@@ -2121,10 +2145,12 @@ insertScanItem(RumScanOpaque so, bool recheck)
 	item->iptr = so->item.iptr;
 	item->recheck = recheck;
 
-	if (AttributeNumberIsValid(so->rumstate.attrnAddToColumn))
+	if (AttributeNumberIsValid(so->rumstate.attrnAddToColumn) || so->willSort)
 	{
 		int			nOrderByAnother = 0,
-					count = 0;
+					nOrderByKey = 0,
+					countByAnother = 0,
+					countByKey = 0;
 
 		for (i = 0; i < so->nkeys; i++)
 		{
@@ -2133,11 +2159,15 @@ insertScanItem(RumScanOpaque so, bool recheck)
 				so->keys[i]->outerAddInfoIsNull = true;
 				nOrderByAnother++;
 			}
+			else if (so->keys[i]->useCurKey)
+				nOrderByKey++;
 		}
 
-		for (i = 0; count < nOrderByAnother && i < so->nkeys; i++)
+		for (i = 0; (countByAnother < nOrderByAnother || countByKey < nOrderByKey) &&
+			 i < so->nkeys; i++)
 		{
-			if (so->keys[i]->attnum == so->rumstate.attrnAddToColumn &&
+			if (countByAnother < nOrderByAnother &&
+				so->keys[i]->attnum == so->rumstate.attrnAddToColumn &&
 				so->keys[i]->outerAddInfoIsNull == false)
 			{
 				Assert(!so->keys[i]->orderBy);
@@ -2150,7 +2180,23 @@ insertScanItem(RumScanOpaque so, bool recheck)
 					{
 						so->keys[j]->outerAddInfoIsNull = false;
 						so->keys[j]->outerAddInfo = so->keys[i]->outerAddInfo;
-						count++;
+						countByAnother++;
+					}
+				}
+			}
+			else if (countByKey < nOrderByKey && so->keys[i]->nentries > 0 &&
+					 so->keys[i]->scanEntry[0]->useCurKey)
+			{
+				Assert(!so->keys[i]->orderBy);
+
+				for (j = i + 1; j < so->nkeys; j++)
+				{
+					if (so->keys[j]->useCurKey)
+					{
+						so->keys[j]->curKey = so->keys[i]->scanEntry[0]->curKey;
+						so->keys[j]->curKeyCategory =
+							so->keys[i]->scanEntry[0]->curKeyCategory;
+						countByKey++;
 					}
 				}
 			}
@@ -2222,10 +2268,7 @@ rumgettuple(IndexScanDesc scan, ScanDirection direction)
 		if (so->naturalOrder == NoMovementScanDirection)
 		{
 			so->sortstate = rum_tuplesort_begin_rum(work_mem, so->norderbys,
-						false,
-						so->totalentries > 0 &&
-						so->entries[0]->queryCategory == RUM_CAT_EMPTY_QUERY &&
-						so->entries[0]->scanWithAddInfo);
+						false, so->scanType == RumFullScan);
 
 
 			while (scanGetItem(scan, &so->item, &so->item, &recheck))

src/rumscan.c

@@ -108,6 +108,9 @@ rumFillScanEntry(RumScanOpaque so, OffsetNumber attnum,
 
 	scanEntry->buffer = InvalidBuffer;
 	RumItemSetMin(&scanEntry->curItem);
+	scanEntry->curKey = (Datum) 0;
+	scanEntry->curKeyCategory = RUM_CAT_NULL_KEY;
+	scanEntry->useCurKey = false;
 	scanEntry->matchSortstate = NULL;
 	scanEntry->stack = NULL;
 	scanEntry->scanWithAddInfo = false;
@@ -153,6 +156,7 @@ rumFillScanKey(RumScanOpaque so, OffsetNumber attnum,
 	key->searchMode = searchMode;
 	key->attnum = key->attnumOrig = attnum;
 	key->useAddToColumn = false;
+	key->useCurKey = false;
 	key->scanDirection = ForwardScanDirection;
 
 	RumItemSetMin(&key->curItem);
@@ -165,22 +169,55 @@ rumFillScanKey(RumScanOpaque so, OffsetNumber attnum,
 
 	if (key->orderBy)
 	{
-		if (key->attnum == rumstate->attrnAttachColumn)
+		/* Add key to order by additional information... */
+		if (key->attnum == rumstate->attrnAttachColumn ||
+			/* ...add key to order by index key value */
+			(strategy == RUM_DISTANCE && rumstate->canOrdering[attnum - 1]))
 		{
 			if (nQueryValues != 1)
 				elog(ERROR, "extractQuery should return only one value for ordering");
-			if (rumstate->canOuterOrdering[attnum - 1] == false)
-				elog(ERROR, "doesn't support ordering as additional info");
 			if (rumstate->origTupdesc->attrs[rumstate->attrnAttachColumn - 1]->attbyval == false)
 				elog(ERROR, "doesn't support order by over pass-by-reference column");
 
-			key->useAddToColumn = true;
-			key->attnum = rumstate->attrnAddToColumn;
+			if (key->attnum == rumstate->attrnAttachColumn)
+			{
+				if (rumstate->canOuterOrdering[attnum - 1] == false)
+					elog(ERROR, "doesn't support ordering as additional info");
+
+				key->useAddToColumn = true;
+				key->outerAddInfoIsNull = true;
+				key->attnum = rumstate->attrnAddToColumn;
+			}
+			else if (strategy == RUM_DISTANCE && rumstate->canOrdering[attnum - 1])
+			{
+				RumScanKey	scanKey = NULL;
+
+				for (i = 0; i < so->nkeys; i++)
+				{
+					if (so->keys[i]->orderBy == false &&
+						so->keys[i]->attnum == key->attnum)
+					{
+						scanKey = so->keys[i];
+						break;
+					}
+				}
+
+				elog(INFO, "test");
+				if (scanKey == NULL)
+					elog(ERROR, "cannot order without attribute %d in WHERE clause",
+						 key->attnum);
+				else if (scanKey->nentries > 1)
+					elog(ERROR, "scan key should contain only one value");
+				else if (scanKey->nentries == 0)	/* Should not happen */
+					elog(ERROR, "scan key should contain key value");
+
+				key->useCurKey = true;
+				scanKey->scanEntry[0]->useCurKey = true;
+			}
+
 			key->nentries = 0;
 			key->nuserentries = 0;
 
-			key->outerAddInfoIsNull = true;
-
 			key->scanEntry = NULL;
 			key->entryRes = NULL;
 			key->addInfo = NULL;

src/rumutil.c

@@ -933,3 +933,50 @@ FunctionCall10Coll(FmgrInfo *flinfo, Oid collation, Datum arg1, Datum arg2,
 
 	return result;
 }
+
+Datum
+FunctionCall12Coll(FmgrInfo *flinfo, Oid collation,
+				   Datum arg1, Datum arg2,
+				   Datum arg3, Datum arg4, Datum arg5,
+				   Datum arg6, Datum arg7, Datum arg8,
+				   Datum arg9, Datum arg10,
+				   Datum arg11, Datum arg12)
+{
+	FunctionCallInfoData fcinfo;
+	Datum		result;
+
+	InitFunctionCallInfoData(fcinfo, flinfo, 12, collation, NULL, NULL);
+
+	fcinfo.arg[0] = arg1;
+	fcinfo.arg[1] = arg2;
+	fcinfo.arg[2] = arg3;
+	fcinfo.arg[3] = arg4;
+	fcinfo.arg[4] = arg5;
+	fcinfo.arg[5] = arg6;
+	fcinfo.arg[6] = arg7;
+	fcinfo.arg[7] = arg8;
+	fcinfo.arg[8] = arg9;
+	fcinfo.arg[9] = arg10;
+	fcinfo.arg[10] = arg11;
+	fcinfo.arg[11] = arg12;
+	fcinfo.argnull[0] = false;
+	fcinfo.argnull[1] = false;
+	fcinfo.argnull[2] = false;
+	fcinfo.argnull[3] = false;
+	fcinfo.argnull[4] = false;
+	fcinfo.argnull[5] = false;
+	fcinfo.argnull[6] = false;
+	fcinfo.argnull[7] = false;
+	fcinfo.argnull[8] = false;
+	fcinfo.argnull[9] = false;
+	fcinfo.argnull[10] = false;
+	fcinfo.argnull[11] = 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;
+}