PostgreSQL Source Code  git master
 All Data Structures Namespaces Files Functions Variables Typedefs Enumerations Enumerator Macros
plpy_cursorobject.c
Go to the documentation of this file.
1 /*
2  * the PLyCursor class
3  *
4  * src/pl/plpython/plpy_cursorobject.c
5  */
6 
7 #include "postgres.h"
8 
9 #include <limits.h>
10 
11 #include "access/xact.h"
12 #include "mb/pg_wchar.h"
13 #include "utils/memutils.h"
14 
15 #include "plpython.h"
16 
17 #include "plpy_cursorobject.h"
18 
19 #include "plpy_elog.h"
20 #include "plpy_main.h"
21 #include "plpy_planobject.h"
22 #include "plpy_procedure.h"
23 #include "plpy_resultobject.h"
24 #include "plpy_spi.h"
25 
26 
27 static PyObject *PLy_cursor_query(const char *query);
28 static PyObject *PLy_cursor_plan(PyObject *ob, PyObject *args);
29 static void PLy_cursor_dealloc(PyObject *arg);
30 static PyObject *PLy_cursor_iternext(PyObject *self);
31 static PyObject *PLy_cursor_fetch(PyObject *self, PyObject *args);
32 static PyObject *PLy_cursor_close(PyObject *self, PyObject *unused);
33 
34 static char PLy_cursor_doc[] = {
35  "Wrapper around a PostgreSQL cursor"
36 };
37 
38 static PyMethodDef PLy_cursor_methods[] = {
39  {"fetch", PLy_cursor_fetch, METH_VARARGS, NULL},
40  {"close", PLy_cursor_close, METH_NOARGS, NULL},
41  {NULL, NULL, 0, NULL}
42 };
43 
44 static PyTypeObject PLy_CursorType = {
46  "PLyCursor", /* tp_name */
47  sizeof(PLyCursorObject), /* tp_size */
48  0, /* tp_itemsize */
49 
50  /*
51  * methods
52  */
53  PLy_cursor_dealloc, /* tp_dealloc */
54  0, /* tp_print */
55  0, /* tp_getattr */
56  0, /* tp_setattr */
57  0, /* tp_compare */
58  0, /* tp_repr */
59  0, /* tp_as_number */
60  0, /* tp_as_sequence */
61  0, /* tp_as_mapping */
62  0, /* tp_hash */
63  0, /* tp_call */
64  0, /* tp_str */
65  0, /* tp_getattro */
66  0, /* tp_setattro */
67  0, /* tp_as_buffer */
68  Py_TPFLAGS_DEFAULT | Py_TPFLAGS_BASETYPE | Py_TPFLAGS_HAVE_ITER, /* tp_flags */
69  PLy_cursor_doc, /* tp_doc */
70  0, /* tp_traverse */
71  0, /* tp_clear */
72  0, /* tp_richcompare */
73  0, /* tp_weaklistoffset */
74  PyObject_SelfIter, /* tp_iter */
75  PLy_cursor_iternext, /* tp_iternext */
76  PLy_cursor_methods, /* tp_tpmethods */
77 };
78 
79 void
81 {
82  if (PyType_Ready(&PLy_CursorType) < 0)
83  elog(ERROR, "could not initialize PLy_CursorType");
84 }
85 
86 PyObject *
87 PLy_cursor(PyObject *self, PyObject *args)
88 {
89  char *query;
90  PyObject *plan;
91  PyObject *planargs = NULL;
92 
93  if (PyArg_ParseTuple(args, "s", &query))
94  return PLy_cursor_query(query);
95 
96  PyErr_Clear();
97 
98  if (PyArg_ParseTuple(args, "O|O", &plan, &planargs))
99  return PLy_cursor_plan(plan, planargs);
100 
101  PLy_exception_set(PLy_exc_error, "plpy.cursor expected a query or a plan");
102  return NULL;
103 }
104 
105 
106 static PyObject *
107 PLy_cursor_query(const char *query)
108 {
110  volatile MemoryContext oldcontext;
111  volatile ResourceOwner oldowner;
112 
113  if ((cursor = PyObject_New(PLyCursorObject, &PLy_CursorType)) == NULL)
114  return NULL;
115  cursor->portalname = NULL;
116  cursor->closed = false;
118  "PL/Python cursor context",
120  PLy_typeinfo_init(&cursor->result, cursor->mcxt);
121 
122  oldcontext = CurrentMemoryContext;
123  oldowner = CurrentResourceOwner;
124 
125  PLy_spi_subtransaction_begin(oldcontext, oldowner);
126 
127  PG_TRY();
128  {
130  SPIPlanPtr plan;
131  Portal portal;
132 
133  pg_verifymbstr(query, strlen(query), false);
134 
135  plan = SPI_prepare(query, 0, NULL);
136  if (plan == NULL)
137  elog(ERROR, "SPI_prepare failed: %s",
139 
140  portal = SPI_cursor_open(NULL, plan, NULL, NULL,
141  exec_ctx->curr_proc->fn_readonly);
142  SPI_freeplan(plan);
143 
144  if (portal == NULL)
145  elog(ERROR, "SPI_cursor_open() failed: %s",
147 
148  cursor->portalname = MemoryContextStrdup(cursor->mcxt, portal->name);
149 
150  PLy_spi_subtransaction_commit(oldcontext, oldowner);
151  }
152  PG_CATCH();
153  {
154  PLy_spi_subtransaction_abort(oldcontext, oldowner);
155  return NULL;
156  }
157  PG_END_TRY();
158 
159  Assert(cursor->portalname != NULL);
160  return (PyObject *) cursor;
161 }
162 
163 static PyObject *
164 PLy_cursor_plan(PyObject *ob, PyObject *args)
165 {
167  volatile int nargs;
168  int i;
169  PLyPlanObject *plan;
170  volatile MemoryContext oldcontext;
171  volatile ResourceOwner oldowner;
172 
173  if (args)
174  {
175  if (!PySequence_Check(args) || PyString_Check(args) || PyUnicode_Check(args))
176  {
177  PLy_exception_set(PyExc_TypeError, "plpy.cursor takes a sequence as its second argument");
178  return NULL;
179  }
180  nargs = PySequence_Length(args);
181  }
182  else
183  nargs = 0;
184 
185  plan = (PLyPlanObject *) ob;
186 
187  if (nargs != plan->nargs)
188  {
189  char *sv;
190  PyObject *so = PyObject_Str(args);
191 
192  if (!so)
193  PLy_elog(ERROR, "could not execute plan");
194  sv = PyString_AsString(so);
195  PLy_exception_set_plural(PyExc_TypeError,
196  "Expected sequence of %d argument, got %d: %s",
197  "Expected sequence of %d arguments, got %d: %s",
198  plan->nargs,
199  plan->nargs, nargs, sv);
200  Py_DECREF(so);
201 
202  return NULL;
203  }
204 
205  if ((cursor = PyObject_New(PLyCursorObject, &PLy_CursorType)) == NULL)
206  return NULL;
207  cursor->portalname = NULL;
208  cursor->closed = false;
210  "PL/Python cursor context",
212  PLy_typeinfo_init(&cursor->result, cursor->mcxt);
213 
214  oldcontext = CurrentMemoryContext;
215  oldowner = CurrentResourceOwner;
216 
217  PLy_spi_subtransaction_begin(oldcontext, oldowner);
218 
219  PG_TRY();
220  {
222  Portal portal;
223  char *volatile nulls;
224  volatile int j;
225 
226  if (nargs > 0)
227  nulls = palloc(nargs * sizeof(char));
228  else
229  nulls = NULL;
230 
231  for (j = 0; j < nargs; j++)
232  {
233  PyObject *elem;
234 
235  elem = PySequence_GetItem(args, j);
236  if (elem != Py_None)
237  {
238  PG_TRY();
239  {
240  plan->values[j] =
241  plan->args[j].out.d.func(&(plan->args[j].out.d),
242  -1,
243  elem,
244  false);
245  }
246  PG_CATCH();
247  {
248  Py_DECREF(elem);
249  PG_RE_THROW();
250  }
251  PG_END_TRY();
252 
253  Py_DECREF(elem);
254  nulls[j] = ' ';
255  }
256  else
257  {
258  Py_DECREF(elem);
259  plan->values[j] =
260  InputFunctionCall(&(plan->args[j].out.d.typfunc),
261  NULL,
262  plan->args[j].out.d.typioparam,
263  -1);
264  nulls[j] = 'n';
265  }
266  }
267 
268  portal = SPI_cursor_open(NULL, plan->plan, plan->values, nulls,
269  exec_ctx->curr_proc->fn_readonly);
270  if (portal == NULL)
271  elog(ERROR, "SPI_cursor_open() failed: %s",
273 
274  cursor->portalname = MemoryContextStrdup(cursor->mcxt, portal->name);
275 
276  PLy_spi_subtransaction_commit(oldcontext, oldowner);
277  }
278  PG_CATCH();
279  {
280  int k;
281 
282  /* cleanup plan->values array */
283  for (k = 0; k < nargs; k++)
284  {
285  if (!plan->args[k].out.d.typbyval &&
286  (plan->values[k] != PointerGetDatum(NULL)))
287  {
288  pfree(DatumGetPointer(plan->values[k]));
289  plan->values[k] = PointerGetDatum(NULL);
290  }
291  }
292 
293  Py_DECREF(cursor);
294 
295  PLy_spi_subtransaction_abort(oldcontext, oldowner);
296  return NULL;
297  }
298  PG_END_TRY();
299 
300  for (i = 0; i < nargs; i++)
301  {
302  if (!plan->args[i].out.d.typbyval &&
303  (plan->values[i] != PointerGetDatum(NULL)))
304  {
305  pfree(DatumGetPointer(plan->values[i]));
306  plan->values[i] = PointerGetDatum(NULL);
307  }
308  }
309 
310  Assert(cursor->portalname != NULL);
311  return (PyObject *) cursor;
312 }
313 
314 static void
316 {
318  Portal portal;
319 
320  cursor = (PLyCursorObject *) arg;
321 
322  if (!cursor->closed)
323  {
324  portal = GetPortalByName(cursor->portalname);
325 
326  if (PortalIsValid(portal))
327  SPI_cursor_close(portal);
328  cursor->closed = true;
329  }
330  if (cursor->mcxt)
331  {
332  MemoryContextDelete(cursor->mcxt);
333  cursor->mcxt = NULL;
334  }
335  arg->ob_type->tp_free(arg);
336 }
337 
338 static PyObject *
339 PLy_cursor_iternext(PyObject *self)
340 {
342  PyObject *ret;
343  volatile MemoryContext oldcontext;
344  volatile ResourceOwner oldowner;
345  Portal portal;
346 
347  cursor = (PLyCursorObject *) self;
348 
349  if (cursor->closed)
350  {
351  PLy_exception_set(PyExc_ValueError, "iterating a closed cursor");
352  return NULL;
353  }
354 
355  portal = GetPortalByName(cursor->portalname);
356  if (!PortalIsValid(portal))
357  {
358  PLy_exception_set(PyExc_ValueError,
359  "iterating a cursor in an aborted subtransaction");
360  return NULL;
361  }
362 
363  oldcontext = CurrentMemoryContext;
364  oldowner = CurrentResourceOwner;
365 
366  PLy_spi_subtransaction_begin(oldcontext, oldowner);
367 
368  PG_TRY();
369  {
370  SPI_cursor_fetch(portal, true, 1);
371  if (SPI_processed == 0)
372  {
373  PyErr_SetNone(PyExc_StopIteration);
374  ret = NULL;
375  }
376  else
377  {
378  if (cursor->result.is_rowtype != 1)
380 
381  ret = PLyDict_FromTuple(&cursor->result, SPI_tuptable->vals[0],
383  }
384 
386 
387  PLy_spi_subtransaction_commit(oldcontext, oldowner);
388  }
389  PG_CATCH();
390  {
391  PLy_spi_subtransaction_abort(oldcontext, oldowner);
392  return NULL;
393  }
394  PG_END_TRY();
395 
396  return ret;
397 }
398 
399 static PyObject *
400 PLy_cursor_fetch(PyObject *self, PyObject *args)
401 {
403  int count;
404  PLyResultObject *ret;
405  volatile MemoryContext oldcontext;
406  volatile ResourceOwner oldowner;
407  Portal portal;
408 
409  if (!PyArg_ParseTuple(args, "i:fetch", &count))
410  return NULL;
411 
412  cursor = (PLyCursorObject *) self;
413 
414  if (cursor->closed)
415  {
416  PLy_exception_set(PyExc_ValueError, "fetch from a closed cursor");
417  return NULL;
418  }
419 
420  portal = GetPortalByName(cursor->portalname);
421  if (!PortalIsValid(portal))
422  {
423  PLy_exception_set(PyExc_ValueError,
424  "iterating a cursor in an aborted subtransaction");
425  return NULL;
426  }
427 
428  ret = (PLyResultObject *) PLy_result_new();
429  if (ret == NULL)
430  return NULL;
431 
432  oldcontext = CurrentMemoryContext;
433  oldowner = CurrentResourceOwner;
434 
435  PLy_spi_subtransaction_begin(oldcontext, oldowner);
436 
437  PG_TRY();
438  {
439  SPI_cursor_fetch(portal, true, count);
440 
441  if (cursor->result.is_rowtype != 1)
443 
444  Py_DECREF(ret->status);
445  ret->status = PyInt_FromLong(SPI_OK_FETCH);
446 
447  Py_DECREF(ret->nrows);
448  ret->nrows = (SPI_processed > (uint64) LONG_MAX) ?
449  PyFloat_FromDouble((double) SPI_processed) :
450  PyInt_FromLong((long) SPI_processed);
451 
452  if (SPI_processed != 0)
453  {
454  uint64 i;
455 
456  /*
457  * PyList_New() and PyList_SetItem() use Py_ssize_t for list size
458  * and list indices; so we cannot support a result larger than
459  * PY_SSIZE_T_MAX.
460  */
461  if (SPI_processed > (uint64) PY_SSIZE_T_MAX)
462  ereport(ERROR,
463  (errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
464  errmsg("query result has too many rows to fit in a Python list")));
465 
466  Py_DECREF(ret->rows);
467  ret->rows = PyList_New(SPI_processed);
468 
469  for (i = 0; i < SPI_processed; i++)
470  {
471  PyObject *row = PLyDict_FromTuple(&cursor->result,
472  SPI_tuptable->vals[i],
474 
475  PyList_SetItem(ret->rows, i, row);
476  }
477  }
478 
480 
481  PLy_spi_subtransaction_commit(oldcontext, oldowner);
482  }
483  PG_CATCH();
484  {
485  PLy_spi_subtransaction_abort(oldcontext, oldowner);
486  return NULL;
487  }
488  PG_END_TRY();
489 
490  return (PyObject *) ret;
491 }
492 
493 static PyObject *
494 PLy_cursor_close(PyObject *self, PyObject *unused)
495 {
497 
498  if (!cursor->closed)
499  {
500  Portal portal = GetPortalByName(cursor->portalname);
501 
502  if (!PortalIsValid(portal))
503  {
504  PLy_exception_set(PyExc_ValueError,
505  "closing a cursor in an aborted subtransaction");
506  return NULL;
507  }
508 
509  SPI_cursor_close(portal);
510  cursor->closed = true;
511  }
512 
513  Py_INCREF(Py_None);
514  return Py_None;
515 }
FmgrInfo typfunc
Definition: plpy_typeio.h:58
static PyMethodDef PLy_cursor_methods[]
void MemoryContextDelete(MemoryContext context)
Definition: mcxt.c:200
void PLy_exception_set_plural(PyObject *exc, const char *fmt_singular, const char *fmt_plural, unsigned long n,...)
Definition: plpy_elog.c:514
void PLy_spi_subtransaction_abort(MemoryContext oldcontext, ResourceOwner oldowner)
Definition: plpy_spi.c:522
PyObject * PLy_cursor(PyObject *self, PyObject *args)
void PLy_typeinfo_init(PLyTypeInfo *arg, MemoryContext mcxt)
Definition: plpy_typeio.c:70
PyObject_HEAD char * portalname
#define PyVarObject_HEAD_INIT(type, size)
Definition: plpython.h:111
#define PointerGetDatum(X)
Definition: postgres.h:564
Portal GetPortalByName(const char *name)
Definition: portalmem.c:129
ResourceOwner CurrentResourceOwner
Definition: resowner.c:138
SPIPlanPtr SPI_prepare(const char *src, int nargs, Oid *argtypes)
Definition: spi.c:481
void PLy_elog(int elevel, const char *fmt,...)
Definition: plpy_elog.c:47
SPITupleTable * SPI_tuptable
Definition: spi.c:41
int errcode(int sqlerrcode)
Definition: elog.c:575
void PLy_input_tuple_funcs(PLyTypeInfo *arg, TupleDesc desc)
Definition: plpy_typeio.c:105
PyObject * PLy_exc_error
Definition: plpy_elog.c:19
Portal SPI_cursor_open(const char *name, SPIPlanPtr plan, Datum *Values, const char *Nulls, bool read_only)
Definition: spi.c:1028
PLyExecutionContext * PLy_current_execution_context(void)
Definition: plpy_main.c:398
void PLy_exception_set(PyObject *exc, const char *fmt,...)
Definition: plpy_elog.c:500
HeapTuple * vals
Definition: spi.h:27
PLyObToDatumFunc func
Definition: plpy_typeio.h:57
uint64 SPI_processed
Definition: spi.c:39
static PyObject * PLy_cursor_query(const char *query)
int SPI_result
Definition: spi.c:42
static PyObject * PLy_cursor_close(PyObject *self, PyObject *unused)
void PLy_spi_subtransaction_begin(MemoryContext oldcontext, ResourceOwner oldowner)
Definition: plpy_spi.c:505
void pfree(void *pointer)
Definition: mcxt.c:992
const char * name
Definition: portal.h:117
static PyObject * PLy_cursor_iternext(PyObject *self)
#define ERROR
Definition: elog.h:43
static void PLy_cursor_dealloc(PyObject *arg)
void PLy_cursor_init_type(void)
MemoryContext mcxt
PLyObToDatum d
Definition: plpy_typeio.h:77
PyObject * PLyDict_FromTuple(PLyTypeInfo *info, HeapTuple tuple, TupleDesc desc)
Definition: plpy_typeio.c:280
PLyTypeInfo * args
const char * SPI_result_code_string(int code)
Definition: spi.c:1509
#define ALLOCSET_DEFAULT_SIZES
Definition: memutils.h:145
static PyObject * PLy_cursor_fetch(PyObject *self, PyObject *args)
PLyTypeOutput out
Definition: plpy_typeio.h:87
Definition: type.h:125
PyObject_HEAD SPIPlanPtr plan
MemoryContext CurrentMemoryContext
Definition: mcxt.c:37
#define ereport(elevel, rest)
Definition: elog.h:122
static char PLy_cursor_doc[]
MemoryContext TopMemoryContext
Definition: mcxt.c:43
PLyProcedure * curr_proc
Definition: plpy_main.h:20
void SPI_freetuptable(SPITupleTable *tuptable)
Definition: spi.c:969
#define PortalIsValid(p)
Definition: portal.h:197
#define SPI_OK_FETCH
Definition: spi.h:49
MemoryContext AllocSetContextCreate(MemoryContext parent, const char *name, Size minContextSize, Size initBlockSize, Size maxBlockSize)
Definition: aset.c:440
TupleDesc tupdesc
Definition: spi.h:26
Datum InputFunctionCall(FmgrInfo *flinfo, char *str, Oid typioparam, int32 typmod)
Definition: fmgr.c:1882
#define PG_CATCH()
Definition: elog.h:293
#define NULL
Definition: c.h:226
#define Assert(condition)
Definition: c.h:670
struct PLyCursorObject PLyCursorObject
int is_rowtype
Definition: plpy_typeio.h:93
#define PG_RE_THROW()
Definition: elog.h:314
static PyObject * PLy_cursor_plan(PyObject *ob, PyObject *args)
#define PY_SSIZE_T_MAX
Definition: plpython.h:66
#define DatumGetPointer(X)
Definition: postgres.h:557
int SPI_freeplan(SPIPlanPtr plan)
Definition: spi.c:608
void SPI_cursor_close(Portal portal)
Definition: spi.c:1399
PyObject * PLy_result_new(void)
void PLy_spi_subtransaction_commit(MemoryContext oldcontext, ResourceOwner oldowner)
Definition: plpy_spi.c:513
void * palloc(Size size)
Definition: mcxt.c:891
int errmsg(const char *fmt,...)
Definition: elog.c:797
char * MemoryContextStrdup(MemoryContext context, const char *string)
Definition: mcxt.c:1152
int i
void * arg
void SPI_cursor_fetch(Portal portal, bool forward, long count)
Definition: spi.c:1343
PyObject_HEAD PyObject * nrows
bool pg_verifymbstr(const char *mbstr, int len, bool noError)
Definition: wchar.c:1866
#define elog
Definition: elog.h:219
#define PG_TRY()
Definition: elog.h:284
static PyTypeObject PLy_CursorType
#define PG_END_TRY()
Definition: elog.h:300
PLyTypeInfo result