/*-------------------------------------------------------------------------
 *
 * executor_util_tuples.c
 *
 * Utility functions for handling tuples during remote execution.
 *
 *-------------------------------------------------------------------------
 */

#include "postgres.h"

#include "funcapi.h"
#include "miscadmin.h"

#include "utils/lsyscache.h"

#include "distributed/executor_util.h"


/*
 * TupleDescGetAttBinaryInMetadata - Build an AttInMetadata structure based on
 * the supplied TupleDesc. AttInMetadata can be used in conjunction with
 * fmStringInfos containing binary encoded types to produce a properly formed
 * tuple.
 *
 * NOTE: This function is a copy of the PG function TupleDescGetAttInMetadata,
 * except that it uses getTypeBinaryInputInfo instead of getTypeInputInfo.
 */
AttInMetadata *
TupleDescGetAttBinaryInMetadata(TupleDesc tupdesc)
{
	int natts = tupdesc->natts;
	int i;
	Oid atttypeid;
	Oid attinfuncid;

	AttInMetadata *attinmeta = (AttInMetadata *) palloc(sizeof(AttInMetadata));

	/* "Bless" the tupledesc so that we can make rowtype datums with it */
	attinmeta->tupdesc = BlessTupleDesc(tupdesc);

	/*
	 * Gather info needed later to call the "in" function for each attribute
	 */
	FmgrInfo *attinfuncinfo = (FmgrInfo *) palloc0(natts * sizeof(FmgrInfo));
	Oid *attioparams = (Oid *) palloc0(natts * sizeof(Oid));
	int32 *atttypmods = (int32 *) palloc0(natts * sizeof(int32));

	for (i = 0; i < natts; i++)
	{
		Form_pg_attribute att = TupleDescAttr(tupdesc, i);

		/* Ignore dropped attributes */
		if (!att->attisdropped)
		{
			atttypeid = att->atttypid;
			getTypeBinaryInputInfo(atttypeid, &attinfuncid, &attioparams[i]);
			fmgr_info(attinfuncid, &attinfuncinfo[i]);
			atttypmods[i] = att->atttypmod;
		}
	}
	attinmeta->attinfuncs = attinfuncinfo;
	attinmeta->attioparams = attioparams;
	attinmeta->atttypmods = atttypmods;

	return attinmeta;
}


/*
 * BuildTupleFromBytes - build a HeapTuple given user data in binary form.
 * values is an array of StringInfos, one for each attribute of the return
 * tuple. A NULL StringInfo pointer indicates we want to create a NULL field.
 *
 * NOTE: This function is a copy of the PG function BuildTupleFromCStrings,
 * except that it uses ReceiveFunctionCall instead of InputFunctionCall.
 */
HeapTuple
BuildTupleFromBytes(AttInMetadata *attinmeta, fmStringInfo *values)
{
	TupleDesc tupdesc = attinmeta->tupdesc;
	int natts = tupdesc->natts;
	int i;

	Datum *dvalues = (Datum *) palloc(natts * sizeof(Datum));
	bool *nulls = (bool *) palloc(natts * sizeof(bool));

	/*
	 * Call the "in" function for each non-dropped attribute, even for nulls,
	 * to support domains.
	 */
	for (i = 0; i < natts; i++)
	{
		if (!TupleDescAttr(tupdesc, i)->attisdropped)
		{
			/* Non-dropped attributes */
			dvalues[i] = ReceiveFunctionCall(&attinmeta->attinfuncs[i],
											 values[i],
											 attinmeta->attioparams[i],
											 attinmeta->atttypmods[i]);
			if (values[i] != NULL)
			{
				nulls[i] = false;
			}
			else
			{
				nulls[i] = true;
			}
		}
		else
		{
			/* Handle dropped attributes by setting to NULL */
			dvalues[i] = (Datum) 0;
			nulls[i] = true;
		}
	}

	/*
	 * Form a tuple
	 */
	HeapTuple tuple = heap_form_tuple(tupdesc, dvalues, nulls);

	/*
	 * Release locally palloc'd space.  XXX would probably be good to pfree
	 * values of pass-by-reference datums, as well.
	 */
	pfree(dvalues);
	pfree(nulls);

	return tuple;
}