Function PostProcessCursorPage

pg_documentdb/src/commands/cursors.c:2515–2590

* Writes the end of the cursor page. Write the end array, overwrite the cursorId * with the actual one (if it's drained, replace it with 0). * Also creates the result tuple that's (document, continuation) and returns it. */

 * Also creates the result tuple that's (document, continuation) and returns it.
 */
Datum
PostProcessCursorPage(pgbson_writer *cursorDoc, pgbson_array_writer *arrayWriter,
					  pgbson_writer *topLevelWriter, int64_t cursorId,
					  pgbson *continuation, bool persistConnection,
					  pgbson *lastContinuationToken,
					  TupleDesc cursorResultTupleDesc)
{
	/* Finish the cursor doc*/
	PgbsonWriterEndArray(cursorDoc, arrayWriter);

	/*
	 * For tailable cursors, append the last continuation token to the cursor.
	 */
	if (lastContinuationToken != NULL)
	{
		AppendLastContinuationTokenToCursor(cursorDoc, lastContinuationToken);
	}

	PgbsonWriterEndDocument(topLevelWriter, cursorDoc);
	PgbsonWriterAppendDouble(topLevelWriter, "ok", 2, 1);
	if (lastContinuationToken != NULL)
	{
		/*
		 * TODO:  Currently, the operationTime field is applicable only for change
		 * stream cursors. In future, if there is another tailable cursor type is
		 * supported, then update the condition above to check specific cursor type.
		 */
		TimestampTz currentTime = GetCurrentTimestamp();
		PgbsonWriterAppendTimestamp(topLevelWriter, "operationTime", 13, currentTime);
	}

	bool queryFullyDrained = continuation == NULL;

	/* If this is a oneshot query (singlePage) mark it as drained. */
	if (cursorId == 0)
	{
		queryFullyDrained = true;
	}
	else if (queryFullyDrained)
	{
		/* Write out the cursor_id given that the cursor is not drained */
		cursorId = 0;
	}

	if (cursorId != 0)
	{
		bson_iter_t cursorDocIter;
		PgbsonWriterGetIterator(topLevelWriter, &cursorDocIter);
		if (!bson_iter_find_descendant(&cursorDocIter, "cursor.id", &cursorDocIter))
		{
			ereport(ERROR, (errmsg(
								"Could not find cursor.id in cursor document. This is a bug")));
		}

		bson_iter_overwrite_int64(&cursorDocIter, cursorId);
	}

	/* Returns (continuation bson, cursorPage bson) */