From 35956cd013e4ee01bfb8a256ffefa1164690ff2d Mon Sep 17 00:00:00 2001 From: Ivan Kush Date: Tue, 12 Dec 2023 01:25:48 +0300 Subject: [PATCH] Autonomous transactions This implements a C API to run SQL statements in a separate from foreground session named autonomous. It runs on background worker, communicating by FE/BE protocol over a DSM message queue. The session looks much like a normal database connection, but it is always to the same database, and there is no authentication needed. Autonomous session's errors are sent to backend. Autonomous sessions are taken from the pool, when autonomous function begins to execute. Each backend lazily creates pool, pools aren't shared between backends. --- src/backend/commands/prepare.c | 2 +- src/backend/commands/variable.c | 5 + src/backend/libpq/pqmq.c | 28 + src/backend/parser/analyze.c | 6 +- src/backend/parser/parse_param.c | 45 +- src/backend/po/de.po | 4 + src/backend/po/es.po | 6 + src/backend/po/fr.po | 4 + src/backend/po/id.po | 4 + src/backend/po/it.po | 4 + src/backend/po/ja.po | 4 + src/backend/po/ko.po | 4 + src/backend/po/pl.po | 4 + src/backend/po/pt_BR.po | 4 + src/backend/po/ru.po | 6 +- src/backend/po/sv.po | 4 + src/backend/po/tr.po | 4 + src/backend/po/uk.po | 4 + src/backend/po/zh_CN.po | 4 + src/backend/postmaster/bgworker.c | 4 + src/backend/tcop/Makefile | 1 + src/backend/tcop/autonomous.c | 1317 +++++++++++++++ src/backend/tcop/postgres.c | 66 +- src/backend/utils/misc/guc_tables.c | 14 + src/backend/utils/misc/postgresql.conf.sample | 4 + src/include/libpq/pqmq.h | 1 + src/include/parser/analyze.h | 2 +- src/include/parser/parse_param.h | 2 +- src/include/tcop/autonomous.h | 48 + src/include/tcop/tcopprot.h | 11 +- src/pl/plpgsql/src/Makefile | 2 +- .../src/expected/plpgsql_autonomous.out | 1423 +++++++++++++++++ src/pl/plpgsql/src/pl_exec.c | 177 +- src/pl/plpgsql/src/pl_gram.y | 17 + src/pl/plpgsql/src/pl_unreserved_kwlist.h | 1 + src/pl/plpgsql/src/plpgsql.h | 5 + src/pl/plpgsql/src/sql/plpgsql_autonomous.sql | 1296 +++++++++++++++ 37 files changed, 4503 insertions(+), 34 deletions(-) create mode 100644 src/backend/tcop/autonomous.c create mode 100644 src/include/tcop/autonomous.h create mode 100644 src/pl/plpgsql/src/expected/plpgsql_autonomous.out create mode 100644 src/pl/plpgsql/src/sql/plpgsql_autonomous.sql diff --git a/src/backend/commands/prepare.c b/src/backend/commands/prepare.c index 18f70319fc..b855066f93 100644 --- a/src/backend/commands/prepare.c +++ b/src/backend/commands/prepare.c @@ -117,7 +117,7 @@ PrepareQuery(ParseState *pstate, PrepareStmt *stmt, * Rewrite the query. The result could be 0, 1, or many queries. */ query_list = pg_analyze_and_rewrite_varparams(rawstmt, pstate->p_sourcetext, - &argtypes, &nargs, NULL); + &argtypes, &nargs, NULL, NULL); /* Finish filling in the CachedPlanSource */ CompleteCachedPlan(plansource, diff --git a/src/backend/commands/variable.c b/src/backend/commands/variable.c index c361bb2079..2fd9cee423 100644 --- a/src/backend/commands/variable.c +++ b/src/backend/commands/variable.c @@ -676,12 +676,17 @@ show_random_seed(void) * SET CLIENT_ENCODING */ +void (*check_client_encoding_hook)(void); + bool check_client_encoding(char **newval, void **extra, GucSource source) { int encoding; const char *canonical_name; + if (check_client_encoding_hook) + check_client_encoding_hook(); + /* Look up the encoding by name */ encoding = pg_valid_client_encoding(*newval); if (encoding < 0) diff --git a/src/backend/libpq/pqmq.c b/src/backend/libpq/pqmq.c index 38b042804c..688db4716b 100644 --- a/src/backend/libpq/pqmq.c +++ b/src/backend/libpq/pqmq.c @@ -45,6 +45,17 @@ static const PQcommMethods PqCommMqMethods = { .putmessage_noblock = mq_putmessage_noblock }; +typedef struct PQcontext { + const PQcommMethods *PqCommMethods; + shm_mq_handle *pq_mq_handle; + CommandDest whereToSendOutput; + ProtocolVersion FrontendProtocol; + dsm_segment *seg; +} PQcontext; +static const PQcontext null_ctx = {.PqCommMethods = NULL, .pq_mq_handle = NULL, .whereToSendOutput = DestNone, .seg = NULL}; +/*same as null_ctx, impossible to assign null_ctx: error: initializer element is not a compile-time constant */ +static PQcontext prev_ctx = {.PqCommMethods = NULL, .pq_mq_handle = NULL, .whereToSendOutput = DestNone, .seg = NULL}; + /* * Arrange to redirect frontend/backend protocol messages to a shared-memory * message queue. @@ -52,6 +63,12 @@ static const PQcommMethods PqCommMqMethods = { void pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh) { + prev_ctx.PqCommMethods = PqCommMethods; + prev_ctx.whereToSendOutput = whereToSendOutput; + prev_ctx.FrontendProtocol = FrontendProtocol; + prev_ctx.pq_mq_handle = pq_mq_handle; + prev_ctx.seg = seg; + PqCommMethods = &PqCommMqMethods; pq_mq_handle = mqh; whereToSendOutput = DestRemote; @@ -59,6 +76,17 @@ pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh) on_dsm_detach(seg, pq_cleanup_redirect_to_shm_mq, (Datum) 0); } +void +pq_stop_redirect_to_shm_mq(void) +{ + cancel_on_dsm_detach(prev_ctx.seg, pq_cleanup_redirect_to_shm_mq, (Datum) 0); + PqCommMethods = prev_ctx.PqCommMethods; + whereToSendOutput = prev_ctx.whereToSendOutput; + FrontendProtocol = prev_ctx.FrontendProtocol; + pq_mq_handle = prev_ctx.pq_mq_handle; + prev_ctx = null_ctx; +} + /* * When the DSM that contains our shm_mq goes away, we need to stop sending * messages to it. diff --git a/src/backend/parser/analyze.c b/src/backend/parser/analyze.c index 7a1dfb6364..e63a338b6c 100644 --- a/src/backend/parser/analyze.c +++ b/src/backend/parser/analyze.c @@ -146,7 +146,7 @@ parse_analyze_fixedparams(RawStmt *parseTree, const char *sourceText, Query * parse_analyze_varparams(RawStmt *parseTree, const char *sourceText, Oid **paramTypes, int *numParams, - QueryEnvironment *queryEnv) + QueryEnvironment *queryEnv, const char *paramNames[]) { ParseState *pstate = make_parsestate(NULL); Query *query; @@ -156,7 +156,7 @@ parse_analyze_varparams(RawStmt *parseTree, const char *sourceText, pstate->p_sourcetext = sourceText; - setup_parse_variable_parameters(pstate, paramTypes, numParams); + setup_parse_variable_parameters(pstate, paramTypes, numParams, paramNames); pstate->p_queryEnv = queryEnv; @@ -2967,7 +2967,7 @@ transformExplainStmt(ParseState *pstate, ExplainStmt *stmt) /* don't "break", as we want the last value */ } if (generic_plan) - setup_parse_variable_parameters(pstate, ¶mTypes, &numParams); + setup_parse_variable_parameters(pstate, ¶mTypes, &numParams, NULL); } /* transform contained query, allowing SELECT INTO */ diff --git a/src/backend/parser/parse_param.c b/src/backend/parser/parse_param.c index 2240284f21..7a91b1fcc7 100644 --- a/src/backend/parser/parse_param.c +++ b/src/backend/parser/parse_param.c @@ -49,8 +49,10 @@ typedef struct VarParamState { Oid **paramTypes; /* array of parameter type OIDs */ int *numParams; /* number of array entries */ + const char **paramNames; } VarParamState; +static Node *variable_post_column_ref_hook(ParseState *pstate, ColumnRef *cref, Node *var); static Node *fixed_paramref_hook(ParseState *pstate, ParamRef *pref); static Node *variable_paramref_hook(ParseState *pstate, ParamRef *pref); static Node *variable_coerce_param_hook(ParseState *pstate, Param *param, @@ -81,17 +83,58 @@ setup_parse_fixed_parameters(ParseState *pstate, */ void setup_parse_variable_parameters(ParseState *pstate, - Oid **paramTypes, int *numParams) + Oid **paramTypes, int *numParams, const char *paramNames[]) { VarParamState *parstate = palloc(sizeof(VarParamState)); parstate->paramTypes = paramTypes; parstate->numParams = numParams; + parstate->paramNames = paramNames; + pstate->p_post_columnref_hook = variable_post_column_ref_hook; pstate->p_ref_hook_state = (void *) parstate; pstate->p_paramref_hook = variable_paramref_hook; pstate->p_coerce_param_hook = variable_coerce_param_hook; } +static Node * +variable_post_column_ref_hook(ParseState *pstate, ColumnRef *cref, Node *var) +{ + VarParamState *parstate = (VarParamState *) pstate->p_ref_hook_state; + + /* already resolved */ + if (var != NULL) + return NULL; + + /* did not supply parameter names */ + if (!parstate->paramNames) + return NULL; + + if (list_length(cref->fields) == 1) + { + Node *field1 = (Node *) linitial(cref->fields); + char *name1; + int i; + Param *param; + + Assert(IsA(field1, String)); + name1 = strVal(field1); + for (i = 0; i < *parstate->numParams; i++) + if (strcmp(name1, parstate->paramNames[i]) == 0) + { + param = makeNode(Param); + param->paramkind = PARAM_EXTERN; + param->paramid = i + 1; + param->paramtype = (*parstate->paramTypes)[i]; + param->paramtypmod = -1; + param->paramcollid = InvalidOid; + param->location = -1; + return (Node *) param; + } + } + + return NULL; +} + /* * Transform a ParamRef using fixed parameter types. */ diff --git a/src/backend/po/de.po b/src/backend/po/de.po index 0a9e668c38..d50451850f 100644 --- a/src/backend/po/de.po +++ b/src/backend/po/de.po @@ -28932,6 +28932,10 @@ msgstr "Setzt die maximale Größe der Pending-Liste eines GIN-Index." msgid "TCP user timeout." msgstr "TCP-User-Timeout." +#: utils/misc/guc.c:3432 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "Maximale Lebensdauer der autonomen Sitzung. In wenigen Minuten. standardmäßig 300." + #: utils/misc/guc_tables.c:3442 msgid "The size of huge page that should be requested." msgstr "Huge-Page-Größe, die angefordert werden soll." diff --git a/src/backend/po/es.po b/src/backend/po/es.po index e50a935033..40c1ba6755 100644 --- a/src/backend/po/es.po +++ b/src/backend/po/es.po @@ -28847,6 +28847,12 @@ msgstr "Define el tamaño máximo de la lista de pendientes de un índice GIN." msgid "TCP user timeout." msgstr "Tiempo de expiración de TCP." +#: utils/misc/guc.c:3437 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "Duración máxima de la sesión autónoma. En minutos. 300 por defecto." + +maximale Lebensdauer der autonomen Sitzung. In wenigen Minuten. standardmäßig 300. + #: utils/misc/guc_tables.c:3447 msgid "The size of huge page that should be requested." msgstr "El tamaño de huge page que se debería solicitar." diff --git a/src/backend/po/fr.po b/src/backend/po/fr.po index fd51500b93..9e869e9eb8 100644 --- a/src/backend/po/fr.po +++ b/src/backend/po/fr.po @@ -29145,6 +29145,10 @@ msgstr "Configure la taille maximale de la pending list d'un index GIN." msgid "TCP user timeout." msgstr "Délai d'attente maximal TCP utilisateur." +#: utils/misc/guc.c:3590 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "Durée de vie maximale de la session autonome. En quelques minutes. 300 par défaut." + #: utils/misc/guc.c:3600 msgid "The size of huge page that should be requested." msgstr "La taille du Huge Page devant être réclamé." diff --git a/src/backend/po/id.po b/src/backend/po/id.po index d5d484132b..25f3bb872d 100644 --- a/src/backend/po/id.po +++ b/src/backend/po/id.po @@ -18196,6 +18196,10 @@ msgstr "Melakukan set jumlah maksimum dari temporary buffer yang digunakan pada msgid "Sets the TCP port the server listens on." msgstr "Melakukan set port TCP yang di-listen." +#: utils/misc/guc.c:1674 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "Masa pakai maksimum sesi otonom. Dalam hitungan menit. 300 secara default." + #: utils/misc/guc.c:1683 msgid "Sets the access permissions of the Unix-domain socket." msgstr "Melakukan set pada hak akses dalam Unix-domain socket." diff --git a/src/backend/po/it.po b/src/backend/po/it.po index 673e2aaf00..bca14d3b60 100644 --- a/src/backend/po/it.po +++ b/src/backend/po/it.po @@ -27349,6 +27349,10 @@ msgstr "Imposta la dimensione massima della lista di attesa per gli indici GIN." msgid "TCP user timeout." msgstr "Timeout utente TCP." +#: utils/misc/guc.c:3600 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "Durata massima della sessione autonoma. In pochi minuti. 300 per impostazione predefinita." + #: utils/misc/guc.c:3610 msgid "The size of huge page that should be requested." msgstr "La dimensione della pagina enorme che dovrebbe essere richiesta." diff --git a/src/backend/po/ja.po b/src/backend/po/ja.po index 1ab9f7f68f..ba81385271 100644 --- a/src/backend/po/ja.po +++ b/src/backend/po/ja.po @@ -28083,6 +28083,10 @@ msgstr "GINインデックスの保留リストの最大サイズを設定。" msgid "TCP user timeout." msgstr "TCPユーザータイムアウト。" +#: utils/misc/guc.c:3432 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "自律セッションの最大有効期間。 数分で デフォルトでは300です。" + #: utils/misc/guc_tables.c:3442 msgid "The size of huge page that should be requested." msgstr "要求が見込まれるヒュージページのサイズ。" diff --git a/src/backend/po/ko.po b/src/backend/po/ko.po index f330f3da7e..794b9efb80 100644 --- a/src/backend/po/ko.po +++ b/src/backend/po/ko.po @@ -27679,6 +27679,10 @@ msgstr "GIN 인덱스를 위한 팬딩(pending) 목록의 최대 크기 지정" msgid "TCP user timeout." msgstr "" +#: utils/misc/guc.c:3376 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "자율 세션의 최대 수명. 몇 분 안에 기본적으로 300 입니다." + #: utils/misc/guc.c:3395 msgid "" "Sets the planner's estimate of the cost of a sequentially fetched disk page." diff --git a/src/backend/po/pl.po b/src/backend/po/pl.po index 3ac9d0451c..a91012f165 100644 --- a/src/backend/po/pl.po +++ b/src/backend/po/pl.po @@ -24083,6 +24083,10 @@ msgstr "Ustawia maksymalną liczbę buforów tymczasowych używanych przez każd msgid "Sets the TCP port the server listens on." msgstr "Ustawia port TCP, na którym nasłuchuje serwer." +#: utils/misc/guc.c:1877 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "Maksymalny czas trwania sesji autonomicznej. Za kilka minut. Domyślnie 300." + #: utils/misc/guc.c:1886 msgid "Sets the access permissions of the Unix-domain socket." msgstr "Ustawia uprawnienia dostępu gniazda domeny Uniksa." diff --git a/src/backend/po/pt_BR.po b/src/backend/po/pt_BR.po index 37e4a28f07..dbc268168c 100644 --- a/src/backend/po/pt_BR.po +++ b/src/backend/po/pt_BR.po @@ -19828,6 +19828,10 @@ msgstr "Define o número máximo de buffers temporários utilizados por cada ses msgid "Sets the TCP port the server listens on." msgstr "Define a porta TCP que o servidor escutará." +#: utils/misc/guc.c:1728 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "Vida útil máxima da sessão autônoma. Em minutos. 300 por padrão." + #: utils/misc/guc.c:1737 msgid "Sets the access permissions of the Unix-domain socket." msgstr "Define as permissões de acesso do soquete de domínio Unix." diff --git a/src/backend/po/ru.po b/src/backend/po/ru.po index ae9c50eed7..5c7a0676da 100644 --- a/src/backend/po/ru.po +++ b/src/backend/po/ru.po @@ -30450,7 +30450,7 @@ msgid "Sets the maximum allowed duration of any statement." msgstr "Задаёт предельную длительность для любого оператора." #: utils/misc/guc.c:2646 utils/misc/guc.c:2657 utils/misc/guc.c:2668 -#: utils/misc/guc.c:2679 +#: utils/misc/guc.c:2679 utils/misc/guc.c:3638 msgid "A value of 0 turns off the timeout." msgstr "Нулевое значение отключает тайм-аут." @@ -31043,6 +31043,10 @@ msgstr "Задаёт максимальный размер списка-очер msgid "TCP user timeout." msgstr "Пользовательский таймаут TCP." +#: utils/misc/guc.c:3590 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "Максимальное время жизни автономной сессии. В минутах. По умолчанию 300." + #: utils/misc/guc.c:3600 msgid "The size of huge page that should be requested." msgstr "Запрашиваемый размер огромных страниц." diff --git a/src/backend/po/sv.po b/src/backend/po/sv.po index 0da20b6d43..03be245871 100644 --- a/src/backend/po/sv.po +++ b/src/backend/po/sv.po @@ -29075,6 +29075,10 @@ msgstr "Sätter maximal storlek på väntelistan för GIN-index." msgid "TCP user timeout." msgstr "Användartimeout för TCP." +#: utils/misc/guc.c:3432 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "Maximal livslängd för autonom session. Om några minuter. 300 som standard." + #: utils/misc/guc_tables.c:3442 msgid "The size of huge page that should be requested." msgstr "Storleken på stora sidor skall hämtas." diff --git a/src/backend/po/tr.po b/src/backend/po/tr.po index b791e886b9..4df1ccd67e 100644 --- a/src/backend/po/tr.po +++ b/src/backend/po/tr.po @@ -25187,6 +25187,10 @@ msgstr "GIN indeksi için olan bekleme (pending) listesinin azami boyutunu ayarl msgid "TCP user timeout." msgstr "" +#: utils/misc/guc.c:3432 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "Maksimum özerk oturum ömrü. Dakikalar içinde. varsayılan olarak 300." + #: utils/misc/guc.c:3208 msgid "Sets the planner's estimate of the cost of a sequentially fetched disk page." msgstr "Dıskten sırayla sayfa okuması için harcanacak işlem zamanı tahminini belirtiyor." diff --git a/src/backend/po/uk.po b/src/backend/po/uk.po index 1095fd9139..a1b3a7477c 100644 --- a/src/backend/po/uk.po +++ b/src/backend/po/uk.po @@ -27794,6 +27794,10 @@ msgstr "Встановлює максимальний розмір списку- msgid "TCP user timeout." msgstr "Таймаут користувача TCP." +#: utils/misc/guc.c:3600 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "Максимальний час автономного сеансу. У хвилинах. За замовчуванням 300." + #: utils/misc/guc.c:3610 msgid "The size of huge page that should be requested." msgstr "Розмір величезної сторінки, яку необхідно затребувати." diff --git a/src/backend/po/zh_CN.po b/src/backend/po/zh_CN.po index 574684d775..48baff46fe 100644 --- a/src/backend/po/zh_CN.po +++ b/src/backend/po/zh_CN.po @@ -25514,6 +25514,10 @@ msgstr "为GIN索引设置待处理列表的最大尺寸。" msgid "TCP user timeout." msgstr "TCP用户超时." +#: utils/misc/guc.c:3191 +msgid "Maximum lifetime of autonomous session. In minutes. 300 by default." +msgstr "自治会话的最长生存期。 几分钟后。 默认情况下为300。" + #: utils/misc/guc.c:3210 msgid "Sets the planner's estimate of the cost of a sequentially fetched disk page." msgstr "设置计划器对顺序获取磁盘页的开销估算" diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c index 3c99cf6047..2227586db3 100644 --- a/src/backend/postmaster/bgworker.c +++ b/src/backend/postmaster/bgworker.c @@ -30,6 +30,7 @@ #include "storage/proc.h" #include "storage/procsignal.h" #include "storage/shmem.h" +#include "tcop/autonomous.h" #include "tcop/tcopprot.h" #include "utils/ascii.h" #include "utils/memutils.h" @@ -132,6 +133,9 @@ static const struct { "ParallelApplyWorkerMain", ParallelApplyWorkerMain }, + { + "AutonomousSessionMain", AutonomousSessionMain + }, { "TablesyncWorkerMain", TablesyncWorkerMain } diff --git a/src/backend/tcop/Makefile b/src/backend/tcop/Makefile index f662a7dd1c..b79bf254b7 100644 --- a/src/backend/tcop/Makefile +++ b/src/backend/tcop/Makefile @@ -13,6 +13,7 @@ top_builddir = ../../.. include $(top_builddir)/src/Makefile.global OBJS = \ + autonomous.o \ cmdtag.o \ dest.o \ fastpath.o \ diff --git a/src/backend/tcop/autonomous.c b/src/backend/tcop/autonomous.c new file mode 100644 index 0000000000..97c1e18e37 --- /dev/null +++ b/src/backend/tcop/autonomous.c @@ -0,0 +1,1317 @@ +/*-------------------------------------------------------------------------- + * + * autonomous.c + * Implementation of autonomous transactions + * + * Copyright (C) 2023-2024, PostgreSQL Global Development Group + * + * IDENTIFICATION + * src/backend/tcop/autonomous.c + * + * + * This implements a C API to launch an autonomous session and run SQL queries + * in it. The session looks much like a normal database connection, but it is + * always to the same database, and there is no authentication needed. The + * "backend" for that connection is a background worker. Dynamic shared memory (dsm) + * is used to store fixed data and message queues used for communication between + * normal backend and the autonomous session. They communicate using 2 message queues + * over the Postgres FE/BE protocol. + * Autonomous session's errors are sent to backend. + * Autonomous sessions are taken from the pool, when autonomous function begins to execute. + * Each backend lazily creates pool, pools aren't shared between backends. + * + * + * Structure of dsm: + * +--------------------------------------------+ + * | table of contents (toc) | + * +-----------------Fixed data-----------------+ + * | database_id; | + * | authenticated_user_id; | + * | current_user_id; | + * | sec_context; | + * +-----------------shm_mqs--------------------+ + * | backend -> autonomous session shm_mq | + * | autonomous session -> backend shm_mq | + * +--------------------------------------------+ + * + * WARNING: currently not all statements are fully supported for execution in autonomous sessions. + * Only 1 autonomous session executes everything sequentially. Autonomous sessions + * create each other sequentially recursively if nested calls of autonomous functions. + * If the support for the execution of a statement in an autonomous session is incomplete, + * then more sessions can be taken from the pool. + * ------------------------------------------------------------------------- + */ + +#include "postgres.h" + +#include "access/htup_details.h" +#include "access/tupdesc.h" +#include "access/xact.h" +#include "commands/async.h" +#include "lib/stringinfo.h" +#include "libpq/libpq.h" +#include "libpq/pqformat.h" +#include "libpq/pqmq.h" +#include "mb/pg_wchar.h" +#include "miscadmin.h" +#include "nodes/pg_list.h" +#include "pgstat.h" +#include "postmaster/bgworker.h" +#include "storage/dsm.h" +#include "storage/ipc.h" +#include "storage/shm_mq.h" +#include "storage/shm_toc.h" +#include "tcop/autonomous.h" +#include "tcop/tcopprot.h" +#include "utils/lsyscache.h" +#include "utils/memutils.h" +#include "utils/snapmgr.h" +#include "utils/resowner.h" + + +/* Unique number to identify autonomous table-of-contents */ +#define AUTONOMOUS_MAGIC 0x50674267 + +/* Table-of-contents constants for our dynamic shared memory segment. */ +#define AUTONOMOUS_KEY_FIXED_DATA 0 +#define AUTONOMOUS_KEY_GUC 1 +#define AUTONOMOUS_KEY_COMMAND_QUEUE 2 +#define AUTONOMOUS_KEY_RESPONSE_QUEUE 3 +#define AUTONOMOUS_NKEYS 4 + +/* Size of shared message queues used for communication */ +#define AUTONOMOUS_QUEUE_SIZE 16384 + +/* Pool constants + Under normal conditions, only 1 autonomous session does everything. + They sequentially recursively create each other if nested calls of autonomous functions. + If the support for the execution of a statement in an autonomous session is incomplete, + then more sessions can be taken from the pool. + */ +static const int autonomous_pool_init_capacity = 1; +static const int autonomous_pool_max_capacity = 100; + +/* + * Restart session by lifetime. Session contains resources: memory, caches. + * In order to prevent infinite grow of resources, clean them using session destroying. + */ +int autonomous_session_lifetime = 0; + +/* Fixed-size data passed via dynamic shared memory segment. */ +typedef struct AutonomousSessionFixedData +{ + Oid database_id; + Oid authenticated_user_id; + Oid current_user_id; + int sec_context; +} AutonomousSessionFixedData; + +/* AutonomousResult -- query result */ + +/* Autonomous session handle */ +struct AutonomousSession +{ + dsm_segment *seg; + BackgroundWorkerHandle *worker_handle; + shm_mq_handle *command_qh; + shm_mq_handle *response_qh; + int transaction_status; + TimestampTz creation_time; + bool in_use; +}; + +/* Autonomous session handle when used protocol Prepare-Execute */ +struct AutonomousPreparedStatement +{ + AutonomousSession *session; + Oid *argtypes; + TupleDesc tupdesc; +}; + +/* Pool of autonomous sessions */ +typedef struct AutonomousPool +{ + unsigned capacity; + unsigned size; + unsigned active; + AutonomousSession **sessions; +} AutonomousPool; +AutonomousPool *autonomous_pool = NULL; + +MemoryContext AutonomousContext = NULL; /* NOTE: move to src/backend/utils/mmgr/mcxt.c */ +ResourceOwner AutonomousResourceOwner = NULL; /* NOTE: move to src/backend/utils/resowner/resowner.c */ + +/* + * Stack stores information what autonomous sessions corresponds to each subblock + * if PL/pgSQL function contains subblocks. + * Some subblocks may be autonomous, others not. Info about autonomicity is contained in estate structure. + * This structure is created for each PL/pgSQL function call and passed through exec_stmt_* functions (see pl_exec.c). + * So estate->autonomous will overwritten if several subblocks are autonomous. + * Stack is used to restore estate->autonomous for parent block after child block is finished. + */ +slist_head AutonomousStack = SLIST_STATIC_INIT(AutonomousStack); + +extern void init_row_description_buf(); +extern void (*check_client_encoding_hook)(void); + +static TimestampTz TimestampTzPlusMinutes(TimestampTz tz, int min); + +static void CreateAutonomousResourceOwner(void); +static void BeforeReleaseAutonomousResourcesCallback(int code, Datum arg); +static void ReleaseAutonomousResourcesCallback(int code, Datum arg); +static void ReleaseAutonomousResources(bool isCommit); + +static void AutonomousPoolInit(void); +static void AutonomousPoolDestroy(void); +static AutonomousSession *AutonomousSessionStart(void); +static void AutonomousSessionEnd(AutonomousSession *session); +static void AutonomousSessionEndError(AutonomousSession *session); +static void AutonomousSessionFreeResources(AutonomousSession *session); + +static void shm_mq_receive_stringinfo(shm_mq_handle *qh, StringInfoData *msg); +static void autonomous_check_client_encoding_hook(void); +static TupleDesc TupleDesc_from_RowDescription(StringInfo msg); +static HeapTuple HeapTuple_from_DataRow(TupleDesc tupdesc, StringInfo msg); +static void forward_NotifyResponse(StringInfo msg); +static void rethrow_errornotice(StringInfo msg, int min_elevel); +static void invalid_protocol_message(char msgtype, int phase) pg_attribute_noreturn(); + + +/* + * Add minutes to TimestampTz values + */ +static TimestampTz TimestampTzPlusMinutes(TimestampTz tz, int min) +{ + return tz + ((TimestampTz)min * 60 * 1000 * 1000); +} + +/* + * ------------------------------------------------------------------------- + */ + +/* + * Establish an AutonomousResourceOwner for the current process. + * NOTE: move to src/backend/utils/resowner/resowner.c + */ +static void +CreateAutonomousResourceOwner(void) +{ + Assert(AutonomousResourceOwner == NULL); + AutonomousResourceOwner = ResourceOwnerCreate(NULL, "Autonomous"); + /* + * Register a shmem-exit callback for cleanup of autonomous-process resource + * owner. (This needs to run after, e.g., ShutdownXLOG.) + */ + before_shmem_exit(BeforeReleaseAutonomousResourcesCallback, (Datum)0); + on_shmem_exit(ReleaseAutonomousResourcesCallback, (Datum)0); +} + +/* + * Convenience routine to release all resources tracked in + * AutonomousResourceOwner (but that resowner is not destroyed here). + * Warn about leaked resources if isCommit is true. + * NOTE: move to src/backend/utils/resowner/resowner.c + */ +static void +ReleaseAutonomousResources(bool isCommit) +{ + /* + * At this writing, the only thing that could actually get released is + * autonomous pool; but we may as well do the full release protocol. + */ + ResourceOwnerRelease(AutonomousResourceOwner, + RESOURCE_RELEASE_BEFORE_LOCKS, + isCommit, true); + ResourceOwnerRelease(AutonomousResourceOwner, + RESOURCE_RELEASE_LOCKS, + isCommit, true); + ResourceOwnerRelease(AutonomousResourceOwner, + RESOURCE_RELEASE_AFTER_LOCKS, + isCommit, true); +} + +/* + * Shmem-exit callback, frees resources. + * Warn about leaked resources if process exit code is zero (ie normal). + * NOTE: move to src/backend/utils/resowner/resowner.c + */ +static void +BeforeReleaseAutonomousResourcesCallback(int code, Datum arg) +{ + AutonomousPoolDestroy(); +} + +/* + * NOTE: move to src/backend/utils/resowner/resowner.c + */ +static void +ReleaseAutonomousResourcesCallback(int code, Datum arg) +{ + bool isCommit = (code == 0); + ReleaseAutonomousResources(isCommit); +} + +/* + * ------------------------------------------------------------------------- + */ + +/* + * Initialize pool of autonomous sessions + */ +static void +AutonomousPoolInit(void) +{ + CreateAutonomousResourceOwner(); + AutonomousContext = AllocSetContextCreate(TopMemoryContext, + "AutonomousContext", + ALLOCSET_DEFAULT_SIZES); + autonomous_pool = (AutonomousPool *) MemoryContextAlloc(AutonomousContext, sizeof(AutonomousPool)); + autonomous_pool->capacity = autonomous_pool_init_capacity; + autonomous_pool->size = 0; + autonomous_pool->active = 0; + autonomous_pool->sessions = (AutonomousSession **) MemoryContextAlloc(AutonomousContext, + autonomous_pool->capacity * sizeof(AutonomousSession *)); +} + +/* + * Pop autonomous session from stack of subblocks after child subblock is finished + */ +AutonomousSession * +AutonomousSessionPopStackSession(bool isBlockAutonomous) +{ + AutonomousSession *session = NULL; + /* Pop PLpgSQL_stmt_block from stack */ + slist_node *node = slist_pop_head_node(&AutonomousStack); + if (isBlockAutonomous) + { + /* End Autonomous session */ + slist_iter iter; + session = slist_container(AutonomousStackNode, node, node)->session; + AutonomousSessionRelease(session); + slist_foreach(iter, &AutonomousStack) + { + session = slist_container(AutonomousStackNode, node, iter.cur)->session; + if(session) { + break; + } + } + } + pfree(node); + return session; +} + +/* + * Unwind stack of autonomous sessions after of SQL-exception. + * After that return first found autonomous session + */ +AutonomousSession * +AutonomousSessionPopStackSessionException(void *block) +{ + slist_node *node = NULL; + void *stack_block = NULL; + AutonomousSession *session = NULL; + slist_iter iter; + + /* Pop PLpgSQL_stmt_blocks from stack after exception + * Also end corresponding Autonomous sessions + */ + while(!slist_is_empty(&AutonomousStack)) { + node = slist_head_node(&AutonomousStack); + stack_block = slist_container(AutonomousStackNode, node, node)->block; + if(stack_block == block) { + break; + } + node = slist_pop_head_node(&AutonomousStack); + session = slist_container(AutonomousStackNode, node, node)->session; + if(session != NULL) { + AutonomousSessionRelease(session); + } + pfree(node); + } + + /* Return first found autonomous session + * Needed for subblocks + */ + slist_foreach(iter, &AutonomousStack) { + session = slist_container(AutonomousStackNode, node, iter.cur)->session; + if(session) { + return session; + } + } + return NULL; +} + +/* + * Destroy pool normally + */ +static void +AutonomousPoolDestroy(void) +{ + if(!autonomous_pool) { + return; + } + for(unsigned i = 0; i < autonomous_pool->size; ++i) { + AutonomousSessionEnd(autonomous_pool->sessions[i]); + } + pfree(autonomous_pool->sessions); + pfree(autonomous_pool); + autonomous_pool = NULL; +} + +/* + * Destroy pool in case of ERROR + */ +void +AutonomousPoolDestroyError(void) +{ + if(!autonomous_pool) { + return; + } + for(unsigned i = 0; i < autonomous_pool->size; ++i) { + AutonomousSessionEndError(autonomous_pool->sessions[i]); + } + pfree(autonomous_pool->sessions); + pfree(autonomous_pool); + autonomous_pool = NULL; +} + +/* + * Get autonomous session from pool + */ +AutonomousSession * +AutonomousSessionGet(void) +{ + TimestampTz curr_time = GetCurrentTimestamp(); + if(!autonomous_pool) { + AutonomousPoolInit(); + } + + /* destroy not used sessions by timeout */ + for(unsigned i = 0;;) + { + if(i >= autonomous_pool->size) + { + break; + } + + if(autonomous_session_lifetime) + { + TimestampTz death_time = TimestampTzPlusMinutes(autonomous_pool->sessions[i]->creation_time, autonomous_session_lifetime); + if(unlikely(death_time < curr_time) && !autonomous_pool->sessions[i]->in_use) + { + AutonomousSessionEnd(autonomous_pool->sessions[i]); + autonomous_pool->sessions[i] = autonomous_pool->sessions[autonomous_pool->size - 1]; + --autonomous_pool->size; + } else { + ++i; + } + } + } + + /* return 1st not used session */ + for (unsigned i = 0; i < autonomous_pool->size; ++i) + { + if(!autonomous_pool->sessions[i]->in_use) { + autonomous_pool->sessions[i]->in_use = true; + ++autonomous_pool->active; + return autonomous_pool->sessions[i]; + } + } + + /* no free sessions. Need to start new or resize a pool at first */ + if(unlikely(autonomous_pool->size == autonomous_pool_max_capacity)) + { + elog(ERROR, "No free autonomous sessions. Max capacity=%d of pool is reached", autonomous_pool_max_capacity); + } + + /* resize pool if it's full */ + if (autonomous_pool->size == autonomous_pool->capacity) + { + unsigned new_capacity = autonomous_pool->capacity * 2; + if(new_capacity > autonomous_pool_max_capacity) + new_capacity = autonomous_pool_max_capacity; + autonomous_pool->capacity = new_capacity; + autonomous_pool->sessions = (AutonomousSession **) + repalloc(autonomous_pool->sessions, autonomous_pool->capacity * sizeof(AutonomousSession *)); + if(!autonomous_pool->sessions) { + elog(ERROR, "Autonomous pool can't resize"); + } + elog(NOTICE, "Autonomous pool was resized, new capacity is %u", autonomous_pool->capacity); + } + + /* create new session */ + { + unsigned i = autonomous_pool->size++; + autonomous_pool->sessions[i] = AutonomousSessionStart(); + ++autonomous_pool->active; + autonomous_pool->sessions[i]->in_use = true; + autonomous_pool->sessions[i]->creation_time = GetCurrentTimestamp(); + return autonomous_pool->sessions[i]; + } +} + +/* + * Return autonomous session to pool + */ +void +AutonomousSessionRelease(AutonomousSession *session) +{ + if(!session) { + elog(ERROR, "Autonomous session is NULL"); + } + if (session->in_use) { + session->in_use = false; + --autonomous_pool->active; + } +} + +/* + * Start autonomous session: allocate memory for dsm, shm_mq, start background worker, etc. + * and return a handle. Launches background worker + */ +static AutonomousSession * +AutonomousSessionStart(void) +{ + BackgroundWorker worker; + pid_t pid; + AutonomousSession *session; + shm_toc_estimator e; + Size segsize; + Size guc_len; + char *gucstate; + dsm_segment *seg; + shm_toc *toc; + AutonomousSessionFixedData *fdata; + shm_mq *command_mq; + shm_mq *response_mq; + BgwHandleStatus bgwstatus; + StringInfoData msg; + char msgtype; + MemoryContext save_ctx; + ResourceOwner save_owner; + + shm_toc_initialize_estimator(&e); + shm_toc_estimate_chunk(&e, sizeof(AutonomousSessionFixedData)); + shm_toc_estimate_chunk(&e, AUTONOMOUS_QUEUE_SIZE); + shm_toc_estimate_chunk(&e, AUTONOMOUS_QUEUE_SIZE); + guc_len = EstimateGUCStateSpace(); + shm_toc_estimate_chunk(&e, guc_len); + shm_toc_estimate_keys(&e, AUTONOMOUS_NKEYS); + segsize = shm_toc_estimate(&e); + + save_ctx = MemoryContextSwitchTo(AutonomousContext); /* XXX move to func start? */ + session = palloc(sizeof(AutonomousSession)); + session->in_use = false; + save_owner = CurrentResourceOwner; + CurrentResourceOwner = AutonomousResourceOwner; + seg = dsm_create(segsize, 0); + CurrentResourceOwner = save_owner; + session->seg = seg; + + toc = shm_toc_create(AUTONOMOUS_MAGIC, dsm_segment_address(seg), segsize); + + /* Store fixed-size data in dynamic shared memory. */ + fdata = shm_toc_allocate(toc, sizeof(*fdata)); + fdata->database_id = MyDatabaseId; + fdata->authenticated_user_id = GetAuthenticatedUserId(); + GetUserIdAndSecContext(&fdata->current_user_id, &fdata->sec_context); + shm_toc_insert(toc, AUTONOMOUS_KEY_FIXED_DATA, fdata); + + /* Store GUC state in dynamic shared memory. */ + gucstate = shm_toc_allocate(toc, guc_len); + SerializeGUCState(guc_len, gucstate); + shm_toc_insert(toc, AUTONOMOUS_KEY_GUC, gucstate); + + command_mq = shm_mq_create(shm_toc_allocate(toc, AUTONOMOUS_QUEUE_SIZE), + AUTONOMOUS_QUEUE_SIZE); + shm_toc_insert(toc, AUTONOMOUS_KEY_COMMAND_QUEUE, command_mq); + shm_mq_set_sender(command_mq, MyProc); + + response_mq = shm_mq_create(shm_toc_allocate(toc, AUTONOMOUS_QUEUE_SIZE), + AUTONOMOUS_QUEUE_SIZE); + shm_toc_insert(toc, AUTONOMOUS_KEY_RESPONSE_QUEUE, response_mq); + shm_mq_set_receiver(response_mq, MyProc); + + session->command_qh = shm_mq_attach(command_mq, seg, NULL); + session->response_qh = shm_mq_attach(response_mq, seg, NULL); + + MemoryContextSwitchTo(save_ctx); + + worker.bgw_flags = + BGWORKER_SHMEM_ACCESS | BGWORKER_BACKEND_DATABASE_CONNECTION; + worker.bgw_start_time = BgWorkerStart_ConsistentState; + worker.bgw_restart_time = BGW_NEVER_RESTART; + sprintf(worker.bgw_library_name, "postgres"); + sprintf(worker.bgw_function_name, "AutonomousSessionMain"); + sprintf(worker.bgw_type, "autonomous_transaction"); + snprintf(worker.bgw_name, BGW_MAXLEN, "autonomous session by PID %d", MyProcPid); + worker.bgw_main_arg = UInt32GetDatum(dsm_segment_handle(seg)); + worker.bgw_notify_pid = MyProcPid; + + if (!RegisterDynamicBackgroundWorker(&worker, &session->worker_handle)) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_RESOURCES), + errmsg("could not register background process"), + errhint("You might need to increase max_worker_processes."))); + + shm_mq_set_handle(session->command_qh, session->worker_handle); + shm_mq_set_handle(session->response_qh, session->worker_handle); + + bgwstatus = WaitForBackgroundWorkerStartup(session->worker_handle, &pid); + if (bgwstatus != BGWH_STARTED) + ereport(ERROR, + (errcode(ERRCODE_INSUFFICIENT_RESOURCES), + errmsg("could not start background worker"))); + + do + { + shm_mq_receive_stringinfo(session->response_qh, &msg); + msgtype = pq_getmsgbyte(&msg); + + switch (msgtype) + { + case 'Z': /* ReadyForQuery */ + session->transaction_status = pq_getmsgbyte(&msg); + pq_getmsgend(&msg); + break; + default: + invalid_protocol_message(msgtype, 1); + break; + case 'N': /* NoticeResponse */ + rethrow_errornotice(&msg, NOTICE); + break; + case 'E': /* ErrorResponse */ + rethrow_errornotice(&msg, ERROR); + break; + } + } + while (msgtype != 'Z'); + + return session; +} + +/* + * Destroy autonomous session normally + */ +static void +AutonomousSessionEnd(AutonomousSession *session) +{ + StringInfoData msg; + if (session->transaction_status == 'T') + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("autonomous session ended with transaction block open"))); + + pq_redirect_to_shm_mq(session->seg, session->command_qh); + pq_beginmessage(&msg, 'X'); /* Terminate */ + pq_endmessage(&msg); + pq_stop_redirect_to_shm_mq(); + + AutonomousSessionFreeResources(session); +} + +/* + * Destroy autonomous session in case of ERROR + */ +static void +AutonomousSessionEndError(AutonomousSession *session) +{ + StringInfoData msg; + pq_redirect_to_shm_mq(session->seg, session->command_qh); + pq_beginmessage(&msg, 'X'); /* Terminate */ + pq_endmessage(&msg); + pq_stop_redirect_to_shm_mq(); + + AutonomousSessionFreeResources(session); +} + +/* + * Free memory of internal structures of autonomous session + */ +static void +AutonomousSessionFreeResources(AutonomousSession *session) +{ + pfree(session->worker_handle); + dsm_detach(session->seg); + pfree(session); +} + +AutonomousResult * +AutonomousSessionExecute(AutonomousSession *session, const char *sql) +{ + StringInfoData msg; + char msgtype; + AutonomousResult *result; + + pq_redirect_to_shm_mq(session->seg, session->command_qh); + pq_beginmessage(&msg, 'Q'); /* Query */ + pq_sendstring(&msg, sql); + pq_endmessage(&msg); + pq_stop_redirect_to_shm_mq(); + + result = palloc0(sizeof(*result)); + + do + { + shm_mq_receive_stringinfo(session->response_qh, &msg); + msgtype = pq_getmsgbyte(&msg); + + switch (msgtype) + { + case 'C': /* CommandComplete */ + { + const char *tag = pq_getmsgstring(&msg); + result->command = pstrdup(tag); + pq_getmsgend(&msg); + break; + } + case 'T': /* RowDescription */ + if (result->tupdesc) + elog(ERROR, "already received a T message"); + result->tupdesc = TupleDesc_from_RowDescription(&msg); + pq_getmsgend(&msg); + break; + case 'D': /* Describe */ + if (!result->tupdesc) + elog(ERROR, "no T message before D"); + result->tuples = lappend(result->tuples, HeapTuple_from_DataRow(result->tupdesc, &msg)); + pq_getmsgend(&msg); + break; + case 'Z': /* ReadyForQuery */ + session->transaction_status = pq_getmsgbyte(&msg); + pq_getmsgend(&msg); + break; + case 'A': /* NotificationResponse */ + forward_NotifyResponse(&msg); + break; + case 'N': /* NoticeResponse */ + rethrow_errornotice(&msg, NOTICE); + break; + case 'E': /* ErrorResponse */ + rethrow_errornotice(&msg, ERROR); + break; + default: + invalid_protocol_message(msgtype, 2); + break; + } + } + while (msgtype != 'Z'); + + return result; +} + +/* + * Prepare an SQL string for subsequent execution + */ +AutonomousPreparedStatement * +AutonomousSessionPrepare(AutonomousSession *session, const char *sql, int16 nargs, + Oid argtypes[], const char *argnames[]) +{ + AutonomousPreparedStatement *result; + StringInfoData msg; + int16 i; + char msgtype; + + pq_redirect_to_shm_mq(session->seg, session->command_qh); + pq_beginmessage(&msg, 'P'); /* Parse */ + pq_sendstring(&msg, ""); + pq_sendstring(&msg, sql); + pq_sendint(&msg, nargs, 2); + for (i = 0; i < nargs; i++) + pq_sendint(&msg, argtypes[i], 4); + if (argnames) + for (i = 0; i < nargs; i++) + pq_sendstring(&msg, argnames[i]); + pq_endmessage(&msg); + pq_stop_redirect_to_shm_mq(); + + result = palloc0(sizeof(*result)); + result->session = session; + result->argtypes = palloc(nargs * sizeof(*result->argtypes)); + memcpy(result->argtypes, argtypes, nargs * sizeof(*result->argtypes)); + + do + { + shm_mq_receive_stringinfo(session->response_qh, &msg); + msgtype = pq_getmsgbyte(&msg); + + switch (msgtype) + { + case '1': /* ParseComplete */ + break; + case 'N': /* NoticeResponse */ + rethrow_errornotice(&msg, NOTICE); + break; + case 'E': /* ErrorResponse */ + rethrow_errornotice(&msg, ERROR); + break; + default: + invalid_protocol_message(msgtype, 3); + break; + } + } + while (msgtype != '1'); + + pq_redirect_to_shm_mq(session->seg, session->command_qh); + pq_beginmessage(&msg, 'D'); /* Describe */ + pq_sendbyte(&msg, 'S'); /* PreparedStatement */ + pq_sendstring(&msg, ""); + pq_endmessage(&msg); + pq_stop_redirect_to_shm_mq(); + + do + { + shm_mq_receive_stringinfo(session->response_qh, &msg); + msgtype = pq_getmsgbyte(&msg); + + switch (msgtype) + { + case 'n': /* NoData */ + break; + case 't': /* ParameterDescription */ + case '1': /* ParseComplete */ + case 'Z': /* ReadyForQuery */ + /* ignore for now */ + break; + case 'T': /* RowDescription */ + if (result->tupdesc) + elog(ERROR, "already received a T message"); + result->tupdesc = TupleDesc_from_RowDescription(&msg); + pq_getmsgend(&msg); + break; + case 'A': /* NotificationResponse */ + forward_NotifyResponse(&msg); + break; + case 'N': /* NoticeResponse */ + rethrow_errornotice(&msg, NOTICE); + break; + case 'E': /* ErrorResponse */ + rethrow_errornotice(&msg, ERROR); + break; + default: + invalid_protocol_message(msgtype, 4); + break; + } + } + while (msgtype != 'n' && msgtype != 'T'); + + return result; +} + +/* + * Execute prepared statement + */ +AutonomousResult * +AutonomousSessionExecutePrepared(AutonomousPreparedStatement *stmt, int16 nargs, Datum *values, bool *nulls) +{ + AutonomousSession *session; + StringInfoData msg; + AutonomousResult *result; + char msgtype; + int16 i; + + session = stmt->session; + + pq_redirect_to_shm_mq(session->seg, session->command_qh); + pq_beginmessage(&msg, 'B'); /* Bind */ + pq_sendstring(&msg, ""); + pq_sendstring(&msg, ""); + pq_sendint(&msg, 1, 2); /* number of parameter format codes */ + pq_sendint(&msg, 1, 2); + pq_sendint(&msg, nargs, 2); /* number of parameter values */ + for (i = 0; i < nargs; i++) + { + if (nulls[i]) + pq_sendint(&msg, -1, 4); + else + { + Oid typsend; + bool typisvarlena; + bytea *outputbytes; + + getTypeBinaryOutputInfo(stmt->argtypes[i], &typsend, &typisvarlena); + outputbytes = OidSendFunctionCall(typsend, values[i]); + pq_sendint(&msg, VARSIZE(outputbytes) - VARHDRSZ, 4); + pq_sendbytes(&msg, VARDATA(outputbytes), VARSIZE(outputbytes) - VARHDRSZ); + pfree(outputbytes); + } + } + pq_sendint(&msg, 1, 2); /* number of result column format codes */ + pq_sendint(&msg, 1, 2); + pq_endmessage(&msg); + pq_stop_redirect_to_shm_mq(); + + do + { + shm_mq_receive_stringinfo(session->response_qh, &msg); + msgtype = pq_getmsgbyte(&msg); + + switch (msgtype) + { + case '2': /* BindComplete */ + break; + case 'N': /* NoticeResponse */ + rethrow_errornotice(&msg, NOTICE); + break; + case 'E': /* ErrorResponse */ + rethrow_errornotice(&msg, ERROR); + break; + default: + invalid_protocol_message(msgtype, 5); + break; + } + } + while (msgtype != '2'); + + pq_redirect_to_shm_mq(session->seg, session->command_qh); + pq_beginmessage(&msg, 'E'); /* Execute */ + pq_sendstring(&msg, ""); + pq_sendint(&msg, 0, 4); + pq_endmessage(&msg); + pq_stop_redirect_to_shm_mq(); + + result = palloc0(sizeof(*result)); + result->tupdesc = stmt->tupdesc; + + do + { + shm_mq_receive_stringinfo(session->response_qh, &msg); + msgtype = pq_getmsgbyte(&msg); + + switch (msgtype) + { + case '2': /* BindComplete */ + break; + case 'C': /* CommandComplete */ + { + const char *tag = pq_getmsgstring(&msg); + result->command = pstrdup(tag); + pq_getmsgend(&msg); + break; + } + case 'D': /* Describe */ + if (!stmt->tupdesc) + elog(ERROR, "did not expect any description rows"); + result->tuples = lappend(result->tuples, HeapTuple_from_DataRow(stmt->tupdesc, &msg)); + pq_getmsgend(&msg); + break; + case 'A': /* NotificationResponse */ + forward_NotifyResponse(&msg); + break; + case 'N': /* NoticeResponse */ + rethrow_errornotice(&msg, NOTICE); + break; + case 'E': /* ErrorResponse */ + rethrow_errornotice(&msg, ERROR); + break; + default: + invalid_protocol_message(msgtype, 6); + break; + } + } + while (msgtype != 'C'); + + pq_redirect_to_shm_mq(session->seg, session->command_qh); + pq_putemptymessage('S'); /* Sync */ + pq_stop_redirect_to_shm_mq(); + + do + { + shm_mq_receive_stringinfo(session->response_qh, &msg); + msgtype = pq_getmsgbyte(&msg); + + switch (msgtype) + { + case 'Z': /* ReadyForQuery */ + session->transaction_status = pq_getmsgbyte(&msg); + pq_getmsgend(&msg); + break; + case 'A': /* NotificationResponse */ + forward_NotifyResponse(&msg); + break; + case 'N': /* NoticeResponse */ + rethrow_errornotice(&msg, NOTICE); + break; + case 'E': /* ErrorResponse */ + rethrow_errornotice(&msg, ERROR); + break; + default: + invalid_protocol_message(msgtype, 7); + break; + } + } + while (msgtype != 'Z'); + + return result; +} + +/* + * Main loop of autonomous session + */ +void +AutonomousSessionMain(Datum main_arg) +{ + dsm_segment *seg; + shm_toc *toc; + AutonomousSessionFixedData *fdata; + char *gucstate; + shm_mq *command_mq; + shm_mq *response_mq; + shm_mq_handle *command_qh; + shm_mq_handle *response_qh; + StringInfoData msg; + char msgtype; + + pqsignal(SIGTERM, die); + + BackgroundWorkerUnblockSignals(); + + /* Set up a memory context and resource owner. */ + Assert(CurrentResourceOwner == NULL); + CurrentResourceOwner = ResourceOwnerCreate(NULL, "autonomous"); + CurrentMemoryContext = AllocSetContextCreate(TopMemoryContext, + "autonomous session", + ALLOCSET_DEFAULT_MINSIZE, + ALLOCSET_DEFAULT_INITSIZE, + ALLOCSET_DEFAULT_MAXSIZE); + + init_row_description_buf(); + + seg = dsm_attach(DatumGetInt32(main_arg)); + if (seg == NULL) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("could not map dynamic shared memory segment"))); + + toc = shm_toc_attach(AUTONOMOUS_MAGIC, dsm_segment_address(seg)); + if (toc == NULL) + ereport(ERROR, + (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE), + errmsg("bad magic number in dynamic shared memory segment"))); + + /* Find data structures in dynamic shared memory. */ + fdata = shm_toc_lookup(toc, AUTONOMOUS_KEY_FIXED_DATA, false); + + gucstate = shm_toc_lookup(toc, AUTONOMOUS_KEY_GUC, false); + + command_mq = shm_toc_lookup(toc, AUTONOMOUS_KEY_COMMAND_QUEUE, false); + shm_mq_set_receiver(command_mq, MyProc); + command_qh = shm_mq_attach(command_mq, seg, NULL); + + response_mq = shm_toc_lookup(toc, AUTONOMOUS_KEY_RESPONSE_QUEUE, false); + shm_mq_set_sender(response_mq, MyProc); + response_qh = shm_mq_attach(response_mq, seg, NULL); + + pq_redirect_to_shm_mq(seg, response_qh); + BackgroundWorkerInitializeConnectionByOid(fdata->database_id, + fdata->authenticated_user_id, 0); + SetClientEncoding(GetDatabaseEncoding()); + + StartTransactionCommand(); + RestoreGUCState(gucstate); + CommitTransactionCommand(); + + process_session_preload_libraries(); + + SetUserIdAndSecContext(fdata->current_user_id, fdata->sec_context); + + whereToSendOutput = DestRemote; + ReadyForQuery(whereToSendOutput); /* Z ReadyForQuery*/ + + MessageContext = AllocSetContextCreate(TopMemoryContext, + "MessageContext", + ALLOCSET_DEFAULT_MINSIZE, + ALLOCSET_DEFAULT_INITSIZE, + ALLOCSET_DEFAULT_MAXSIZE); + + do + { + MemoryContextSwitchTo(MessageContext); + MemoryContextReset(MessageContext); + InvalidateCatalogSnapshotConditionally(); + + pgstat_report_stat(false); + pgstat_report_activity(STATE_IDLE, NULL); + + shm_mq_receive_stringinfo(command_qh, &msg); + msgtype = pq_getmsgbyte(&msg); + + switch (msgtype) + { + case 'B': + { + SetCurrentStatementStartTimestamp(); + exec_bind_message(&msg); + break; + } + case 'D': /* Describe */ + { + int describe_type; + const char *describe_target; + + SetCurrentStatementStartTimestamp(); + + describe_type = pq_getmsgbyte(&msg); + describe_target = pq_getmsgstring(&msg); + pq_getmsgend(&msg); + + switch (describe_type) + { + case 'S': /* PreparedStatement */ + exec_describe_statement_message(describe_target); + break; +#ifdef XXX + case 'P': /* Portal */ + exec_describe_portal_message(describe_target); + break; +#endif + default: + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("invalid DESCRIBE message subtype %d", + describe_type))); + break; + } + } + break; + case 'E': /* Execute */ + { + const char *portal_name; + int max_rows; + + SetCurrentStatementStartTimestamp(); + + portal_name = pq_getmsgstring(&msg); + max_rows = pq_getmsgint(&msg, 4); + pq_getmsgend(&msg); + + exec_execute_message(portal_name, max_rows); + } + break; + + case 'P': /* Parse */ + { + const char *stmt_name; + const char *query_string; + int numParams; + Oid *paramTypes = NULL; + const char **paramNames = NULL; + + SetCurrentStatementStartTimestamp(); + + stmt_name = pq_getmsgstring(&msg); + query_string = pq_getmsgstring(&msg); + numParams = pq_getmsgint(&msg, 2); + if (numParams > 0) + { + int i; + + paramTypes = palloc(numParams * sizeof(Oid)); + for (i = 0; i < numParams; i++) + paramTypes[i] = pq_getmsgint(&msg, 4); + } + /* If data left in message, read parameter names. */ + if (msg.cursor != msg.len) + { + int i; + + paramNames = palloc(numParams * sizeof(char *)); + for (i = 0; i < numParams; i++) + paramNames[i] = pq_getmsgstring(&msg); + } + pq_getmsgend(&msg); + + exec_parse_message(query_string, stmt_name, + paramTypes, numParams, paramNames); + break; + } + case 'Q': /* Query */ + { + const char *sql; + int save_log_statement; + bool save_log_duration; + int save_log_min_duration_statement; + + sql = pq_getmsgstring(&msg); + pq_getmsgend(&msg); + + /* XXX room for improvement */ + save_log_statement = log_statement; + save_log_duration = log_duration; + save_log_min_duration_statement = log_min_duration_statement; + + check_client_encoding_hook = autonomous_check_client_encoding_hook; + log_statement = LOGSTMT_NONE; + log_duration = false; + log_min_duration_statement = -1; + + SetCurrentStatementStartTimestamp(); + exec_simple_query(sql, 1); + + log_statement = save_log_statement; + log_duration = save_log_duration; + log_min_duration_statement = save_log_min_duration_statement; + check_client_encoding_hook = NULL; + + ReadyForQuery(whereToSendOutput); + break; + } + case 'S': /* Sync */ + { + pq_getmsgend(&msg); + finish_xact_command(); + ReadyForQuery(whereToSendOutput); + break; + } + case 'X': /* Terminate */ + break; + default: + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("invalid protocol message type from autonomous session leader: %c", + msgtype))); + break; + } + } + while (msgtype != 'X'); +} + + +static void +shm_mq_receive_stringinfo(shm_mq_handle *qh, StringInfoData *msg) +{ + Size nbytes; + void *data; + + shm_mq_result res = shm_mq_receive(qh, &nbytes, &data, false); + if (res != SHM_MQ_SUCCESS) + elog(ERROR, "shm_mq_receive failed: %d", res); + + initStringInfo(msg); + appendBinaryStringInfo(msg, data, nbytes); +} + + +static void +autonomous_check_client_encoding_hook(void) +{ + elog(ERROR, "cannot set client encoding in autonomous session"); +} + + +static TupleDesc +TupleDesc_from_RowDescription(StringInfo msg) +{ + int16 natts = pq_getmsgint(msg, 2); + TupleDesc tupdesc = CreateTemplateTupleDesc(natts); + + for (int16 i = 0; i < natts; i++) + { + const char *colname; + Oid type_oid; + int32 typmod; + int16 format; + + colname = pq_getmsgstring(msg); + (void) pq_getmsgint(msg, 4); /* table OID */ + (void) pq_getmsgint(msg, 2); /* table attnum */ + type_oid = pq_getmsgint(msg, 4); + (void) pq_getmsgint(msg, 2); /* type length */ + typmod = pq_getmsgint(msg, 4); + format = pq_getmsgint(msg, 2); + (void) format; + /* XXX The protocol sometimes sends 0 (text) if the format is not + * determined yet. Fix why it can't be determined. + * We always use binary, 1. + */ + + TupleDescInitEntry(tupdesc, i + 1, colname, type_oid, typmod, 0); + } + return tupdesc; +} + + +static HeapTuple +HeapTuple_from_DataRow(TupleDesc tupdesc, StringInfo msg) +{ + int16 natts = pq_getmsgint(msg, 2); + Datum *values; + bool *nulls; + StringInfoData buf; + + Assert(tupdesc); + + if (natts != tupdesc->natts) + elog(ERROR, "malformed DataRow"); + + values = palloc(natts * sizeof(*values)); + nulls = palloc(natts * sizeof(*nulls)); + initStringInfo(&buf); + + for (int16 i = 0; i < natts; i++) + { + int32 len = pq_getmsgint(msg, 4); + + if (len < 0) + nulls[i] = true; + else + { + Oid recvid; + Oid typioparams; + + nulls[i] = false; + + getTypeBinaryInputInfo(tupdesc->attrs[i].atttypid, + &recvid, + &typioparams); + resetStringInfo(&buf); + appendBinaryStringInfo(&buf, pq_getmsgbytes(msg, len), len); + values[i] = OidReceiveFunctionCall(recvid, &buf, typioparams, + tupdesc->attrs[i].atttypmod); + } + } + + return heap_form_tuple(tupdesc, values, nulls); +} + + +static void +forward_NotifyResponse(StringInfo msg) +{ + int32 pid; + const char *channel; + const char *payload; + + pid = pq_getmsgint(msg, 4); + channel = pq_getmsgrawstring(msg); + payload = pq_getmsgrawstring(msg); + pq_endmessage(msg); + + NotifyMyFrontEnd(channel, payload, pid); +} + +static void +rethrow_errornotice(StringInfo msg, int min_elevel) +{ + ErrorData edata; + + pq_parse_errornotice(msg, &edata); + edata.elevel = Min(edata.elevel, min_elevel); + if(edata.elevel >= ERROR) { + slist_node *node = slist_pop_head_node(&AutonomousStack); + AutonomousSession *session = slist_container(AutonomousStackNode, node, node)->session; + AutonomousSessionRelease(session); + pfree(node); + } + ThrowErrorData(&edata); +} + + +static void +invalid_protocol_message(char msgtype, int phase) +{ + ereport(ERROR, + (errcode(ERRCODE_PROTOCOL_VIOLATION), + errmsg("invalid protocol message type %c from autonomous session during phase %d", + msgtype, phase))); +} diff --git a/src/backend/tcop/postgres.c b/src/backend/tcop/postgres.c index 7298a187d1..44a809d4c5 100644 --- a/src/backend/tcop/postgres.c +++ b/src/backend/tcop/postgres.c @@ -67,6 +67,7 @@ #include "storage/proc.h" #include "storage/procsignal.h" #include "storage/sinval.h" +#include "tcop/autonomous.h" #include "tcop/fastpath.h" #include "tcop/pquery.h" #include "tcop/tcopprot.h" @@ -169,6 +170,24 @@ static volatile sig_atomic_t RecoveryConflictPendingReasons[NUM_PROCSIGNALS]; static MemoryContext row_description_context = NULL; static StringInfoData row_description_buf; +extern void init_row_description_buf(void); /* keep patch minimal */ +void init_row_description_buf() +{ + MemoryContext old_context = CurrentMemoryContext; + /* + * Create memory context and buffer used for RowDescription messages. As + * SendRowDescriptionMessage(), via exec_describe_statement_message(), is + * frequently executed for ever single statement, we don't want to + * allocate a separate buffer every time. + */ + row_description_context = AllocSetContextCreate(TopMemoryContext, + "RowDescriptionContext", + ALLOCSET_DEFAULT_SIZES); + MemoryContextSwitchTo(row_description_context); + initStringInfo(&row_description_buf); + MemoryContextSwitchTo(old_context); +} + /* ---------------------------------------------------------------- * decls for routines only used in this file * ---------------------------------------------------------------- @@ -183,8 +202,8 @@ static int errdetail_execute(List *raw_parsetree_list); static int errdetail_params(ParamListInfo params); static int errdetail_abort(void); static void bind_param_error_callback(void *arg); -static void start_xact_command(void); -static void finish_xact_command(void); +void start_xact_command(void); +void finish_xact_command(void); static bool IsTransactionExitStmt(Node *parsetree); static bool IsTransactionExitStmtList(List *pstmts); static bool IsTransactionStmtList(List *pstmts); @@ -710,7 +729,7 @@ pg_analyze_and_rewrite_varparams(RawStmt *parsetree, const char *query_string, Oid **paramTypes, int *numParams, - QueryEnvironment *queryEnv) + QueryEnvironment *queryEnv, const char *paramNames[]) { Query *query; List *querytree_list; @@ -724,7 +743,7 @@ pg_analyze_and_rewrite_varparams(RawStmt *parsetree, ResetUsage(); query = parse_analyze_varparams(parsetree, query_string, paramTypes, numParams, - queryEnv); + queryEnv, paramNames); /* * Check all parameter types got determined. @@ -1008,8 +1027,8 @@ pg_plan_queries(List *querytrees, const char *query_string, int cursorOptions, * * Execute a "simple Query" protocol message. */ -static void -exec_simple_query(const char *query_string) +void +exec_simple_query(const char *query_string, int16 format) { CommandDest dest = whereToSendOutput; MemoryContext oldcontext; @@ -1103,7 +1122,6 @@ exec_simple_query(const char *query_string) *plantree_list; Portal portal; DestReceiver *receiver; - int16 format; const char *cmdtagname; size_t cmdtaglen; @@ -1233,6 +1251,8 @@ exec_simple_query(const char *query_string) */ PortalStart(portal, NULL, 0, InvalidSnapshot); + if (format < 0) + { /* * Select the appropriate output format: text unless we are doing a * FETCH from a binary cursor. (Pretty grotty to have to do this here @@ -1253,6 +1273,7 @@ exec_simple_query(const char *query_string) format = 1; /* BINARY */ } } + } PortalSetResultFormat(portal, 1, &format); /* @@ -1386,11 +1407,12 @@ exec_simple_query(const char *query_string) * * Execute a "Parse" protocol message. */ -static void +void exec_parse_message(const char *query_string, /* string to execute */ const char *stmt_name, /* name for prepared stmt */ Oid *paramTypes, /* parameter types */ - int numParams) /* number of parameters */ + int numParams, /* number of parameters */ + const char *paramNames[]) { MemoryContext unnamed_stmt_context = NULL; MemoryContext oldcontext; @@ -1521,7 +1543,7 @@ exec_parse_message(const char *query_string, /* string to execute */ query_string, ¶mTypes, &numParams, - NULL); + NULL, paramNames); /* Done with the snapshot used for parsing */ if (snapshot_set) @@ -1621,7 +1643,7 @@ exec_parse_message(const char *query_string, /* string to execute */ * * Process a "Bind" message to create a portal from a prepared statement */ -static void +void exec_bind_message(StringInfo input_message) { const char *portal_name; @@ -2080,7 +2102,7 @@ exec_bind_message(StringInfo input_message) * * Process an "Execute" message for a portal */ -static void +void exec_execute_message(const char *portal_name, long max_rows) { CommandDest dest; @@ -2592,7 +2614,7 @@ bind_param_error_callback(void *arg) * * Process a "Describe" message for a prepared statement */ -static void +void exec_describe_statement_message(const char *stmt_name) { CachedPlanSource *psrc; @@ -2738,7 +2760,7 @@ exec_describe_portal_message(const char *portal_name) /* * Convenience routines for starting/committing a single command. */ -static void +void start_xact_command(void) { if (!xact_started) @@ -2766,7 +2788,7 @@ start_xact_command(void) client_connection_check_interval); } -static void +void finish_xact_command(void) { /* cancel active statement timeout after each command */ @@ -4313,6 +4335,14 @@ PostgresMain(const char *dbname, const char *username) if (sigsetjmp(local_sigjmp_buf, 1) != 0) { + /* Autonomous transactions */ + /* Clear stack and pool */ + while(!slist_is_empty(&AutonomousStack)) { + slist_node *node = slist_pop_head_node(&AutonomousStack); + pfree(node); + } + AutonomousPoolDestroyError(); + /* * NOTE: if you are tempted to add more code in this if-block, * consider the high probability that it should be in @@ -4647,10 +4677,10 @@ PostgresMain(const char *dbname, const char *username) if (am_walsender) { if (!exec_replication_command(query_string)) - exec_simple_query(query_string); + exec_simple_query(query_string, -1); } else - exec_simple_query(query_string); + exec_simple_query(query_string, -1); valgrind_report_error_query(query_string); @@ -4682,7 +4712,7 @@ PostgresMain(const char *dbname, const char *username) pq_getmsgend(&input_message); exec_parse_message(query_string, stmt_name, - paramTypes, numParams); + paramTypes, numParams, NULL); valgrind_report_error_query(query_string); } diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index f7c9882f7c..7aac1471a4 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -72,6 +72,7 @@ #include "storage/pg_shmem.h" #include "storage/predicate.h" #include "storage/standby.h" +#include "tcop/autonomous.h" #include "tcop/tcopprot.h" #include "tsearch/ts_cache.h" #include "utils/builtins.h" @@ -2662,6 +2663,19 @@ struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"autonomous_session_lifetime", + PGC_SIGHUP, + CUSTOM_OPTIONS, + gettext_noop("Maximum lifetime of autonomous session. In minutes. 300 by default."), + gettext_noop("A value of 0 turns off the timeout."), + GUC_UNIT_MIN + }, + &autonomous_session_lifetime, + 300, 0, INT_MAX, + NULL, NULL, NULL + }, + { {"authentication_timeout", PGC_SIGHUP, CONN_AUTH_AUTH, gettext_noop("Sets the maximum allowed time to complete client authentication."), diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index cf9f283cfe..c5f9816ff5 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -632,6 +632,10 @@ #log_planner_stats = off #log_executor_stats = off +#------------------------------------------------------------------------------ +# AUTONOMOUS TRANSACTIONS +#------------------------------------------------------------------------------ +#autonomous_session_lifetime = 300min # lifetime of autonomous session #------------------------------------------------------------------------------ # AUTOVACUUM diff --git a/src/include/libpq/pqmq.h b/src/include/libpq/pqmq.h index af607edf4c..35b5edce92 100644 --- a/src/include/libpq/pqmq.h +++ b/src/include/libpq/pqmq.h @@ -18,6 +18,7 @@ extern void pq_redirect_to_shm_mq(dsm_segment *seg, shm_mq_handle *mqh); extern void pq_set_parallel_leader(pid_t pid, BackendId backend_id); +extern void pq_stop_redirect_to_shm_mq(void); extern void pq_parse_errornotice(StringInfo msg, ErrorData *edata); diff --git a/src/include/parser/analyze.h b/src/include/parser/analyze.h index c96483ae78..9d07d7416d 100644 --- a/src/include/parser/analyze.h +++ b/src/include/parser/analyze.h @@ -28,7 +28,7 @@ extern PGDLLIMPORT post_parse_analyze_hook_type post_parse_analyze_hook; extern Query *parse_analyze_fixedparams(RawStmt *parseTree, const char *sourceText, const Oid *paramTypes, int numParams, QueryEnvironment *queryEnv); extern Query *parse_analyze_varparams(RawStmt *parseTree, const char *sourceText, - Oid **paramTypes, int *numParams, QueryEnvironment *queryEnv); + Oid **paramTypes, int *numParams, QueryEnvironment *queryEnv, const char *paramNames[]); extern Query *parse_analyze_withcb(RawStmt *parseTree, const char *sourceText, ParserSetupHook parserSetup, void *parserSetupArg, diff --git a/src/include/parser/parse_param.h b/src/include/parser/parse_param.h index d4865e50f6..a00a530d3c 100644 --- a/src/include/parser/parse_param.h +++ b/src/include/parser/parse_param.h @@ -18,7 +18,7 @@ extern void setup_parse_fixed_parameters(ParseState *pstate, const Oid *paramTypes, int numParams); extern void setup_parse_variable_parameters(ParseState *pstate, - Oid **paramTypes, int *numParams); + Oid **paramTypes, int *numParams, const char *paramNames[]); extern void check_variable_parameters(ParseState *pstate, Query *query); extern bool query_contains_extern_params(Query *query); diff --git a/src/include/tcop/autonomous.h b/src/include/tcop/autonomous.h new file mode 100644 index 0000000000..156fb43f9c --- /dev/null +++ b/src/include/tcop/autonomous.h @@ -0,0 +1,48 @@ +#ifndef AUTONOMOUS_H +#define AUTONOMOUS_H + +#include "access/tupdesc.h" +#include "lib/ilist.h" +#include "nodes/pg_list.h" +#include "utils/guc.h" +#include "utils/resowner.h" + +struct AutonomousSession; +typedef struct AutonomousSession AutonomousSession; + +struct AutonomousPreparedStatement; +typedef struct AutonomousPreparedStatement AutonomousPreparedStatement; + +typedef struct AutonomousResult +{ + TupleDesc tupdesc; + List *tuples; + char *command; +} AutonomousResult; + +AutonomousSession *AutonomousSessionGet(void); +void AutonomousSessionRelease(AutonomousSession *session); +void AutonomousPoolDestroyError(void); +AutonomousSession *AutonomousSessionPopStackSessionException(void *block); +AutonomousSession *AutonomousSessionPopStackSession(bool isBlockAutonomous); +AutonomousResult *AutonomousSessionExecute(AutonomousSession *session, const char *sql); +AutonomousPreparedStatement *AutonomousSessionPrepare(AutonomousSession *session, const char *sql, int16 nargs, + Oid argtypes[], const char *argnames[]); +AutonomousResult *AutonomousSessionExecutePrepared(AutonomousPreparedStatement *stmt, int16 nargs, Datum *values, bool *nulls); + +void AutonomousSessionMain(Datum main_arg); + +typedef struct AutonomousStackNode { + slist_node node; + AutonomousSession* session; + void *block; +} AutonomousStackNode; + +extern PGDLLIMPORT int autonomous_session_lifetime; + +extern PGDLLIMPORT slist_head AutonomousStack; +extern PGDLLIMPORT MemoryContext AutonomousContext; /* NOTE: move to src/include/utils/memutils.h */ +extern PGDLLIMPORT ResourceOwner AutonomousResourceOwner; /* NOTE: move to src/include/utils/resowner.h */ + + +#endif /* AUTONOMOUS_H */ diff --git a/src/include/tcop/tcopprot.h b/src/include/tcop/tcopprot.h index b694d85974..03cdf91e47 100644 --- a/src/include/tcop/tcopprot.h +++ b/src/include/tcop/tcopprot.h @@ -52,7 +52,7 @@ extern List *pg_analyze_and_rewrite_varparams(RawStmt *parsetree, const char *query_string, Oid **paramTypes, int *numParams, - QueryEnvironment *queryEnv); + QueryEnvironment *queryEnv, const char *paramNames[]); extern List *pg_analyze_and_rewrite_withcb(RawStmt *parsetree, const char *query_string, ParserSetupHook parserSetup, @@ -64,6 +64,12 @@ extern PlannedStmt *pg_plan_query(Query *querytree, const char *query_string, extern List *pg_plan_queries(List *querytrees, const char *query_string, int cursorOptions, ParamListInfo boundParams); +extern void exec_simple_query(const char *query_string, int16 format); +extern void exec_parse_message(const char *query_string, const char *stmt_name, + Oid paramTypes[], int numParams, const char *paramNames[]); +extern void exec_bind_message(StringInfo input_message); +extern void exec_execute_message(const char *portal_name, long max_rows); +extern void exec_describe_statement_message(const char *stmt_name); extern void die(SIGNAL_ARGS); extern void quickdie(SIGNAL_ARGS) pg_attribute_noreturn(); @@ -73,6 +79,9 @@ extern void HandleRecoveryConflictInterrupt(ProcSignalReason reason); extern void ProcessClientReadInterrupt(bool blocked); extern void ProcessClientWriteInterrupt(bool blocked); +extern void start_xact_command(void); +extern void finish_xact_command(void); + extern void process_postgres_switches(int argc, char *argv[], GucContext ctx, const char **dbname); extern void PostgresSingleUserMain(int argc, char *argv[], diff --git a/src/pl/plpgsql/src/Makefile b/src/pl/plpgsql/src/Makefile index dfb815212f..d3155f1c47 100644 --- a/src/pl/plpgsql/src/Makefile +++ b/src/pl/plpgsql/src/Makefile @@ -32,7 +32,7 @@ DATA = plpgsql.control plpgsql--1.0.sql REGRESS_OPTS = --dbname=$(PL_TESTDB) -REGRESS = plpgsql_array plpgsql_call plpgsql_control plpgsql_copy plpgsql_domain \ +REGRESS = plpgsql_autonomous plpgsql_array plpgsql_call plpgsql_control plpgsql_copy plpgsql_domain \ plpgsql_record plpgsql_cache plpgsql_simple plpgsql_transaction \ plpgsql_trap plpgsql_trigger plpgsql_varprops diff --git a/src/pl/plpgsql/src/expected/plpgsql_autonomous.out b/src/pl/plpgsql/src/expected/plpgsql_autonomous.out new file mode 100644 index 0000000000..2e64b841b6 --- /dev/null +++ b/src/pl/plpgsql/src/expected/plpgsql_autonomous.out @@ -0,0 +1,1423 @@ +-------------------------------------------------------------- +------ auton_func simple ------------------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_func simple"'; END; $$; +NOTICE: Test "auton_func simple" +CREATE TABLE tbl (a int); +CREATE OR REPLACE FUNCTION func() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (1); +END; +$$; +START TRANSACTION; +SELECT func(); + func +------ + +(1 row) + +ROLLBACK; +SELECT * FROM tbl; + a +--- + 1 +(1 row) + +DROP FUNCTION func; +DROP TABLE tbl; +-------------------------------------------------------------- +----- auton_func -> func call through SELECT ----------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_func -> func call through SELECT"'; END; $$; +NOTICE: Test "auton_func -> func call through SELECT" +CREATE TABLE tbl (a varchar); +CREATE OR REPLACE FUNCTION func() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE +BEGIN + INSERT INTO tbl VALUES ('in func'); +END; +$$; +CREATE OR REPLACE FUNCTION auton_func() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('before func'); + START TRANSACTION; + SELECT func(); + COMMIT; + INSERT INTO tbl VALUES ('after func'); +END; +$$; +START TRANSACTION; +SELECT auton_func(); + auton_func +------------ + +(1 row) + +ROLLBACK; +SELECT * FROM tbl; + a +------------- + before func + in func + after func +(3 rows) + +DROP FUNCTION func; +DROP FUNCTION auton_func; +DROP TABLE tbl; +-------------------------------------------------------------- +----- auton_func -> func call through PERFORM ---------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_func -> func call through PERFORM"'; END; $$; +NOTICE: Test "auton_func -> func call through PERFORM" +CREATE TABLE tbl (a varchar); +CREATE OR REPLACE FUNCTION func() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE +BEGIN + INSERT INTO tbl VALUES ('in func'); +END; +$$; +CREATE OR REPLACE FUNCTION auton_func() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('before func'); + START TRANSACTION; + PERFORM func(); + COMMIT; + INSERT INTO tbl VALUES ('after func'); +END; +$$; +START TRANSACTION; +SELECT auton_func(); + auton_func +------------ + +(1 row) + +ROLLBACK; +SELECT * FROM tbl; + a +------------- + before func + in func + after func +(3 rows) + +DROP FUNCTION func; +DROP FUNCTION auton_func; +DROP TABLE tbl; +-------------------------------------------------------------- +----- auton_func -> auton_func call through SELECT ----------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_func -> auton_func call through SELECT"'; END; $$; +NOTICE: Test "auton_func -> auton_func call through SELECT" +CREATE TABLE tbl (a varchar); +CREATE OR REPLACE FUNCTION auton_func1() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('in auton_func1'); +END; +$$; +CREATE OR REPLACE FUNCTION auton_func2() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('before auton_func1'); + START TRANSACTION; + SELECT auton_func1(); + COMMIT; + INSERT INTO tbl VALUES ('after auton_func1'); +END; +$$; +START TRANSACTION; +SELECT auton_func2(); + auton_func2 +------------- + +(1 row) + +ROLLBACK; +SELECT * FROM tbl; + a +-------------------- + before auton_func1 + in auton_func1 + after auton_func1 +(3 rows) + +DROP FUNCTION auton_func1; +DROP FUNCTION auton_func2; +DROP TABLE tbl; +-------------------------------------------------------------- +----- auton_func -> auton_func call through PERFORM----------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_func -> auton_func call through PERFORM"'; END; $$; +NOTICE: Test "auton_func -> auton_func call through PERFORM" +CREATE TABLE tbl (a varchar); +CREATE OR REPLACE FUNCTION auton_func1() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('in auton_func1'); +END; +$$; +CREATE OR REPLACE FUNCTION auton_func2() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('before auton_func1'); + START TRANSACTION; + PERFORM auton_func1(); + COMMIT; + INSERT INTO tbl VALUES ('after auton_func1'); +END; +$$; +START TRANSACTION; +SELECT auton_func2(); + auton_func2 +------------- + +(1 row) + +ROLLBACK; +SELECT * FROM tbl; + a +-------------------- + before auton_func1 + in auton_func1 + after auton_func1 +(3 rows) + +DROP FUNCTION auton_func1; +DROP FUNCTION auton_func2; +DROP TABLE tbl; +-------------------------------------------------------------- +----- 4 auton func calls, simple SQL ------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "4 auton func calls, simple SQL"'; END; $$; +NOTICE: Test "4 auton func calls, simple SQL" +CREATE TABLE tbl (a int); +CREATE OR REPLACE FUNCTION func1() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (1); +END; +$$; +CREATE OR REPLACE FUNCTION func2() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + SELECT func1(); + INSERT INTO tbl VALUES (2); +END; +$$; +CREATE OR REPLACE FUNCTION func3() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + SELECT func2(); + INSERT INTO tbl VALUES (3); +END; +$$; +CREATE OR REPLACE FUNCTION func4() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (0); + SELECT func3(); + INSERT INTO tbl VALUES (4); +END; +$$; +START TRANSACTION; +SELECT func4(); + func4 +------- + +(1 row) + +ROLLBACK; +SELECT * FROM tbl; + a +--- + 0 + 1 + 2 + 3 + 4 +(5 rows) + +TRUNCATE tbl; +DROP FUNCTION func1; +DROP FUNCTION func2; +DROP FUNCTION func3; +DROP FUNCTION func4; +DROP TABLE tbl; +-------------------------------------------------------------- +----- 5 func calls №1, simple SQL ---------------------------- +-------------------------------------------------------------- +-- auton_func -> func -> auton_func -> func -> auton_func +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "5 func calls 1, simple SQL"'; END; $$; +NOTICE: Test "5 func calls 1, simple SQL" +CREATE TABLE tbl (a int); +CREATE OR REPLACE FUNCTION func1() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (1); +END; +$$; +CREATE OR REPLACE FUNCTION func2() RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + PERFORM func1(); + INSERT INTO tbl VALUES (2); +END; +$$; +CREATE OR REPLACE FUNCTION func3() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + SELECT func2(); + INSERT INTO tbl VALUES (3); +END; +$$; +CREATE OR REPLACE FUNCTION func4() RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + PERFORM func3(); + INSERT INTO tbl VALUES (3); +END; +$$; +CREATE OR REPLACE FUNCTION func5() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (0); + SELECT func4(); + INSERT INTO tbl VALUES (5); +END; +$$; +START TRANSACTION; +SELECT func4(); + func4 +------- + +(1 row) + +ROLLBACK; +SELECT * FROM tbl; + a +--- + 1 + 2 + 3 +(3 rows) + +TRUNCATE tbl; +DROP FUNCTION func1; +DROP FUNCTION func2; +DROP FUNCTION func3; +DROP FUNCTION func4; +DROP FUNCTION func5; +DROP TABLE tbl; +-------------------------------------------------------------- +----- 5 func calls №2, simple SQL ---------------------------- +-------------------------------------------------------------- +-- func -> auton_func -> func -> auton_func -> func +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "5 func calls 2, simple SQL"'; END; $$; +NOTICE: Test "5 func calls 2, simple SQL" +CREATE TABLE tbl (a int); +CREATE OR REPLACE FUNCTION func1() RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES (1); +END; +$$; +CREATE OR REPLACE FUNCTION func2() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + SELECT func1(); + INSERT INTO tbl VALUES (2); +END; +$$; +CREATE OR REPLACE FUNCTION func3() RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + PERFORM func2(); + INSERT INTO tbl VALUES (3); +END; +$$; +CREATE OR REPLACE FUNCTION func4() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + SELECT func3(); + INSERT INTO tbl VALUES (3); +END; +$$; +CREATE OR REPLACE FUNCTION func5() RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES (0); + PERFORM func4(); + INSERT INTO tbl VALUES (5); +END; +$$; +SELECT func4(); + func4 +------- + +(1 row) + +SELECT * FROM tbl; + a +--- + 1 + 2 + 3 + 3 +(4 rows) + +TRUNCATE tbl; +DROP FUNCTION func1; +DROP FUNCTION func2; +DROP FUNCTION func3; +DROP FUNCTION func4; +DROP FUNCTION func5; +DROP TABLE tbl; +-------------------------------------------------------------- +----- auton_func -> auton_func call, dynamic SQL --------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton -> auton call, dynamic SQL"'; END; $$; +NOTICE: Test "auton -> auton call, dynamic SQL" +CREATE TABLE tbl (a int); +CREATE OR REPLACE FUNCTION func1() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + EXECUTE 'INSERT INTO tbl VALUES (1)'; +END; +$$; +CREATE OR REPLACE FUNCTION func2() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + EXECUTE 'INSERT INTO tbl VALUES (0)'; + SELECT func1(); + EXECUTE 'INSERT INTO tbl VALUES (2)'; +END; +$$; +START TRANSACTION; +SELECT func2(); + func2 +------- + +(1 row) + +ROLLBACK; +SELECT * FROM tbl; + a +--- + 0 + 1 + 2 +(3 rows) + +TRUNCATE tbl; +DROP FUNCTION func1; +DROP FUNCTION func2; +DROP TABLE tbl; +-------------------------------------------------------------- +----- for loop with transaction statements ------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "for loop with transaction statements"'; END; $$; +NOTICE: Test "for loop with transaction statements" +CREATE TABLE tbl (a int); +CREATE FUNCTION dynamic() RETURNS integer +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + FOR i IN 0..9 LOOP + START TRANSACTION; + EXECUTE 'INSERT INTO tbl VALUES (' || i::text || ')'; + IF i % 2 = 0 THEN + COMMIT; + ELSE + ROLLBACK; + END IF; + END LOOP; + + RETURN 42; +END; +$$; +SELECT dynamic(); + dynamic +--------- + 42 +(1 row) + +SELECT * FROM tbl; + a +--- + 0 + 2 + 4 + 6 + 8 +(5 rows) + +TRUNCATE tbl; +CREATE FUNCTION simple() RETURNS integer +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + FOR i IN 0..9 LOOP + START TRANSACTION; + INSERT INTO tbl VALUES (i); + IF i % 2 = 0 THEN + COMMIT; + ELSE + ROLLBACK; + END IF; + END LOOP; + + RETURN 42; +END; +$$; +SELECT simple(); + simple +-------- + 42 +(1 row) + +SELECT * FROM tbl; + a +--- + 0 + 2 + 4 + 6 + 8 +(5 rows) + +TRUNCATE tbl; +DROP FUNCTION dynamic; +DROP FUNCTION simple; +DROP TABLE tbl; +-------------------------------------------------------------- +----- transaction scope -------------------------------------- +-------------------------------------------------------------- +-- if error is in autonomous function, then exit from it +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "transaction scope"'; END; $$; +NOTICE: Test "transaction scope" +CREATE TABLE tbl (a integer CHECK (a > 500)); +CREATE FUNCTION sep_trans() +RETURNS void +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl(a) VALUES(700); -- 1st autonomous transaction + INSERT INTO tbl(a) VALUES(100); -- 2nd autonomous transaction + INSERT INTO tbl(a) VALUES(800); -- no transaction, as in 2nd violation +END +$$ LANGUAGE plpgsql; +SELECT sep_trans(); +ERROR: new row for relation "tbl" violates check constraint "tbl_a_check" +DETAIL: Failing row contains (100). +CONTEXT: PL/pgSQL function sep_trans() line 6 at SQL statement +SELECT * FROM tbl; + a +----- + 700 +(1 row) + +TRUNCATE TABLE tbl; +CREATE FUNCTION one_trans() +RETURNS void +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + START TRANSACTION; -- 1 autonomous transaction + INSERT INTO tbl(a) VALUES(1700); + INSERT INTO tbl(a) VALUES(200); -- violation of constraint + INSERT INTO tbl(a) VALUES(1800); + COMMIT; +END +$$ LANGUAGE plpgsql; +SELECT one_trans(); +ERROR: new row for relation "tbl" violates check constraint "tbl_a_check" +DETAIL: Failing row contains (200). +CONTEXT: PL/pgSQL function one_trans() line 7 at SQL statement +SELECT * FROM tbl; + a +--- +(0 rows) + +TRUNCATE tbl; +DROP FUNCTION sep_trans(); +DROP FUNCTION one_trans(); +DROP TABLE tbl; +-------------------------------------------------------------- +----- non transaction commands ------------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "non transaction commands"'; END; $$; +NOTICE: Test "non transaction commands" +CREATE TABLE tbl (a integer); +DO $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + EXECUTE 'VACUUM tbl'; +END $$; +DROP TABLE tbl; +-------------------------------------------------------------- +----- audit -------------------------------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "audit"'; END; $$; +NOTICE: Test "audit" +CREATE TABLE table_tracking(log_id serial, username text, event_date timestamp, msg text); +CREATE FUNCTION log_action_atx ( + username text, event_date timestamp, msg text +) RETURNS void AS +$body$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + START TRANSACTION; + INSERT INTO table_tracking VALUES (nextval('table_tracking_log_id_seq'::regclass),username, event_date, msg); + COMMIT; +END; +$body$ +LANGUAGE plpgsql; +CREATE FUNCTION log_action ( + username text, event_date timestamp, msg text +) RETURNS void AS +$body$ +DECLARE + v_query text; +BEGIN + -- Call the autonomous function + v_query := 'SELECT log_action_atx ( ' || quote_nullable(username) || + ',' || quote_nullable(event_date) || ',' || quote_nullable(msg) || ' )'; + EXECUTE v_query; + + -- Do something else + RAISE NOTICE 'Division by zero'; + PERFORM 5/0; +END; +$body$ +LANGUAGE PLPGSQL; +SELECT * FROM log_action(current_user::text, now()::timestamp, 'Test'::text); +NOTICE: Division by zero +ERROR: division by zero +CONTEXT: SQL statement "SELECT 5/0" +PL/pgSQL function log_action(text,timestamp without time zone,text) line 12 at PERFORM +SELECT msg FROM table_tracking; + msg +------ + Test +(1 row) + +DROP FUNCTION log_action; +DROP FUNCTION log_action_atx; +DROP TABLE table_tracking; +-------------------------------------------------------------- +----- audit through trigger ---------------------------------- +-------------------------------------------------------------- +-- users are in pg_user table +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "audit through trigger"'; END; $$; +NOTICE: Test "audit through trigger" +CREATE SCHEMA audtrig; +CREATE TABLE audtrig.tbl ( + id integer, + user_name varchar +); +CREATE TABLE audtrig.log ( + user_name varchar, + event varchar +); +CREATE OR REPLACE FUNCTION audtrig.log_func() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + v_action varchar; +BEGIN + IF (TG_OP = 'INSERT') THEN + v_action := 'Added id=' || NEW.id::text; + ELSIF (TG_OP = 'UPDATE') THEN + v_action := 'Updated id=' || NEW.id::text; + ELSIF (TG_OP = 'DELETE') THEN + v_action := 'Deleted id=' || OLD.id::text; + END IF; + INSERT INTO audtrig.log VALUES (current_user, v_action); + RETURN NEW; +END; +$$; +CREATE OR REPLACE TRIGGER audtrig_trig_func +AFTER INSERT OR UPDATE OR DELETE +ON audtrig.tbl +FOR EACH ROW +EXECUTE PROCEDURE audtrig.log_func(); +SELECT * FROM audtrig.log; + user_name | event +-----------+------- +(0 rows) + +SELECT * FROM audtrig.tbl; + id | user_name +----+----------- +(0 rows) + +INSERT INTO audtrig.tbl VALUES (9001,'SMITH'); +INSERT INTO audtrig.tbl VALUES (9002,'JONES'); +DELETE FROM audtrig.tbl WHERE id=9001; +SELECT * FROM audtrig.log; + user_name | event +-----------+----------------- + postgres | Added id=9001 + postgres | Added id=9002 + postgres | Deleted id=9001 +(3 rows) + +SELECT * FROM audtrig.tbl; + id | user_name +------+----------- + 9002 | JONES +(1 row) + +TRUNCATE audtrig.log; +TRUNCATE audtrig.tbl; +START TRANSACTION; +INSERT INTO audtrig.tbl VALUES (9001,'SMITH'); +INSERT INTO audtrig.tbl VALUES (9002,'JONES'); +DELETE FROM audtrig.tbl WHERE id=9001; +ROLLBACK; +SELECT * FROM audtrig.log; + user_name | event +-----------+----------------- + postgres | Added id=9001 + postgres | Added id=9002 + postgres | Deleted id=9001 +(3 rows) + +SELECT * FROM audtrig.tbl; + id | user_name +----+----------- +(0 rows) + +DROP TABLE audtrig.log; +DROP TABLE audtrig.tbl; +-------------------------------------------------------------- +----- function and procedure call in autonomous sessions ----- +-------------------------------------------------------------- +-- procedure call is forbidden in autonomous functions +-- but possible in simple functions, that are called in autonomous functions +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "function and procedure call in autonomous sessions"'; END; $$; +NOTICE: Test "function and procedure call in autonomous sessions" +CREATE TABLE tbl (a int); +CREATE FUNCTION func() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES (1); +END; +$$; +CREATE PROCEDURE proc() +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES (2); + ROLLBACK; + INSERT INTO tbl VALUES (3); +END; +$$; +-- check: auton_func -> func +-- -> proc +CREATE FUNCTION auton_func() +RETURNS void +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (10); + SELECT * FROM func(); + CALL proc(); + INSERT INTO tbl VALUES (20); +END; +$$ LANGUAGE plpgsql; +SELECT auton_func(); +ERROR: call of procedures is forbidden currently in autonomous sessions, CALL proc() +CONTEXT: PL/pgSQL function auton_func() line 7 at CALL +SELECT * FROM tbl; + a +---- + 10 + 1 +(2 rows) + +TRUNCATE TABLE tbl; +-- check: auton_proc -> func +-- -> proc +CREATE PROCEDURE auton_proc() +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (10); + SELECT * FROM func(); + CALL proc(); + INSERT INTO tbl VALUES (20); +END; +$$; +CALL auton_proc(); +ERROR: call of procedures is forbidden currently in autonomous sessions, CALL proc() +CONTEXT: PL/pgSQL function auton_proc() line 7 at CALL +SELECT * FROM tbl; + a +---- + 10 + 1 +(2 rows) + +TRUNCATE TABLE tbl; +-- check: auton_func_2 -> func_2 -> proc_no_trans +CREATE PROCEDURE proc_no_trans() +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES (2); + INSERT INTO tbl VALUES (3); +END; +$$; +CREATE FUNCTION func_2() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES (30); + CALL proc_no_trans(); + INSERT INTO tbl VALUES (40); +END; +$$; +CREATE FUNCTION auton_func_2() +RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (10); + SELECT * FROM func_2(); + INSERT INTO tbl VALUES (20); +END; +$$; +SELECT auton_func_2(); + auton_func_2 +-------------- + +(1 row) + +SELECT * FROM tbl; + a +---- + 10 + 30 + 2 + 3 + 40 + 20 +(6 rows) + +TRUNCATE TABLE tbl; +DROP FUNCTION func; +DROP FUNCTION func_2; +DROP PROCEDURE proc; +DROP PROCEDURE proc_no_trans; +DROP FUNCTION auton_func; +DROP FUNCTION auton_func_2; +DROP PROCEDURE auton_proc; +DROP TABLE tbl; +-------------------------------------------------------------- +----- exception simple --------------------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "exception simple"'; END; $$; +NOTICE: Test "exception simple" +CREATE FUNCTION f_exception() +RETURNS VOID AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + RAISE NOTICE 'Call f_exception'; + RAISE EXCEPTION 'Test exception' USING ERRCODE = 'AB001'; +END; +$$ +LANGUAGE plpgsql; +CREATE OR REPLACE FUNCTION f_main () +RETURNS VOID AS $$ +BEGIN + BEGIN + PERFORM f_exception(); + EXCEPTION + WHEN OTHERS THEN + IF SQLSTATE = 'AB001' THEN + RAISE NOTICE 'Caught custom exception with SQLSTATE AB001.'; + ELSE + RAISE NOTICE 'An exception occurred: %', SQLERRM; + END IF; + END; +END; +$$ +LANGUAGE plpgsql; +SELECT * FROM f_main(); +NOTICE: Call f_exception +NOTICE: Caught custom exception with SQLSTATE AB001. + f_main +-------- + +(1 row) + +DROP FUNCTION f_exception; +DROP FUNCTION f_main; +-------------------------------------------------------------- +----- exception with many functions -------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "exception with many functions"'; END; $$; +NOTICE: Test "exception with many functions" +CREATE TABLE tbl (a varchar); +CREATE FUNCTION f_exception_auton() +RETURNS VOID AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('before raise'); + RAISE EXCEPTION 'Test exception' USING ERRCODE = 'AB001'; + INSERT INTO tbl VALUES ('after raise'); +END; +$$ +LANGUAGE plpgsql; +CREATE OR REPLACE FUNCTION func0() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES ('before f_exception_auton'); + PERFORM f_exception_auton(); + INSERT INTO tbl VALUES ('after f_exception_auton'); +END; +$$; +CREATE OR REPLACE FUNCTION func0_auton() +RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('before func0'); + PERFORM func0(); + INSERT INTO tbl VALUES ('after func0'); +END; +$$; +CREATE OR REPLACE FUNCTION func1() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES ('before func0_auton'); + PERFORM func0_auton(); + INSERT INTO tbl VALUES ('after func0_auton'); +END; +$$; +CREATE OR REPLACE FUNCTION catch_exc() +RETURNS void +LANGUAGE plpgsql +AS $$ + BEGIN + INSERT INTO tbl VALUES ('before func1'); + PERFORM func1(); + INSERT INTO tbl VALUES ('after func1'); + EXCEPTION + WHEN OTHERS THEN + IF SQLSTATE = 'AB001' THEN + RAISE NOTICE 'Caught custom exception with SQLSTATE AB001.'; + ELSE + RAISE NOTICE 'An exception occurred: %', SQLERRM; + END IF; + INSERT INTO tbl VALUES ('after func1 in exc'); + END; +$$; +CREATE OR REPLACE FUNCTION func1_auton() +RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES ('before catch_exc'); + PERFORM catch_exc(); + INSERT INTO tbl VALUES ('after catch_exc'); + COMMIT; +END; +$$; +CREATE OR REPLACE FUNCTION func2() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES ('before func1_auton'); + PERFORM func1_auton(); + INSERT INTO tbl VALUES ('after func1_auton'); +END; +$$; +CREATE OR REPLACE FUNCTION func2_auton() +RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES ('before func2'); + PERFORM func2(); + INSERT INTO tbl VALUES ('after func2'); + ROLLBACK; +END; +$$; +CREATE OR REPLACE FUNCTION main() +RETURNS void AS $$ +BEGIN + INSERT INTO tbl VALUES ('before func2_auton'); + PERFORM func2_auton(); + INSERT INTO tbl VALUES ('after func2_auton'); +END; +$$ LANGUAGE plpgsql; +SELECT main(); +NOTICE: Caught custom exception with SQLSTATE AB001. + main +------ + +(1 row) + +SELECT * FROM tbl; + a +-------------------- + before func2_auton + before catch_exc + before func0 + before raise + after func1 in exc + after catch_exc + after func2_auton +(7 rows) + +DROP FUNCTION func0; +DROP FUNCTION func0_auton; +DROP FUNCTION func1; +DROP FUNCTION func1_auton; +DROP FUNCTION func2; +DROP FUNCTION func2_auton; +DROP FUNCTION catch_exc; +DROP FUNCTION main; +DROP FUNCTION f_exception_auton; +DROP TABLE tbl; +-------------------------------------------------------------- +----- subblock -> auton_subblock ----------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "subblock -> auton_subblock"'; END; $$; +NOTICE: Test "subblock -> auton_subblock" +CREATE TABLE tbl (a varchar); +CREATE OR REPLACE FUNCTION func() +RETURNS void AS $$ +BEGIN + INSERT INTO tbl VALUES('subblock, before auton_subblock'); + -- begin subblock + DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES('rollback in auton_subblock'); + ROLLBACK; + + START TRANSACTION; + INSERT INTO tbl VALUES('commit in auton_subblock'); + COMMIT; + END; + -- end subblock + INSERT INTO tbl VALUES('subblock, after auton_subblock'); +END; +$$ LANGUAGE plpgsql; +START TRANSACTION; +SELECT func(); + func +------ + +(1 row) + +ROLLBACK; +SELECT * FROM tbl; + a +-------------------------- + commit in auton_subblock +(1 row) + +DROP FUNCTION func; +DROP TABLE tbl; +-------------------------------------------------------------- +----- auton_subblock -> subblock ----------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_subblock -> subblock"'; END; $$; +NOTICE: Test "auton_subblock -> subblock" +CREATE TABLE tbl (a varchar); +CREATE OR REPLACE FUNCTION func() +RETURNS void AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES('auton_subblock, before subblock'); + START TRANSACTION; + -- begin subblock + BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES('rollback in subblock'); + ROLLBACK; + + START TRANSACTION; + INSERT INTO tbl VALUES('commit in subblock'); + COMMIT; + END; + -- end subblock + ROLLBACK; + INSERT INTO tbl VALUES('auton_subblock, after subblock'); +END; +$$ LANGUAGE plpgsql; +START TRANSACTION; +SELECT func(); +NOTICE: there is already a transaction in progress +ERROR: invalid transaction termination +CONTEXT: PL/pgSQL function func() line 18 at ROLLBACK +ROLLBACK; +SELECT * FROM tbl; + a +--------------------------------- + auton_subblock, before subblock + commit in subblock +(2 rows) + +DROP FUNCTION func; +DROP TABLE tbl; +-------------------------------------------------------------- +----- auton_subblock -> auton_subblock ----------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_subblock -> auton_subblock"'; END; $$; +NOTICE: Test "auton_subblock -> auton_subblock" +CREATE TABLE tbl (a varchar); +CREATE OR REPLACE FUNCTION func() +RETURNS void AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES('auton_subblock 1, before auton_subblock'); + START TRANSACTION; + -- begin subblock + DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES('rollback in auton_subblock'); + ROLLBACK; + + START TRANSACTION; + INSERT INTO tbl VALUES('commit in auton_subblock'); + COMMIT; + END; + -- end subblock + ROLLBACK; + INSERT INTO tbl VALUES('auton_subblock 1, after auton_subblock'); +END; +$$ LANGUAGE plpgsql; +START TRANSACTION; +SELECT func(); +NOTICE: Autonomous pool was resized, new capacity is 2 + func +------ + +(1 row) + +ROLLBACK; +SELECT * FROM tbl; + a +----------------------------------------- + auton_subblock 1, before auton_subblock + commit in auton_subblock + auton_subblock 1, after auton_subblock +(3 rows) + +DROP FUNCTION func; +DROP TABLE tbl; +-------------------------------------------------------------- +----- many subblocks with exception -------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "many subblocks with exception"'; END; $$; +NOTICE: Test "many subblocks with exception" +CREATE TABLE tbl (a varchar); +CREATE FUNCTION f_exception() +RETURNS VOID AS $$ +BEGIN + RAISE EXCEPTION 'Test exception' USING ERRCODE = 'AB001'; +END; +$$ +LANGUAGE plpgsql; +CREATE OR REPLACE FUNCTION func1() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES ('before f_exception'); + PERFORM f_exception(); + INSERT INTO tbl VALUES ('after f_exception'); +END; +$$; +CREATE OR REPLACE FUNCTION main() +RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('main start'); + START TRANSACTION; + INSERT INTO tbl VALUES ('trans 1 in start'); + + DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES ('trans 2 in start'); + + BEGIN + INSERT INTO tbl VALUES ('empty 1 begin'); + + BEGIN + INSERT INTO tbl VALUES ('exc begin'); + + DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + BEGIN + INSERT INTO tbl VALUES ('before func1'); + PERFORM func1(); + INSERT INTO tbl VALUES ('after func1'); + END; + + EXCEPTION + WHEN OTHERS THEN + IF SQLSTATE = 'AB001' THEN + RAISE NOTICE 'Caught custom exception with SQLSTATE AB001.'; + ELSE + RAISE NOTICE 'An exception occurred: %', SQLERRM; + END IF; + INSERT INTO tbl VALUES ('exc end'); + END; + + INSERT INTO tbl VALUES ('empty 1 end'); + END; + + INSERT INTO tbl VALUES ('trans 2 in rollback'); + ROLLBACK; + END; + + INSERT INTO tbl VALUES ('trans 1 in commit'); + COMMIT; + + INSERT INTO tbl VALUES ('main end'); +END; +$$; +SELECT main(); +NOTICE: Autonomous pool was resized, new capacity is 4 +NOTICE: Caught custom exception with SQLSTATE AB001. +ERROR: invalid transaction termination +CONTEXT: PL/pgSQL function main() line 43 at ROLLBACK +SELECT * FROM tbl; + a +-------------- + main start + before func1 +(2 rows) + +DROP FUNCTION func1; +DROP FUNCTION main; +DROP FUNCTION f_exception; +DROP TABLE tbl; +-------------------------------------------------------------- +----- proc in many subblocks many with exception ------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "proc in many subblocks many with exception"'; END; $$; +NOTICE: Test "proc in many subblocks many with exception" +CREATE TABLE tbl (a varchar); +CREATE FUNCTION f_exception() +RETURNS VOID AS $$ +BEGIN + RAISE EXCEPTION 'Test exception' USING ERRCODE = 'AB001'; +END; +$$ +LANGUAGE plpgsql; +CREATE OR REPLACE FUNCTION func1() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES ('before f_exception'); + PERFORM f_exception(); + INSERT INTO tbl VALUES ('after f_exception'); +END; +$$; +CREATE OR REPLACE PROCEDURE proc() +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES ('proc before rollback'); + ROLLBACK; + INSERT INTO tbl VALUES ('proc after rollback'); +END; +$$; +CREATE OR REPLACE FUNCTION main() +RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('main start'); + START TRANSACTION; + INSERT INTO tbl VALUES ('trans 1 in start'); + + DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES ('trans 2 in start'); + + BEGIN + INSERT INTO tbl VALUES ('empty 1 begin'); + + BEGIN + INSERT INTO tbl VALUES ('exc begin'); + + DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + BEGIN + INSERT INTO tbl VALUES ('before func1'); + PERFORM func1(); + INSERT INTO tbl VALUES ('after func1'); + END; + + EXCEPTION + WHEN OTHERS THEN + IF SQLSTATE = 'AB001' THEN + RAISE NOTICE 'Caught custom exception with SQLSTATE AB001.'; + ELSE + RAISE NOTICE 'An exception occurred: %', SQLERRM; + END IF; + INSERT INTO tbl VALUES ('exc before proc'); + CALL proc(); + INSERT INTO tbl VALUES ('exc after proc'); + INSERT INTO tbl VALUES ('exc end'); + END; + + INSERT INTO tbl VALUES ('empty 1 end'); + END; + + INSERT INTO tbl VALUES ('trans 2 in rollback'); + ROLLBACK; + END; + + INSERT INTO tbl VALUES ('trans 1 in commit'); + COMMIT; + + INSERT INTO tbl VALUES ('main end'); +END; +$$; +SELECT main(); +NOTICE: Autonomous pool was resized, new capacity is 2 +NOTICE: Autonomous pool was resized, new capacity is 4 +NOTICE: Caught custom exception with SQLSTATE AB001. +ERROR: call of procedures is forbidden currently in autonomous sessions, CALL proc() +CONTEXT: PL/pgSQL function main() line 37 at CALL +SELECT * FROM tbl; + a +-------------- + main start + before func1 +(2 rows) + +DROP FUNCTION func1; +DROP FUNCTION main; +DROP FUNCTION f_exception; +DROP TABLE tbl; diff --git a/src/pl/plpgsql/src/pl_exec.c b/src/pl/plpgsql/src/pl_exec.c index f8c7f48747..cc21f9eb94 100644 --- a/src/pl/plpgsql/src/pl_exec.c +++ b/src/pl/plpgsql/src/pl_exec.c @@ -38,6 +38,7 @@ #include "plpgsql.h" #include "storage/proc.h" #include "tcop/cmdtag.h" +#include "tcop/autonomous.h" #include "tcop/pquery.h" #include "tcop/tcopprot.h" #include "tcop/utility.h" @@ -259,6 +260,10 @@ static HTAB *shared_cast_hash = NULL; /************************************************************ * Local function forward declarations ************************************************************/ +static void build_symbol_table(PLpgSQL_execstate *estate, + PLpgSQL_nsitem *ns_start, int *ret_nitems, + const char ***ret_names, Oid **ret_types, + Datum **ret_values, bool **ret_isnull); static void coerce_function_result_tuple(PLpgSQL_execstate *estate, TupleDesc tupdesc); static void plpgsql_exec_error_callback(void *arg); @@ -1652,6 +1657,26 @@ exec_stmt_block(PLpgSQL_execstate *estate, PLpgSQL_stmt_block *block) volatile int rc = -1; int i; + /* Autonomous transactions */ + { + AutonomousStackNode* node = NULL; + MemoryContext oldctx = MemoryContextSwitchTo(TopMemoryContext); + + /* Store PLpgSQL_stmt_block to stack */ + node = palloc(sizeof(AutonomousStackNode)); + node->block = (void*)block; + node->session = NULL; + if (block->autonomous) { + estate->autonomous_session = AutonomousSessionGet(); + if(!estate->autonomous_session) { + elog(ERROR, "No free autonomous workers, %s can't be executed", estate->func->fn_signature); + } + node->session = estate->autonomous_session; + } + slist_push_head(&AutonomousStack, &node->node); + MemoryContextSwitchTo(oldctx); + } + /* * First initialize all variables declared in this block */ @@ -1830,6 +1855,9 @@ exec_stmt_block(PLpgSQL_execstate *estate, PLpgSQL_stmt_block *block) ErrorData *edata; ListCell *e; + /* Autonomous transactions */ + estate->autonomous_session = AutonomousSessionPopStackSessionException(block); + estate->err_text = gettext_noop("during exception cleanup"); /* Save error info in our stmt_mcontext */ @@ -1944,6 +1972,9 @@ exec_stmt_block(PLpgSQL_execstate *estate, PLpgSQL_stmt_block *block) estate->err_text = NULL; + estate->autonomous_session = AutonomousSessionPopStackSession(block->autonomous); + + /* * Handle the return code. This is intentionally different from * LOOP_RC_PROCESSING(): CONTINUE never matches a block, and EXIT matches @@ -2168,6 +2199,24 @@ exec_stmt_perform(PLpgSQL_execstate *estate, PLpgSQL_stmt_perform *stmt) { PLpgSQL_expr *expr = stmt->expr; + if (estate->autonomous_session) + { + int nparams = 0; + const char **param_names = NULL; + Oid *param_types = NULL; + AutonomousPreparedStatement *astmt; + Datum *values; + bool *nulls; + AutonomousResult *aresult; + + build_symbol_table(estate, expr->ns, &nparams, ¶m_names, ¶m_types, &values, &nulls); + astmt = AutonomousSessionPrepare(estate->autonomous_session, expr->query, nparams, param_types, param_names); + + aresult = AutonomousSessionExecutePrepared(astmt, nparams, values, nulls); + exec_set_found(estate, (list_length(aresult->tuples) != 0)); + return PLPGSQL_RC_OK; + } + (void) exec_run_select(estate, expr, 0, NULL); exec_set_found(estate, (estate->eval_processed != 0)); exec_eval_cleanup(estate); @@ -2190,6 +2239,14 @@ exec_stmt_call(PLpgSQL_execstate *estate, PLpgSQL_stmt_call *stmt) SPIExecuteOptions options; int rc; + if (estate->autonomous_session) + { + ereport(ERROR, + (errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION), + errmsg("call of procedures is forbidden currently in autonomous sessions, %s", stmt->expr->query))); + } + + /* * Make a plan if we don't have one already. */ @@ -4019,6 +4076,8 @@ plpgsql_estate_setup(PLpgSQL_execstate *estate, } estate->rsi = rsi; + estate->autonomous_session = NULL; + estate->found_varno = func->found_varno; estate->ndatums = func->ndatums; estate->datums = NULL; @@ -4203,6 +4262,72 @@ exec_prepare_plan(PLpgSQL_execstate *estate, } +static +void build_symbol_table(PLpgSQL_execstate *estate, + PLpgSQL_nsitem *ns_start, + int *ret_nitems, + const char ***ret_names, + Oid **ret_types, + Datum **ret_values, + bool **ret_isnull) +{ + PLpgSQL_nsitem *nsitem; + List *names = NIL; + List *itemnos = NIL; + + ListCell *l_name, *l_itemno; + int i, nitems; + const char **names_vector; + Oid *types_vector; + Datum *values_vector; + bool *isnull_vector; + + for (nsitem = ns_start; + nsitem; + nsitem = nsitem->prev) + { + if (nsitem->itemtype == PLPGSQL_NSTYPE_VAR) + { + String *name; + + if (strcmp(nsitem->name, "found") == 0) + continue; // XXX + elog(LOG, "namespace item variable itemno %d, name %s", + nsitem->itemno, nsitem->name); + name = makeString(nsitem->name); + if (!list_member(names, name)) + { + names = lappend(names, name); + itemnos = lappend_oid(itemnos, nsitem->itemno); + } + } + } + + nitems = list_length(names); + names_vector = palloc(nitems * sizeof(char *)); + types_vector = palloc(nitems * sizeof(Oid)); + values_vector = palloc(nitems * sizeof(Datum)); + isnull_vector = palloc(nitems * sizeof(bool)); + i = 0; + forboth(l_name, names, l_itemno, itemnos) + { + int itemno = lfirst_oid(l_itemno); + PLpgSQL_datum *datum = estate->datums[itemno]; + PLpgSQL_var *var = (PLpgSQL_var *) datum; + names_vector[i] = pstrdup(strVal(lfirst(l_name))); + types_vector[i] = var->datatype->typoid; + values_vector[i] = var->value; + isnull_vector[i] = var->isnull; + ++i; + } + + *ret_nitems = nitems; + *ret_names = names_vector; + *ret_types = types_vector; + *ret_values = values_vector; + *ret_isnull = isnull_vector; +} + /* ---------- * exec_stmt_execsql Execute an SQL statement (possibly with INTO). * @@ -4225,6 +4350,24 @@ exec_stmt_execsql(PLpgSQL_execstate *estate, else if (plpgsql_extra_warnings & PLPGSQL_XCHECK_TOOMANYROWS) too_many_rows_level = WARNING; + if (estate->autonomous_session) + { + int nparams = 0; + const char **param_names = NULL; + Oid *param_types = NULL; + AutonomousPreparedStatement *astmt; + Datum *values; + bool *nulls; + AutonomousResult *aresult; + + build_symbol_table(estate, stmt->sqlstmt->ns, &nparams, ¶m_names, ¶m_types, &values, &nulls); + astmt = AutonomousSessionPrepare(estate->autonomous_session, stmt->sqlstmt->query, nparams, param_types, param_names); + + aresult = AutonomousSessionExecutePrepared(astmt, nparams, values, nulls); + exec_set_found(estate, (list_length(aresult->tuples) != 0)); + return PLPGSQL_RC_OK; + } + /* * On the first call for this statement generate the plan, and detect * whether the statement is INSERT/UPDATE/DELETE/MERGE @@ -4472,6 +4615,12 @@ exec_stmt_dynexecute(PLpgSQL_execstate *estate, exec_eval_cleanup(estate); + if (estate->autonomous_session) + { + AutonomousSessionExecute(estate->autonomous_session, querystr); + return PLPGSQL_RC_OK; + } + /* * Execute the query without preparing a saved plan. */ @@ -4958,10 +5107,18 @@ exec_stmt_close(PLpgSQL_execstate *estate, PLpgSQL_stmt_close *stmt) static int exec_stmt_commit(PLpgSQL_execstate *estate, PLpgSQL_stmt_commit *stmt) { - if (stmt->chain) - SPI_commit_and_chain(); + if (estate->autonomous_session) + { + AutonomousSessionExecute(estate->autonomous_session, "COMMIT"); + return PLPGSQL_RC_OK; + } else - SPI_commit(); + { + if (stmt->chain) + SPI_commit_and_chain(); + else + SPI_commit(); + } /* * We need to build new simple-expression infrastructure, since the old @@ -4982,10 +5139,18 @@ exec_stmt_commit(PLpgSQL_execstate *estate, PLpgSQL_stmt_commit *stmt) static int exec_stmt_rollback(PLpgSQL_execstate *estate, PLpgSQL_stmt_rollback *stmt) { - if (stmt->chain) - SPI_rollback_and_chain(); + if (estate->autonomous_session) + { + AutonomousSessionExecute(estate->autonomous_session, "ROLLBACK"); + return PLPGSQL_RC_OK; + } else - SPI_rollback(); + { + if (stmt->chain) + SPI_rollback_and_chain(); + else + SPI_rollback(); + } /* * We need to build new simple-expression infrastructure, since the old diff --git a/src/pl/plpgsql/src/pl_gram.y b/src/pl/plpgsql/src/pl_gram.y index 6a09bfdd67..cb2af56584 100644 --- a/src/pl/plpgsql/src/pl_gram.y +++ b/src/pl/plpgsql/src/pl_gram.y @@ -107,6 +107,8 @@ static PLpgSQL_expr *read_cursor_args(PLpgSQL_var *cursor, static List *read_raise_options(void); static void check_raise_parameters(PLpgSQL_stmt_raise *stmt); +static bool last_pragma; + %} %expect 0 @@ -143,6 +145,7 @@ static void check_raise_parameters(PLpgSQL_stmt_raise *stmt); char *label; int n_initvars; int *initvarnos; + bool autonomous; } declhdr; struct { @@ -323,6 +326,7 @@ static void check_raise_parameters(PLpgSQL_stmt_raise *stmt); %token K_PG_EXCEPTION_DETAIL %token K_PG_EXCEPTION_HINT %token K_PG_ROUTINE_OID +%token K_PRAGMA %token K_PRINT_STRICT_PARAMS %token K_PRIOR %token K_QUERY @@ -416,6 +420,7 @@ pl_block : decl_sect K_BEGIN proc_sect exception_sect K_END opt_label new->lineno = plpgsql_location_to_lineno(@2); new->stmtid = ++plpgsql_curr_compile->nstatements; new->label = $1.label; + new->autonomous = $1.autonomous; new->n_initvars = $1.n_initvars; new->initvarnos = $1.initvarnos; new->body = $3; @@ -436,6 +441,7 @@ decl_sect : opt_block_label $$.label = $1; $$.n_initvars = 0; $$.initvarnos = NULL; + $$.autonomous = false; } | opt_block_label decl_start { @@ -443,6 +449,7 @@ decl_sect : opt_block_label $$.label = $1; $$.n_initvars = 0; $$.initvarnos = NULL; + $$.autonomous = false; } | opt_block_label decl_start decl_stmts { @@ -450,6 +457,8 @@ decl_sect : opt_block_label $$.label = $1; /* Remember variables declared in decl_stmts */ $$.n_initvars = plpgsql_add_initdatums(&($$.initvarnos)); + $$.autonomous = last_pragma; + last_pragma = false; } ; @@ -457,6 +466,7 @@ decl_start : K_DECLARE { /* Forget any variables created before block */ plpgsql_add_initdatums(NULL); + last_pragma = false; /* * Disable scanner lookup of identifiers while * we process the decl_stmts @@ -554,6 +564,13 @@ decl_statement : decl_varname decl_const decl_datatype decl_collate decl_notnull new->cursor_explicit_argrow = $5->dno; new->cursor_options = CURSOR_OPT_FAST_PLAN | $2; } + | K_PRAGMA any_identifier ';' + { + if (pg_strcasecmp($2, "autonomous_transaction") == 0) + last_pragma = true; + else + elog(ERROR, "invalid pragma"); + } ; opt_scrollable : diff --git a/src/pl/plpgsql/src/pl_unreserved_kwlist.h b/src/pl/plpgsql/src/pl_unreserved_kwlist.h index 3e258a6bb9..cf8e2aa69d 100644 --- a/src/pl/plpgsql/src/pl_unreserved_kwlist.h +++ b/src/pl/plpgsql/src/pl_unreserved_kwlist.h @@ -86,6 +86,7 @@ PG_KEYWORD("pg_exception_context", K_PG_EXCEPTION_CONTEXT) PG_KEYWORD("pg_exception_detail", K_PG_EXCEPTION_DETAIL) PG_KEYWORD("pg_exception_hint", K_PG_EXCEPTION_HINT) PG_KEYWORD("pg_routine_oid", K_PG_ROUTINE_OID) +PG_KEYWORD("pragma", K_PRAGMA) PG_KEYWORD("print_strict_params", K_PRINT_STRICT_PARAMS) PG_KEYWORD("prior", K_PRIOR) PG_KEYWORD("query", K_QUERY) diff --git a/src/pl/plpgsql/src/plpgsql.h b/src/pl/plpgsql/src/plpgsql.h index 9f0a912115..5a958f0636 100644 --- a/src/pl/plpgsql/src/plpgsql.h +++ b/src/pl/plpgsql/src/plpgsql.h @@ -23,6 +23,7 @@ #include "utils/expandedrecord.h" #include "utils/typcache.h" +#include "tcop/autonomous.h" /********************************************************************** * Definitions @@ -502,6 +503,7 @@ typedef struct PLpgSQL_stmt_block int lineno; unsigned int stmtid; char *label; + bool autonomous; List *body; /* List of statements */ int n_initvars; /* Length of initvarnos[] */ int *initvarnos; /* dnos of variables declared in this block */ @@ -1046,6 +1048,9 @@ typedef struct PLpgSQL_execstate ResourceOwner tuple_store_owner; ReturnSetInfo *rsi; + AutonomousSession *autonomous_session; + + /* the datums representing the function's local variables */ int found_varno; /* diff --git a/src/pl/plpgsql/src/sql/plpgsql_autonomous.sql b/src/pl/plpgsql/src/sql/plpgsql_autonomous.sql new file mode 100644 index 0000000000..15c7e9c519 --- /dev/null +++ b/src/pl/plpgsql/src/sql/plpgsql_autonomous.sql @@ -0,0 +1,1296 @@ +-------------------------------------------------------------- +------ auton_func simple ------------------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_func simple"'; END; $$; + +CREATE TABLE tbl (a int); + +CREATE OR REPLACE FUNCTION func() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (1); +END; +$$; + +START TRANSACTION; +SELECT func(); +ROLLBACK; +SELECT * FROM tbl; + +DROP FUNCTION func; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- auton_func -> func call through SELECT ----------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_func -> func call through SELECT"'; END; $$; + +CREATE TABLE tbl (a varchar); + +CREATE OR REPLACE FUNCTION func() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE +BEGIN + INSERT INTO tbl VALUES ('in func'); +END; +$$; + +CREATE OR REPLACE FUNCTION auton_func() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('before func'); + START TRANSACTION; + SELECT func(); + COMMIT; + INSERT INTO tbl VALUES ('after func'); +END; +$$; + +START TRANSACTION; +SELECT auton_func(); +ROLLBACK; +SELECT * FROM tbl; + +DROP FUNCTION func; +DROP FUNCTION auton_func; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- auton_func -> func call through PERFORM ---------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_func -> func call through PERFORM"'; END; $$; + +CREATE TABLE tbl (a varchar); + +CREATE OR REPLACE FUNCTION func() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE +BEGIN + INSERT INTO tbl VALUES ('in func'); +END; +$$; + +CREATE OR REPLACE FUNCTION auton_func() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('before func'); + START TRANSACTION; + PERFORM func(); + COMMIT; + INSERT INTO tbl VALUES ('after func'); +END; +$$; + +START TRANSACTION; +SELECT auton_func(); +ROLLBACK; +SELECT * FROM tbl; + +DROP FUNCTION func; +DROP FUNCTION auton_func; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- auton_func -> auton_func call through SELECT ----------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_func -> auton_func call through SELECT"'; END; $$; + +CREATE TABLE tbl (a varchar); + +CREATE OR REPLACE FUNCTION auton_func1() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('in auton_func1'); +END; +$$; + +CREATE OR REPLACE FUNCTION auton_func2() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('before auton_func1'); + START TRANSACTION; + SELECT auton_func1(); + COMMIT; + INSERT INTO tbl VALUES ('after auton_func1'); +END; +$$; + +START TRANSACTION; +SELECT auton_func2(); +ROLLBACK; +SELECT * FROM tbl; + +DROP FUNCTION auton_func1; +DROP FUNCTION auton_func2; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- auton_func -> auton_func call through PERFORM----------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_func -> auton_func call through PERFORM"'; END; $$; + +CREATE TABLE tbl (a varchar); + +CREATE OR REPLACE FUNCTION auton_func1() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('in auton_func1'); +END; +$$; + +CREATE OR REPLACE FUNCTION auton_func2() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('before auton_func1'); + START TRANSACTION; + PERFORM auton_func1(); + COMMIT; + INSERT INTO tbl VALUES ('after auton_func1'); +END; +$$; + +START TRANSACTION; +SELECT auton_func2(); +ROLLBACK; +SELECT * FROM tbl; + +DROP FUNCTION auton_func1; +DROP FUNCTION auton_func2; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- 4 auton func calls, simple SQL ------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "4 auton func calls, simple SQL"'; END; $$; + +CREATE TABLE tbl (a int); + +CREATE OR REPLACE FUNCTION func1() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (1); +END; +$$; + +CREATE OR REPLACE FUNCTION func2() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + SELECT func1(); + INSERT INTO tbl VALUES (2); +END; +$$; + +CREATE OR REPLACE FUNCTION func3() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + SELECT func2(); + INSERT INTO tbl VALUES (3); +END; +$$; + +CREATE OR REPLACE FUNCTION func4() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (0); + SELECT func3(); + INSERT INTO tbl VALUES (4); +END; +$$; + +START TRANSACTION; +SELECT func4(); +ROLLBACK; +SELECT * FROM tbl; +TRUNCATE tbl; + +DROP FUNCTION func1; +DROP FUNCTION func2; +DROP FUNCTION func3; +DROP FUNCTION func4; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- 5 func calls №1, simple SQL ---------------------------- +-------------------------------------------------------------- +-- auton_func -> func -> auton_func -> func -> auton_func +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "5 func calls 1, simple SQL"'; END; $$; + +CREATE TABLE tbl (a int); + +CREATE OR REPLACE FUNCTION func1() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (1); +END; +$$; + +CREATE OR REPLACE FUNCTION func2() RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + PERFORM func1(); + INSERT INTO tbl VALUES (2); +END; +$$; + +CREATE OR REPLACE FUNCTION func3() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + SELECT func2(); + INSERT INTO tbl VALUES (3); +END; +$$; + +CREATE OR REPLACE FUNCTION func4() RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + PERFORM func3(); + INSERT INTO tbl VALUES (3); +END; +$$; + +CREATE OR REPLACE FUNCTION func5() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (0); + SELECT func4(); + INSERT INTO tbl VALUES (5); +END; +$$; + +START TRANSACTION; +SELECT func4(); +ROLLBACK; +SELECT * FROM tbl; +TRUNCATE tbl; + +DROP FUNCTION func1; +DROP FUNCTION func2; +DROP FUNCTION func3; +DROP FUNCTION func4; +DROP FUNCTION func5; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- 5 func calls №2, simple SQL ---------------------------- +-------------------------------------------------------------- +-- func -> auton_func -> func -> auton_func -> func +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "5 func calls 2, simple SQL"'; END; $$; + +CREATE TABLE tbl (a int); + +CREATE OR REPLACE FUNCTION func1() RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES (1); +END; +$$; + +CREATE OR REPLACE FUNCTION func2() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + SELECT func1(); + INSERT INTO tbl VALUES (2); +END; +$$; + +CREATE OR REPLACE FUNCTION func3() RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + PERFORM func2(); + INSERT INTO tbl VALUES (3); +END; +$$; + +CREATE OR REPLACE FUNCTION func4() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + SELECT func3(); + INSERT INTO tbl VALUES (3); +END; +$$; + +CREATE OR REPLACE FUNCTION func5() RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES (0); + PERFORM func4(); + INSERT INTO tbl VALUES (5); +END; +$$; + +SELECT func4(); +SELECT * FROM tbl; +TRUNCATE tbl; + +DROP FUNCTION func1; +DROP FUNCTION func2; +DROP FUNCTION func3; +DROP FUNCTION func4; +DROP FUNCTION func5; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- auton_func -> auton_func call, dynamic SQL --------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton -> auton call, dynamic SQL"'; END; $$; + +CREATE TABLE tbl (a int); + +CREATE OR REPLACE FUNCTION func1() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + EXECUTE 'INSERT INTO tbl VALUES (1)'; +END; +$$; + +CREATE OR REPLACE FUNCTION func2() RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + EXECUTE 'INSERT INTO tbl VALUES (0)'; + SELECT func1(); + EXECUTE 'INSERT INTO tbl VALUES (2)'; +END; +$$; + +START TRANSACTION; +SELECT func2(); +ROLLBACK; +SELECT * FROM tbl; +TRUNCATE tbl; + +DROP FUNCTION func1; +DROP FUNCTION func2; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- for loop with transaction statements ------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "for loop with transaction statements"'; END; $$; + +CREATE TABLE tbl (a int); + +CREATE FUNCTION dynamic() RETURNS integer +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + FOR i IN 0..9 LOOP + START TRANSACTION; + EXECUTE 'INSERT INTO tbl VALUES (' || i::text || ')'; + IF i % 2 = 0 THEN + COMMIT; + ELSE + ROLLBACK; + END IF; + END LOOP; + + RETURN 42; +END; +$$; + + +SELECT dynamic(); +SELECT * FROM tbl; +TRUNCATE tbl; + + +CREATE FUNCTION simple() RETURNS integer +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + FOR i IN 0..9 LOOP + START TRANSACTION; + INSERT INTO tbl VALUES (i); + IF i % 2 = 0 THEN + COMMIT; + ELSE + ROLLBACK; + END IF; + END LOOP; + + RETURN 42; +END; +$$; + + +SELECT simple(); +SELECT * FROM tbl; +TRUNCATE tbl; + +DROP FUNCTION dynamic; +DROP FUNCTION simple; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- transaction scope -------------------------------------- +-------------------------------------------------------------- +-- if error is in autonomous function, then exit from it +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "transaction scope"'; END; $$; + +CREATE TABLE tbl (a integer CHECK (a > 500)); + +CREATE FUNCTION sep_trans() +RETURNS void +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl(a) VALUES(700); -- 1st autonomous transaction + INSERT INTO tbl(a) VALUES(100); -- 2nd autonomous transaction + INSERT INTO tbl(a) VALUES(800); -- no transaction, as in 2nd violation +END +$$ LANGUAGE plpgsql; + +SELECT sep_trans(); +SELECT * FROM tbl; +TRUNCATE TABLE tbl; + +CREATE FUNCTION one_trans() +RETURNS void +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + START TRANSACTION; -- 1 autonomous transaction + INSERT INTO tbl(a) VALUES(1700); + INSERT INTO tbl(a) VALUES(200); -- violation of constraint + INSERT INTO tbl(a) VALUES(1800); + COMMIT; +END +$$ LANGUAGE plpgsql; + +SELECT one_trans(); +SELECT * FROM tbl; +TRUNCATE tbl; + +DROP FUNCTION sep_trans(); +DROP FUNCTION one_trans(); +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- non transaction commands ------------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "non transaction commands"'; END; $$; + +CREATE TABLE tbl (a integer); + +DO $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + EXECUTE 'VACUUM tbl'; +END $$; + +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- audit -------------------------------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "audit"'; END; $$; + +CREATE TABLE table_tracking(log_id serial, username text, event_date timestamp, msg text); + +CREATE FUNCTION log_action_atx ( + username text, event_date timestamp, msg text +) RETURNS void AS +$body$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + START TRANSACTION; + INSERT INTO table_tracking VALUES (nextval('table_tracking_log_id_seq'::regclass),username, event_date, msg); + COMMIT; +END; +$body$ +LANGUAGE plpgsql; + +CREATE FUNCTION log_action ( + username text, event_date timestamp, msg text +) RETURNS void AS +$body$ +DECLARE + v_query text; +BEGIN + -- Call the autonomous function + v_query := 'SELECT log_action_atx ( ' || quote_nullable(username) || + ',' || quote_nullable(event_date) || ',' || quote_nullable(msg) || ' )'; + EXECUTE v_query; + + -- Do something else + RAISE NOTICE 'Division by zero'; + PERFORM 5/0; +END; +$body$ +LANGUAGE PLPGSQL; + + +SELECT * FROM log_action(current_user::text, now()::timestamp, 'Test'::text); +SELECT msg FROM table_tracking; + +DROP FUNCTION log_action; +DROP FUNCTION log_action_atx; +DROP TABLE table_tracking; + + +-------------------------------------------------------------- +----- audit through trigger ---------------------------------- +-------------------------------------------------------------- +-- users are in pg_user table +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "audit through trigger"'; END; $$; + +CREATE SCHEMA audtrig; + +CREATE TABLE audtrig.tbl ( + id integer, + user_name varchar +); + +CREATE TABLE audtrig.log ( + user_name varchar, + event varchar +); + +CREATE OR REPLACE FUNCTION audtrig.log_func() +RETURNS TRIGGER +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + v_action varchar; +BEGIN + IF (TG_OP = 'INSERT') THEN + v_action := 'Added id=' || NEW.id::text; + ELSIF (TG_OP = 'UPDATE') THEN + v_action := 'Updated id=' || NEW.id::text; + ELSIF (TG_OP = 'DELETE') THEN + v_action := 'Deleted id=' || OLD.id::text; + END IF; + INSERT INTO audtrig.log VALUES (current_user, v_action); + RETURN NEW; +END; +$$; + +CREATE OR REPLACE TRIGGER audtrig_trig_func +AFTER INSERT OR UPDATE OR DELETE +ON audtrig.tbl +FOR EACH ROW +EXECUTE PROCEDURE audtrig.log_func(); + +SELECT * FROM audtrig.log; +SELECT * FROM audtrig.tbl; + +INSERT INTO audtrig.tbl VALUES (9001,'SMITH'); +INSERT INTO audtrig.tbl VALUES (9002,'JONES'); +DELETE FROM audtrig.tbl WHERE id=9001; +SELECT * FROM audtrig.log; +SELECT * FROM audtrig.tbl; + + +TRUNCATE audtrig.log; +TRUNCATE audtrig.tbl; +START TRANSACTION; +INSERT INTO audtrig.tbl VALUES (9001,'SMITH'); +INSERT INTO audtrig.tbl VALUES (9002,'JONES'); +DELETE FROM audtrig.tbl WHERE id=9001; +ROLLBACK; + +SELECT * FROM audtrig.log; +SELECT * FROM audtrig.tbl; + +DROP TABLE audtrig.log; +DROP TABLE audtrig.tbl; + + +-------------------------------------------------------------- +----- function and procedure call in autonomous sessions ----- +-------------------------------------------------------------- +-- procedure call is forbidden in autonomous functions +-- but possible in simple functions, that are called in autonomous functions +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "function and procedure call in autonomous sessions"'; END; $$; + +CREATE TABLE tbl (a int); + +CREATE FUNCTION func() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES (1); +END; +$$; + +CREATE PROCEDURE proc() +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES (2); + ROLLBACK; + INSERT INTO tbl VALUES (3); +END; +$$; + + +-- check: auton_func -> func +-- -> proc + +CREATE FUNCTION auton_func() +RETURNS void +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (10); + SELECT * FROM func(); + CALL proc(); + INSERT INTO tbl VALUES (20); +END; +$$ LANGUAGE plpgsql; + +SELECT auton_func(); +SELECT * FROM tbl; +TRUNCATE TABLE tbl; + + +-- check: auton_proc -> func +-- -> proc + +CREATE PROCEDURE auton_proc() +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (10); + SELECT * FROM func(); + CALL proc(); + INSERT INTO tbl VALUES (20); +END; +$$; + +CALL auton_proc(); +SELECT * FROM tbl; +TRUNCATE TABLE tbl; + + +-- check: auton_func_2 -> func_2 -> proc_no_trans + +CREATE PROCEDURE proc_no_trans() +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES (2); + INSERT INTO tbl VALUES (3); +END; +$$; + +CREATE FUNCTION func_2() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES (30); + CALL proc_no_trans(); + INSERT INTO tbl VALUES (40); +END; +$$; + +CREATE FUNCTION auton_func_2() +RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES (10); + SELECT * FROM func_2(); + INSERT INTO tbl VALUES (20); +END; +$$; + +SELECT auton_func_2(); +SELECT * FROM tbl; +TRUNCATE TABLE tbl; + +DROP FUNCTION func; +DROP FUNCTION func_2; +DROP PROCEDURE proc; +DROP PROCEDURE proc_no_trans; +DROP FUNCTION auton_func; +DROP FUNCTION auton_func_2; +DROP PROCEDURE auton_proc; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- exception simple --------------------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "exception simple"'; END; $$; + +CREATE FUNCTION f_exception() +RETURNS VOID AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + RAISE NOTICE 'Call f_exception'; + RAISE EXCEPTION 'Test exception' USING ERRCODE = 'AB001'; +END; +$$ +LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION f_main () +RETURNS VOID AS $$ +BEGIN + BEGIN + PERFORM f_exception(); + EXCEPTION + WHEN OTHERS THEN + IF SQLSTATE = 'AB001' THEN + RAISE NOTICE 'Caught custom exception with SQLSTATE AB001.'; + ELSE + RAISE NOTICE 'An exception occurred: %', SQLERRM; + END IF; + END; +END; +$$ +LANGUAGE plpgsql; + +SELECT * FROM f_main(); + +DROP FUNCTION f_exception; +DROP FUNCTION f_main; + + +-------------------------------------------------------------- +----- exception with many functions -------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "exception with many functions"'; END; $$; + +CREATE TABLE tbl (a varchar); + +CREATE FUNCTION f_exception_auton() +RETURNS VOID AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('before raise'); + RAISE EXCEPTION 'Test exception' USING ERRCODE = 'AB001'; + INSERT INTO tbl VALUES ('after raise'); +END; +$$ +LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION func0() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES ('before f_exception_auton'); + PERFORM f_exception_auton(); + INSERT INTO tbl VALUES ('after f_exception_auton'); +END; +$$; + +CREATE OR REPLACE FUNCTION func0_auton() +RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('before func0'); + PERFORM func0(); + INSERT INTO tbl VALUES ('after func0'); +END; +$$; + + +CREATE OR REPLACE FUNCTION func1() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES ('before func0_auton'); + PERFORM func0_auton(); + INSERT INTO tbl VALUES ('after func0_auton'); +END; +$$; + + +CREATE OR REPLACE FUNCTION catch_exc() +RETURNS void +LANGUAGE plpgsql +AS $$ + BEGIN + INSERT INTO tbl VALUES ('before func1'); + PERFORM func1(); + INSERT INTO tbl VALUES ('after func1'); + EXCEPTION + WHEN OTHERS THEN + IF SQLSTATE = 'AB001' THEN + RAISE NOTICE 'Caught custom exception with SQLSTATE AB001.'; + ELSE + RAISE NOTICE 'An exception occurred: %', SQLERRM; + END IF; + INSERT INTO tbl VALUES ('after func1 in exc'); + END; +$$; + +CREATE OR REPLACE FUNCTION func1_auton() +RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES ('before catch_exc'); + PERFORM catch_exc(); + INSERT INTO tbl VALUES ('after catch_exc'); + COMMIT; +END; +$$; + +CREATE OR REPLACE FUNCTION func2() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES ('before func1_auton'); + PERFORM func1_auton(); + INSERT INTO tbl VALUES ('after func1_auton'); +END; +$$; + +CREATE OR REPLACE FUNCTION func2_auton() +RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES ('before func2'); + PERFORM func2(); + INSERT INTO tbl VALUES ('after func2'); + ROLLBACK; +END; +$$; + +CREATE OR REPLACE FUNCTION main() +RETURNS void AS $$ +BEGIN + INSERT INTO tbl VALUES ('before func2_auton'); + PERFORM func2_auton(); + INSERT INTO tbl VALUES ('after func2_auton'); +END; +$$ LANGUAGE plpgsql; + +SELECT main(); +SELECT * FROM tbl; + +DROP FUNCTION func0; +DROP FUNCTION func0_auton; +DROP FUNCTION func1; +DROP FUNCTION func1_auton; +DROP FUNCTION func2; +DROP FUNCTION func2_auton; +DROP FUNCTION catch_exc; +DROP FUNCTION main; +DROP FUNCTION f_exception_auton; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- subblock -> auton_subblock ----------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "subblock -> auton_subblock"'; END; $$; + +CREATE TABLE tbl (a varchar); + +CREATE OR REPLACE FUNCTION func() +RETURNS void AS $$ +BEGIN + INSERT INTO tbl VALUES('subblock, before auton_subblock'); + -- begin subblock + DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES('rollback in auton_subblock'); + ROLLBACK; + + START TRANSACTION; + INSERT INTO tbl VALUES('commit in auton_subblock'); + COMMIT; + END; + -- end subblock + INSERT INTO tbl VALUES('subblock, after auton_subblock'); +END; +$$ LANGUAGE plpgsql; + +START TRANSACTION; +SELECT func(); +ROLLBACK; + +SELECT * FROM tbl; + +DROP FUNCTION func; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- auton_subblock -> subblock ----------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_subblock -> subblock"'; END; $$; + +CREATE TABLE tbl (a varchar); + +CREATE OR REPLACE FUNCTION func() +RETURNS void AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES('auton_subblock, before subblock'); + START TRANSACTION; + -- begin subblock + BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES('rollback in subblock'); + ROLLBACK; + + START TRANSACTION; + INSERT INTO tbl VALUES('commit in subblock'); + COMMIT; + END; + -- end subblock + ROLLBACK; + INSERT INTO tbl VALUES('auton_subblock, after subblock'); +END; +$$ LANGUAGE plpgsql; + +START TRANSACTION; +SELECT func(); +ROLLBACK; + +SELECT * FROM tbl; + +DROP FUNCTION func; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- auton_subblock -> auton_subblock ----------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "auton_subblock -> auton_subblock"'; END; $$; + +CREATE TABLE tbl (a varchar); + +CREATE OR REPLACE FUNCTION func() +RETURNS void AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES('auton_subblock 1, before auton_subblock'); + START TRANSACTION; + -- begin subblock + DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES('rollback in auton_subblock'); + ROLLBACK; + + START TRANSACTION; + INSERT INTO tbl VALUES('commit in auton_subblock'); + COMMIT; + END; + -- end subblock + ROLLBACK; + INSERT INTO tbl VALUES('auton_subblock 1, after auton_subblock'); +END; +$$ LANGUAGE plpgsql; + +START TRANSACTION; +SELECT func(); +ROLLBACK; +SELECT * FROM tbl; + +DROP FUNCTION func; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- many subblocks with exception -------------------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "many subblocks with exception"'; END; $$; + +CREATE TABLE tbl (a varchar); + +CREATE FUNCTION f_exception() +RETURNS VOID AS $$ +BEGIN + RAISE EXCEPTION 'Test exception' USING ERRCODE = 'AB001'; +END; +$$ +LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION func1() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES ('before f_exception'); + PERFORM f_exception(); + INSERT INTO tbl VALUES ('after f_exception'); +END; +$$; + + +CREATE OR REPLACE FUNCTION main() +RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('main start'); + START TRANSACTION; + INSERT INTO tbl VALUES ('trans 1 in start'); + + DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES ('trans 2 in start'); + + BEGIN + INSERT INTO tbl VALUES ('empty 1 begin'); + + BEGIN + INSERT INTO tbl VALUES ('exc begin'); + + DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + BEGIN + INSERT INTO tbl VALUES ('before func1'); + PERFORM func1(); + INSERT INTO tbl VALUES ('after func1'); + END; + + EXCEPTION + WHEN OTHERS THEN + IF SQLSTATE = 'AB001' THEN + RAISE NOTICE 'Caught custom exception with SQLSTATE AB001.'; + ELSE + RAISE NOTICE 'An exception occurred: %', SQLERRM; + END IF; + INSERT INTO tbl VALUES ('exc end'); + END; + + INSERT INTO tbl VALUES ('empty 1 end'); + END; + + INSERT INTO tbl VALUES ('trans 2 in rollback'); + ROLLBACK; + END; + + INSERT INTO tbl VALUES ('trans 1 in commit'); + COMMIT; + + INSERT INTO tbl VALUES ('main end'); +END; +$$; + +SELECT main(); +SELECT * FROM tbl; + +DROP FUNCTION func1; +DROP FUNCTION main; +DROP FUNCTION f_exception; +DROP TABLE tbl; + + +-------------------------------------------------------------- +----- proc in many subblocks many with exception ------------- +-------------------------------------------------------------- +DO $$ BEGIN RAISE NOTICE 'Test "proc in many subblocks many with exception"'; END; $$; + +CREATE TABLE tbl (a varchar); + +CREATE FUNCTION f_exception() +RETURNS VOID AS $$ +BEGIN + RAISE EXCEPTION 'Test exception' USING ERRCODE = 'AB001'; +END; +$$ +LANGUAGE plpgsql; + +CREATE OR REPLACE FUNCTION func1() +RETURNS void +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES ('before f_exception'); + PERFORM f_exception(); + INSERT INTO tbl VALUES ('after f_exception'); +END; +$$; + +CREATE OR REPLACE PROCEDURE proc() +LANGUAGE plpgsql +AS $$ +BEGIN + INSERT INTO tbl VALUES ('proc before rollback'); + ROLLBACK; + INSERT INTO tbl VALUES ('proc after rollback'); +END; +$$; + + +CREATE OR REPLACE FUNCTION main() +RETURNS void +LANGUAGE plpgsql +AS $$ +DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; +BEGIN + INSERT INTO tbl VALUES ('main start'); + START TRANSACTION; + INSERT INTO tbl VALUES ('trans 1 in start'); + + DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + BEGIN + START TRANSACTION; + INSERT INTO tbl VALUES ('trans 2 in start'); + + BEGIN + INSERT INTO tbl VALUES ('empty 1 begin'); + + BEGIN + INSERT INTO tbl VALUES ('exc begin'); + + DECLARE + PRAGMA AUTONOMOUS_TRANSACTION; + BEGIN + INSERT INTO tbl VALUES ('before func1'); + PERFORM func1(); + INSERT INTO tbl VALUES ('after func1'); + END; + + EXCEPTION + WHEN OTHERS THEN + IF SQLSTATE = 'AB001' THEN + RAISE NOTICE 'Caught custom exception with SQLSTATE AB001.'; + ELSE + RAISE NOTICE 'An exception occurred: %', SQLERRM; + END IF; + INSERT INTO tbl VALUES ('exc before proc'); + CALL proc(); + INSERT INTO tbl VALUES ('exc after proc'); + INSERT INTO tbl VALUES ('exc end'); + END; + + INSERT INTO tbl VALUES ('empty 1 end'); + END; + + INSERT INTO tbl VALUES ('trans 2 in rollback'); + ROLLBACK; + END; + + INSERT INTO tbl VALUES ('trans 1 in commit'); + COMMIT; + + INSERT INTO tbl VALUES ('main end'); +END; +$$; + +SELECT main(); +SELECT * FROM tbl; + +DROP FUNCTION func1; +DROP FUNCTION main; +DROP FUNCTION f_exception; +DROP TABLE tbl; + -- 2.34.1