explain_dr.c

Go to the documentation of this file.

/*-------------------------------------------------------------------------
 *
 * 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 "commands/explain_state.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;
}