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

 *

 * tstoreReceiver.c

 *    An implementation of DestReceiver that stores the result tuples in

 *    a Tuplestore.

 *

 * Optionally, we can force detoasting (but not decompression) of out-of-line

 * toasted values.  This is to support cursors WITH HOLD, which must retain

 * data even if the underlying table is dropped.

 *

 * Also optionally, we can apply a tuple conversion map before storing.

 *

 *

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

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

 *

 * IDENTIFICATION

 *    src/backend/executor/tstoreReceiver.c

 *

 *-------------------------------------------------------------------------

 */

#include "postgres.h"

#include "access/detoast.h"

#include "access/tupconvert.h"

#include "executor/tstoreReceiver.h"

typedef struct

{

  DestReceiver pub;

  /* parameters: */

  Tuplestorestate *tstore;  /* where to put the data */

  MemoryContext cxt;      /* context containing tstore */

  bool    detoast;    /* were we told to detoast? */

  TupleDesc target_tupdesc; /* target tupdesc, or NULL if none */

  const char *map_failure_msg;  /* tupdesc mapping failure message */

  /* workspace: */

  Datum    *outvalues;    /* values array for result tuple */

  Datum    *tofree;     /* temp values to be pfree'd */

  TupleConversionMap *tupmap; /* conversion map, if needed */

  TupleTableSlot *mapslot;  /* slot for mapped tuples */

} TStoreState;

static bool tstoreReceiveSlot_notoast(TupleTableSlot *slot, DestReceiver *self);

static bool tstoreReceiveSlot_detoast(TupleTableSlot *slot, DestReceiver *self);

static bool tstoreReceiveSlot_tupmap(TupleTableSlot *slot, DestReceiver *self);

/*

 * Prepare to receive tuples from executor.

 */

static void

tstoreStartupReceiver(DestReceiver *self, int operation, TupleDesc typeinfo)

{

  TStoreState *myState = (TStoreState *) self;

  bool    needtoast = false;

  int     natts = typeinfo->natts;

  int     i;

  /* Check if any columns require detoast work */

  if (myState->detoast)

  {

    for (i = 0; i < natts; i++)

    {

      CompactAttribute *attr = TupleDescCompactAttr(typeinfo, i);

      if (attr->attisdropped)

        continue;

      if (attr->attlen == -1)

      {

        needtoast = true;

        break;

      }

    }

  }

  /* Check if tuple conversion is needed */

  if (myState->target_tupdesc)

    myState->tupmap = convert_tuples_by_position(typeinfo,

                           myState->target_tupdesc,

                           myState->map_failure_msg);

  else

    myState->tupmap = NULL;

  /* Set up appropriate callback */

  if (needtoast)

  {

    Assert(!myState->tupmap);

    myState->pub.receiveSlot = tstoreReceiveSlot_detoast;

    /* Create workspace */

    myState->outvalues = (Datum *)

      MemoryContextAlloc(myState->cxt, natts * sizeof(Datum));

    myState->tofree = (Datum *)

      MemoryContextAlloc(myState->cxt, natts * sizeof(Datum));

    myState->mapslot = NULL;

  }

  else if (myState->tupmap)

  {

    myState->pub.receiveSlot = tstoreReceiveSlot_tupmap;

    myState->outvalues = NULL;

    myState->tofree = NULL;

    myState->mapslot = MakeSingleTupleTableSlot(myState->target_tupdesc,

                          &TTSOpsVirtual);

  }

  else

  {

    myState->pub.receiveSlot = tstoreReceiveSlot_notoast;

    myState->outvalues = NULL;

    myState->tofree = NULL;

    myState->mapslot = NULL;

  }

}

/*

 * Receive a tuple from the executor and store it in the tuplestore.

 * This is for the easy case where we don't have to detoast nor map anything.

 */

static bool

tstoreReceiveSlot_notoast(TupleTableSlot *slot, DestReceiver *self)

{

  TStoreState *myState = (TStoreState *) self;

  tuplestore_puttupleslot(myState->tstore, slot);

  return true;

}

/*

 * Receive a tuple from the executor and store it in the tuplestore.

 * This is for the case where we have to detoast any toasted values.

 */

static bool

tstoreReceiveSlot_detoast(TupleTableSlot *slot, DestReceiver *self)

{

  TStoreState *myState = (TStoreState *) self;

  TupleDesc typeinfo = slot->tts_tupleDescriptor;

  int     natts = typeinfo->natts;

  int     nfree;

  int     i;

  MemoryContext oldcxt;

  /* Make sure the tuple is fully deconstructed */

  slot_getallattrs(slot);

  /*

   * Fetch back any out-of-line datums.  We build the new datums array in

   * myState->outvalues[] (but we can re-use the slot's isnull array). Also,

   * remember the fetched values to free afterwards.

   */

  nfree = 0;

  for (i = 0; i < natts; i++)

  {

    Datum   val = slot->tts_values[i];

    CompactAttribute *attr = TupleDescCompactAttr(typeinfo, i);

    if (!attr->attisdropped && attr->attlen == -1 && !slot->tts_isnull[i])

    {

      if (VARATT_IS_EXTERNAL(DatumGetPointer(val)))

      {

        val = PointerGetDatum(detoast_external_attr((struct varlena *)

                              DatumGetPointer(val)));

        myState->tofree[nfree++] = val;

      }

    }

    myState->outvalues[i] = val;

  }

  /*

   * Push the modified tuple into the tuplestore.

   */

  oldcxt = MemoryContextSwitchTo(myState->cxt);

  tuplestore_putvalues(myState->tstore, typeinfo,

             myState->outvalues, slot->tts_isnull);

  MemoryContextSwitchTo(oldcxt);

  /* And release any temporary detoasted values */

  for (i = 0; i < nfree; i++)

    pfree(DatumGetPointer(myState->tofree[i]));

  return true;

}

/*

 * Receive a tuple from the executor and store it in the tuplestore.

 * This is for the case where we must apply a tuple conversion map.

 */

static bool

tstoreReceiveSlot_tupmap(TupleTableSlot *slot, DestReceiver *self)

{

  TStoreState *myState = (TStoreState *) self;

  execute_attr_map_slot(myState->tupmap->attrMap, slot, myState->mapslot);

  tuplestore_puttupleslot(myState->tstore, myState->mapslot);

  return true;

}

/*

 * Clean up at end of an executor run

 */

static void

tstoreShutdownReceiver(DestReceiver *self)

{

  TStoreState *myState = (TStoreState *) self;

  /* Release workspace if any */

  if (myState->outvalues)

    pfree(myState->outvalues);

  myState->outvalues = NULL;

  if (myState->tofree)

    pfree(myState->tofree);

  myState->tofree = NULL;

  if (myState->tupmap)

    free_conversion_map(myState->tupmap);

  myState->tupmap = NULL;

  if (myState->mapslot)

    ExecDropSingleTupleTableSlot(myState->mapslot);

  myState->mapslot = NULL;

}

/*

 * Destroy receiver when done with it

 */

static void

tstoreDestroyReceiver(DestReceiver *self)

{

  pfree(self);

}

/*

 * Initially create a DestReceiver object.

 */

DestReceiver *

CreateTuplestoreDestReceiver(void)

{

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

  self->pub.receiveSlot = tstoreReceiveSlot_notoast;  /* might change */

  self->pub.rStartup = tstoreStartupReceiver;

  self->pub.rShutdown = tstoreShutdownReceiver;

  self->pub.rDestroy = tstoreDestroyReceiver;

  self->pub.mydest = DestTuplestore;

  /* private fields will be set by SetTuplestoreDestReceiverParams */

  return (DestReceiver *) self;

}

/*

 * Set parameters for a TuplestoreDestReceiver

 *

 * tStore: where to store the tuples

 * tContext: memory context containing tStore

 * detoast: forcibly detoast contained data?

 * target_tupdesc: if not NULL, forcibly convert tuples to this rowtype

 * map_failure_msg: error message to use if mapping to target_tupdesc fails

 *

 * We don't currently support both detoast and target_tupdesc at the same

 * time, just because no existing caller needs that combination.

 */

void

SetTuplestoreDestReceiverParams(DestReceiver *self,

                Tuplestorestate *tStore,

                MemoryContext tContext,

                bool detoast,

                TupleDesc target_tupdesc,

                const char *map_failure_msg)

{

  TStoreState *myState = (TStoreState *) self;

  Assert(!(detoast && target_tupdesc));

  Assert(myState->pub.mydest == DestTuplestore);

  myState->tstore = tStore;

  myState->cxt = tContext;

  myState->detoast = detoast;

  myState->target_tupdesc = target_tupdesc;

  myState->map_failure_msg = map_failure_msg;

}