/*-------------------------------------------------------------------------

 *

 * explain_dr.c

 *    Explain DestReceiver to measure serialization overhead

 *

 * Portions Copyright (c) 1996-2026, PostgreSQL Global Development Group

 * Portions Copyright (c) 1994-5, Regents of the University of California

 *

 * IDENTIFICATION

 *    src/backend/commands/explain_dr.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"

#include "varatt.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 = palloc0_array(FmgrInfo, nattrs);

  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 = palloc0_object(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;

}