Invent SERIALIZE option for EXPLAIN.

EXPLAIN (ANALYZE, SERIALIZE) allows collection of statistics about
the volume of data emitted by a query, as well as the time taken
to convert the data to the on-the-wire format.  Previously there
was no way to investigate this without actually sending the data
to the client, in which case network transmission costs might
swamp what you wanted to see.  In particular this feature allows
investigating the costs of de-TOASTing compressed or out-of-line
data during formatting.

Stepan Rutz and Matthias van de Meent,
reviewed by Tomas Vondra and myself

Discussion: https://postgr.es/m/ca0adb0e-fa4e-c37e-1cd7-91170b18cae1@gmx.de

1 files changed, 409 insertions, 2 deletions

diff --git a/src/backend/commands/explain.c b/src/backend/commands/explain.c
index 926d70afaf8..895d18ebd59 100644
--- a/src/backend/commands/explain.c
+++ b/src/backend/commands/explain.c
@@ -20,6 +20,7 @@
 #include "commands/prepare.h"
 #include "foreign/fdwapi.h"
 #include "jit/jit.h"
+#include "libpq/pqformat.h"
 #include "nodes/extensible.h"
 #include "nodes/makefuncs.h"
 #include "nodes/nodeFuncs.h"
@@ -47,6 +48,14 @@ ExplainOneQuery_hook_type ExplainOneQuery_hook = NULL;
 explain_get_index_name_hook_type explain_get_index_name_hook = NULL;
 
 
+/* Instrumentation data for SERIALIZE option */
+typedef struct SerializeMetrics
+{
+	uint64		bytesSent;		/* # of bytes serialized */
+	instr_time	timeSpent;		/* time spent serializing */
+	BufferUsage bufferUsage;	/* buffers accessed during serialization */
+} SerializeMetrics;
+
 /* OR-able flags for ExplainXMLTag() */
 #define X_OPENING 0
 #define X_CLOSING 1
@@ -59,6 +68,8 @@ static void ExplainOneQuery(Query *query, int cursorOptions,
 							QueryEnvironment *queryEnv);
 static void ExplainPrintJIT(ExplainState *es, int jit_flags,
 							JitInstrumentation *ji);
+static void ExplainPrintSerialize(ExplainState *es,
+								  SerializeMetrics *metrics);
 static void report_triggers(ResultRelInfo *rInfo, bool show_relname,
 							ExplainState *es);
 static double elapsed_time(instr_time *starttime);
@@ -153,6 +164,7 @@ static void ExplainIndentText(ExplainState *es);
 static void ExplainJSONLineEnding(ExplainState *es);
 static void ExplainYAMLLineStarting(ExplainState *es);
 static void escape_yaml(StringInfo buf, const char *str);
+static SerializeMetrics GetSerializationMetrics(DestReceiver *dest);
 
 
 
@@ -204,6 +216,31 @@ ExplainQuery(ParseState *pstate, ExplainStmt *stmt,
 		}
 		else if (strcmp(opt->defname, "memory") == 0)
 			es->memory = defGetBoolean(opt);
+		else if (strcmp(opt->defname, "serialize") == 0)
+		{
+			if (opt->arg)
+			{
+				char	   *p = defGetString(opt);
+
+				if (strcmp(p, "off") == 0 || strcmp(p, "none") == 0)
+					es->serialize = EXPLAIN_SERIALIZE_NONE;
+				else if (strcmp(p, "text") == 0)
+					es->serialize = EXPLAIN_SERIALIZE_TEXT;
+				else if (strcmp(p, "binary") == 0)
+					es->serialize = EXPLAIN_SERIALIZE_BINARY;
+				else
+					ereport(ERROR,
+							(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+							 errmsg("unrecognized value for EXPLAIN option \"%s\": \"%s\"",
+									opt->defname, p),
+							 parser_errposition(pstate, opt->location)));
+			}
+			else
+			{
+				/* SERIALIZE without an argument is taken as 'text' */
+				es->serialize = EXPLAIN_SERIALIZE_TEXT;
+			}
+		}
 		else if (strcmp(opt->defname, "format") == 0)
 		{
 			char	   *p = defGetString(opt);
@@ -246,6 +283,12 @@ ExplainQuery(ParseState *pstate, ExplainStmt *stmt,
 				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
 				 errmsg("EXPLAIN option TIMING requires ANALYZE")));
 
+	/* check that serialize is used with EXPLAIN ANALYZE */
+	if (es->serialize != EXPLAIN_SERIALIZE_NONE && !es->analyze)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("EXPLAIN option SERIALIZE requires ANALYZE")));
+
 	/* check that GENERIC_PLAN is not used with EXPLAIN ANALYZE */
 	if (es->generic && es->analyze)
 		ereport(ERROR,
@@ -576,6 +619,7 @@ ExplainOnePlan(PlannedStmt *plannedstmt, IntoClause *into, ExplainState *es,
 	double		totaltime = 0;
 	int			eflags;
 	int			instrument_option = 0;
+	SerializeMetrics serializeMetrics = {0};
 
 	Assert(plannedstmt->commandType != CMD_UTILITY);
 
@@ -604,11 +648,17 @@ ExplainOnePlan(PlannedStmt *plannedstmt, IntoClause *into, ExplainState *es,
 	UpdateActiveSnapshotCommandId();
 
 	/*
-	 * Normally we discard the query's output, but if explaining CREATE TABLE
-	 * AS, we'd better use the appropriate tuple receiver.
+	 * We discard the output if we have no use for it.  If we're explaining
+	 * CREATE TABLE AS, we'd better use the appropriate tuple receiver, while
+	 * the SERIALIZE option requires its own tuple receiver.  (If you specify
+	 * SERIALIZE while explaining CREATE TABLE AS, you'll see zeroes for the
+	 * results, which is appropriate since no data would have gone to the
+	 * client.)
 	 */
 	if (into)
 		dest = CreateIntoRelDestReceiver(into);
+	else if (es->serialize != EXPLAIN_SERIALIZE_NONE)
+		dest = CreateExplainSerializeDestReceiver(es);
 	else
 		dest = None_Receiver;
 
@@ -651,6 +701,13 @@ ExplainOnePlan(PlannedStmt *plannedstmt, IntoClause *into, ExplainState *es,
 		totaltime += elapsed_time(&starttime);
 	}
 
+	/* grab serialization metrics before we destroy the DestReceiver */
+	if (es->serialize != EXPLAIN_SERIALIZE_NONE)
+		serializeMetrics = GetSerializationMetrics(dest);
+
+	/* call the DestReceiver's destroy method even during explain */
+	dest->rDestroy(dest);
+
 	ExplainOpenGroup("Query", NULL, true, es);
 
 	/* Create textual dump of plan tree */
@@ -700,6 +757,10 @@ ExplainOnePlan(PlannedStmt *plannedstmt, IntoClause *into, ExplainState *es,
 	if (es->costs)
 		ExplainPrintJITSummary(es, queryDesc);
 
+	/* Print info about serialization of output */
+	if (es->serialize != EXPLAIN_SERIALIZE_NONE)
+		ExplainPrintSerialize(es, &serializeMetrics);
+
 	/*
 	 * Close down the query and free resources.  Include time for this in the
 	 * total execution time (although it should be pretty minimal).
@@ -1034,6 +1095,62 @@ ExplainPrintJIT(ExplainState *es, int jit_flags, JitInstrumentation *ji)
 }
 
 /*
+ * ExplainPrintSerialize -
+ *	  Append information about query output volume to es->str.
+ */
+static void
+ExplainPrintSerialize(ExplainState *es, SerializeMetrics *metrics)
+{
+	const char *format;
+
+	/* We shouldn't get called for EXPLAIN_SERIALIZE_NONE */
+	if (es->serialize == EXPLAIN_SERIALIZE_TEXT)
+		format = "text";
+	else
+	{
+		Assert(es->serialize == EXPLAIN_SERIALIZE_BINARY);
+		format = "binary";
+	}
+
+	ExplainOpenGroup("Serialization", "Serialization", true, es);
+
+	if (es->format == EXPLAIN_FORMAT_TEXT)
+	{
+		ExplainIndentText(es);
+		if (es->timing)
+			appendStringInfo(es->str, "Serialization: time=%.3f ms  output=" INT64_FORMAT "kB  format=%s\n",
+							 1000.0 * INSTR_TIME_GET_DOUBLE(metrics->timeSpent),
+							 (metrics->bytesSent + 512) / 1024,
+							 format);
+		else
+			appendStringInfo(es->str, "Serialization: output=" INT64_FORMAT "kB  format=%s\n",
+							 (metrics->bytesSent + 512) / 1024,
+							 format);
+
+		if (es->buffers && peek_buffer_usage(es, &metrics->bufferUsage))
+		{
+			es->indent++;
+			show_buffer_usage(es, &metrics->bufferUsage);
+			es->indent--;
+		}
+	}
+	else
+	{
+		if (es->timing)
+			ExplainPropertyFloat("Time", "ms",
+								 1000.0 * INSTR_TIME_GET_DOUBLE(metrics->timeSpent),
+								 3, es);
+		ExplainPropertyUInteger("Output Volume", "kB",
+								(metrics->bytesSent + 512) / 1024, es);
+		ExplainPropertyText("Format", format, es);
+		if (es->buffers)
+			show_buffer_usage(es, &metrics->bufferUsage);
+	}
+
+	ExplainCloseGroup("Serialization", "Serialization", true, es);
+}
+
+/*
  * ExplainQueryText -
  *	  add a "Query Text" node that contains the actual text of the query
  *
@@ -5161,3 +5278,293 @@ escape_yaml(StringInfo buf, const char *str)
 {
 	escape_json(buf, str);
 }
+
+
+/*
+ * 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, 'D');
+
+	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.
+ */
+static 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;
+}