Function DrainPersistedFileCursor

pg_documentdb/src/commands/cursors.c:1670–1715

bytea *
DrainPersistedFileCursor(const char *cursorName, int batchSize,
						 int32_t *numIterations, uint32_t accumulatedSize,
						 pgbson_array_writer *arrayWriter, bytea *cursorFileState)
{
	if (!UseFileBasedPersistedCursors)
	{
		ereport(ERROR, (errcode(ERRCODE_DOCUMENTDB_INTERNALERROR),
						errmsg("File based cursor is not enabled")));
	}

	CursorFileState *cursorState = DeserializeFileState(cursorFileState);

	bool closeCursor = true;
	bool isSingleResult = false;
	PersistentTupleDestReceiver *destReceiver = CreatePersistentTupleDestReceiver(
		arrayWriter,
		CurrentMemoryContext,
		batchSize, cursorName,
		accumulatedSize,
		closeCursor,
		isSingleResult);
	destReceiver->cursorFileState = cursorState;

	pgbson *nextDocument = ReadFromCursorFile(cursorState);
	while (nextDocument != NULL)
	{
		if (!PersistentDestReceiveCore(nextDocument, destReceiver))
		{
			/* Batch size limit reached */
			break;
		}

		pfree(nextDocument);
		nextDocument = ReadFromCursorFile(cursorState);
	}

	PersistentDestReceiverShutdown((DestReceiver *) destReceiver);

	if (nextDocument == NULL)
	{
		return NULL;
	}

	return destReceiver->continuationState;
}


/*