postgres/src/backend/commands/explain_dr.c

/*-------------------------------------------------------------------------
 *
 * explain_dr.c
 *	  Explain DestReceiver to measure serialization overhead
 *
 * Portions Copyright (c) 1996-2025, PostgreSQL Global Development Group
 * Portions Copyright (c) 1994-5, Regents of the University of California
 *
 * IDENTIFICATION
 *	  src/backend/commands/explain.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include "commands/explain.h"
#include "commands/explain_dr.h"
#include "libpq/pqformat.h"
#include "libpq/protocol.h"
#include "utils/lsyscache.h"

/*
 * DestReceiver functions for SERIALIZE option
 *
 * A DestReceiver for query tuples, that serializes passed rows into RowData
 * messages while measuring the resources expended and total serialized size,
 * while never sending the data to the client.  This allows measuring the
 * overhead of deTOASTing and datatype out/sendfuncs, which are not otherwise
 * exercisable without actually hitting the network.
 */
typedef struct SerializeDestReceiver
{
	DestReceiver pub;
	ExplainState *es;			/* this EXPLAIN statement's ExplainState */
	int8		format;			/* text or binary, like pq wire protocol */
	TupleDesc	attrinfo;		/* the output tuple desc */
	int			nattrs;			/* current number of columns */
	FmgrInfo   *finfos;			/* precomputed call info for output fns */
	MemoryContext tmpcontext;	/* per-row temporary memory context */
	StringInfoData buf;			/* buffer to hold the constructed message */
	SerializeMetrics metrics;	/* collected metrics */
} SerializeDestReceiver;

/*
 * Get the function lookup info that we'll need for output.
 *
 * This is a subset of what printtup_prepare_info() does.  We don't need to
 * cope with format choices varying across columns, so it's slightly simpler.
 */
static void
serialize_prepare_info(SerializeDestReceiver *receiver,
					   TupleDesc typeinfo, int nattrs)
{
	/* get rid of any old data */
	if (receiver->finfos)
		pfree(receiver->finfos);
	receiver->finfos = NULL;

	receiver->attrinfo = typeinfo;
	receiver->nattrs = nattrs;
	if (nattrs <= 0)
		return;

	receiver->finfos = (FmgrInfo *) palloc0(nattrs * sizeof(FmgrInfo));

	for (int i = 0; i < nattrs; i++)
	{
		FmgrInfo   *finfo = receiver->finfos + i;
		Form_pg_attribute attr = TupleDescAttr(typeinfo, i);
		Oid			typoutput;
		Oid			typsend;
		bool		typisvarlena;

		if (receiver->format == 0)
		{
			/* wire protocol format text */
			getTypeOutputInfo(attr->atttypid,
							  &typoutput,
							  &typisvarlena);
			fmgr_info(typoutput, finfo);
		}
		else if (receiver->format == 1)
		{
			/* wire protocol format binary */
			getTypeBinaryOutputInfo(attr->atttypid,
									&typsend,
									&typisvarlena);
			fmgr_info(typsend, finfo);
		}
		else
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("unsupported format code: %d", receiver->format)));
	}
}

/*
 * serializeAnalyzeReceive - collect tuples for EXPLAIN (SERIALIZE)
 *
 * This should match printtup() in printtup.c as closely as possible,
 * except for the addition of measurement code.
 */
static bool
serializeAnalyzeReceive(TupleTableSlot *slot, DestReceiver *self)
{
	TupleDesc	typeinfo = slot->tts_tupleDescriptor;
	SerializeDestReceiver *myState = (SerializeDestReceiver *) self;
	MemoryContext oldcontext;
	StringInfo	buf = &myState->buf;
	int			natts = typeinfo->natts;
	instr_time	start,
				end;
	BufferUsage instr_start;

	/* only measure time, buffers if requested */
	if (myState->es->timing)
		INSTR_TIME_SET_CURRENT(start);
	if (myState->es->buffers)
		instr_start = pgBufferUsage;

	/* Set or update my derived attribute info, if needed */
	if (myState->attrinfo != typeinfo || myState->nattrs != natts)
		serialize_prepare_info(myState, typeinfo, natts);

	/* Make sure the tuple is fully deconstructed */
	slot_getallattrs(slot);

	/* Switch into per-row context so we can recover memory below */
	oldcontext = MemoryContextSwitchTo(myState->tmpcontext);

	/*
	 * Prepare a DataRow message (note buffer is in per-query context)
	 *
	 * Note that we fill a StringInfo buffer the same as printtup() does, so
	 * as to capture the costs of manipulating the strings accurately.
	 */
	pq_beginmessage_reuse(buf, PqMsg_DataRow);

	pq_sendint16(buf, natts);

	/*
	 * send the attributes of this tuple
	 */
	for (int i = 0; i < natts; i++)
	{
		FmgrInfo   *finfo = myState->finfos + i;
		Datum		attr = slot->tts_values[i];

		if (slot->tts_isnull[i])
		{
			pq_sendint32(buf, -1);
			continue;
		}

		if (myState->format == 0)
		{
			/* Text output */
			char	   *outputstr;

			outputstr = OutputFunctionCall(finfo, attr);
			pq_sendcountedtext(buf, outputstr, strlen(outputstr));
		}
		else
		{
			/* Binary output */
			bytea	   *outputbytes;

			outputbytes = SendFunctionCall(finfo, attr);
			pq_sendint32(buf, VARSIZE(outputbytes) - VARHDRSZ);
			pq_sendbytes(buf, VARDATA(outputbytes),
						 VARSIZE(outputbytes) - VARHDRSZ);
		}
	}

	/*
	 * We mustn't call pq_endmessage_reuse(), since that would actually send
	 * the data to the client.  Just count the data, instead.  We can leave
	 * the buffer alone; it'll be reset on the next iteration (as would also
	 * happen in printtup()).
	 */
	myState->metrics.bytesSent += buf->len;

	/* Return to caller's context, and flush row's temporary memory */
	MemoryContextSwitchTo(oldcontext);
	MemoryContextReset(myState->tmpcontext);

	/* Update timing data */
	if (myState->es->timing)
	{
		INSTR_TIME_SET_CURRENT(end);
		INSTR_TIME_ACCUM_DIFF(myState->metrics.timeSpent, end, start);
	}

	/* Update buffer metrics */
	if (myState->es->buffers)
		BufferUsageAccumDiff(&myState->metrics.bufferUsage,
							 &pgBufferUsage,
							 &instr_start);

	return true;
}

/*
 * serializeAnalyzeStartup - start up the serializeAnalyze receiver
 */
static void
serializeAnalyzeStartup(DestReceiver *self, int operation, TupleDesc typeinfo)
{
	SerializeDestReceiver *receiver = (SerializeDestReceiver *) self;

	Assert(receiver->es != NULL);

	switch (receiver->es->serialize)
	{
		case EXPLAIN_SERIALIZE_NONE:
			Assert(false);
			break;
		case EXPLAIN_SERIALIZE_TEXT:
			receiver->format = 0;	/* wire protocol format text */
			break;
		case EXPLAIN_SERIALIZE_BINARY:
			receiver->format = 1;	/* wire protocol format binary */
			break;
	}

	/* Create per-row temporary memory context */
	receiver->tmpcontext = AllocSetContextCreate(CurrentMemoryContext,
												 "SerializeTupleReceive",
												 ALLOCSET_DEFAULT_SIZES);

	/* The output buffer is re-used across rows, as in printtup.c */
	initStringInfo(&receiver->buf);

	/* Initialize results counters */
	memset(&receiver->metrics, 0, sizeof(SerializeMetrics));
	INSTR_TIME_SET_ZERO(receiver->metrics.timeSpent);
}

/*
 * serializeAnalyzeShutdown - shut down the serializeAnalyze receiver
 */
static void
serializeAnalyzeShutdown(DestReceiver *self)
{
	SerializeDestReceiver *receiver = (SerializeDestReceiver *) self;

	if (receiver->finfos)
		pfree(receiver->finfos);
	receiver->finfos = NULL;

	if (receiver->buf.data)
		pfree(receiver->buf.data);
	receiver->buf.data = NULL;

	if (receiver->tmpcontext)
		MemoryContextDelete(receiver->tmpcontext);
	receiver->tmpcontext = NULL;
}

/*
 * serializeAnalyzeDestroy - destroy the serializeAnalyze receiver
 */
static void
serializeAnalyzeDestroy(DestReceiver *self)
{
	pfree(self);
}

/*
 * Build a DestReceiver for EXPLAIN (SERIALIZE) instrumentation.
 */
DestReceiver *
CreateExplainSerializeDestReceiver(ExplainState *es)
{
	SerializeDestReceiver *self;

	self = (SerializeDestReceiver *) palloc0(sizeof(SerializeDestReceiver));

	self->pub.receiveSlot = serializeAnalyzeReceive;
	self->pub.rStartup = serializeAnalyzeStartup;
	self->pub.rShutdown = serializeAnalyzeShutdown;
	self->pub.rDestroy = serializeAnalyzeDestroy;
	self->pub.mydest = DestExplainSerialize;

	self->es = es;

	return (DestReceiver *) self;
}

/*
 * GetSerializationMetrics - collect metrics
 *
 * We have to be careful here since the receiver could be an IntoRel
 * receiver if the subject statement is CREATE TABLE AS.  In that
 * case, return all-zeroes stats.
 */
SerializeMetrics
GetSerializationMetrics(DestReceiver *dest)
{
	SerializeMetrics empty;

	if (dest->mydest == DestExplainSerialize)
		return ((SerializeDestReceiver *) dest)->metrics;

	memset(&empty, 0, sizeof(SerializeMetrics));
	INSTR_TIME_SET_ZERO(empty.timeSpent);

	return empty;
}