Function StreamingDestReceiverReceive

pg_documentdb/src/commands/cursors.c:1011–1076

static bool
StreamingDestReceiverReceive(TupleTableSlot *slot,
							 DestReceiver *destReceiver)
{
	StreamingTupleDestReceiver *receiver =
		(StreamingTupleDestReceiver *) destReceiver;
	BsonStoreTupleDestReceiverBase *base = &receiver->base;

	bool isNull = false;
	Datum result = slot_getattr(slot, 1, &isNull);
	if (isNull)
	{
		/*
		 * For streaming receivers: NULL data rows are skipped.
		 * Return true (continue) so the executor keeps sending rows.
		 */
		return true;
	}

	pgbson *documentValue = DatumGetPgBsonPacked(result);
	uint32_t datumSize = VARSIZE_ANY_EXHDR(documentValue);

	if (datumSize > BSON_MAX_ALLOWED_SIZE)
	{
		ereport(ERROR, (errcode(ERRCODE_DOCUMENTDB_BSONOBJECTTOOLARGE),
						errmsg("Size %u is larger than MaxDocumentSize %u",
							   datumSize, BSON_MAX_ALLOWED_SIZE)));
	}

	receiver->streamingAccumulatedSize += datumSize;

	int64_t totalSize = base->currentAccumulatedSize + datumSize +
						PER_DOC_OVERHEAD;

	bool sizeLimitReached = (totalSize >= BSON_MAX_ALLOWED_SIZE &&
							 base->numRowsFetched > 0);
	if (sizeLimitReached ||
		base->numRowsFetched >= (uint32_t) base->batchSize)
	{
		receiver->terminationReason = sizeLimitReached ?
									  TerminationReason_BatchSizeLimit :
									  TerminationReason_BatchItemLimit;
		return false;
	}

	MemoryContext oldContext = MemoryContextSwitchTo(base->writerContext);
	PgbsonArrayWriterWriteDocument(base->writer, documentValue);
	MemoryContextSwitchTo(oldContext);

	base->numRowsFetched++;
	base->currentAccumulatedSize += (datumSize + PER_DOC_OVERHEAD);

	/* Process continuation token (column 2) and update the cursor map. */
	bool isContinuationNull = false;
	Datum continuationDatum = slot_getattr(slot, 2, &isContinuationNull);
	if (!isContinuationNull)
	{
		pgbson *continuation = DatumGetPgBsonPacked(continuationDatum);