/*

* Python 3.x procedural language extension

*/

#include <setjmp.h>

#define PY_SSIZE_T_CLEAN

#include <Python.h>

#include <compile.h>

#include <structmember.h>

#include "postgres.h"

#include "fmgr.h"

#include "funcapi.h"

#include "libpq/libpq-be.h"

#include "libpq/pqsignal.h"

#include "miscadmin.h"

#include "access/htup.h"

#include "access/heapam.h"

#include "access/xact.h"

#include "access/transam.h"

#include "catalog/namespace.h"

#include "catalog/pg_class.h"

#include "catalog/pg_database.h"

#include "catalog/pg_proc.h"

#include "catalog/pg_type.h"

#include "catalog/pg_language.h"

#include "catalog/indexing.h"

#include "storage/block.h"

#include "storage/off.h"

#include "storage/ipc.h"

#include "commands/trigger.h"

#include "executor/spi.h"

#include "nodes/memnodes.h"

#include "tcop/tcopprot.h"

#include "utils/memutils.h"

#include "utils/array.h"

#include "utils/datum.h"

#include "utils/elog.h"

#include "utils/builtins.h"

#include "utils/hsearch.h"

#include "utils/syscache.h"

#include "utils/relcache.h"

#include "utils/typcache.h"

#include "mb/pg_wchar.h"

#include "pypg/python.h"

#include "pypg/postgres.h"

#include "pypg/extension.h"

#include "pypg/pl.h"

#include "pypg/errordata.h"

#include "pypg/triggerdata.h"

#include "pypg/errcodes.h"

#include "pypg/error.h"

#include "pypg/ist.h"

#include "pypg/type/type.h"

#include "pypg/type/object.h"

#include "pypg/type/record.h"

#include "pypg/type/array.h"

#include "pypg/type/bitwise.h"

#include "pypg/type/numeric.h"

#include "pypg/type/string.h"

#include "pypg/type/system.h"

#include "pypg/type/timewise.h"

#include "pypg/function.h"

#include "pypg/tupledesc.h"

#include "pypg/statement.h"

#include "pypg/cursor.h"

#include "pypg/module.h"

#include "pypg/xact.h"

struct pl_exec_state *pl_execution_context = NULL;

/*

* SXD() - Used to provide *some* information about what the PL is doing.

*

* Often, a Python error can occur outside of the function module.

* When this happens, there is no traceback. The execution context description

* attempts to fill some of that void with what the PL was trying to do at the

* time.

*/

#define SXD(DESCR) pl_execution_context->description = DESCR

/*

* Used by SRFs

*/

PyObj Py_ReturnArgs = NULL;

/*

* Postgres.StopEvent (triggers)

*/

PyObj PyExc_PostgresStopEvent = NULL;

/*

* common, persistent, global strings(PyUnicode).

*

* main_str_ob = "main"

* before_insert_str_ob = "before_insert"

* INSERT_str_ob = "INSERT"

* BEFORE_str_ob = "BEFORE"

* STATEMENT_str_ob = "STATEMENT"

* etc..

*/

#define IDSTR(NAME) PyObj NAME##_str_ob = NULL;

PL_INTERNAL()

PL_ENTRY_POINTS()

PL_MANIPULATIONS()

PL_TRIGGER_ORIENTATIONS()

PL_TRIGGER_TIMINGS()

#undef IDSTR

/*

* Call the function's load_module() method.

*

* This does *not* execute main.

* It only runs the module code in preparation for main.

*/

static PyObj

run_PyPgFunction_module(PyObj func)

{

unsigned long stored_ist_count = ist_count;

PyObj rob;

Assert(func != NULL);

Assert(PyPgFunction_CheckExact(func));

SXD("loading function module");

rob = PyPgFunction_load_module(func);

if (rob == NULL)

{

/*

* Error is being indicated via a thrown Python exception,

* only correct and warn about the issue, if necessary.

*/

if (ext_state > 0)

ext_state = ext_ready;

ext_check_state(WARNING, stored_ist_count);

PyErr_ThrowPostgresError(

"could not load Python function's module object");

}

else

{

/*

* Add the function module to the transaction scope.

* This is kept in the transaction scope for now until

* it actually gets into sys.modules.

*/

if (Py_XACTREF(rob) == -1)

{

Py_DECREF(rob);

PyErr_ThrowPostgresError(

"failed to add function module to transaction scope");

}

Py_DECREF(rob);

/*

* The module body can start ISTs, so the count and state needs to be

* validated.

*/

ext_check_state(ERROR, stored_ist_count);

}

return(rob);

}

/*

* Get the currently cached module object from sys.modules or create one.

*

* >>> getattr(sys.modules.get(str(fn_oid)), '__func__', None) or Postgres.Function(fn_oid)

*/

static PyObj

get_PyPgFunction_from_oid(Oid fn_oid, PyObj *module)

{

PyObj modules, func = NULL;

Assert(OidIsValid(fn_oid));

Assert(module != NULL);

modules = PyImport_GetModuleDict(); /* borrowed */

if (PyErr_Occurred())

return(NULL);

Py_ALLOCATE_OWNER();

{

PyObj so;

so = PyUnicode_FromFormat("%lu", fn_oid);

if (so == NULL)

{

/* Nothing has been acquired so it's safe to return here. */

return(NULL);

}

Py_ACQUIRE(so);

/*

* Check if requested function exists in sys.modules.

*/

if (PyMapping_HasKey(modules, so) == 1)

{

/*

* Function has already been loaded. Check for protocol consistency.

*/

*module = PyObject_GetItem(modules, so);

if (*module != NULL)

{

Py_ACQUIRE(*module);

func = PyObject_GetAttr(*module, __func___str_ob);

if (func != NULL)

{

Py_ACQUIRE(func);

if (!PyPgFunction_CheckExact(func))

{

PyErr_SetString(PyExc_TypeError,

"module's '__func__' attribute is not a Postgres.Function object");

func = NULL;

}

else if (PyPgFunction_GetOid(func) != fn_oid)

{

PyErr_SetString(PyExc_ValueError,

"module's '__func__' attribute does not have the expected object identifier");

func = NULL;

}

else

{

/*

* It's good, INCREF for *return*.

*/

Py_INCREF(func);

Py_INCREF(*module);

}

}

}

}

else

{

*module = NULL;

func = PyPgFunction_FromOid(fn_oid);

}

}

Py_DEALLOCATE_OWNER();

return(func);

}

static PyObj

build_args(PyObj input, int nargs, Datum *arg, bool *argnull)

{

PyObj rob;

SXD("building arguments");

/*

* elog as this expects proper argument counts.

*/

if (PyObject_Length(input) != nargs)

elog(ERROR, "invalid number of argument for Python function");

rob = PyTuple_New(nargs);

if (rob == NULL)

PyErr_ThrowPostgresError(

"failed to create arguments tuple for function invocation");

if (nargs > 0)

{

int i;

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

{

PyObj typ = NULL, ob = NULL;

if (argnull[i])

{

ob = Py_None;

Py_INCREF(ob);

}

else

{

typ = PyPgTupleDesc_GetAttributeType(input, i);

if (typ == NULL)

{

Py_DECREF(rob);

PyErr_ThrowPostgresError("failed to lookup argument type");

}

ob = PyPgObject_New(typ, arg[i]);

}

if (ob == NULL)

{

Py_DECREF(rob);

PyErr_ThrowPostgresError(

"failed to build arguments for function invocation");

}

PyTuple_SET_ITEM(rob, i, ob);

}

}

return(rob);

}

/*

* invoke the "main" object in the given module object using the given args

*/

static PyObj

invoke_main(PyObj module, PyObj args)

{

PyObj main_ob, rob;

/*

* Yes, get the attribute everytime.

*/

main_ob = PyObject_GetAttr(module, main_str_ob);

if (main_ob == NULL)

{

Py_DECREF(args);

PyErr_ThrowPostgresErrorWithCode(

ERRCODE_PYTHON_PROTOCOL_VIOLATION,

"function module has no \"main\" object");

}

SXD("executing main");

rob = PyObject_CallObject(main_ob, args);

Py_DECREF(main_ob);

Py_DECREF(args);

SXD(NULL);

if (rob == NULL)

PyErr_ThrowPostgresErrorWithCode(

ERRCODE_PYTHON_EXCEPTION,

"function's \"main\" raised a Python exception");

return(rob);

}

/*

* SRF ExprContext CallBack function to clean up after VPC-SRFs

*/

static void

srf_eccb(Datum arg)

{

struct pl_fn_info *fn_info = (struct pl_fn_info *) DatumGetPointer(arg);

Assert(fn_info != NULL);

Assert(fn_info->fi_internal_state != NULL);

/*

* XXX: Perhaps too optimistic about this not failing.

*/

Py_DEXTREF(fn_info->fi_internal_state);

fn_info->fi_internal_state = NULL;

}

/*

* pl_validator - compile the function to validate its syntax.

*/

PG_FUNCTION_INFO_V1(pl_validator);

Datum

pl_validator(PG_FUNCTION_ARGS)

{

Oid fn_oid = PG_GETARG_OID(0);

PyObj func, code;

PyGILState_STATE gs;

struct pl_exec_state pl_ctx = {NULL, NULL, NULL,};

Assert(fn_oid != InvalidOid);

Assert(Py_IsInitialized());

if (ext_state == init_pending)

ext_entry();

gs = PyGILState_Ensure();

PG_TRY();

{

func = PyPgFunction_FromOid(fn_oid);

if (func == NULL)

{

PyErr_ThrowPostgresErrorWithContext(

ERRCODE_PYTHON_ERROR,

"could not create Postgres.Function from Oid",

&pl_ctx);

}

code = PyPgFunction_get_code(func);

if (code == NULL)

{

Py_DECREF(func);

PyErr_ThrowPostgresErrorWithContext(

ERRCODE_PYTHON_ERROR,

"cannot compile Python function",

&pl_ctx);

}

Py_DECREF(code);

/*

* All functions that are looked up are cached. This is mere validation,

* so there's no need to actually keep it around.

*/

if (PyPgFunction_RemoveModule(func))

{

Py_DECREF(func);

PyErr_ThrowPostgresErrorWithContext(

ERRCODE_PYTHON_ERROR,

"could not remove of function module from sys.modules",

&pl_ctx);

}

Py_DECREF(func);

}

PG_CATCH();

{

PyGILState_Release(gs);

PG_RE_THROW();

}

PG_END_TRY();

/*

* If a Python error occurred, it should have been raised by now.

*/

Assert(!PyErr_Occurred());

PyGILState_Release(gs);

return(0);

}

/*

* Given a TriggerEvent, return the PyUnicode object

* that is used to represent the handler.

*/

static PyObj

select_trigger_handler(TriggerEvent tev)

{

#define Py_ROW_TRIGGER_TEMPLATE(ev,name) \

case TRIGGER_EVENT_##ev|TRIGGER_EVENT_ROW|TRIGGER_EVENT_BEFORE: \

return(before_##name##_str_ob); \

break; \

case TRIGGER_EVENT_##ev|TRIGGER_EVENT_ROW: \

return(after_##name##_str_ob); \

break;

#define Py_STATEMENT_TRIGGER_TEMPLATE(ev,name) \

case TRIGGER_EVENT_##ev|TRIGGER_EVENT_BEFORE: \

return(before_##name##_statement_str_ob); \

break; \

case TRIGGER_EVENT_##ev: \

return(after_##name##_statement_str_ob); \

break;

switch (tev & (TRIGGER_EVENT_OPMASK|TRIGGER_EVENT_ROW|TRIGGER_EVENT_BEFORE))

{

Py_ROW_TRIGGER_TEMPLATE(INSERT,insert)

Py_STATEMENT_TRIGGER_TEMPLATE(INSERT,insert)

Py_ROW_TRIGGER_TEMPLATE(UPDATE,update)

Py_STATEMENT_TRIGGER_TEMPLATE(UPDATE,update)

Py_ROW_TRIGGER_TEMPLATE(DELETE,delete)

Py_STATEMENT_TRIGGER_TEMPLATE(DELETE,delete)

#if TRIGGER_EVENT_TRUNCATE != 0xDEADBEEF

Py_STATEMENT_TRIGGER_TEMPLATE(TRUNCATE,truncate)

#endif

default:

return(NULL);

break;

}

#undef Py_ROW_TRIGGER_TEMPLATE

#undef Py_STATEMENT_TRIGGER_TEMPLATE

}

/*

* row_trigger - execute the row handler for the event

*/

static Datum

row_trigger(PyObj handler, PyObj trigger_data,

HeapTuple ht_old, HeapTuple ht_new)

{

Datum rd = 0;

MemoryContext former;

PyObj rob, args, old = NULL, new = NULL;

PyObj reltype;

PyObj timing;

/*

* If both are not NULL, it's an update.

*/

if (ht_new != NULL && ht_old != NULL)

args = PyTuple_New(3);

else

args = PyTuple_New(2);

if (args == NULL)

PyErr_ThrowPostgresError("failed to build arguments tuple for trigger");

/*

* Using the pl_handler's reference owner.

*/

Py_ACQUIRE(args);

Py_INCREF(trigger_data);

PyTuple_SET_ITEM(args, 0, trigger_data);

reltype = PyPgTriggerData_GetRelationType(trigger_data);

timing = PyPgTriggerData_GetTiming(trigger_data);

/*

* Need to be in PythonMemoryContext for "PyPgObject_FromPyPgTypeAndHeapTuple".

*

* These *can* elog-out.

*/

former = MemoryContextSwitchTo(PythonMemoryContext);

if (ht_old == NULL)

{

/* INSERT */

new = PyPgObject_FromPyPgTypeAndHeapTuple(reltype, ht_new);

PyTuple_SET_ITEM(args, 1, new);

}

else if (ht_new == NULL)

{

/* DELETE */

old = PyPgObject_FromPyPgTypeAndHeapTuple(reltype, ht_old);

PyTuple_SET_ITEM(args, 1, old);

}

else

{

/* UPDATE */

old = PyPgObject_FromPyPgTypeAndHeapTuple(reltype, ht_old);

PyTuple_SET_ITEM(args, 1, old);

new = PyPgObject_FromPyPgTypeAndHeapTuple(reltype, ht_new);

PyTuple_SET_ITEM(args, 2, new);

}

MemoryContextSwitchTo(former);

rob = PyObject_CallObject(handler, args);

/*

* ReferenceOwner will handle args DECREF, and subseqently new &| old.

*/

if (rob == NULL)

{

/*

* Python exception raised.

*/

if (PyErr_ExceptionMatches(PyExc_PostgresStopEvent))

{

/*

* Indicate that they should not throw StopEvent in AFTER triggers.

*

* Note that ERRCODE_PYTHON_PROTOCOL_VIOLATION is used.

*/

if (timing == AFTER_str_ob)

PyErr_ThrowPostgresErrorWithCode(

ERRCODE_PYTHON_PROTOCOL_VIOLATION,

"cannot stop events that have already occurred");

/*

* Return a zero-Datum to stop the event.

*/

PyErr_Clear();

return(0);

}

PyErr_ThrowPostgresError("trigger function raised Python exception");

}

else

{

if (rob == Py_None || rob == new)

{

/*

* Returned `None`; do default action.

*/

Py_DECREF(rob);

rd = PointerGetDatum(ht_new ? ht_new : ht_old);

}

else if (rob == old)

{

Py_DECREF(rob);

rd = PointerGetDatum(ht_old);

}

else if (timing == AFTER_str_ob)

{

/*

* Be picky about the return value. This should help identify

* potential problems early.

*/

Py_DECREF(rob);

ereport(ERROR,

(errcode(ERRCODE_PYTHON_PROTOCOL_VIOLATION),

errmsg("non-None value returned by trigger fired after"))

);

}

else

{

MemoryContext former;

Datum hthd;

bool isnull = false;

HeapTupleData ht;

PyObj row;

row = Py_NormalizeRow(

PyPgTupleDesc_GetNatts(PyPgType_GetPyPgTupleDesc(reltype)),

PyPgType_GetTupleDesc(reltype),

PyPgTupleDesc_GetNameMap(PyPgType_GetPyPgTupleDesc(reltype)),

rob);

Py_DECREF(rob);

if (row == NULL)

PyErr_ThrowPostgresError("could not normalize replacement row");

rob = row;

PyPgType_DatumNew(reltype, rob, -1, &hthd, &isnull);

ht.t_data = (HeapTupleHeader) DatumGetPointer(hthd);

ht.t_len = HeapTupleHeaderGetDatumLength(ht.t_data);

ht.t_tableOid = PyPgType_GetTableOid(reltype);

former = MemoryContextSwitchTo(pl_execution_context->return_memory_context);

rd = PointerGetDatum(heap_copytuple(&ht));

MemoryContextSwitchTo(former);

pfree(DatumGetPointer(hthd));

}

}

return(rd);

}

/*

* statement_trigger - execute the statement handler for the event

*/

static Datum

statement_trigger(PyObj handler, PyObj trigger_data)

{

PyObj rob, args;

if (PyPgTriggerData_GetTiming(trigger_data) == AFTER_str_ob)

{

/*

* If Postgres ever gets transition tables, this will look much different.

*/

if (PyPgTriggerData_GetManipulation(trigger_data) == UPDATE_str_ob)

{

args = PyTuple_New(3);

Py_INCREF(Py_None);

Py_INCREF(Py_None);

PyTuple_SET_ITEM(args, 1, Py_None);

PyTuple_SET_ITEM(args, 2, Py_None);

}

else

{

args = PyTuple_New(2);

Py_INCREF(Py_None);

PyTuple_SET_ITEM(args, 1, Py_None);

}

}

else

{

/*

* BEFORE the statements run, there are no transition tables.

*/

args = PyTuple_New(1);

}

Py_INCREF(trigger_data);

PyTuple_SET_ITEM(args, 0, trigger_data);

rob = PyObject_CallObject(handler, args);

Py_DECREF(args);

if (rob == NULL)

{

PyErr_ThrowPostgresError("statement trigger raised Python exception");

}

else

{

Py_DECREF(rob); /* whatcha gonna do with it? */

if (rob != Py_None)

{

ereport(ERROR,(

errcode(ERRCODE_E_R_I_E_TRIGGER_PROTOCOL_VIOLATED),

errmsg("statement trigger did not return None")

));

}

}

return(0);

}

/*

* Execute a trigger function.

*/

static Datum

pull_trigger(PG_FUNCTION_ARGS)

{

struct pl_fn_info *fn_info = fcinfo->flinfo->fn_extra;

Datum rd;

PyObj handler_str_ob, handler;

TriggerData *td = (TriggerData *) (fcinfo->context);

TriggerEvent ev = td->tg_event;

Assert(PyPgFunction_IsTrigger(fn_info->fi_func));

Assert(PyPgTriggerData_Check(fn_info->fi_input));

SXD("pulling trigger");

/*

* Select the handler string: "after_insert", "before_insert",

* "after_delete_statement", etc.

*/

handler_str_ob = select_trigger_handler(ev);

if (handler_str_ob == NULL)

elog(ERROR, "unknown trigger event");

/*

* Get the module object that will handle the specific event.

*/

handler = PyObject_GetAttr(fn_info->fi_module, handler_str_ob);

if (handler == NULL)

{

PyErr_ThrowPostgresErrorWithCode(

ERRCODE_TRIGGERED_ACTION_EXCEPTION,

"trigger function does not support event");

}

/* borrow reference from module */

Py_DECREF(handler);

/*

* At this point, the code is no longer common between statement and row

* triggers, so choose the appropriate path.

*/

if (TRIGGER_FIRED_FOR_ROW(ev))

rd = row_trigger(handler, fn_info->fi_input,

td->tg_trigtuple, td->tg_newtuple);

else if (TRIGGER_FIRED_FOR_STATEMENT(td->tg_event))

rd = statement_trigger(handler, fn_info->fi_input);

else

{

elog(ERROR, "unknown trigger event");

/*

* Keep compiler quiet.

*/

Assert(false);

return(0);

}

return(rd);

}

static Datum

create_result_datum(PyObj output, PyObj rob, bool *isnull)

{

Datum rd;

MemoryContext former = CurrentMemoryContext;

bool dont_free = false;

Assert(output != NULL);

Assert(rob != NULL);

Assert(isnull != NULL);

SXD("creating result");

Py_ACQUIRE(rob);

if (output != (PyObj) Py_TYPE(rob))

{

/*

* elog()'s on failure; Also handles the Py_None case.

*/

PyPgType_DatumNew(output, rob, (int32) -1, &rd, isnull);

}

else

{

/*

* Exact type.

*/

rd = PyPgObject_GetDatum(rob);

dont_free = true;

}

/*

* Not NULL and !typbyval, so copy the Datum out.

*/

if (!(*isnull) && !PyPgType_Get_typbyval(output))

{

Datum tmpd;

/*

* Iff !typbyval, the datum is currently in the CurrentMemoryContext,

* so be sure to copy it to the handler context

*/

MemoryContextSwitchTo(pl_execution_context->return_memory_context);

tmpd = datumCopy(rd, false, PyPgType_Get_typlen(output));

MemoryContextSwitchTo(former);

/* bein' kind */

if (!dont_free)

pfree(DatumGetPointer(rd));

rd = tmpd;

}

return(rd);

}

/*

* srf_materialize_cursor - materialize a Postgres.Cursor

*/

static Datum

srf_materialize_cursor(FunctionCallInfo fcinfo, PyObj cursor)

{

struct pl_fn_info *fn_info = fcinfo->flinfo->fn_extra;

ReturnSetInfo *rsi = (ReturnSetInfo *) fcinfo->resultinfo;

MemoryContext former = NULL;

Tuplestorestate *tss;

TupleDesc srcdesc;

PyObj buf, row;

Portal p;

bool forward;

int i;

/*

* Fast path for PyPgCursor's

*/

Assert(PyPgCursor_Check(cursor));

SXD("materializing cursor");

srcdesc = PyPgType_GetTupleDesc(PyPgCursor_GetOutput(cursor));

p = PyPgCursor_GetPortal(cursor);

buf = PyPgCursor_GetBuffer(cursor);

/*

* Note the direction. (Backward specifies a scrollable cursor)

*/

if (PyPgCursor_GetChunksize(cursor) == CUR_SCROLL_BACKWARD)

forward = false;

else

forward = true;

former = MemoryContextSwitchTo(rsi->econtext->ecxt_per_query_memory);

rsi->returnMode = SFRM_Materialize;

rsi->isDone = ExprSingleResult;

rsi->setDesc = CreateTupleDescCopy(srcdesc);

rsi->setResult = tss = tuplestore_begin_heap(

rsi->allowedModes, false, work_mem

);

/*

* If the cursor had a buffer(rows()), write those to the store

* first.

*/

if (buf != NULL)

{

Py_ACQUIRE_SPACE();

{

while ((row = PyIter_Next(buf)) != NULL)

{

HeapTupleData ht;

Py_XREPLACE(row);

ht.t_data = (HeapTupleHeader) DatumGetPointer(PyPgObject_GetDatum(row));

ht.t_len = HeapTupleHeaderGetDatumLength(ht.t_data);

ht.t_tableOid = PyPgType_GetTableOid(fn_info->fi_input);

tuplestore_puttuple(tss, &ht);

}

}

Py_RELEASE_SPACE();

/*

* If failure was caused by a Postgres error, flow should never

* get here. Otherwise, watch for a Python exception.

*/

if (PyErr_Occurred())

{

PyErr_ThrowPostgresError(

"could not materialize cursor buffer");

}

}

/*

* Write the remaining tuples.

*/

do

{

/*

* We assume that the tuples are of reasonable size.

*/

MemoryContextSwitchTo(former);

SPI_cursor_fetch(p, forward, 30);

former = MemoryContextSwitchTo(rsi->econtext->ecxt_per_query_memory);

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

tuplestore_puttuple(tss, SPI_tuptable->vals[i]);

SPI_freetuptable(SPI_tuptable);

}

while (SPI_processed == 30);

MemoryContextSwitchTo(former);

return(0);

}

/*

* srf_materialize_iter - materialize an arbitrary iterator

*/

static Datum

srf_materialize_iter(FunctionCallInfo fcinfo, PyObj iter)

{

struct pl_fn_info *fn_info = fcinfo->flinfo->fn_extra;

ReturnSetInfo *rsi = (ReturnSetInfo *) fcinfo->resultinfo;

MemoryContext former = NULL;

Tuplestorestate *tss;

PyObj row, tdo;

Datum *datums;

bool *nulls;

PyObj namemap, typs;

int rnatts, *freemap;

SXD("materializing");

if (PyPgType_IsComposite(fn_info->fi_output))

tdo = PyPgType_GetPyPgTupleDesc(fn_info->fi_output);

else

{

tdo = PyPgTupleDesc_FromCopy(rsi->expectedDesc);

Py_ACQUIRE(tdo); /* owned by the _handler call */

}

rnatts = PyPgTupleDesc_GetNatts(tdo);

namemap = PyPgTupleDesc_GetNameMap(tdo);

typs = PyPgTupleDesc_GetTypesTuple(tdo);

freemap = PyPgTupleDesc_GetFreeMap(tdo);

rsi->returnMode = SFRM_Materialize;

rsi->isDone = ExprSingleResult;

former = MemoryContextSwitchTo(rsi->econtext->ecxt_per_query_memory);

rsi->setDesc = CreateTupleDescCopy(PyPgTupleDesc_GetTupleDesc(tdo));

rsi->setResult = tss = tuplestore_begin_heap(

rsi->allowedModes & SFRM_Materialize_Random, false, work_mem

);

MemoryContextSwitchTo(former);

/*

* Allocate memory for building tuples.

*/

datums = palloc(sizeof(Datum) * rsi->setDesc->natts);

nulls = palloc(sizeof(bool) * rsi->setDesc->natts);

Py_ACQUIRE_SPACE();

{

while ((row = PyIter_Next(iter)) != NULL)

{

HeapTuple ht;

Py_XREPLACE(row); /** managed reference **/

row = Py_NormalizeRow(rnatts, rsi->setDesc, namemap, row);

if (row == NULL)

break;

Py_XREPLACE(row); /** replace managed reference **/

Py_BuildDatumsAndNulls(rsi->setDesc, typs, row, datums, nulls);

ht = heap_form_tuple(rsi->setDesc, datums, nulls);

/*

* Any memory allocated for datums & nulls needs to be freed.

* Likely, it would be wise to run Py_BuildDatumsAndNulls in a memory

* context that gets reset every N iterations, but for now, explicitly

* pfree the memory.

*/

FreeReferences(freemap, datums, nulls);

former = MemoryContextSwitchTo(rsi->econtext->ecxt_per_query_memory);

tuplestore_puttuple(tss, ht);

MemoryContextSwitchTo(former);

heap_freetuple(ht);

}

}

Py_RELEASE_SPACE();

pfree(datums);

datums = NULL;

pfree(nulls);

nulls = NULL;

MemoryContextSwitchTo(former);

if (PyErr_Occurred())

{

PyErr_ThrowPostgresError(

"could not materialize result from returned iterable");

}

return(0);

}

/*

* srf_materialize - call the function and materialize the result

*/