Star214
315
代码
代码
Issues2230
Pull Requests489
项目讨论
Wiki
分析
Star214
315
已合并
【特性】【Opengauss回合】存储过程支持COMMIT/ROLLBACK事务控制 #123
AtomGit-Bot创建于 2020年8月21日
【特性】【Opengauss回合】存储过程支持COMMIT/ROLLBACK事务控制 #123
已合并
AtomGit-Bot创建于 2020年8月21日
从master合入到master
共 27 个文件变更+1665-129
Msrc/common/backend/utils/fmgr/fmgr.cpp+7-0
@@ -1105,6 +1105,10 @@ static Datum fmgr_security_definer(PG_FUNCTION_ARGS)
1105 volatile int save_nestlevel;1105 volatile int save_nestlevel;
1106 PgStat_FunctionCallUsage fcusage;1106 PgStat_FunctionCallUsage fcusage;
1107 1107 
1108+ /* Does not allow commit in pre setting scenary */
1109+ bool savedisTopLevelForSTP = u_sess->SPI_cxt.is_toplevel_stp;
1110+ u_sess->SPI_cxt.is_toplevel_stp = false;
1111+ 
1108 if (!fcinfo->flinfo->fn_extra) {1112 if (!fcinfo->flinfo->fn_extra) {
1109 HeapTuple tuple;1113 HeapTuple tuple;
1110 Form_pg_proc procedureStruct;1114 Form_pg_proc procedureStruct;
@@ -1216,6 +1220,9 @@ static Datum fmgr_security_definer(PG_FUNCTION_ARGS)
1216 (*fmgr_hook)(FHET_END, &(fcache->flinfo), &(fcache->arg));1220 (*fmgr_hook)(FHET_END, &(fcache->flinfo), &(fcache->arg));
1217 }1221 }
1218 1222 
1223+ /* restore is_toplevel_stp */
1224+ u_sess->SPI_cxt.is_toplevel_stp = savedisTopLevelForSTP;
1225+ 
1219 return result;1226 return result;
1220}1227}
1221 1228 
Msrc/common/backend/utils/mmgr/portalmem.cpp+174-41
@@ -27,6 +27,8 @@
27#include "utils/memutils.h"27#include "utils/memutils.h"
28#include "utils/plpgsql.h"28#include "utils/plpgsql.h"
29#include "utils/timestamp.h"29#include "utils/timestamp.h"
30+#include "utils/resowner.h"
31+#include "nodes/execnodes.h"
30 32 
31#ifdef PGXC33#ifdef PGXC
32#include "pgxc/pgxc.h"34#include "pgxc/pgxc.h"
@@ -594,6 +596,36 @@ void PortalHashTableDeleteAll(void)
594 }596 }
595}597}
596 598 
599+/*
600+ * "Hold" a portal. Prepare it for access by later transactions.
601+ */
602+static void HoldPortal(Portal portal)
603+{
604+ /*
605+ * Note that PersistHoldablePortal() must release all resources
606+ * used by the portal that are local to the creating transaction.
607+ */
608+ PortalCreateHoldStore(portal);
609+ PersistHoldablePortal(portal);
610+ 
611+ /* drop cached plan reference, if any */
612+ PortalReleaseCachedPlan(portal);
613+ 
614+ /*
615+ * Any resources belonging to the portal will be released in the
616+ * upcoming transaction-wide cleanup; the portal will no longer
617+ * have its own resources.
618+ */
619+ portal->resowner = NULL;
620+ 
621+ /*
622+ * Having successfully exported the holdable cursor, mark it as
623+ * not belonging to this transaction.
624+ */
625+ portal->createSubid = InvalidSubTransactionId;
626+ portal->activeSubid = InvalidSubTransactionId;
627+}
628+ 
597/*629/*
598 * Pre-commit processing for portals.630 * Pre-commit processing for portals.
599 *631 *
@@ -606,7 +638,7 @@ void PortalHashTableDeleteAll(void)
606 * Returns TRUE if any portals changed state (possibly causing user-defined638 * Returns TRUE if any portals changed state (possibly causing user-defined
607 * code to be run), FALSE if not.639 * code to be run), FALSE if not.
608 */640 */
609-bool PreCommit_Portals(bool isPrepare)641+bool PreCommit_Portals(bool isPrepare, bool stpCommit)
610{642{
611 bool result = false;643 bool result = false;
612 HASH_SEQ_STATUS status;644 HASH_SEQ_STATUS status;
@@ -619,12 +651,14 @@ bool PreCommit_Portals(bool isPrepare)
619 651 
620 while ((hentry = (PortalHashEnt*)hash_seq_search(&status)) != NULL) {652 while ((hentry = (PortalHashEnt*)hash_seq_search(&status)) != NULL) {
621 Portal portal = hentry->portal;653 Portal portal = hentry->portal;
654+ ResourceOwner owner = portal->resowner;
622 655 
623 /*656 /*
624 * There should be no pinned portals anymore. Complain if someone657 * There should be no pinned portals anymore. Complain if someone
625- * leaked one.658+ * leaked one. Auto-held portals are allowed; we assume that whoever
659+ * pinned them is managing them.
626 */660 */
627- if (portal->portalPinned)661+ if (portal->portalPinned && !portal->autoHeld)
628 ereport(ERROR, (errcode(ERRCODE_OPERATE_NOT_SUPPORTED), errmsg("cannot commit while a portal is pinned")));662 ereport(ERROR, (errcode(ERRCODE_OPERATE_NOT_SUPPORTED), errmsg("cannot commit while a portal is pinned")));
629 663 
630 /*664 /*
@@ -635,7 +669,22 @@ bool PreCommit_Portals(bool isPrepare)
635 * still going to go away, so don't leave a dangling pointer.669 * still going to go away, so don't leave a dangling pointer.
636 */670 */
637 if (portal->status == PORTAL_ACTIVE) {671 if (portal->status == PORTAL_ACTIVE) {
638- portal->resowner = NULL;672+ /*
673+ * If we are in multi commit and we also have owner, then need to cleanup all the snapshots
674+ * during commit time. Otherwise it will cause leak snapshots reference warning.
675+ */
676+ if (owner && stpCommit) {
677+ ResourceOwnerDecrementNsnapshots(owner, portal->queryDesc);
678+ }
679+ 
680+ /*
681+ * If we are in commit within stored procedure need to keep resowner and it will be used to
682+ * connect new local resources. Otherwise it will cause leak snapshots reference warning, because
683+ * the new snapshot does not have owner.
684+ */
685+ if (!stpCommit) {
686+ portal->resowner = NULL;
687+ }
639 continue;688 continue;
640 }689 }
641 690 
@@ -655,28 +704,7 @@ bool PreCommit_Portals(bool isPrepare)
655 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),704 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
656 errmsg("cannot PREPARE a transaction that has created a cursor WITH HOLD")));705 errmsg("cannot PREPARE a transaction that has created a cursor WITH HOLD")));
657 706 
658- /*707+ HoldPortal(portal);
659- * Note that PersistHoldablePortal() must release all resources
660- * used by the portal that are local to the creating transaction.
661- */
662- PortalCreateHoldStore(portal);
663- PersistHoldablePortal(portal);
664- 
665- /* drop cached plan reference, if any */
666- PortalReleaseCachedPlan(portal);
667- 
668- /*
669- * Any resources belonging to the portal will be released in the
670- * upcoming transaction-wide cleanup; the portal will no longer
671- * have its own resources.
672- */
673- portal->resowner = NULL;
674- 
675- /*
676- * Having successfully exported the holdable cursor, mark it as
677- * not belonging to this transaction.
678- */
679- portal->createSubid = InvalidSubTransactionId;
680 708 
681 /* Report we changed state */709 /* Report we changed state */
682 result = true;710 result = true;
@@ -709,13 +737,13 @@ bool PreCommit_Portals(bool isPrepare)
709/*737/*
710 * Abort processing for portals.738 * Abort processing for portals.
711 *739 *
712- * At this point we reset "active" status and run the cleanup hook if740+ * At this point we run the cleanup hook if present, but we can't release the
713- * present, but we can't release the portal's memory until the cleanup call.741+ * portal's memory until the cleanup call.
714 *742 *
715 * The reason we need to reset active is so that we can replace the unnamed743 * The reason we need to reset active is so that we can replace the unnamed
716 * portal, else we'll fail to execute ROLLBACK when it arrives.744 * portal, else we'll fail to execute ROLLBACK when it arrives.
717 */745 */
718-void AtAbort_Portals(void)746+void AtAbort_Portals(bool stpRollback)
719{747{
720 HASH_SEQ_STATUS status;748 HASH_SEQ_STATUS status;
721 PortalHashEnt* hentry = NULL;749 PortalHashEnt* hentry = NULL;
@@ -728,16 +756,28 @@ void AtAbort_Portals(void)
728 while ((hentry = (PortalHashEnt*)hash_seq_search(&status)) != NULL) {756 while ((hentry = (PortalHashEnt*)hash_seq_search(&status)) != NULL) {
729 Portal portal = hentry->portal;757 Portal portal = hentry->portal;
730 758 
731- /* Any portal that was actually running has to be considered broken */
732- if (portal->status == PORTAL_ACTIVE)
733- MarkPortalFailed(portal);
734- 
735 /*759 /*
736 * Do nothing else to cursors held over from a previous transaction.760 * Do nothing else to cursors held over from a previous transaction.
737 */761 */
738 if (portal->createSubid == InvalidSubTransactionId)762 if (portal->createSubid == InvalidSubTransactionId)
739 continue;763 continue;
740 764 
765+ /*
766+ * Do nothing to auto-held cursors. This is similar to the case of a
767+ * cursor from a previous transaction, but it could also be that the
768+ * cursor was auto-held in this trasnaction, so it wants to live on.
769+ */
770+ if (portal->autoHeld)
771+ continue;
772+ 
773+ /*
774+ * Within multi rollback need to clean up snapshots before release
775+ * its resource owner.
776+ */
777+ if (portal->resowner && stpRollback) {
778+ ResourceOwnerDecrementNsnapshots(portal->resowner, portal->queryDesc);
779+ }
780+ 
741 /*781 /*
742 * If it was created in the current transaction, we can't do normal782 * If it was created in the current transaction, we can't do normal
743 * shutdown on a READY portal either; it might refer to objects783 * shutdown on a READY portal either; it might refer to objects
@@ -749,9 +789,10 @@ void AtAbort_Portals(void)
749 789 
750 /*790 /*
751 * Allow portalcmds.c to clean up the state it knows about, if we791 * Allow portalcmds.c to clean up the state it knows about, if we
752- * haven't already.792+ * haven't already. Should haven't already. Should not clean up portal
793+ * during multi commit and rollback.
753 */794 */
754- if (PointerIsValid(portal->cleanup)) {795+ if (PointerIsValid(portal->cleanup) && !stpRollback) {
755 (*portal->cleanup)(portal);796 (*portal->cleanup)(portal);
756 portal->cleanup = NULL;797 portal->cleanup = NULL;
757 }798 }
@@ -762,17 +803,24 @@ void AtAbort_Portals(void)
762 /*803 /*
763 * Any resources belonging to the portal will be released in the804 * Any resources belonging to the portal will be released in the
764 * upcoming transaction-wide cleanup; they will be gone before we run805 * upcoming transaction-wide cleanup; they will be gone before we run
765- * PortalDrop.806+ * PortalDrop. Can not reset resowner NULL if
807+ * we are in stpRollback, because the Portal is still alive. If we set
808+ * resowner to NULL it will cause leak snapshots reference error, because
809+ * the new snaphosts does not have owner.
766 */810 */
767- portal->resowner = NULL;811+ if (!stpRollback) {
812+ portal->resowner = NULL;
813+ }
768 814 
769 /*815 /*
770 * Although we can't delete the portal data structure proper, we can816 * Although we can't delete the portal data structure proper, we can
771 * release any memory in subsidiary contexts, such as executor state.817 * release any memory in subsidiary contexts, such as executor state.
772 * The cleanup hook was the last thing that might have needed data818 * The cleanup hook was the last thing that might have needed data
773- * there.819+ * there. But leave active portals alone.
774 */820 */
775- MemoryContextDeleteChildren(PortalGetHeapMemory(portal));821+ if (portal->status != PORTAL_ACTIVE) {
822+ MemoryContextDeleteChildren(PortalGetHeapMemory(portal));
823+ }
776 }824 }
777}825}
778 826 
@@ -793,8 +841,18 @@ void AtCleanup_Portals(void)
793 while ((hentry = (PortalHashEnt*)hash_seq_search(&status)) != NULL) {841 while ((hentry = (PortalHashEnt*)hash_seq_search(&status)) != NULL) {
794 Portal portal = hentry->portal;842 Portal portal = hentry->portal;
795 843 
796- /* Do nothing to cursors held over from a previous transaction */844+ /*
797- if (portal->createSubid == InvalidSubTransactionId) {845+ * Do not touch active portals --- this can only happen in the case of
846+ * a multi-transaction command.
847+ */
848+ if (portal->status == PORTAL_ACTIVE)
849+ continue;
850+ 
851+ /*
852+ * Do nothing to cursors held over from a previous transaction or
853+ * auto-held ones.
854+ */
855+ if (portal->createSubid == InvalidSubTransactionId || portal->autoHeld) {
798 Assert(portal->status != PORTAL_ACTIVE);856 Assert(portal->status != PORTAL_ACTIVE);
799 Assert(portal->resowner == NULL);857 Assert(portal->resowner == NULL);
800 continue;858 continue;
@@ -820,6 +878,29 @@ void AtCleanup_Portals(void)
820 }878 }
821}879}
822 880 
881+/*
882+ * Potal-related cleanup when we return to the main loop on error.
883+ *
884+ * This is different from the cleanup at transaction abort. Auto-held portals
885+ * are cleaned up on error but not on transaction abort.
886+ */
887+void PortalErrorCleanup(void)
888+{
889+ HASH_SEQ_STATUS status;
890+ PortalHashEnt* hentry = NULL;
891+ 
892+ hash_seq_init(&status, u_sess->exec_cxt.PortalHashTable);
893+ 
894+ while ((hentry = (PortalHashEnt *)hash_seq_search(&status)) != NULL) {
895+ Portal portal = hentry->portal;
896+ 
897+ if (portal->autoHeld) {
898+ portal->portalPinned = false;
899+ PortalDrop(portal, false);
900+ }
901+ }
902+}
903+ 
823/*904/*
824 * Pre-subcommit processing for portals.905 * Pre-subcommit processing for portals.
825 *906 *
@@ -1132,3 +1213,55 @@ void ResetPortalCursor(SubTransactionId mySubid, Oid funOid, int funUseCount)
1132 ResetCursorOption(portal, true);1213 ResetCursorOption(portal, true);
1133 }1214 }
1134}1215}
1216+ 
1217+/*
1218+ * Hold all pinned portals.
1219+ *
1220+ * When initialing a COMMIT or ROLLBACK insise a procedure, this must be
1221+ * called to protect internally-generated cursors from being dropped during
1222+ * the transaction shutdown. Currently, SPI calls this automatically; PLs
1223+ * that initiate COMMIT or ROLLBACK some other way are on the hook to do it
1224+ * themselves. (Note that we couldn't do this in, say, AtAbort_Portals
1225+ * because we need to run user-defined code while persisting a portal.
1226+ * It's too late to do that once transaction abort has started.)
1227+ *
1228+ * We protect such portals by converting them to held cursors. We mark them
1229+ * as "auto-held" so that exception exit knowns to clean them up. (In normal,
1230+ * non-exception code paths, the PL needs to clean such portals itself, since
1231+ * transaction end won't do it anymore; but that should be normal practice
1232+ * anyway.)
1233+ */
1234+void HoldPinnedPortals(void)
1235+{
1236+ HASH_SEQ_STATUS status;
1237+ PortalHashEnt* hentry = NULL;
1238+ 
1239+ hash_seq_init(&status, u_sess->exec_cxt.PortalHashTable);
1240+ 
1241+ while ((hentry = (PortalHashEnt *) hash_seq_search(&status)) != NULL) {
1242+ Portal portal = hentry->portal;
1243+ 
1244+ if (portal->portalPinned && !portal->autoHeld) {
1245+ /*
1246+ * Doing transaction control, especially abort, inside a cursor
1247+ * loop that is not read-only, for example using UPDATE
1248+ * ... RETURNING, has weird semantics issues. Also, this
1249+ * implementation wouldn't work, because such portals cannot be
1250+ * held. (The core grammer enforces that only SELECT statements
1251+ * can drive a cursor, but for example PL/pgSQL does not restrict
1252+ * it.)
1253+ */
1254+ if (portal->strategy != PORTAL_ONE_SELECT)
1255+ ereport(ERROR,
1256+ (errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
1257+ errmsg("cannot perform transaction commands inside a cursor loop that is not read-only")));
1258+ 
1259+ /* Verify it's in a suitable state to be held */
1260+ if (portal->status != PORTAL_READY)
1261+ elog(ERROR, "pinned portal is not ready to be auto-held");
1262+ 
1263+ HoldPortal(portal);
1264+ portal->autoHeld = true;
1265+ }
1266+ }
1267+}
Msrc/common/backend/utils/resowner/resowner.cpp+77-0
@@ -431,6 +431,10 @@ void ResourceOwnerDelete(ResourceOwner owner)
431 */431 */
432 ResourceOwnerNewParent(owner, NULL);432 ResourceOwnerNewParent(owner, NULL);
433 433 
434+ if (owner == t_thrd.utils_cxt.StpSavedResourceOwner) {
435+ return;
436+ }
437+ 
434 /* And free the object. */438 /* And free the object. */
435 if (owner->buffers)439 if (owner->buffers)
436 pfree(owner->buffers);440 pfree(owner->buffers);
@@ -469,6 +473,30 @@ ResourceOwner ResourceOwnerGetParent(ResourceOwner owner)
469 return owner->parent;473 return owner->parent;
470}474}
471 475 
476+/*
477+ * Fetch nextchild of a ResourceOwner (returns)
478+ */
479+ResourceOwner ResourceOwnerGetNextChild(ResourceOwner owner)
480+{
481+ return owner->nextchild;
482+}
483+ 
484+/*
485+ * Fetch name of a ResourceOwner (should always has value)
486+ */
487+const char* ResourceOwnerGetName(ResourceOwner owner)
488+{
489+ return owner->name;
490+}
491+ 
492+/*
493+ * Fetch firstchild of a ResourceOwner.
494+ */
495+ResourceOwner ResourceOwnerGetFirstChild(ResourceOwner owner)
496+{
497+ return owner->firstchild;
498+}
499+ 
472/*500/*
473 * Reassign a ResourceOwner to have a new parent501 * Reassign a ResourceOwner to have a new parent
474 */502 */
@@ -1320,6 +1348,55 @@ void ResourceOwnerForgetSnapshot(ResourceOwner owner, const Snapshot snapshot)
1320 errmsg("snapshot is not owned by resource owner %s", owner->name)));1348 errmsg("snapshot is not owned by resource owner %s", owner->name)));
1321}1349}
1322 1350 
1351+/*
1352+ * This function is used to clean up the snapshots.
1353+ * It will be called by PreCommit_Portals and Abort_Portals.
1354+ */
1355+void ResourceOwnerDecrementNsnapshots(ResourceOwner owner, void* queryDesc)
1356+{
1357+ QueryDesc* queryDescTemp = (QueryDesc*)queryDesc;
1358+ 
1359+ while (owner->nsnapshots > 0) {
1360+ if (queryDescTemp) {
1361+ /*
1362+ * check if owner's snapshot is same as queryDesc's snapshot, need to set queryDesc
1363+ * snapshot to null, because this function will clean up those snapshots.
1364+ */
1365+ if (owner->snapshots[owner->nsnapshots - 1] == queryDescTemp->estate->es_snapshot) {
1366+ queryDescTemp->estate->es_snapshot = NULL;
1367+ }
1368+ 
1369+ if (owner->snapshots[owner->nsnapshots - 1] == queryDescTemp->estate->es_crosscheck_snapshot) {
1370+ queryDescTemp->estate->es_crosscheck_snapshot = NULL;
1371+ }
1372+ 
1373+ if (owner->snapshots[owner->nsnapshots - 1] == queryDescTemp->snapshot) {
1374+ queryDescTemp->snapshot = NULL;
1375+ }
1376+ 
1377+ if (owner->snapshots[owner->nsnapshots - 1] == queryDescTemp->crosscheck_snapshot) {
1378+ queryDescTemp->crosscheck_snapshot = NULL;
1379+ }
1380+ }
1381+ 
1382+ UnregisterSnapshotFromOwner(owner->snapshots[owner->nsnapshots - 1], owner);
1383+ }
1384+}
1385+ 
1386+/*
1387+ * This function is used to clean up the cached plan.
1388+ * It will ba called by CommitTransaction.
1389+ */
1390+void ResourceOwnerDecrementNPlanRefs(ResourceOwner owner, bool useResOwner)
1391+{
1392+ if (!owner) {
1393+ return;
1394+ }
1395+ 
1396+ while (owner->nplanrefs > 0) {
1397+ ReleaseCachedPlan(owner->planrefs[owner->nplanrefs - 1], useResOwner);
1398+ }
1399+}
1323/*1400/*
1324 * Debugging subroutine1401 * Debugging subroutine
1325 */1402 */
Msrc/common/backend/utils/time/snapmgr.cpp+8-0
@@ -534,6 +534,14 @@ void UpdateActiveSnapshotCommandId(void)
534 */534 */
535void PopActiveSnapshot(void)535void PopActiveSnapshot(void)
536{536{
537+ /*
538+ * In multi commit/rollback within stored procedure, the ActiveSnapshot already poped.
539+ * Therefore, no need to pop the active snapshot. Otherwise it will cause seg fault.
540+ */
541+ if (!u_sess->utils_cxt.ActiveSnapshot) {
542+ return;
543+ }
544+ 
537 ActiveSnapshotElt* newstack = NULL;545 ActiveSnapshotElt* newstack = NULL;
538 546 
539 newstack = u_sess->utils_cxt.ActiveSnapshot->as_next;547 newstack = u_sess->utils_cxt.ActiveSnapshot->as_next;
Msrc/common/pl/plpgsql/src/pl_exec.cpp+201-13
@@ -130,6 +130,8 @@ static bool is_anonymous_block(const char* query);
130static int exec_stmt_dynfors(PLpgSQL_execstate* estate, PLpgSQL_stmt_dynfors* stmt);130static int exec_stmt_dynfors(PLpgSQL_execstate* estate, PLpgSQL_stmt_dynfors* stmt);
131 131 
132static void plpgsql_estate_setup(PLpgSQL_execstate* estate, PLpgSQL_function* func, ReturnSetInfo* rsi);132static void plpgsql_estate_setup(PLpgSQL_execstate* estate, PLpgSQL_function* func, ReturnSetInfo* rsi);
133+static int exec_stmt_commit(PLpgSQL_execstate* estate, PLpgSQL_stmt_commit* stmt);
134+static int exec_stmt_rollback(PLpgSQL_execstate* estate, PLpgSQL_stmt_rollback* stmt);
133static void exec_eval_cleanup(PLpgSQL_execstate* estate);135static void exec_eval_cleanup(PLpgSQL_execstate* estate);
134 136 
135static void exec_prepare_plan(PLpgSQL_execstate* estate, PLpgSQL_expr* expr, int cursorOptions);137static void exec_prepare_plan(PLpgSQL_execstate* estate, PLpgSQL_expr* expr, int cursorOptions);
@@ -291,6 +293,7 @@ Datum plpgsql_exec_function(PLpgSQL_function* func, FunctionCallInfo fcinfo, boo
291{293{
292 PLpgSQL_execstate estate;294 PLpgSQL_execstate estate;
293 ErrorContextCallback plerrcontext;295 ErrorContextCallback plerrcontext;
296+ bool savedIsStp;
294 int i;297 int i;
295 int rc;298 int rc;
296 299 
@@ -412,7 +415,9 @@ Datum plpgsql_exec_function(PLpgSQL_function* func, FunctionCallInfo fcinfo, boo
412 */415 */
413 estate.err_text = NULL;416 estate.err_text = NULL;
414 estate.err_stmt = (PLpgSQL_stmt*)(func->action);417 estate.err_stmt = (PLpgSQL_stmt*)(func->action);
418+ savedIsStp = u_sess->SPI_cxt.is_stp;
415 rc = exec_stmt_block(&estate, func->action);419 rc = exec_stmt_block(&estate, func->action);
420+ u_sess->SPI_cxt.is_stp = savedIsStp;
416 if (rc != PLPGSQL_RC_RETURN) {421 if (rc != PLPGSQL_RC_RETURN) {
417 estate.err_stmt = NULL;422 estate.err_stmt = NULL;
418 estate.err_text = NULL;423 estate.err_text = NULL;
@@ -633,6 +638,7 @@ HeapTuple plpgsql_exec_trigger(PLpgSQL_function* func, TriggerData* trigdata)
633 PLpgSQL_rec *rec_new = NULL;638 PLpgSQL_rec *rec_new = NULL;
634 PLpgSQL_rec *rec_old = NULL;639 PLpgSQL_rec *rec_old = NULL;
635 HeapTuple rettup;640 HeapTuple rettup;
641+ bool saveIsStp;
636 642 
637 /*643 /*
638 * Setup the execution state644 * Setup the execution state
@@ -842,7 +848,10 @@ HeapTuple plpgsql_exec_trigger(PLpgSQL_function* func, TriggerData* trigdata)
842 */848 */
843 estate.err_text = NULL;849 estate.err_text = NULL;
844 estate.err_stmt = (PLpgSQL_stmt*)(func->action);850 estate.err_stmt = (PLpgSQL_stmt*)(func->action);
851+ saveIsStp = u_sess->SPI_cxt.is_stp;
852+ u_sess->SPI_cxt.is_stp = false;
845 rc = exec_stmt_block(&estate, func->action);853 rc = exec_stmt_block(&estate, func->action);
854+ u_sess->SPI_cxt.is_stp = saveIsStp;
846 if (rc != PLPGSQL_RC_RETURN) {855 if (rc != PLPGSQL_RC_RETURN) {
847 estate.err_stmt = NULL;856 estate.err_stmt = NULL;
848 estate.err_text = NULL;857 estate.err_text = NULL;
@@ -1399,6 +1408,10 @@ static int exec_stmt_block(PLpgSQL_execstate* estate, PLpgSQL_stmt_block* block)
1399 int i;1408 int i;
1400 int n;1409 int n;
1401 SubTransactionId subXid = InvalidSubTransactionId;1410 SubTransactionId subXid = InvalidSubTransactionId;
1411+ bool savedIsTopLevelForStp = u_sess->SPI_cxt.is_toplevel_stp;
1412+ bool savedIsStp = u_sess->SPI_cxt.is_stp;
1413+ TransactionId oldTransactionId = InvalidTransactionId;
1414+ 
1402 /*1415 /*
1403 * First initialize all variables declared in this block1416 * First initialize all variables declared in this block
1404 */1417 */
@@ -1477,13 +1490,17 @@ static int exec_stmt_block(PLpgSQL_execstate* estate, PLpgSQL_stmt_block* block)
1477 }1490 }
1478 1491 
1479 if (block->exceptions != NULL) {1492 if (block->exceptions != NULL) {
1493+ u_sess->SPI_cxt.portal_stp_exception_counter++;
1494+ 
1480 /*1495 /*
1481 * Execute the statements in the block's body inside a sub-transaction1496 * Execute the statements in the block's body inside a sub-transaction
1482 */1497 */
1483 MemoryContext oldcontext = CurrentMemoryContext;1498 MemoryContext oldcontext = CurrentMemoryContext;
1484 ResourceOwner oldowner = t_thrd.utils_cxt.CurrentResourceOwner;1499 ResourceOwner oldowner = t_thrd.utils_cxt.CurrentResourceOwner;
1485- ExprContext* old_eval_econtext = estate->eval_econtext;
1486 ErrorData* save_cur_error = estate->cur_error;1500 ErrorData* save_cur_error = estate->cur_error;
1501+ if (!RecoveryInProgress()) {
1502+ oldTransactionId = GetTopTransactionId();
1503+ }
1487 1504 
1488 estate->err_text = gettext_noop("during statement block entry");1505 estate->err_text = gettext_noop("during statement block entry");
1489 1506 
@@ -1539,13 +1556,25 @@ static int exec_stmt_block(PLpgSQL_execstate* estate, PLpgSQL_stmt_block* block)
1539 }1556 }
1540 1557 
1541 MemoryContextSwitchTo(oldcontext);1558 MemoryContextSwitchTo(oldcontext);
1559+ if ((!RecoveryInProgress()) && (oldTransactionId != GetTopTransactionId())) {
1560+ if (ResourceOwnerGetNextChild(t_thrd.utils_cxt.CurrentResourceOwner)) {
1561+ oldowner = ResourceOwnerGetNextChild(t_thrd.utils_cxt.CurrentResourceOwner);
1562+ } else {
1563+ if (ResourceOwnerGetParent(t_thrd.utils_cxt.CurrentResourceOwner)) {
1564+ oldowner = ResourceOwnerGetParent(t_thrd.utils_cxt.CurrentResourceOwner);
1565+ } else {
1566+ oldowner = ResourceOwnerGetFirstChild(t_thrd.utils_cxt.CurrentResourceOwner);
1567+ }
1568+ }
1569+ }
1542 t_thrd.utils_cxt.CurrentResourceOwner = oldowner;1570 t_thrd.utils_cxt.CurrentResourceOwner = oldowner;
1571+ u_sess->SPI_cxt.portal_stp_exception_counter--;
1543 1572 
1544 /*1573 /*
1545 * Revert to outer eval_econtext. (The inner one was1574 * Revert to outer eval_econtext. (The inner one was
1546 * automatically cleaned up during subxact exit.)1575 * automatically cleaned up during subxact exit.)
1547 */1576 */
1548- estate->eval_econtext = old_eval_econtext;1577+ estate->eval_econtext = u_sess->plsql_cxt.simple_econtext_stack->stack_econtext;
1549 1578 
1550 /*1579 /*
1551 * AtEOSubXact_SPI() should not have popped any SPI context, but1580 * AtEOSubXact_SPI() should not have popped any SPI context, but
@@ -1557,6 +1586,10 @@ static int exec_stmt_block(PLpgSQL_execstate* estate, PLpgSQL_stmt_block* block)
1557 {1586 {
1558 ErrorData* edata = NULL;1587 ErrorData* edata = NULL;
1559 ListCell* e = NULL;1588 ListCell* e = NULL;
1589+ 
1590+ u_sess->SPI_cxt.is_toplevel_stp = savedIsTopLevelForStp;
1591+ u_sess->SPI_cxt.is_stp = savedIsStp;
1592+ 
1560 estate->cursor_return_data = saved_cursor_data;1593 estate->cursor_return_data = saved_cursor_data;
1561 1594 
1562 /* gs_signal_handle maybe block sigusr2 when accept SIGINT */1595 /* gs_signal_handle maybe block sigusr2 when accept SIGINT */
@@ -1601,10 +1634,22 @@ static int exec_stmt_block(PLpgSQL_execstate* estate, PLpgSQL_stmt_block* block)
1601 }1634 }
1602 1635 
1603 MemoryContextSwitchTo(oldcontext);1636 MemoryContextSwitchTo(oldcontext);
1637+ if ((!RecoveryInProgress()) && (oldTransactionId != GetTopTransactionId())) {
1638+ if (ResourceOwnerGetNextChild(t_thrd.utils_cxt.CurrentResourceOwner)) {
1639+ oldowner = ResourceOwnerGetNextChild(t_thrd.utils_cxt.CurrentResourceOwner);
1640+ } else {
1641+ if (ResourceOwnerGetParent(t_thrd.utils_cxt.CurrentResourceOwner)) {
1642+ oldowner = ResourceOwnerGetParent(t_thrd.utils_cxt.CurrentResourceOwner);
1643+ } else {
1644+ oldowner = ResourceOwnerGetFirstChild(t_thrd.utils_cxt.CurrentResourceOwner);
1645+ }
1646+ }
1647+ }
1604 t_thrd.utils_cxt.CurrentResourceOwner = oldowner;1648 t_thrd.utils_cxt.CurrentResourceOwner = oldowner;
1605 1649 
1606 /* Revert to outer eval_econtext */1650 /* Revert to outer eval_econtext */
1607- estate->eval_econtext = old_eval_econtext;1651+ estate->eval_econtext = u_sess->plsql_cxt.simple_econtext_stack->stack_econtext;
1652+ u_sess->SPI_cxt.portal_stp_exception_counter--;
1608 1653 
1609 /*1654 /*
1610 * If AtEOSubXact_SPI() popped any SPI context of the subxact, it1655 * If AtEOSubXact_SPI() popped any SPI context of the subxact, it
@@ -2061,14 +2106,28 @@ static int exec_stmt_assign(PLpgSQL_execstate* estate, PLpgSQL_stmt_assign* stmt
2061static int exec_stmt_perform(PLpgSQL_execstate* estate, PLpgSQL_stmt_perform* stmt)2106static int exec_stmt_perform(PLpgSQL_execstate* estate, PLpgSQL_stmt_perform* stmt)
2062{2107{
2063 PLpgSQL_expr* expr = stmt->expr;2108 PLpgSQL_expr* expr = stmt->expr;
2109+ TransactionId oldTransactionId = InvalidTransactionId;
2064 int rc;2110 int rc;
2065 2111 
2112+ if (!RecoveryInProgress()) {
2113+ oldTransactionId = GetTopTransactionId();
2114+ }
2115+ 
2066 rc = exec_run_select(estate, expr, 0, NULL);2116 rc = exec_run_select(estate, expr, 0, NULL);
2067 if (rc != SPI_OK_SELECT) {2117 if (rc != SPI_OK_SELECT) {
2068 ereport(DEBUG1,2118 ereport(DEBUG1,
2069 (errcode(ERRCODE_WRONG_OBJECT_TYPE), errmodule(MOD_PLSQL), errmsg("exec_run_select returns %d", rc)));2119 (errcode(ERRCODE_WRONG_OBJECT_TYPE), errmodule(MOD_PLSQL), errmsg("exec_run_select returns %d", rc)));
2070 }2120 }
2071 2121 
2122+ /*
2123+ * This is used for nested STP. If the transaction Id changed,
2124+ * then need to create new econtext for the TopTransaction.
2125+ */
2126+ if ((!RecoveryInProgress()) && (oldTransactionId != GetTopTransactionId())) {
2127+ u_sess->plsql_cxt.simple_eval_estate = NULL;
2128+ plpgsql_create_econtext(estate);
2129+ }
2130+ 
2072 exec_set_found(estate, (estate->eval_processed != 0));2131 exec_set_found(estate, (estate->eval_processed != 0));
2073 exec_eval_cleanup(estate);2132 exec_eval_cleanup(estate);
2074 2133 
@@ -3695,7 +3754,7 @@ static void exec_eval_cleanup(PLpgSQL_execstate* estate)
3695 estate->eval_tuptable = NULL;3754 estate->eval_tuptable = NULL;
3696 3755 
3697 /* Clear result of exec_eval_simple_expr (but keep the econtext) */3756 /* Clear result of exec_eval_simple_expr (but keep the econtext) */
3698- if (estate->eval_econtext != NULL) {3757+ if (estate->eval_econtext != NULL && estate->eval_econtext->ecxt_per_tuple_memory != NULL) {
3699 ResetExprContext(estate->eval_econtext);3758 ResetExprContext(estate->eval_econtext);
3700 }3759 }
3701}3760}
@@ -3762,6 +3821,11 @@ static int exec_stmt_execsql(PLpgSQL_execstate* estate, PLpgSQL_stmt_execsql* st
3762 int rc;3821 int rc;
3763 PLpgSQL_expr* expr = stmt->sqlstmt;3822 PLpgSQL_expr* expr = stmt->sqlstmt;
3764 Cursor_Data* saved_cursor_data = NULL;3823 Cursor_Data* saved_cursor_data = NULL;
3824+ TransactionId oldTransactionId = InvalidTransactionId;
3825+ 
3826+ if (!RecoveryInProgress()) {
3827+ oldTransactionId = GetTopTransactionId();
3828+ }
3765 3829 
3766 /*3830 /*
3767 * On the first call for this statement generate the plan, and detect3831 * On the first call for this statement generate the plan, and detect
@@ -3836,6 +3900,16 @@ static int exec_stmt_execsql(PLpgSQL_execstate* estate, PLpgSQL_stmt_execsql* st
3836 * Execute the plan3900 * Execute the plan
3837 */3901 */
3838 rc = SPI_execute_plan_with_paramlist(expr->plan, paramLI, estate->readonly_func, tcount);3902 rc = SPI_execute_plan_with_paramlist(expr->plan, paramLI, estate->readonly_func, tcount);
3903+ 
3904+ /*
3905+ * This is used for nested STP. If the transaction Id changed,
3906+ * then need to create new econtext for the TopTransaction.
3907+ */
3908+ if ((!RecoveryInProgress()) && (oldTransactionId != GetTopTransactionId())) {
3909+ u_sess->plsql_cxt.simple_eval_estate = NULL;
3910+ plpgsql_create_econtext(estate);
3911+ }
3912+ 
3839 plpgsql_estate = NULL;3913 plpgsql_estate = NULL;
3840 3914 
3841 /*3915 /*
@@ -4179,6 +4253,11 @@ static int exec_stmt_dynexecute(PLpgSQL_execstate* estate, PLpgSQL_stmt_dynexecu
4179 FmgrInfo flinfo;4253 FmgrInfo flinfo;
4180 int ppdindex = 0;4254 int ppdindex = 0;
4181 int datumindex = 0;4255 int datumindex = 0;
4256+ TransactionId oldTransactionId = InvalidTransactionId;
4257+ 
4258+ if (!RecoveryInProgress()) {
4259+ oldTransactionId = GetTopTransactionId();
4260+ }
4182 4261 
4183 /* Compile the anonymous code block */4262 /* Compile the anonymous code block */
4184 /* support pass external parameter in anonymous block */4263 /* support pass external parameter in anonymous block */
@@ -4207,6 +4286,16 @@ static int exec_stmt_dynexecute(PLpgSQL_execstate* estate, PLpgSQL_stmt_dynexecu
4207 flinfo.fn_mcxt = CurrentMemoryContext;4286 flinfo.fn_mcxt = CurrentMemoryContext;
4208 4287 
4209 (void)plpgsql_exec_function(func, &fake_fcinfo, true);4288 (void)plpgsql_exec_function(func, &fake_fcinfo, true);
4289+ 
4290+ /*
4291+ * This is used for nested STP. If the transaction Id changed,
4292+ * then need to create new econtext for the TopTransaction.
4293+ */
4294+ if ((!RecoveryInProgress()) && (oldTransactionId != GetTopTransactionId())) {
4295+ u_sess->plsql_cxt.simple_eval_estate = NULL;
4296+ plpgsql_create_econtext(estate);
4297+ }
4298+ 
4210 exec_set_sql_isopen(estate, false);4299 exec_set_sql_isopen(estate, false);
4211 exec_set_sql_cursor_found(estate, PLPGSQL_TRUE);4300 exec_set_sql_cursor_found(estate, PLPGSQL_TRUE);
4212 exec_set_sql_notfound(estate, PLPGSQL_FALSE);4301 exec_set_sql_notfound(estate, PLPGSQL_FALSE);
@@ -4895,6 +4984,58 @@ static int exec_stmt_null(PLpgSQL_execstate* estate, PLpgSQL_stmt* stmt)
4895 */4984 */
4896static int exec_stmt_commit(PLpgSQL_execstate* estate, PLpgSQL_stmt_commit* stmt)4985static int exec_stmt_commit(PLpgSQL_execstate* estate, PLpgSQL_stmt_commit* stmt)
4897{4986{
4987+ const char* PORTAL = "Portal";
4988+ int subTransactionCount = u_sess->SPI_cxt.portal_stp_exception_counter;
4989+ 
4990+ if (u_sess->SPI_cxt.portal_stp_exception_counter == 0) {
4991+ t_thrd.utils_cxt.StpSavedResourceOwner = t_thrd.utils_cxt.CurrentResourceOwner;
4992+ }
4993+ 
4994+ if (strcmp(PORTAL, ResourceOwnerGetName(t_thrd.utils_cxt.CurrentResourceOwner)) == 0) {
4995+ if (ResourceOwnerGetNextChild(t_thrd.utils_cxt.CurrentResourceOwner)
4996+ && (strcmp(PORTAL, ResourceOwnerGetName(ResourceOwnerGetNextChild(t_thrd.utils_cxt.CurrentResourceOwner))) == 0))
4997+ ereport(ERROR,
4998+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
4999+ errmsg("commit with PE is not supported")));
5000+ }
5001+ 
5002+ SPI_commit();
5003+ SPI_start_transaction();
5004+ 
5005+ u_sess->plsql_cxt.simple_eval_estate = NULL;
5006+ plpgsql_create_econtext(estate);
5007+ 
5008+ /* link portal to new TopTransaction */
5009+ ResourceOwnerNewParent(t_thrd.utils_cxt.StpSavedResourceOwner, t_thrd.utils_cxt.CurrentResourceOwner);
5010+ t_thrd.utils_cxt.CurrentResourceOwner = t_thrd.utils_cxt.StpSavedResourceOwner;
5011+ 
5012+ while (subTransactionCount > 0) {
5013+ if (u_sess->SPI_cxt.portal_stp_exception_counter > 0) {
5014+ MemoryContext oldcontext = CurrentMemoryContext;
5015+ 
5016+ estate->err_text = gettext_noop("during statement block entry");
5017+ 
5018+ /* CN should send savesopint command to remote nodes to begin sub transaction remotely */
5019+ if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
5020+ pgxc_node_remote_savepoint("Savepoint s1", EXEC_ON_ALL_NODES, true, true);
5021+ }
5022+ 
5023+ BeginInternalSubTransaction(NULL);
5024+ 
5025+ /* Want to run statements inside function's memory context */
5026+ MemoryContextSwitchTo(oldcontext);
5027+ 
5028+ plpgsql_create_econtext(estate);
5029+ estate->err_text = NULL;
5030+ }
5031+ 
5032+ subTransactionCount--;
5033+ }
5034+ 
5035+ if (u_sess->SPI_cxt.portal_stp_exception_counter == 0) {
5036+ t_thrd.utils_cxt.CurrentResourceOwner = t_thrd.utils_cxt.StpSavedResourceOwner;
5037+ }
5038+ 
4898 return PLPGSQL_RC_OK;5039 return PLPGSQL_RC_OK;
4899}5040}
4900 5041 
@@ -4905,6 +5046,58 @@ static int exec_stmt_commit(PLpgSQL_execstate* estate, PLpgSQL_stmt_commit* stmt
4905 */5046 */
4906static int exec_stmt_rollback(PLpgSQL_execstate* estate, PLpgSQL_stmt_rollback* stmt)5047static int exec_stmt_rollback(PLpgSQL_execstate* estate, PLpgSQL_stmt_rollback* stmt)
4907{5048{
5049+ const char* PORTAL = "Portal";
5050+ int subTransactionCount = u_sess->SPI_cxt.portal_stp_exception_counter;
5051+ 
5052+ if (u_sess->SPI_cxt.portal_stp_exception_counter == 0) {
5053+ t_thrd.utils_cxt.StpSavedResourceOwner = t_thrd.utils_cxt.CurrentResourceOwner;
5054+ }
5055+ 
5056+ if (strcmp(PORTAL, ResourceOwnerGetName(t_thrd.utils_cxt.CurrentResourceOwner)) == 0) {
5057+ if (ResourceOwnerGetNextChild(t_thrd.utils_cxt.CurrentResourceOwner)
5058+ && (strcmp(PORTAL, ResourceOwnerGetName(ResourceOwnerGetNextChild(t_thrd.utils_cxt.CurrentResourceOwner))) == 0))
5059+ ereport(ERROR,
5060+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
5061+ errmsg("commit with PE is not supported")));
5062+ }
5063+ 
5064+ SPI_rollback();
5065+ SPI_start_transaction();
5066+ 
5067+ u_sess->plsql_cxt.simple_eval_estate = NULL;
5068+ plpgsql_create_econtext(estate);
5069+ 
5070+ /* link portal to new TopTransaction */
5071+ ResourceOwnerNewParent(t_thrd.utils_cxt.StpSavedResourceOwner, t_thrd.utils_cxt.CurrentResourceOwner);
5072+ t_thrd.utils_cxt.CurrentResourceOwner = t_thrd.utils_cxt.StpSavedResourceOwner;
5073+ 
5074+ while (subTransactionCount > 0) {
5075+ if (u_sess->SPI_cxt.portal_stp_exception_counter > 0) {
5076+ MemoryContext oldcontext = CurrentMemoryContext;
5077+ 
5078+ estate->err_text = gettext_noop("during statement block entry");
5079+ 
5080+ /* CN should send savesopint command to remote nodes to begin sub transaction remotely */
5081+ if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
5082+ pgxc_node_remote_savepoint("Savepoint s1", EXEC_ON_ALL_NODES, true, true);
5083+ }
5084+ 
5085+ BeginInternalSubTransaction(NULL);
5086+ 
5087+ /* Want to run statements inside function's memory context */
5088+ MemoryContextSwitchTo(oldcontext);
5089+ 
5090+ plpgsql_create_econtext(estate);
5091+ estate->err_text = NULL;
5092+ }
5093+ 
5094+ subTransactionCount--;
5095+ }
5096+ 
5097+ if (u_sess->SPI_cxt.portal_stp_exception_counter == 0) {
5098+ t_thrd.utils_cxt.CurrentResourceOwner = t_thrd.utils_cxt.StpSavedResourceOwner;
5099+ }
5100+ 
4908 return PLPGSQL_RC_OK;5101 return PLPGSQL_RC_OK;
4909}5102}
4910 5103 
@@ -7555,26 +7748,21 @@ static void plpgsql_destroy_econtext(PLpgSQL_execstate* estate)
7555 */7748 */
7556void plpgsql_xact_cb(XactEvent event, void* arg)7749void plpgsql_xact_cb(XactEvent event, void* arg)
7557{7750{
7751+ u_sess->plsql_cxt.simple_eval_estate = NULL;
7752+ 
7558 /*7753 /*
7559 * If we are doing a clean transaction shutdown, free the EState (so that7754 * If we are doing a clean transaction shutdown, free the EState (so that
7560 * any remaining resources will be released correctly). In an abort, we7755 * any remaining resources will be released correctly). In an abort, we
7561 * expect the regular abort recovery procedures to release everything of7756 * expect the regular abort recovery procedures to release everything of
7562 * interest.7757 * interest.
7563 */7758 */
7564- if (event == XACT_EVENT_PREROLLBACK_CLEANUP) {7759+ u_sess->plsql_cxt.simple_econtext_stack = NULL;
7565- return;7760+ if (event != XACT_EVENT_ABORT) {
7566- } else if (event != XACT_EVENT_ABORT) {
7567- /* Shouldn't be any econtext stack entries left at commit */
7568- AssertEreport(u_sess->plsql_cxt.simple_econtext_stack == NULL,
7569- MOD_PLSQL,
7570- "Shouldn't be any econtext stack entries left at commit");
7571- 
7572 if (u_sess->plsql_cxt.simple_eval_estate) {7761 if (u_sess->plsql_cxt.simple_eval_estate) {
7573 FreeExecutorState(u_sess->plsql_cxt.simple_eval_estate);7762 FreeExecutorState(u_sess->plsql_cxt.simple_eval_estate);
7574 }7763 }
7575 u_sess->plsql_cxt.simple_eval_estate = NULL;7764 u_sess->plsql_cxt.simple_eval_estate = NULL;
7576 } else {7765 } else {
7577- u_sess->plsql_cxt.simple_econtext_stack = NULL;
7578 u_sess->plsql_cxt.simple_eval_estate = NULL;7766 u_sess->plsql_cxt.simple_eval_estate = NULL;
7579 }7767 }
7580}7768}
Msrc/common/pl/plpgsql/src/pl_handler.cpp+12-2
@@ -179,6 +179,7 @@ Datum plpgsql_call_handler(PG_FUNCTION_ARGS)
179 // PGSTAT_INIT_PLSQL_TIME_RECORD179 // PGSTAT_INIT_PLSQL_TIME_RECORD
180 int64 startTime = 0;180 int64 startTime = 0;
181 bool needRecord = false;181 bool needRecord = false;
182+ bool nonatomic = false;
182#ifdef STREAMPLAN183#ifdef STREAMPLAN
183 bool outer_is_stream = false;184 bool outer_is_stream = false;
184 bool outer_is_stream_support = false;185 bool outer_is_stream_support = false;
@@ -193,11 +194,20 @@ Datum plpgsql_call_handler(PG_FUNCTION_ARGS)
193 }194 }
194#endif195#endif
195 196 
197+ /*
198+ * If the atomic stored in fcinfo is false means allow
199+ * commit/rollback within stord procedure.
200+ * set the noatomic and will be reused within function.
201+ */
202+ nonatomic = fcinfo->context &&
203+ IsA(fcinfo->context, FunctionScanState) &&
204+ !castNode(FunctionScanState, fcinfo->context)->atomic;
205+ 
196 _PG_init();206 _PG_init();
197 /*207 /*
198 * Connect to SPI manager208 * Connect to SPI manager
199 */209 */
200- if ((rc = SPI_connect()) != SPI_OK_CONNECT) {210+ if ((rc = SPI_connect_ext(DestSPI, NULL, NULL, nonatomic ? SPI_OPT_NOATOMIC : 0)) != SPI_OK_CONNECT) {
201 ereport(ERROR,211 ereport(ERROR,
202 (errmodule(MOD_PLSQL),212 (errmodule(MOD_PLSQL),
203 errcode(ERRCODE_UNDEFINED_OBJECT),213 errcode(ERRCODE_UNDEFINED_OBJECT),
@@ -338,7 +348,7 @@ Datum plpgsql_inline_handler(PG_FUNCTION_ARGS)
338 /*348 /*
339 * Connect to SPI manager349 * Connect to SPI manager
340 */350 */
341- if ((rc = SPI_connect()) != SPI_OK_CONNECT) {351+ if ((rc = SPI_connect_ext(DestSPI, NULL, NULL, codeblock->atomic ? 0 : SPI_OPT_NOATOMIC)) != SPI_OK_CONNECT) {
342 ereport(ERROR,352 ereport(ERROR,
343 (errmodule(MOD_PLSQL),353 (errmodule(MOD_PLSQL),
344 errcode(ERRCODE_SPI_CONNECTION_FAILURE),354 errcode(ERRCODE_SPI_CONNECTION_FAILURE),
Msrc/gausskernel/optimizer/commands/functioncmds.cpp+2-1
@@ -2133,7 +2133,7 @@ Oid AlterFunctionNamespace_oid(Oid procOid, Oid nspOid)
2133 * ExecuteDoStmt2133 * ExecuteDoStmt
2134 * Execute inline procedural-language code2134 * Execute inline procedural-language code
2135 */2135 */
2136-void ExecuteDoStmt(const DoStmt* stmt)2136+void ExecuteDoStmt(const DoStmt* stmt, bool atomic)
2137{2137{
2138 InlineCodeBlock* codeblock = makeNode(InlineCodeBlock);2138 InlineCodeBlock* codeblock = makeNode(InlineCodeBlock);
2139 ListCell* arg = NULL;2139 ListCell* arg = NULL;
@@ -2200,6 +2200,7 @@ void ExecuteDoStmt(const DoStmt* stmt)
2200 2200 
2201 /* get the handler function's OID */2201 /* get the handler function's OID */
2202 laninline = languageStruct->laninline;2202 laninline = languageStruct->laninline;
2203+ codeblock->atomic = atomic;
2203 if (!OidIsValid(laninline))2204 if (!OidIsValid(laninline))
2204 ereport(ERROR,2205 ereport(ERROR,
2205 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),2206 (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
Msrc/gausskernel/process/tcop/postgres.cpp+21-0
@@ -44,6 +44,7 @@
44#include "catalog/pg_authid.h"44#include "catalog/pg_authid.h"
45#include "commands/async.h"45#include "commands/async.h"
46#include "commands/prepare.h"46#include "commands/prepare.h"
47+#include "executor/spi.h"
47#include "commands/user.h"48#include "commands/user.h"
48#include "commands/vacuum.h"49#include "commands/vacuum.h"
49#ifdef PGXC50#ifdef PGXC
@@ -93,6 +94,7 @@
93#include "mb/pg_wchar.h"94#include "mb/pg_wchar.h"
94#include "pgaudit.h"95#include "pgaudit.h"
95#include "auditfuncs.h"96#include "auditfuncs.h"
97+#include "funcapi.h"
96#ifdef PGXC98#ifdef PGXC
97#include "storage/procarray.h"99#include "storage/procarray.h"
98#include "pgxc/pgxc.h"100#include "pgxc/pgxc.h"
@@ -2039,6 +2041,7 @@ static void exec_simple_query(const char* query_string, MessageType messageType,
2039 * significant to PreventTransactionChain.)2041 * significant to PreventTransactionChain.)
2040 */2042 */
2041 isTopLevel = (list_length(parsetree_list) == 1);2043 isTopLevel = (list_length(parsetree_list) == 1);
2044+ u_sess->SPI_cxt.is_toplevel_stp = isTopLevel;
2042 2045 
2043 if (isTopLevel != 1)2046 if (isTopLevel != 1)
2044 t_thrd.explain_cxt.explain_perf_mode = EXPLAIN_NORMAL;2047 t_thrd.explain_cxt.explain_perf_mode = EXPLAIN_NORMAL;
@@ -2478,6 +2481,12 @@ static void exec_simple_query(const char* query_string, MessageType messageType,
2478 2481 
2479 MemoryContextDelete(OptimizerContext);2482 MemoryContextDelete(OptimizerContext);
2480 2483 
2484+ /* Reset store procedure's session variables. */
2485+ u_sess->SPI_cxt.is_toplevel_stp = false;
2486+ u_sess->SPI_cxt.is_stp = true;
2487+ u_sess->SPI_cxt.is_proconfig_set = false;
2488+ u_sess->SPI_cxt.portal_stp_exception_counter = 0;
2489+ 
2481 /*2490 /*
2482 * Close down transaction statement, if one is open.2491 * Close down transaction statement, if one is open.
2483 */2492 */
@@ -4331,6 +4340,7 @@ static void exec_execute_message(const char* portal_name, long max_rows)
4331 bool execute_is_fetch = false;4340 bool execute_is_fetch = false;
4332 bool was_logged = false;4341 bool was_logged = false;
4333 char msec_str[32];4342 char msec_str[32];
4343+ bool savedIsTopLevelForSTP = false;
4334 4344 
4335 gstrace_entry(GS_TRC_ID_exec_execute_message);4345 gstrace_entry(GS_TRC_ID_exec_execute_message);
4336 /* Adjust destination to tell printtup.c what to do */4346 /* Adjust destination to tell printtup.c what to do */
@@ -4469,6 +4479,9 @@ static void exec_execute_message(const char* portal_name, long max_rows)
4469 /* Check for cancel signal before we start execution */4479 /* Check for cancel signal before we start execution */
4470 CHECK_FOR_INTERRUPTS();4480 CHECK_FOR_INTERRUPTS();
4471 4481 
4482+ savedIsTopLevelForSTP = u_sess->SPI_cxt.is_toplevel_stp;
4483+ u_sess->SPI_cxt.is_toplevel_stp = true;
4484+ 
4472 /*4485 /*
4473 * Okay to run the portal.4486 * Okay to run the portal.
4474 */4487 */
@@ -4482,6 +4495,7 @@ static void exec_execute_message(const char* portal_name, long max_rows)
4482 receiver,4495 receiver,
4483 completionTag);4496 completionTag);
4484 4497 
4498+ u_sess->SPI_cxt.is_toplevel_stp = savedIsTopLevelForSTP;
4485 (*receiver->rDestroy)(receiver);4499 (*receiver->rDestroy)(receiver);
4486 4500 
4487 if (completed) {4501 if (completed) {
@@ -7371,6 +7385,10 @@ int PostgresMain(int argc, char* argv[], const char* dbname, const char* usernam
7371 if (sigsetjmp(local_sigjmp_buf, 1) != 0) {7385 if (sigsetjmp(local_sigjmp_buf, 1) != 0) {
7372 gstrace_tryblock_exit(true, oldTryCounter);7386 gstrace_tryblock_exit(true, oldTryCounter);
7373 7387 
7388+ u_sess->SPI_cxt.is_stp = true;
7389+ u_sess->SPI_cxt.is_proconfig_set = false;
7390+ u_sess->SPI_cxt.portal_stp_exception_counter = 0;
7391+ 
7374 (void)pgstat_report_waitstatus(STATE_WAIT_UNDEFINED);7392 (void)pgstat_report_waitstatus(STATE_WAIT_UNDEFINED);
7375 t_thrd.pgxc_cxt.GlobalNetInstr = NULL;7393 t_thrd.pgxc_cxt.GlobalNetInstr = NULL;
7376 /* output the memory tracking information when error happened */7394 /* output the memory tracking information when error happened */
@@ -7494,6 +7512,9 @@ int PostgresMain(int argc, char* argv[], const char* dbname, const char* usernam
7494 */7512 */
7495 AbortCurrentTransaction();7513 AbortCurrentTransaction();
7496 7514 
7515+ PortalErrorCleanup();
7516+ SPICleanup();
7517+ 
7497 /* Notice: at the most time it isn't necessary to call because7518 /* Notice: at the most time it isn't necessary to call because
7498 * all the LWLocks are released in AbortCurrentTransaction().7519 * all the LWLocks are released in AbortCurrentTransaction().
7499 * but in some rare exception not in one transaction (for7520 * but in some rare exception not in one transaction (for
Msrc/gausskernel/process/tcop/utility.cpp+1-1
@@ -4425,7 +4425,7 @@ void standard_ProcessUtility(Node* parse_tree, const char* query_string, ParamLi
4425 break;4425 break;
4426 4426 
4427 case T_DoStmt:4427 case T_DoStmt:
4428- ExecuteDoStmt((DoStmt*)parse_tree);4428+ ExecuteDoStmt((DoStmt*)parse_tree, (!u_sess->SPI_cxt.is_toplevel_stp || IsTransactionBlock()));
4429 break;4429 break;
4430 4430 
4431 case T_CreatedbStmt:4431 case T_CreatedbStmt:
Msrc/gausskernel/process/threadpool/knl_session.cpp+4-0
@@ -239,6 +239,10 @@ static void knl_u_SPI_init(knl_u_SPI_context* spi)
239 spi->_curid = -1;239 spi->_curid = -1;
240 spi->_stack = NULL;240 spi->_stack = NULL;
241 spi->_current = NULL;241 spi->_current = NULL;
242+ spi->is_toplevel_stp = false;
243+ spi->is_stp = true;
244+ spi->is_proconfig_set = false;
245+ spi->portal_stp_exception_counter = 0;
242}246}
243 247 
244static void knl_u_trigger_init(knl_u_trigger_context* tri_cxt)248static void knl_u_trigger_init(knl_u_trigger_context* tri_cxt)
Msrc/gausskernel/process/threadpool/knl_thread.cpp+1-0
@@ -845,6 +845,7 @@ static void knl_t_utils_init(knl_t_utils_context* utils_cxt)
845 utils_cxt->CurrentResourceOwner = NULL;845 utils_cxt->CurrentResourceOwner = NULL;
846 utils_cxt->CurTransactionResourceOwner = NULL;846 utils_cxt->CurTransactionResourceOwner = NULL;
847 utils_cxt->TopTransactionResourceOwner = NULL;847 utils_cxt->TopTransactionResourceOwner = NULL;
848+ utils_cxt->StpSavedResourceOwner = NULL;
848 utils_cxt->ResourceRelease_callbacks = NULL;849 utils_cxt->ResourceRelease_callbacks = NULL;
849 utils_cxt->SortColumnOptimize = false;850 utils_cxt->SortColumnOptimize = false;
850 utils_cxt->pRelatedRel = NULL;851 utils_cxt->pRelatedRel = NULL;
Msrc/gausskernel/runtime/executor/execQual.cpp+105-1
@@ -2042,6 +2042,52 @@ static Datum ExecMakeFunctionResultNoSets(
2042 PgStat_FunctionCallUsage fcusage;2042 PgStat_FunctionCallUsage fcusage;
2043 int i;2043 int i;
2044 int* var_dno = NULL;2044 int* var_dno = NULL;
2045+ FunctionScanState* node = NULL;
2046+ HeapTuple tup;
2047+ FuncExpr* fexpr = NULL;
2048+ bool savedIsStp = u_sess->SPI_cxt.is_stp;
2049+ bool savedProConfigIsSet = u_sess->SPI_cxt.is_proconfig_set;
2050+ bool proIsProcedure = false;
2051+ bool supportTransaction = false;
2052+ 
2053+#ifdef ENABLE_MULTIPLE_NODES
2054+ if (IS_PGXC_COORDINATOR) {
2055+ supportTransaction = true;
2056+ }
2057+#else
2058+ supportTransaction = true;
2059+#endif
2060+ 
2061+ if (supportTransaction && IsA(fcache->xprstate.expr, FuncExpr)) {
2062+ fexpr = (FuncExpr*)(fcache->xprstate.expr);
2063+ node = makeNode(FunctionScanState);
2064+ node->atomic = (!u_sess->SPI_cxt.is_toplevel_stp || IsTransactionBlock());
2065+ 
2066+ /*
2067+ * If proconfig is set we can't allow transaction commands because of the
2068+ * way the GUC stacking works. The transaction boundary would have to pop
2069+ * the proconfig setting off the stack. That restriction could be lefted
2070+ * by redesigning the GUC nesting mechanism a bit.
2071+ */
2072+ Relation relation = heap_open(ProcedureRelationId, AccessShareLock);
2073+ tup = SearchSysCache1(PROCOID, ObjectIdGetDatum(fexpr->funcid));
2074+ if (!HeapTupleIsValid(tup)) {
2075+ elog(ERROR, "cache lookup failed for function %u", fexpr->funcid);
2076+ }
2077+ 
2078+ if (!heap_attisnull(tup, Anum_pg_proc_proconfig, NULL) || u_sess->SPI_cxt.is_proconfig_set) {
2079+ u_sess->SPI_cxt.is_proconfig_set = true;
2080+ node->atomic = true;
2081+ }
2082+ 
2083+ proIsProcedure = PROC_IS_PRO(((Form_pg_proc)GETSTRUCT(tup))->prokind);
2084+ 
2085+ heap_close(relation, AccessShareLock);
2086+ 
2087+ /* If proisprocedure is true means it was a stored procedure. */
2088+ u_sess->SPI_cxt.is_stp = savedIsStp && proIsProcedure;
2089+ ReleaseSysCache(tup);
2090+ }
2045 2091 
2046 /* Guard against stack overflow due to overly complex expressions */2092 /* Guard against stack overflow due to overly complex expressions */
2047 check_stack_depth();2093 check_stack_depth();
@@ -2058,6 +2104,10 @@ static Datum ExecMakeFunctionResultNoSets(
2058 /* init the number of arguments to a function*/2104 /* init the number of arguments to a function*/
2059 InitFunctionCallInfoArgs(*fcinfo, list_length(fcache->args), 1);2105 InitFunctionCallInfoArgs(*fcinfo, list_length(fcache->args), 1);
2060 2106 
2107+ if (supportTransaction) {
2108+ fcinfo->context = (Node*)node;
2109+ }
2110+ 
2061 if (has_cursor_return) {2111 if (has_cursor_return) {
2062 /* init returnCursor to store out-args cursor info on ExprContext*/2112 /* init returnCursor to store out-args cursor info on ExprContext*/
2063 fcinfo->refcursor_data.returnCursor =2113 fcinfo->refcursor_data.returnCursor =
@@ -2152,6 +2202,9 @@ static Datum ExecMakeFunctionResultNoSets(
2152 pfree_ext(var_dno);2202 pfree_ext(var_dno);
2153 }2203 }
2154 2204 
2205+ u_sess->SPI_cxt.is_stp = savedIsStp;
2206+ u_sess->SPI_cxt.is_proconfig_set = savedProConfigIsSet;
2207+ 
2155 return result;2208 return result;
2156}2209}
2157 2210 
@@ -2234,9 +2287,53 @@ Tuplestorestate* ExecMakeTableFunctionResult(
2234 bool first_time = true;2287 bool first_time = true;
2235 int* var_dno = NULL;2288 int* var_dno = NULL;
2236 bool has_refcursor = false;2289 bool has_refcursor = false;
2290+ HeapTuple tup;
2291+ FuncExpr* fexpr = NULL;
2292+ bool savedIsStp = u_sess->SPI_cxt.is_stp;
2293+ bool savedProConfigIsSet = u_sess->SPI_cxt.is_proconfig_set;
2294+ bool proIsProcedure = false;
2295+ bool supportTransaction = false;
2237 2296 
2238 callerContext = CurrentMemoryContext;2297 callerContext = CurrentMemoryContext;
2239 2298 
2299+#ifdef ENABLE_MULTIPLE_NODES
2300+ if (IS_PGXC_COORDINATOR) {
2301+ supportTransaction = true;
2302+ }
2303+#else
2304+ supportTransaction = true;
2305+#endif
2306+ 
2307+ if (supportTransaction && IsA(funcexpr->expr, FuncExpr)) {
2308+ fexpr = (FuncExpr*)(funcexpr->expr);
2309+ node->atomic = (!u_sess->SPI_cxt.is_toplevel_stp || IsTransactionBlock());
2310+ 
2311+ /*
2312+ * If proconfig is set we can't allow transaction commands because of the
2313+ * way the GUC stacking works. The transaction boundary would have to pop
2314+ * the proconfig setting off the stack. That restriction could be lefted
2315+ * by redesigning the GUC nesting mechanism a bit.
2316+ */
2317+ Relation relation = heap_open(ProcedureRelationId, AccessShareLock);
2318+ tup = SearchSysCache1(PROCOID, ObjectIdGetDatum(fexpr->funcid));
2319+ if (!HeapTupleIsValid(tup)) {
2320+ elog(ERROR, "cache lookup failed for function %u", fexpr->funcid);
2321+ }
2322+ 
2323+ if (!heap_attisnull(tup, Anum_pg_proc_proconfig, NULL) || u_sess->SPI_cxt.is_proconfig_set) {
2324+ u_sess->SPI_cxt.is_proconfig_set = true;
2325+ node->atomic = true;
2326+ }
2327+ 
2328+ proIsProcedure = PROC_IS_PRO(((Form_pg_proc)GETSTRUCT(tup))->prokind);
2329+ 
2330+ heap_close(relation, AccessShareLock);
2331+ 
2332+ /* If proisprocedure is true means it was a stored procedure. */
2333+ u_sess->SPI_cxt.is_stp = savedIsStp && proIsProcedure;
2334+ ReleaseSysCache(tup);
2335+ }
2336+ 
2240 if (unlikely(funcexpr == NULL)) {2337 if (unlikely(funcexpr == NULL)) {
2241 ereport(ERROR, (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED), errmsg("The input function expression is NULL.")));2338 ereport(ERROR, (errcode(ERRCODE_NULL_VALUE_NOT_ALLOWED), errmsg("The input function expression is NULL.")));
2242 }2339 }
@@ -2351,7 +2448,11 @@ Tuplestorestate* ExecMakeTableFunctionResult(
2351 } else {2448 } else {
2352 /* Treat funcexpr as a generic expression */2449 /* Treat funcexpr as a generic expression */
2353 direct_function_call = false;2450 direct_function_call = false;
2354- InitFunctionCallInfoData(fcinfo, NULL, 0, InvalidOid, NULL, NULL);2451+ if (supportTransaction) {
2452+ InitFunctionCallInfoData(fcinfo, NULL, 0, InvalidOid, (Node*)node, NULL);
2453+ } else {
2454+ InitFunctionCallInfoData(fcinfo, NULL, 0, InvalidOid, NULL, NULL);
2455+ }
2355 }2456 }
2356 2457 
2357 /*2458 /*
@@ -2592,6 +2693,9 @@ no_function_result:
2592 pfree_ext(var_dno);2693 pfree_ext(var_dno);
2593 }2694 }
2594 2695 
2696+ u_sess->SPI_cxt.is_stp = savedIsStp;
2697+ u_sess->SPI_cxt.is_proconfig_set = savedProConfigIsSet;
2698+ 
2595 /* All done, pass back the tuplestore */2699 /* All done, pass back the tuplestore */
2596 return rsinfo.setResult;2700 return rsinfo.setResult;
2597}2701}
Msrc/gausskernel/runtime/executor/spi.cpp+187-16
@@ -77,6 +77,11 @@ static void CopySPI_Plan(SPIPlanPtr newplan, SPIPlanPtr plan, MemoryContext plan
77 77 
78/* =================== interface functions =================== */78/* =================== interface functions =================== */
79int SPI_connect(CommandDest dest, void (*spiCallbackfn)(void *), void *clientData)79int SPI_connect(CommandDest dest, void (*spiCallbackfn)(void *), void *clientData)
80+{
81+ return SPI_connect_ext(dest, spiCallbackfn, clientData, 0);
82+}
83+ 
84+int SPI_connect_ext(CommandDest dest, void (*spiCallbackfn)(void *), void *clientData, int options)
80{85{
81 int new_depth;86 int new_depth;
82 /*87 /*
@@ -93,8 +98,12 @@ int SPI_connect(CommandDest dest, void (*spiCallbackfn)(void *), void *clientDat
93 u_sess->SPI_cxt._connected != -1 ? "init level is not -1." : "stack depth is not zero.")));98 u_sess->SPI_cxt._connected != -1 ? "init level is not -1." : "stack depth is not zero.")));
94 }99 }
95 new_depth = 16;100 new_depth = 16;
101+ /*
102+ * Need TopMemoryContext because commit is allowed in stored procedure and it will clear all memory
103+ * context from TopTransaction. Therefor,need to use TopMemoryContext.
104+ */
96 u_sess->SPI_cxt._stack =105 u_sess->SPI_cxt._stack =
97- (_SPI_connection *)MemoryContextAlloc(u_sess->top_transaction_mem_cxt, new_depth * sizeof(_SPI_connection));106+ (_SPI_connection *)MemoryContextAlloc(u_sess->top_mem_cxt, new_depth * sizeof(_SPI_connection));
98 u_sess->SPI_cxt._stack_depth = new_depth;107 u_sess->SPI_cxt._stack_depth = new_depth;
99 } else {108 } else {
100 if (u_sess->SPI_cxt._stack_depth <= 0 || u_sess->SPI_cxt._stack_depth <= u_sess->SPI_cxt._connected) {109 if (u_sess->SPI_cxt._stack_depth <= 0 || u_sess->SPI_cxt._stack_depth <= u_sess->SPI_cxt._connected) {
@@ -125,18 +134,28 @@ int SPI_connect(CommandDest dest, void (*spiCallbackfn)(void *), void *clientDat
125 u_sess->SPI_cxt._current->dest = dest;134 u_sess->SPI_cxt._current->dest = dest;
126 u_sess->SPI_cxt._current->spiCallback = (void (*)(void *))spiCallbackfn;135 u_sess->SPI_cxt._current->spiCallback = (void (*)(void *))spiCallbackfn;
127 u_sess->SPI_cxt._current->clientData = clientData;136 u_sess->SPI_cxt._current->clientData = clientData;
137+ u_sess->SPI_cxt._current->atomic = (options & SPI_OPT_NOATOMIC) ? false : true;
138+ u_sess->SPI_cxt._current->internal_xact = false;
128 139 
129 /*140 /*
130 * Create memory contexts for this procedure141 * Create memory contexts for this procedure
142+ *
143+ * In atomic contexts (the normal case), we use TopTransactionContext,
144+ * otherwise PortalContext, so that it lives across transaction
145+ * boundaries.
131 *146 *
132- * XXX it would be better to use t_thrd.mem_cxt.portal_mem_cxt as the parent context, but147+ * XXX it would be better to use PortalContext as the parent context in
133- * we may not be inside a portal (consider deferred-trigger execution).148+ * all cases, but we may not be inside a portal (consider deferred-trigger
134- * Perhaps t_thrd.mem_cxt.cur_transaction_mem_cxt would do? For now it doesn't matter149+ * execution). Perhaps CurTransactionContext could be an option? For now
135- * because we clean up explicitly in AtEOSubXact_SPI().150+ * it doesn't matter because we clean up explicitly in ATEOSubXact_SPI().
136 */151 */
137- u_sess->SPI_cxt._current->procCxt = AllocSetContextCreate(u_sess->top_transaction_mem_cxt, "SPI Proc",152+ u_sess->SPI_cxt._current->procCxt =
153+ AllocSetContextCreate(u_sess->SPI_cxt._current->atomic ? u_sess->top_transaction_mem_cxt :
154+ t_thrd.mem_cxt.portal_mem_cxt, "SPI Proc",
138 ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE);155 ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE);
139- u_sess->SPI_cxt._current->execCxt = AllocSetContextCreate(u_sess->top_transaction_mem_cxt, "SPI Exec",156+ u_sess->SPI_cxt._current->execCxt =
157+ AllocSetContextCreate(u_sess->SPI_cxt._current->atomic ? u_sess->top_transaction_mem_cxt :
158+ u_sess->SPI_cxt._current->procCxt, "SPI Exec",
140 ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE);159 ALLOCSET_DEFAULT_MINSIZE, ALLOCSET_DEFAULT_INITSIZE, ALLOCSET_DEFAULT_MAXSIZE);
141 /* ... and switch to procedure's context */160 /* ... and switch to procedure's context */
142 u_sess->SPI_cxt._current->savedcxt = MemoryContextSwitchTo(u_sess->SPI_cxt._current->procCxt);161 u_sess->SPI_cxt._current->savedcxt = MemoryContextSwitchTo(u_sess->SPI_cxt._current->procCxt);
@@ -185,11 +204,138 @@ int SPI_finish(void)
185 return SPI_OK_FINISH;204 return SPI_OK_FINISH;
186}205}
187 206 
207+void SPI_start_transaction(void)
208+{
209+ MemoryContext oldContext = CurrentMemoryContext;
210+ 
211+ StartTransactionCommand(true);
212+ MemoryContextSwitchTo(oldContext);
213+}
214+ 
215+void SPI_commit(void)
216+{
217+ MemoryContext oldContext = CurrentMemoryContext;
218+ 
219+#ifdef ENABLE_MULTIPLE_NODES
220+ /* Can not commit at non-CN nodes */
221+ if (!IS_PGXC_COORDINATOR || IsConnFromCoord()) {
222+ ereport(ERROR,
223+ (errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
224+ errmsg("cannot commit at non-CN node")));
225+ }
226+#endif
227+ 
228+ /* If commit is not within stored procedure report error */
229+ if (!u_sess->SPI_cxt.is_stp) {
230+ ereport(ERROR,
231+ (errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
232+ errmsg("cannot commit within function")));
233+ }
234+ 
235+ /* Cannot commit if it's atomic is true */
236+ if (u_sess->SPI_cxt._current->atomic) {
237+ ereport(ERROR,
238+ (errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
239+ errmsg("invalid transaction termination")));
240+ }
241+ 
242+ /*
243+ * Hold any pinned portals that any PLs might be using. We have to do
244+ * this before changing trasnaction state, since this will run
245+ * user-defined code that might throw an error.
246+ */
247+ HoldPinnedPortals();
248+ 
249+ /*
250+ * This restriction is required by PLs implemented on top of SPI. They
251+ * use subtransactions to establish exception blocks that are supposed to
252+ * be rolled back together if there is an error. Terminating the
253+ * top-level transaction in such a block violates that idea. A future PL
254+ * implementation might have different ideas about this, in which case
255+ * this restriction would have to be refined or the check possibly be
256+ * moved out of SPI into the PLs.
257+ */
258+ u_sess->SPI_cxt._current->internal_xact = true;
259+ 
260+ while (ActiveSnapshotSet()) {
261+ PopActiveSnapshot();
262+ }
263+ 
264+ CommitTransactionCommand(true);
265+ MemoryContextSwitchTo(oldContext);
266+ 
267+ u_sess->SPI_cxt._current->internal_xact = false;
268+}
269+ 
270+void SPI_rollback(void)
271+{
272+ MemoryContext oldContext = CurrentMemoryContext;
273+ 
274+#ifdef ENABLE_MULTIPLE_NODES
275+ /* Can not commit at non-CN nodes */
276+ if (!IS_PGXC_COORDINATOR || IsConnFromCoord()) {
277+ ereport(ERROR,
278+ (errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
279+ errmsg("cannot rollback at non-CN node")));
280+ }
281+#endif
282+ 
283+ /* If commit is not within stored procedure report error */
284+ if (!u_sess->SPI_cxt.is_stp) {
285+ ereport(ERROR,
286+ (errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
287+ errmsg("cannot rollback within function")));
288+ }
289+ 
290+ /* Cannot commit if it's atomic is true */
291+ if (u_sess->SPI_cxt._current->atomic) {
292+ ereport(ERROR,
293+ (errcode(ERRCODE_INVALID_TRANSACTION_TERMINATION),
294+ errmsg("invalid transaction termination")));
295+ }
296+ 
297+ /*
298+ * Hold any pinned portals that any PLs might be using. We have to do
299+ * this before changing trasnaction state, since this will run
300+ * user-defined code that might throw an error.
301+ */
302+ HoldPinnedPortals();
303+ 
304+ /* see under SPI_commit() */
305+ u_sess->SPI_cxt._current->internal_xact = true;
306+ 
307+ AbortCurrentTransaction(true);
308+ MemoryContextSwitchTo(oldContext);
309+ 
310+ u_sess->SPI_cxt._current->internal_xact = false;
311+}
312+ 
313+/*
314+ * Clean up SPI state. Called on trasnaction end (of non-SPI-internal
315+ * trasnactions) and when retruning to the main loop on error.
316+ */
317+void SPICleanup(void)
318+{
319+ u_sess->SPI_cxt._current = u_sess->SPI_cxt._stack = NULL;
320+ u_sess->SPI_cxt._stack_depth = 0;
321+ u_sess->SPI_cxt._connected = u_sess->SPI_cxt._curid = -1;
322+ SPI_processed = 0;
323+ u_sess->SPI_cxt.lastoid = InvalidOid;
324+ SPI_tuptable = NULL;
325+}
326+ 
188/*327/*
189 * Clean up SPI state at transaction commit or abort.328 * Clean up SPI state at transaction commit or abort.
190 */329 */
191-void AtEOXact_SPI(bool isCommit)330+void AtEOXact_SPI(bool isCommit, bool stpRollback, bool stpCommit)
192{331{
332+ /*
333+ * Do nothing if the trasnaction end was initiated by SPI.
334+ */
335+ if (stpRollback || stpCommit) {
336+ return;
337+ }
338+ 
193 /*339 /*
194 * Note that memory contexts belonging to SPI stack entries will be freed340 * Note that memory contexts belonging to SPI stack entries will be freed
195 * automatically, so we can ignore them here. We just need to restore our341 * automatically, so we can ignore them here. We just need to restore our
@@ -200,12 +346,7 @@ void AtEOXact_SPI(bool isCommit)
200 errhint("Check for missing \"SPI_finish\" calls.")));346 errhint("Check for missing \"SPI_finish\" calls.")));
201 }347 }
202 348 
203- u_sess->SPI_cxt._current = u_sess->SPI_cxt._stack = NULL;349+ SPICleanup();
204- u_sess->SPI_cxt._stack_depth = 0;
205- u_sess->SPI_cxt._connected = u_sess->SPI_cxt._curid = -1;
206- SPI_processed = 0;
207- u_sess->SPI_cxt.lastoid = InvalidOid;
208- SPI_tuptable = NULL;
209}350}
210 351 
211/*352/*
@@ -214,8 +355,12 @@ void AtEOXact_SPI(bool isCommit)
214 * During commit, there shouldn't be any unclosed entries remaining from355 * During commit, there shouldn't be any unclosed entries remaining from
215 * the current subtransaction; we emit a warning if any are found.356 * the current subtransaction; we emit a warning if any are found.
216 */357 */
217-void AtEOSubXact_SPI(bool isCommit, SubTransactionId mySubid)358+void AtEOSubXact_SPI(bool isCommit, SubTransactionId mySubid, bool stpRollback, bool stpCommit)
218{359{
360+ if (stpRollback || stpCommit) {
361+ return;
362+ }
363+ 
219 bool found = false;364 bool found = false;
220 365 
221 while (u_sess->SPI_cxt._connected >= 0) {366 while (u_sess->SPI_cxt._connected >= 0) {
@@ -225,6 +370,10 @@ void AtEOSubXact_SPI(bool isCommit, SubTransactionId mySubid)
225 break; /* couldn't be any underneath it either */370 break; /* couldn't be any underneath it either */
226 }371 }
227 372 
373+ if (connection->internal_xact) {
374+ break;
375+ }
376+ 
228 found = true;377 found = true;
229 /*378 /*
230 * Release procedure memory explicitly (see note in SPI_connect)379 * Release procedure memory explicitly (see note in SPI_connect)
@@ -1873,6 +2022,11 @@ static int _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI, Snapshot sn
1873 CachedPlan *cplan = NULL;2022 CachedPlan *cplan = NULL;
1874 ListCell *lc1 = NULL;2023 ListCell *lc1 = NULL;
1875 bool tmp_enable_light_proxy = u_sess->attr.attr_sql.enable_light_proxy;2024 bool tmp_enable_light_proxy = u_sess->attr.attr_sql.enable_light_proxy;
2025+ TransactionId oldTransactionId = InvalidTransactionId;
2026+ 
2027+ if (!RecoveryInProgress()) {
2028+ oldTransactionId = GetTopTransactionId();
2029+ }
1876 2030 
1877 /* not allow Light CN */2031 /* not allow Light CN */
1878 u_sess->attr.attr_sql.enable_light_proxy = false;2032 u_sess->attr.attr_sql.enable_light_proxy = false;
@@ -2118,7 +2272,9 @@ static int _SPI_execute_plan(SPIPlanPtr plan, ParamListInfo paramLI, Snapshot sn
2118 }2272 }
2119 2273 
2120 /* Done with this plan, so release refcount */2274 /* Done with this plan, so release refcount */
2121- ReleaseCachedPlan(cplan, plan->saved);2275+ if ((!RecoveryInProgress()) && (oldTransactionId == GetTopTransactionId())) {
2276+ ReleaseCachedPlan(cplan, plan->saved);
2277+ }
2122 cplan = NULL;2278 cplan = NULL;
2123 2279 
2124 /*2280 /*
@@ -2228,6 +2384,7 @@ static int _SPI_pquery(QueryDesc *queryDesc, bool fire_triggers, long tcount, bo
2228 int operation = queryDesc->operation;2384 int operation = queryDesc->operation;
2229 int eflags;2385 int eflags;
2230 int res;2386 int res;
2387+ TransactionId oldTransactionId = InvalidTransactionId;
2231 2388 
2232 switch (operation) {2389 switch (operation) {
2233 case CMD_SELECT:2390 case CMD_SELECT:
@@ -2275,6 +2432,9 @@ static int _SPI_pquery(QueryDesc *queryDesc, bool fire_triggers, long tcount, bo
2275 eflags = EXEC_FLAG_SKIP_TRIGGERS;2432 eflags = EXEC_FLAG_SKIP_TRIGGERS;
2276 }2433 }
2277 2434 
2435+ if (!RecoveryInProgress()) {
2436+ oldTransactionId = GetTopTransactionId();
2437+ }
2278 ExecutorStart(queryDesc, eflags);2438 ExecutorStart(queryDesc, eflags);
2279 2439 
2280 bool forced_control = !from_lock && IS_PGXC_COORDINATOR &&2440 bool forced_control = !from_lock && IS_PGXC_COORDINATOR &&
@@ -2351,6 +2511,17 @@ static int _SPI_pquery(QueryDesc *queryDesc, bool fire_triggers, long tcount, bo
2351 }2511 }
2352 }2512 }
2353 2513 
2514+ /*
2515+ * If there are commit/rollback within stored proedure. Snapshot has already free during commit/rollback process.
2516+ * Therefor, need to set queryDesc snapshots to NULL. Otherwise the reference will be stale pointers.
2517+ */
2518+ if ((!RecoveryInProgress()) && (oldTransactionId != GetTopTransactionId())) {
2519+ queryDesc->snapshot = NULL;
2520+ queryDesc->crosscheck_snapshot = NULL;
2521+ queryDesc->estate->es_snapshot = NULL;
2522+ queryDesc->estate->es_crosscheck_snapshot = NULL;
2523+ }
2524+ 
2354 ExecutorFinish(queryDesc);2525 ExecutorFinish(queryDesc);
2355 ExecutorEnd(queryDesc);2526 ExecutorEnd(queryDesc);
2356 /* FreeQueryDesc is done by the caller */2527 /* FreeQueryDesc is done by the caller */
Msrc/gausskernel/storage/access/transam/xact.cpp+135-45
@@ -277,7 +277,7 @@ typedef struct GTMCallbackItem {
277 277 
278/* local function prototypes */278/* local function prototypes */
279static void AssignTransactionId(TransactionState s);279static void AssignTransactionId(TransactionState s);
280-static void AbortTransaction(bool PerfectRollback);280+static void AbortTransaction(bool PerfectRollback = false, bool stpRollback = false);
281static void AtAbort_Memory(void);281static void AtAbort_Memory(void);
282static void AtCleanup_Memory(void);282static void AtCleanup_Memory(void);
283static void AtAbort_ResourceOwner(void);283static void AtAbort_ResourceOwner(void);
@@ -294,13 +294,13 @@ static void CleanSequenceCallbacks(void);
294static void CallSequenceCallbacks(GTMEvent event);294static void CallSequenceCallbacks(GTMEvent event);
295#endif295#endif
296static void CleanupTransaction(void);296static void CleanupTransaction(void);
297-static void CommitTransaction(void);297+static void CommitTransaction(bool stpCommit = false);
298static TransactionId RecordTransactionAbort(bool isSubXact);298static TransactionId RecordTransactionAbort(bool isSubXact);
299static void StartTransaction(bool begin_on_gtm);299static void StartTransaction(bool begin_on_gtm);
300 300 
301static void StartSubTransaction(void);301static void StartSubTransaction(void);
302-static void CommitSubTransaction(void);302+static void CommitSubTransaction(bool stpCommit = false);
303-static void AbortSubTransaction(void);303+static void AbortSubTransaction(bool stpRollback = false);
304static void CleanupSubTransaction(void);304static void CleanupSubTransaction(void);
305static void PushTransaction(void);305static void PushTransaction(void);
306static void PopTransaction(void);306static void PopTransaction(void);
@@ -316,7 +316,7 @@ static void ShowTransactionState(const char* str);
316static void ShowTransactionStateRec(TransactionState state);316static void ShowTransactionStateRec(TransactionState state);
317static const char* BlockStateAsString(TBlockState blockState);317static const char* BlockStateAsString(TBlockState blockState);
318static const char* TransStateAsString(TransState state);318static const char* TransStateAsString(TransState state);
319-static void PrepareTransaction(void);319+static void PrepareTransaction(bool stpCommit = false);
320 320 
321extern void print_leak_warning_at_commit();321extern void print_leak_warning_at_commit();
322#ifndef ENABLE_LLT322#ifndef ENABLE_LLT
@@ -2365,7 +2365,7 @@ void ThreadLocalFlagCleanUp()
2365 *2365 *
2366 * NB: if you change this routine, better look at PrepareTransaction too!2366 * NB: if you change this routine, better look at PrepareTransaction too!
2367 */2367 */
2368-static void CommitTransaction(void)2368+static void CommitTransaction(bool stpCommit)
2369{2369{
2370 u_sess->need_report_top_xid = false;2370 u_sess->need_report_top_xid = false;
2371 TransactionState s = CurrentTransactionState;2371 TransactionState s = CurrentTransactionState;
@@ -2410,6 +2410,14 @@ static void CommitTransaction(void)
2410 t_thrd.xact_cxt.handlesDestroyedInCancelQuery = false;2410 t_thrd.xact_cxt.handlesDestroyedInCancelQuery = false;
2411 ThreadLocalFlagCleanUp();2411 ThreadLocalFlagCleanUp();
2412 2412 
2413+ /*
2414+ * When commit within nested store procedure, it will create a plan cache.
2415+ * During commit time, need to clean up those plan cache.
2416+ */
2417+ if (stpCommit) {
2418+ ResourceOwnerDecrementNPlanRefs(t_thrd.utils_cxt.CurrentResourceOwner, true);
2419+ }
2420+ 
2413#ifdef PGXC2421#ifdef PGXC
2414 /*2422 /*
2415 * If we are a Coordinator and currently serving the client,2423 * If we are a Coordinator and currently serving the client,
@@ -2486,7 +2494,7 @@ static void CommitTransaction(void)
2486 */2494 */
2487 Assert(GlobalTransactionIdIsValid(s->transactionId));2495 Assert(GlobalTransactionIdIsValid(s->transactionId));
2488 2496 
2489- PrepareTransaction();2497+ PrepareTransaction(stpCommit);
2490 s->blockState = TBLOCK_DEFAULT;2498 s->blockState = TBLOCK_DEFAULT;
2491 2499 
2492 /*2500 /*
@@ -2519,7 +2527,7 @@ static void CommitTransaction(void)
2519 * If there weren't any, we are done ... otherwise loop back to check2527 * If there weren't any, we are done ... otherwise loop back to check
2520 * if they queued deferred triggers. Lather, rinse, repeat.2528 * if they queued deferred triggers. Lather, rinse, repeat.
2521 */2529 */
2522- if (!PreCommit_Portals(false))2530+ if (!PreCommit_Portals(false, stpCommit))
2523 break;2531 break;
2524 }2532 }
2525 2533 
@@ -2848,9 +2856,11 @@ static void CommitTransaction(void)
2848 2856 
2849 AtCommit_Notify();2857 AtCommit_Notify();
2850 AtEOXact_GUC(true, 1);2858 AtEOXact_GUC(true, 1);
2851- AtEOXact_SPI(true);2859+ AtEOXact_SPI(true, false, stpCommit);
2852 AtEOXact_on_commit_actions(true);2860 AtEOXact_on_commit_actions(true);
2853- AtEOXact_Namespace(true);2861+ if (!stpCommit){
2862+ AtEOXact_Namespace(true);
2863+ }
2854 AtEOXact_SMgr();2864 AtEOXact_SMgr();
2855 AtEOXact_Files();2865 AtEOXact_Files();
2856 AtEOXact_ComboCid();2866 AtEOXact_ComboCid();
@@ -3093,7 +3103,7 @@ bool AtEOXact_GlobalTxn(bool commit, bool is_write)
3093 * If PrepareTransaction is called during an implicit 2PC, do not release ressources,3103 * If PrepareTransaction is called during an implicit 2PC, do not release ressources,
3094 * this is made by CommitTransaction when transaction has been committed on Nodes.3104 * this is made by CommitTransaction when transaction has been committed on Nodes.
3095 */3105 */
3096-static void PrepareTransaction(void)3106+static void PrepareTransaction(bool stpCommit)
3097{3107{
3098 u_sess->need_report_top_xid = false;3108 u_sess->need_report_top_xid = false;
3099 TransactionState s = CurrentTransactionState;3109 TransactionState s = CurrentTransactionState;
@@ -3198,7 +3208,7 @@ static void PrepareTransaction(void)
3198 * If there weren't any, we are done ... otherwise loop back to check3208 * If there weren't any, we are done ... otherwise loop back to check
3199 * if they queued deferred triggers. Lather, rinse, repeat.3209 * if they queued deferred triggers. Lather, rinse, repeat.
3200 */3210 */
3201- if (!PreCommit_Portals(true))3211+ if (!PreCommit_Portals(true, stpCommit))
3202 break;3212 break;
3203 }3213 }
3204 3214 
@@ -3392,9 +3402,16 @@ static void PrepareTransaction(void)
3392 3402 
3393 /* PREPARE acts the same as COMMIT as far as GUC is concerned */3403 /* PREPARE acts the same as COMMIT as far as GUC is concerned */
3394 AtEOXact_GUC(true, 1);3404 AtEOXact_GUC(true, 1);
3395- AtEOXact_SPI(true);3405+ AtEOXact_SPI(true, false, stpCommit);
3396 AtEOXact_on_commit_actions(true);3406 AtEOXact_on_commit_actions(true);
3397- AtEOXact_Namespace(true);3407+ /*
3408+ * For commit within stored procedure don't clean up namespace.
3409+ * Otherwise it will throw warning leaked override search path,
3410+ * since we push the search path hasn't pop yet.
3411+ */
3412+ if (!stpCommit) {
3413+ AtEOXact_Namespace(true);
3414+ }
3398 AtEOXact_SMgr();3415 AtEOXact_SMgr();
3399 AtEOXact_Files();3416 AtEOXact_Files();
3400 AtEOXact_ComboCid();3417 AtEOXact_ComboCid();
@@ -3468,7 +3485,7 @@ static void PrepareTransaction(void)
3468#endif3485#endif
3469}3486}
3470 3487 
3471-static void AbortTransaction(bool PerfectRollback = false)3488+static void AbortTransaction(bool PerfectRollback, bool stpRollback)
3472{3489{
3473 u_sess->need_report_top_xid = false;3490 u_sess->need_report_top_xid = false;
3474 TransactionState s = CurrentTransactionState;3491 TransactionState s = CurrentTransactionState;
@@ -3723,7 +3740,7 @@ static void AbortTransaction(bool PerfectRollback = false)
3723 */3740 */
3724 AfterTriggerEndXact(false); /* 'false' means it's abort */3741 AfterTriggerEndXact(false); /* 'false' means it's abort */
3725 CallXactCallbacks(XACT_EVENT_PREROLLBACK_CLEANUP);3742 CallXactCallbacks(XACT_EVENT_PREROLLBACK_CLEANUP);
3726- AtAbort_Portals();3743+ AtAbort_Portals(stpRollback);
3727 AtEOXact_LargeObject(false);3744 AtEOXact_LargeObject(false);
3728 AtAbort_Notify();3745 AtAbort_Notify();
3729 AtEOXact_RelationMap(false);3746 AtEOXact_RelationMap(false);
@@ -3787,9 +3804,11 @@ static void AbortTransaction(bool PerfectRollback = false)
3787 if (change_user_name)3804 if (change_user_name)
3788 u_sess->misc_cxt.CurrentUserName = NULL;3805 u_sess->misc_cxt.CurrentUserName = NULL;
3789 3806 
3790- AtEOXact_SPI(false);3807+ AtEOXact_SPI(false, stpRollback, false);
3791 AtEOXact_on_commit_actions(false);3808 AtEOXact_on_commit_actions(false);
3792- AtEOXact_Namespace(false);3809+ if (!stpRollback) {
3810+ AtEOXact_Namespace(false);
3811+ }
3793 AtEOXact_SMgr();3812 AtEOXact_SMgr();
3794 AtEOXact_Files();3813 AtEOXact_Files();
3795 AtEOXact_ComboCid();3814 AtEOXact_ComboCid();
@@ -3876,7 +3895,7 @@ static void CleanupTransaction(void)
3876#endif3895#endif
3877}3896}
3878 3897 
3879-void StartTransactionCommand(void)3898+void StartTransactionCommand(bool stpRollback)
3880{3899{
3881 TransactionState s = CurrentTransactionState;3900 TransactionState s = CurrentTransactionState;
3882 3901 
@@ -3918,6 +3937,9 @@ void StartTransactionCommand(void)
3918 */3937 */
3919 case TBLOCK_ABORT:3938 case TBLOCK_ABORT:
3920 case TBLOCK_SUBABORT:3939 case TBLOCK_SUBABORT:
3940+ if (stpRollback) {
3941+ s->blockState = TBLOCK_DEFAULT;
3942+ }
3921 break;3943 break;
3922 3944 
3923 /* These cases are invalid. */3945 /* These cases are invalid. */
@@ -3949,10 +3971,11 @@ void StartTransactionCommand(void)
3949 (void)MemoryContextSwitchTo(t_thrd.mem_cxt.cur_transaction_mem_cxt);3971 (void)MemoryContextSwitchTo(t_thrd.mem_cxt.cur_transaction_mem_cxt);
3950}3972}
3951 3973 
3952-void CommitTransactionCommand(void)3974+void CommitTransactionCommand(bool stpCommit)
3953{3975{
3954 TransactionState s = CurrentTransactionState;3976 TransactionState s = CurrentTransactionState;
3955 TBlockState oldstate = s->blockState;3977 TBlockState oldstate = s->blockState;
3978+ const int stpownerlevel = 2;
3956 3979 
3957 switch (s->blockState) {3980 switch (s->blockState) {
3958 /*3981 /*
@@ -3971,7 +3994,7 @@ void CommitTransactionCommand(void)
3971 * transaction commit, and return to the idle state.3994 * transaction commit, and return to the idle state.
3972 */3995 */
3973 case TBLOCK_STARTED:3996 case TBLOCK_STARTED:
3974- CommitTransaction();3997+ CommitTransaction(stpCommit);
3975 s->blockState = TBLOCK_DEFAULT;3998 s->blockState = TBLOCK_DEFAULT;
3976 break;3999 break;
3977 4000 
@@ -3993,6 +4016,29 @@ void CommitTransactionCommand(void)
3993 case TBLOCK_INPROGRESS:4016 case TBLOCK_INPROGRESS:
3994 case TBLOCK_SUBINPROGRESS:4017 case TBLOCK_SUBINPROGRESS:
3995 CommandCounterIncrement();4018 CommandCounterIncrement();
4019+ 
4020+ if (stpCommit && u_sess->SPI_cxt.portal_stp_exception_counter > 0) {
4021+ int subTransactionCounter = 0;
4022+ Assert(!StreamThreadAmI());
4023+ 
4024+ do {
4025+ MemoryContextSwitchTo(t_thrd.mem_cxt.cur_transaction_mem_cxt);
4026+ CommitSubTransaction(stpCommit);
4027+ s = CurrentTransactionState;
4028+ subTransactionCounter++;
4029+ 
4030+ if (IS_PGXC_COORDINATOR && !IsConnFromCoord()) {
4031+ /* CN should send release savepoint command to remote nodes for savepoint name reuse */
4032+ pgxc_node_remote_savepoint("release s1", EXEC_ON_ALL_NODES, true, false);
4033+ }
4034+ } while (s->blockState == TBLOCK_SUBINPROGRESS);
4035+ 
4036+ /* If we had a commit command, finish off the main xact too */
4037+ Assert(subTransactionCounter == u_sess->SPI_cxt.portal_stp_exception_counter);
4038+ t_thrd.utils_cxt.CurrentResourceOwner = t_thrd.utils_cxt.StpSavedResourceOwner;
4039+ CommitTransaction(stpCommit);
4040+ s->blockState = TBLOCK_DEFAULT;
4041+ }
3996 break;4042 break;
3997 4043 
3998 /*4044 /*
@@ -4000,7 +4046,7 @@ void CommitTransactionCommand(void)
4000 * idle state.4046 * idle state.
4001 */4047 */
4002 case TBLOCK_END:4048 case TBLOCK_END:
4003- CommitTransaction();4049+ CommitTransaction(stpCommit);
4004 s->blockState = TBLOCK_DEFAULT;4050 s->blockState = TBLOCK_DEFAULT;
4005 break;4051 break;
4006 4052 
@@ -4029,7 +4075,7 @@ void CommitTransactionCommand(void)
4029 * and then clean up.4075 * and then clean up.
4030 */4076 */
4031 case TBLOCK_ABORT_PENDING:4077 case TBLOCK_ABORT_PENDING:
4032- AbortTransaction(true);4078+ AbortTransaction(true, false);
4033 CleanupTransaction();4079 CleanupTransaction();
4034 s->blockState = TBLOCK_DEFAULT;4080 s->blockState = TBLOCK_DEFAULT;
4035 break;4081 break;
@@ -4050,6 +4096,9 @@ void CommitTransactionCommand(void)
4050 * state.)4096 * state.)
4051 */4097 */
4052 case TBLOCK_SUBBEGIN:4098 case TBLOCK_SUBBEGIN:
4099+ if (CurrentTransactionState->nestingLevel == stpownerlevel) {
4100+ t_thrd.utils_cxt.StpSavedResourceOwner = t_thrd.utils_cxt.CurrentResourceOwner;
4101+ }
4053 StartSubTransaction();4102 StartSubTransaction();
4054 s->blockState = TBLOCK_SUBINPROGRESS;4103 s->blockState = TBLOCK_SUBINPROGRESS;
4055 break;4104 break;
@@ -4112,7 +4161,7 @@ void CommitTransactionCommand(void)
4112 4161 
4113 /* As above, but it's not dead yet, so abort first. */4162 /* As above, but it's not dead yet, so abort first. */
4114 case TBLOCK_SUBABORT_PENDING:4163 case TBLOCK_SUBABORT_PENDING:
4115- AbortSubTransaction();4164+ AbortSubTransaction(stpCommit);
4116 CleanupSubTransaction();4165 CleanupSubTransaction();
4117 if (t_thrd.xact_cxt.handlesDestroyedInCancelQuery) {4166 if (t_thrd.xact_cxt.handlesDestroyedInCancelQuery) {
4118 ereport(WARNING,4167 ereport(WARNING,
@@ -4137,7 +4186,7 @@ void CommitTransactionCommand(void)
4137 s->name = NULL;4186 s->name = NULL;
4138 savepointLevel = s->savepointLevel;4187 savepointLevel = s->savepointLevel;
4139 4188 
4140- AbortSubTransaction();4189+ AbortSubTransaction(stpCommit);
4141 CleanupSubTransaction();4190 CleanupSubTransaction();
4142 if (t_thrd.xact_cxt.handlesDestroyedInCancelQuery) {4191 if (t_thrd.xact_cxt.handlesDestroyedInCancelQuery) {
4143 ereport(WARNING,4192 ereport(WARNING,
@@ -4198,7 +4247,7 @@ void CommitTransactionCommand(void)
4198 }4247 }
4199}4248}
4200 4249 
4201-void AbortCurrentTransaction(void)4250+void AbortCurrentTransaction(bool stpRollback)
4202{4251{
4203 TransactionState s = CurrentTransactionState;4252 TransactionState s = CurrentTransactionState;
4204 4253 
@@ -4216,7 +4265,7 @@ void AbortCurrentTransaction(void)
4216 */4265 */
4217 if (s->state == TRANS_START)4266 if (s->state == TRANS_START)
4218 s->state = TRANS_INPROGRESS;4267 s->state = TRANS_INPROGRESS;
4219- AbortTransaction();4268+ AbortTransaction(false, stpRollback);
4220 CleanupTransaction();4269 CleanupTransaction();
4221 }4270 }
4222 break;4271 break;
@@ -4226,7 +4275,7 @@ void AbortCurrentTransaction(void)
4226 * & cleanup transaction.4275 * & cleanup transaction.
4227 */4276 */
4228 case TBLOCK_STARTED:4277 case TBLOCK_STARTED:
4229- AbortTransaction();4278+ AbortTransaction(false, stpRollback);
4230 CleanupTransaction();4279 CleanupTransaction();
4231 s->blockState = TBLOCK_DEFAULT;4280 s->blockState = TBLOCK_DEFAULT;
4232 break;4281 break;
@@ -4239,7 +4288,7 @@ void AbortCurrentTransaction(void)
4239 * state.4288 * state.
4240 */4289 */
4241 case TBLOCK_BEGIN:4290 case TBLOCK_BEGIN:
4242- AbortTransaction();4291+ AbortTransaction(false, stpRollback);
4243 CleanupTransaction();4292 CleanupTransaction();
4244 s->blockState = TBLOCK_DEFAULT;4293 s->blockState = TBLOCK_DEFAULT;
4245 break;4294 break;
@@ -4250,7 +4299,7 @@ void AbortCurrentTransaction(void)
4250 * ABORT state. We will stay in ABORT until we get a ROLLBACK.4299 * ABORT state. We will stay in ABORT until we get a ROLLBACK.
4251 */4300 */
4252 case TBLOCK_INPROGRESS:4301 case TBLOCK_INPROGRESS:
4253- AbortTransaction();4302+ AbortTransaction(false, stpRollback);
4254 s->blockState = TBLOCK_ABORT;4303 s->blockState = TBLOCK_ABORT;
4255 /* CleanupTransaction happens when we exit TBLOCK_ABORT_END */4304 /* CleanupTransaction happens when we exit TBLOCK_ABORT_END */
4256 break;4305 break;
@@ -4261,7 +4310,7 @@ void AbortCurrentTransaction(void)
4261 * the transaction).4310 * the transaction).
4262 */4311 */
4263 case TBLOCK_END:4312 case TBLOCK_END:
4264- AbortTransaction();4313+ AbortTransaction(false, stpRollback);
4265 CleanupTransaction();4314 CleanupTransaction();
4266 s->blockState = TBLOCK_DEFAULT;4315 s->blockState = TBLOCK_DEFAULT;
4267 break;4316 break;
@@ -4290,7 +4339,7 @@ void AbortCurrentTransaction(void)
4290 * Abort, cleanup, go to idle state.4339 * Abort, cleanup, go to idle state.
4291 */4340 */
4292 case TBLOCK_ABORT_PENDING:4341 case TBLOCK_ABORT_PENDING:
4293- AbortTransaction();4342+ AbortTransaction(false, stpRollback);
4294 CleanupTransaction();4343 CleanupTransaction();
4295 s->blockState = TBLOCK_DEFAULT;4344 s->blockState = TBLOCK_DEFAULT;
4296 break;4345 break;
@@ -4301,7 +4350,7 @@ void AbortCurrentTransaction(void)
4301 * the transaction).4350 * the transaction).
4302 */4351 */
4303 case TBLOCK_PREPARE:4352 case TBLOCK_PREPARE:
4304- AbortTransaction();4353+ AbortTransaction(false, stpRollback);
4305 CleanupTransaction();4354 CleanupTransaction();
4306 s->blockState = TBLOCK_DEFAULT;4355 s->blockState = TBLOCK_DEFAULT;
4307 break;4356 break;
@@ -4312,8 +4361,29 @@ void AbortCurrentTransaction(void)
4312 * we get ROLLBACK.4361 * we get ROLLBACK.
4313 */4362 */
4314 case TBLOCK_SUBINPROGRESS:4363 case TBLOCK_SUBINPROGRESS:
4315- AbortSubTransaction();4364+ if (stpRollback && u_sess->SPI_cxt.portal_stp_exception_counter > 0) {
4316- s->blockState = TBLOCK_SUBABORT;4365+ int subTransactionCounter = 0;
4366+ do {
4367+ AbortSubTransaction(stpRollback);
4368+ s->blockState = TBLOCK_SUBABORT;
4369+ CleanupSubTransaction();
4370+ s = CurrentTransactionState;
4371+ subTransactionCounter++;
4372+ } while (s->blockState == TBLOCK_SUBINPROGRESS);
4373+ 
4374+ Assert(subTransactionCounter == u_sess->SPI_cxt.portal_stp_exception_counter);
4375+ if (s->state == TRANS_START) {
4376+ s->state = TRANS_INPROGRESS;
4377+ }
4378+ 
4379+ AbortTransaction(false, stpRollback);
4380+ CleanupTransaction();
4381+ s->blockState = TBLOCK_DEFAULT;
4382+ } else {
4383+ AbortSubTransaction();
4384+ s->blockState = TBLOCK_SUBABORT;
4385+ }
4386+
4317 if (t_thrd.xact_cxt.handlesDestroyedInCancelQuery) {4387 if (t_thrd.xact_cxt.handlesDestroyedInCancelQuery) {
4318 ereport(WARNING,4388 ereport(WARNING,
4319 (errmsg(4389 (errmsg(
@@ -4332,7 +4402,7 @@ void AbortCurrentTransaction(void)
4332 case TBLOCK_SUBCOMMIT:4402 case TBLOCK_SUBCOMMIT:
4333 case TBLOCK_SUBABORT_PENDING:4403 case TBLOCK_SUBABORT_PENDING:
4334 case TBLOCK_SUBRESTART:4404 case TBLOCK_SUBRESTART:
4335- AbortSubTransaction();4405+ AbortSubTransaction(stpRollback);
4336 CleanupSubTransaction();4406 CleanupSubTransaction();
4337 if (t_thrd.xact_cxt.handlesDestroyedInCancelQuery) {4407 if (t_thrd.xact_cxt.handlesDestroyedInCancelQuery) {
4338 ereport(WARNING,4408 ereport(WARNING,
@@ -4347,7 +4417,7 @@ void AbortCurrentTransaction(void)
4347 case TBLOCK_SUBABORT_END:4417 case TBLOCK_SUBABORT_END:
4348 case TBLOCK_SUBABORT_RESTART:4418 case TBLOCK_SUBABORT_RESTART:
4349 CleanupSubTransaction();4419 CleanupSubTransaction();
4350- AbortCurrentTransaction();4420+ AbortCurrentTransaction(stpRollback);
4351 break;4421 break;
4352 default:4422 default:
4353 ereport(FATAL,4423 ereport(FATAL,
@@ -5473,7 +5543,7 @@ void RollbackAndReleaseCurrentSubTransaction(void)
5473 5543 
5474 /* Abort the current subtransaction, if needed. */5544 /* Abort the current subtransaction, if needed. */
5475 if (s->blockState == TBLOCK_SUBINPROGRESS) {5545 if (s->blockState == TBLOCK_SUBINPROGRESS) {
5476- AbortSubTransaction();5546+ AbortSubTransaction(false);
5477 }5547 }
5478 5548 
5479 /* And clean it up, too */5549 /* And clean it up, too */
@@ -5586,7 +5656,7 @@ void AbortOutOfAnyTransaction(bool reserve_topxact_abort)
5586 case TBLOCK_SUBCOMMIT:5656 case TBLOCK_SUBCOMMIT:
5587 case TBLOCK_SUBABORT_PENDING:5657 case TBLOCK_SUBABORT_PENDING:
5588 case TBLOCK_SUBRESTART:5658 case TBLOCK_SUBRESTART:
5589- AbortSubTransaction();5659+ AbortSubTransaction(false);
5590 CleanupSubTransaction();5660 CleanupSubTransaction();
5591 s = CurrentTransactionState; /* changed by pop */5661 s = CurrentTransactionState; /* changed by pop */
5592 break;5662 break;
@@ -5627,6 +5697,10 @@ bool IsTransactionBlock(void)
5627{5697{
5628 TransactionState s = CurrentTransactionState;5698 TransactionState s = CurrentTransactionState;
5629 5699 
5700+ if (u_sess->SPI_cxt.portal_stp_exception_counter > 0 && s->blockState == TBLOCK_SUBINPROGRESS) {
5701+ return false;
5702+ }
5703+ 
5630 if (s->blockState == TBLOCK_DEFAULT || s->blockState == TBLOCK_STARTED) {5704 if (s->blockState == TBLOCK_DEFAULT || s->blockState == TBLOCK_STARTED) {
5631 return false;5705 return false;
5632 }5706 }
@@ -5770,7 +5844,7 @@ static void StartSubTransaction(void)
5770 * The caller has to make sure to always reassign CurrentTransactionState5844 * The caller has to make sure to always reassign CurrentTransactionState
5771 * if it has a local pointer to it after calling this function.5845 * if it has a local pointer to it after calling this function.
5772 */5846 */
5773-static void CommitSubTransaction(void)5847+static void CommitSubTransaction(bool stpCommit)
5774{5848{
5775 TransactionState s = CurrentTransactionState;5849 TransactionState s = CurrentTransactionState;
5776 5850 
@@ -5862,14 +5936,27 @@ static void CommitSubTransaction(void)
5862 XactLockTableDelete(s->transactionId);5936 XactLockTableDelete(s->transactionId);
5863 }5937 }
5864 5938 
5939+ /*
5940+ * When commit within nedted store procedure, it will create a plan cache.
5941+ * During commit time, need to clean up those plan cache.
5942+ */
5943+ if (stpCommit) {
5944+ ResourceOwnerDecrementNPlanRefs(t_thrd.utils_cxt.CurrentResourceOwner, true);
5945+ ResourceOwnerDecrementNsnapshots(t_thrd.utils_cxt.CurrentResourceOwner, NULL);
5946+ }
5947+ 
5865 /* Other locks should get transferred to their parent resource owner. */5948 /* Other locks should get transferred to their parent resource owner. */
5866 ResourceOwnerRelease(s->curTransactionOwner, RESOURCE_RELEASE_LOCKS, true, false);5949 ResourceOwnerRelease(s->curTransactionOwner, RESOURCE_RELEASE_LOCKS, true, false);
5867 ResourceOwnerRelease(s->curTransactionOwner, RESOURCE_RELEASE_AFTER_LOCKS, true, false);5950 ResourceOwnerRelease(s->curTransactionOwner, RESOURCE_RELEASE_AFTER_LOCKS, true, false);
5868 5951 
5869 AtEOXact_GUC(true, s->gucNestLevel);5952 AtEOXact_GUC(true, s->gucNestLevel);
5870- AtEOSubXact_SPI(true, s->subTransactionId);5953+ if (!stpCommit) {
5954+ AtEOSubXact_SPI(true, s->subTransactionId, false, stpCommit);
5955+ }
5871 AtEOSubXact_on_commit_actions(true, s->subTransactionId, s->parent->subTransactionId);5956 AtEOSubXact_on_commit_actions(true, s->subTransactionId, s->parent->subTransactionId);
5872- AtEOSubXact_Namespace(true, s->subTransactionId, s->parent->subTransactionId);5957+ if (!stpCommit) {
5958+ AtEOSubXact_Namespace(true, s->subTransactionId, s->parent->subTransactionId);
5959+ }
5873 AtEOSubXact_Files(true, s->subTransactionId, s->parent->subTransactionId);5960 AtEOSubXact_Files(true, s->subTransactionId, s->parent->subTransactionId);
5874 AtEOSubXact_HashTables(true, s->nestingLevel);5961 AtEOSubXact_HashTables(true, s->nestingLevel);
5875 AtEOSubXact_PgStat(true, s->nestingLevel);5962 AtEOSubXact_PgStat(true, s->nestingLevel);
@@ -5895,7 +5982,7 @@ static void CommitSubTransaction(void)
5895 PopTransaction();5982 PopTransaction();
5896}5983}
5897 5984 
5898-static void AbortSubTransaction(void)5985+static void AbortSubTransaction(bool stpRollback)
5899{5986{
5900 TransactionState s = CurrentTransactionState;5987 TransactionState s = CurrentTransactionState;
5901 t_thrd.xact_cxt.bInAbortTransaction = true;5988 t_thrd.xact_cxt.bInAbortTransaction = true;
@@ -6045,9 +6132,11 @@ static void AbortSubTransaction(void)
6045 ResourceOwnerRelease(s->curTransactionOwner, RESOURCE_RELEASE_AFTER_LOCKS, false, false);6132 ResourceOwnerRelease(s->curTransactionOwner, RESOURCE_RELEASE_AFTER_LOCKS, false, false);
6046 6133 
6047 AtEOXact_GUC(false, s->gucNestLevel);6134 AtEOXact_GUC(false, s->gucNestLevel);
6048- AtEOSubXact_SPI(false, s->subTransactionId);6135+ AtEOSubXact_SPI(false, s->subTransactionId, stpRollback, false);
6049 AtEOSubXact_on_commit_actions(false, s->subTransactionId, s->parent->subTransactionId);6136 AtEOSubXact_on_commit_actions(false, s->subTransactionId, s->parent->subTransactionId);
6050- AtEOSubXact_Namespace(false, s->subTransactionId, s->parent->subTransactionId);6137+ if (!stpRollback) {
6138+ AtEOSubXact_Namespace(false, s->subTransactionId, s->parent->subTransactionId);
6139+ }
6051 AtEOSubXact_Files(false, s->subTransactionId, s->parent->subTransactionId);6140 AtEOSubXact_Files(false, s->subTransactionId, s->parent->subTransactionId);
6052 AtEOSubXact_HashTables(false, s->nestingLevel);6141 AtEOSubXact_HashTables(false, s->nestingLevel);
6053 AtEOSubXact_PgStat(false, s->nestingLevel);6142 AtEOSubXact_PgStat(false, s->nestingLevel);
@@ -6217,6 +6306,7 @@ static void PopTransaction(void)
6217 /* Ditto for ResourceOwner links */6306 /* Ditto for ResourceOwner links */
6218 t_thrd.utils_cxt.CurTransactionResourceOwner = s->parent->curTransactionOwner;6307 t_thrd.utils_cxt.CurTransactionResourceOwner = s->parent->curTransactionOwner;
6219 t_thrd.utils_cxt.CurrentResourceOwner = s->parent->curTransactionOwner;6308 t_thrd.utils_cxt.CurrentResourceOwner = s->parent->curTransactionOwner;
6309+ t_thrd.xact_cxt.currentSubTransactionId = s->parent->subTransactionId;
6220 6310 
6221 /* Free the old child structure */6311 /* Free the old child structure */
6222 if (s->name) {6312 if (s->name) {
Msrc/include/access/xact.h+3-3
@@ -311,12 +311,12 @@ extern void CopyTransactionIdLoggedIfAny(TransactionState state);
311extern bool TransactionIdIsCurrentTransactionId(TransactionId xid);311extern bool TransactionIdIsCurrentTransactionId(TransactionId xid);
312extern void CommandCounterIncrement(void);312extern void CommandCounterIncrement(void);
313extern void ForceSyncCommit(void);313extern void ForceSyncCommit(void);
314-extern void StartTransactionCommand(void);314+extern void StartTransactionCommand(bool stpRollback = false);
315-extern void CommitTransactionCommand(void);315+extern void CommitTransactionCommand(bool stpCommit = false);
316#ifdef PGXC316#ifdef PGXC
317extern void AbortCurrentTransactionOnce(void);317extern void AbortCurrentTransactionOnce(void);
318#endif318#endif
319-extern void AbortCurrentTransaction(void);319+extern void AbortCurrentTransaction(bool stpRollback = false);
320extern void BeginTransactionBlock(void);320extern void BeginTransactionBlock(void);
321extern bool EndTransactionBlock(void);321extern bool EndTransactionBlock(void);
322extern bool PrepareTransactionBlock(const char* gid);322extern bool PrepareTransactionBlock(const char* gid);
Msrc/include/commands/defrem.h+1-1
@@ -52,7 +52,7 @@ extern void CreateCast(CreateCastStmt* stmt);
52extern void DropCastById(Oid castOid);52extern void DropCastById(Oid castOid);
53extern void AlterFunctionNamespace(List* name, List* argtypes, bool isagg, const char* newschema);53extern void AlterFunctionNamespace(List* name, List* argtypes, bool isagg, const char* newschema);
54extern Oid AlterFunctionNamespace_oid(Oid procOid, Oid nspOid);54extern Oid AlterFunctionNamespace_oid(Oid procOid, Oid nspOid);
55-extern void ExecuteDoStmt(const DoStmt* stmt);55+extern void ExecuteDoStmt(const DoStmt* stmt, bool atomic);
56extern Oid get_cast_oid(Oid sourcetypeid, Oid targettypeid, bool missing_ok);56extern Oid get_cast_oid(Oid sourcetypeid, Oid targettypeid, bool missing_ok);
57 57 
58/* commands/operatorcmds.c */58/* commands/operatorcmds.c */
Msrc/include/executor/spi.h+10-2
@@ -56,12 +56,15 @@ typedef struct _SPI_plan* SPIPlanPtr;
56#define SPI_OK_UPDATE_RETURNING 1356#define SPI_OK_UPDATE_RETURNING 13
57#define SPI_OK_REWRITTEN 1457#define SPI_OK_REWRITTEN 14
58#define SPI_OK_MERGE 1558#define SPI_OK_MERGE 15
59+#define SPI_OPT_NOATOMIC (1 << 0)
59 60 
60extern THR_LOCAL PGDLLIMPORT uint32 SPI_processed;61extern THR_LOCAL PGDLLIMPORT uint32 SPI_processed;
61extern THR_LOCAL PGDLLIMPORT SPITupleTable* SPI_tuptable;62extern THR_LOCAL PGDLLIMPORT SPITupleTable* SPI_tuptable;
62extern THR_LOCAL PGDLLIMPORT int SPI_result;63extern THR_LOCAL PGDLLIMPORT int SPI_result;
63 64 
64extern int SPI_connect(CommandDest dest = DestSPI, void (*spiCallbackfn)(void*) = NULL, void* clientData = NULL);65extern int SPI_connect(CommandDest dest = DestSPI, void (*spiCallbackfn)(void*) = NULL, void* clientData = NULL);
66+extern int SPI_connect_ext(CommandDest dest = DestSPI, void (*spiCallbackfn)(void*) = NULL,
67+ void* clientData = NULL, int options =0);
65extern int SPI_finish(void);68extern int SPI_finish(void);
66extern void SPI_push(void);69extern void SPI_push(void);
67extern void SPI_pop(void);70extern void SPI_pop(void);
@@ -123,8 +126,13 @@ extern void SPI_scroll_cursor_fetch(Portal, FetchDirection direction, long count
123extern void SPI_scroll_cursor_move(Portal, FetchDirection direction, long count);126extern void SPI_scroll_cursor_move(Portal, FetchDirection direction, long count);
124extern void SPI_cursor_close(Portal portal);127extern void SPI_cursor_close(Portal portal);
125 128 
126-extern void AtEOXact_SPI(bool isCommit);129+extern void SPI_start_transaction(void);
127-extern void AtEOSubXact_SPI(bool isCommit, SubTransactionId mySubid);130+extern void SPI_commit(void);
131+extern void SPI_rollback(void);
132+extern void SPICleanup(void);
133+ 
134+extern void AtEOXact_SPI(bool isCommit, bool stpRollback, bool stpCommit);
135+extern void AtEOSubXact_SPI(bool isCommit, SubTransactionId mySubid, bool stpRollback, bool stpCommit);
128extern DestReceiver* createAnalyzeSPIDestReceiver(CommandDest dest);136extern DestReceiver* createAnalyzeSPIDestReceiver(CommandDest dest);
129/* SPI execution helpers */137/* SPI execution helpers */
130extern void spi_exec_with_callback(CommandDest dest, const char* src, bool read_only, long tcount, bool direct_call,138extern void spi_exec_with_callback(CommandDest dest, const char* src, bool read_only, long tcount, bool direct_call,
Msrc/include/executor/spi_priv.h+5-0
@@ -28,6 +28,11 @@ typedef struct _SPI_connection {
28 MemoryContext savedcxt; /* context of SPI_connect's caller */28 MemoryContext savedcxt; /* context of SPI_connect's caller */
29 SubTransactionId connectSubid; /* ID of connecting subtransaction */29 SubTransactionId connectSubid; /* ID of connecting subtransaction */
30 CommandDest dest; /* identify which is the orientated caller of spi interface, analyze or normal */30 CommandDest dest; /* identify which is the orientated caller of spi interface, analyze or normal */
31+ 
32+ /* transaction management suppoort */
33+ bool atomic; /* atomic execution context, does not allow transactions */
34+ bool internal_xact; /* SPI-managed transaction boundary, skip cleanup */
35+ 
31 void* clientData; /* argument to call back function */36 void* clientData; /* argument to call back function */
32 void (*spiCallback)(void*); /* callback for process received data. */37 void (*spiCallback)(void*); /* callback for process received data. */
33} _SPI_connection;38} _SPI_connection;
Msrc/include/knl/knl_session.h+8-0
@@ -183,6 +183,14 @@ typedef struct knl_u_SPI_context {
183 struct _SPI_connection* _stack;183 struct _SPI_connection* _stack;
184 184 
185 struct _SPI_connection* _current;185 struct _SPI_connection* _current;
186+ 
187+ bool is_toplevel_stp;
188+ 
189+ bool is_stp;
190+ 
191+ bool is_proconfig_set;
192+ 
193+ int portal_stp_exception_counter;
186} knl_u_SPI_context;194} knl_u_SPI_context;
187 195 
188typedef struct knl_u_index_context {196typedef struct knl_u_index_context {
Msrc/include/knl/knl_thread.h+1-0
@@ -1578,6 +1578,7 @@ typedef struct knl_t_utils_context {
1578 struct ResourceOwnerData* CurrentResourceOwner;1578 struct ResourceOwnerData* CurrentResourceOwner;
1579 struct ResourceOwnerData* CurTransactionResourceOwner;1579 struct ResourceOwnerData* CurTransactionResourceOwner;
1580 struct ResourceOwnerData* TopTransactionResourceOwner;1580 struct ResourceOwnerData* TopTransactionResourceOwner;
1581+ struct ResourceOwnerData* StpSavedResourceOwner;
1581 struct ResourceReleaseCallbackItem* ResourceRelease_callbacks;1582 struct ResourceReleaseCallbackItem* ResourceRelease_callbacks;
1582 bool SortColumnOptimize;1583 bool SortColumnOptimize;
1583 struct RelationData* pRelatedRel;1584 struct RelationData* pRelatedRel;
Msrc/include/nodes/execnodes.h+1-0
@@ -1734,6 +1734,7 @@ typedef struct FunctionScanState {
1734 TupleDesc tupdesc;1734 TupleDesc tupdesc;
1735 Tuplestorestate* tuplestorestate;1735 Tuplestorestate* tuplestorestate;
1736 ExprState* funcexpr;1736 ExprState* funcexpr;
1737+ bool atomic;
1737} FunctionScanState;1738} FunctionScanState;
1738 1739 
1739/* ----------------1740/* ----------------
Msrc/include/nodes/parsenodes.h+6-0
@@ -2683,8 +2683,14 @@ typedef struct InlineCodeBlock {
2683 char* source_text; /* source text of anonymous code block */2683 char* source_text; /* source text of anonymous code block */
2684 Oid langOid; /* OID of selected language */2684 Oid langOid; /* OID of selected language */
2685 bool langIsTrusted; /* trusted property of the language */2685 bool langIsTrusted; /* trusted property of the language */
2686+ bool atomic; /* atomic execution context */
2686} InlineCodeBlock;2687} InlineCodeBlock;
2687 2688 
2689+typedef struct CallContext {
2690+ NodeTag type;
2691+ bool atomic;
2692+} CallContext;
2693+ 
2688/* ----------------------2694/* ----------------------
2689 * Alter Object Rename Statement2695 * Alter Object Rename Statement
2690 * ----------------------2696 * ----------------------
Msrc/include/utils/portal.h+5-2
@@ -145,6 +145,7 @@ typedef struct PortalData {
145 /* Status data */145 /* Status data */
146 PortalStatus status; /* see above */146 PortalStatus status; /* see above */
147 bool portalPinned; /* a pinned portal can't be dropped */147 bool portalPinned; /* a pinned portal can't be dropped */
148+ bool autoHeld; /* was automatically converted from pinned to held */
148 149 
149 /* If not NULL, Executor is active; call ExecutorEnd eventually: */150 /* If not NULL, Executor is active; call ExecutorEnd eventually: */
150 QueryDesc* queryDesc; /* info needed for executor invocation */151 QueryDesc* queryDesc; /* info needed for executor invocation */
@@ -206,9 +207,10 @@ typedef struct PortalData {
206 207 
207/* Prototypes for functions in utils/mmgr/portalmem.c */208/* Prototypes for functions in utils/mmgr/portalmem.c */
208extern void EnablePortalManager(void);209extern void EnablePortalManager(void);
209-extern bool PreCommit_Portals(bool isPrepare);210+extern bool PreCommit_Portals(bool isPrepare, bool stpCommit);
210-extern void AtAbort_Portals(void);211+extern void AtAbort_Portals(bool stpRollback);
211extern void AtCleanup_Portals(void);212extern void AtCleanup_Portals(void);
213+extern void PortalErrorCleanup(void);
212extern void AtSubCommit_Portals(SubTransactionId mySubid, SubTransactionId parentSubid, ResourceOwner parentXactOwner);214extern void AtSubCommit_Portals(SubTransactionId mySubid, SubTransactionId parentSubid, ResourceOwner parentXactOwner);
213extern void AtSubAbort_Portals(215extern void AtSubAbort_Portals(
214 SubTransactionId mySubid, SubTransactionId parentSubid, ResourceOwner myXactOwner, ResourceOwner parentXactOwner);216 SubTransactionId mySubid, SubTransactionId parentSubid, ResourceOwner myXactOwner, ResourceOwner parentXactOwner);
@@ -229,5 +231,6 @@ extern void PortalCreateHoldStore(Portal portal);
229extern void PortalHashTableDeleteAll(void);231extern void PortalHashTableDeleteAll(void);
230extern bool ThereAreNoReadyPortals(void);232extern bool ThereAreNoReadyPortals(void);
231extern void ResetPortalCursor(SubTransactionId mySubid, Oid funOid, int funUseCount);233extern void ResetPortalCursor(SubTransactionId mySubid, Oid funOid, int funUseCount);
234+extern void HoldPinnedPortals(void);
232 235 
233#endif /* PORTAL_H */236#endif /* PORTAL_H */
Msrc/include/utils/resowner.h+5-0
@@ -63,6 +63,9 @@ extern ResourceOwner ResourceOwnerCreate(ResourceOwner parent, const char* name)
63extern void ResourceOwnerRelease(ResourceOwner owner, ResourceReleasePhase phase, bool isCommit, bool isTopLevel);63extern void ResourceOwnerRelease(ResourceOwner owner, ResourceReleasePhase phase, bool isCommit, bool isTopLevel);
64extern void ResourceOwnerDelete(ResourceOwner owner);64extern void ResourceOwnerDelete(ResourceOwner owner);
65extern ResourceOwner ResourceOwnerGetParent(ResourceOwner owner);65extern ResourceOwner ResourceOwnerGetParent(ResourceOwner owner);
66+extern ResourceOwner ResourceOwnerGetNextChild(ResourceOwner owner);
67+extern ResourceOwner ResourceOwnerGetFirstChild(ResourceOwner owner);
68+extern const char* ResourceOwnerGetName(ResourceOwner owner);
66extern void ResourceOwnerNewParent(ResourceOwner owner, ResourceOwner newparent);69extern void ResourceOwnerNewParent(ResourceOwner owner, ResourceOwner newparent);
67extern void RegisterResourceReleaseCallback(ResourceReleaseCallback callback, void* arg);70extern void RegisterResourceReleaseCallback(ResourceReleaseCallback callback, void* arg);
68extern void UnregisterResourceReleaseCallback(ResourceReleaseCallback callback, void* arg);71extern void UnregisterResourceReleaseCallback(ResourceReleaseCallback callback, void* arg);
@@ -116,6 +119,8 @@ extern void ResourceOwnerForgetTupleDesc(ResourceOwner owner, const TupleDesc tu
116extern void ResourceOwnerEnlargeSnapshots(ResourceOwner owner);119extern void ResourceOwnerEnlargeSnapshots(ResourceOwner owner);
117extern void ResourceOwnerRememberSnapshot(ResourceOwner owner, Snapshot snapshot);120extern void ResourceOwnerRememberSnapshot(ResourceOwner owner, Snapshot snapshot);
118extern void ResourceOwnerForgetSnapshot(ResourceOwner owner, const Snapshot snapshot);121extern void ResourceOwnerForgetSnapshot(ResourceOwner owner, const Snapshot snapshot);
122+extern void ResourceOwnerDecrementNsnapshots(ResourceOwner owner, void* queryDesc);
123+extern void ResourceOwnerDecrementNPlanRefs(ResourceOwner owner, bool useResOwner);
119 124 
120/* support for temporary file management */125/* support for temporary file management */
121extern void ResourceOwnerEnlargeFiles(ResourceOwner owner);126extern void ResourceOwnerEnlargeFiles(ResourceOwner owner);
Asrc/test/regress/expected/transactions_control.out+379-0
@@ -0,0 +1,379 @@
1+CREATE TABLE test1 (a int, b text);
2+CREATE PROCEDURE transaction_test1()
3+AS
4+BEGIN
5+ FOR i IN 0..9 LOOP
6+ INSERT INTO test1 (a) VALUES (i);
7+ IF i % 2 = 0 THEN
8+ COMMIT;
9+ ELSE
10+ ROLLBACK;
11+ END IF;
12+ END LOOP;
13+END;
14+/
15+CALL transaction_test1();
16+ transaction_test1
17+-------------------
18+
19+(1 row)
20+ 
21+SELECT * FROM test1;
22+ a | b
23+---+---
24+ 0 |
25+ 2 |
26+ 4 |
27+ 6 |
28+ 8 |
29+(5 rows)
30+ 
31+TRUNCATE test1;
32+DO
33+LANGUAGE plpgsql
34+$$
35+BEGIN
36+ FOR i IN 0..9 LOOP
37+ INSERT INTO test1 (a) VALUES (i);
38+ IF i % 2 = 0 THEN
39+ COMMIT;
40+ ELSE
41+ ROLLBACK;
42+ END IF;
43+ END LOOP;
44+END
45+$$;
46+SELECT * FROM test1;
47+ a | b
48+---+---
49+ 0 |
50+ 2 |
51+ 4 |
52+ 6 |
53+ 8 |
54+(5 rows)
55+ 
56+-- transaction commands not allowed when called in transaction block
57+START TRANSACTION;
58+CALL transaction_test1();
59+ERROR: invalid transaction termination
60+CONTEXT: PL/pgSQL function transaction_test1() line 6 at COMMIT
61+COMMIT;
62+START TRANSACTION;
63+DO LANGUAGE plpgsql $$ BEGIN COMMIT; END $$;
64+ERROR: invalid transaction termination
65+CONTEXT: PL/pgSQL function inline_code_block line 1 at COMMIT
66+COMMIT;
67+TRUNCATE test1;
68+-- not allowed in a function
69+CREATE FUNCTION transaction_test2() RETURNS int
70+LANGUAGE plpgsql
71+AS $$
72+BEGIN
73+ FOR i IN 0..9 LOOP
74+ INSERT INTO test1 (a) VALUES (i);
75+ IF i % 2 = 0 THEN
76+ COMMIT;
77+ ELSE
78+ ROLLBACK;
79+ END IF;
80+ END LOOP;
81+ RETURN 1;
82+END
83+$$;
84+SELECT transaction_test2();
85+ERROR: cannot commit within function
86+CONTEXT: PL/pgSQL function transaction_test2() line 6 at COMMIT
87+referenced column: transaction_test2
88+SELECT * FROM test1;
89+ a | b
90+---+---
91+(0 rows)
92+ 
93+-- also not allowed if procedure is called from a function
94+CREATE FUNCTION transaction_test3() RETURNS int
95+LANGUAGE plpgsql
96+AS $$
97+BEGIN
98+ CALL transaction_test1();
99+ RETURN 1;
100+END;
101+$$;
102+SELECT transaction_test3();
103+ERROR: cannot commit within function
104+CONTEXT: PL/pgSQL function transaction_test1() line 6 at COMMIT
105+SQL statement "CALL transaction_test1()"
106+PL/pgSQL function transaction_test3() line 3 at SQL statement
107+referenced column: transaction_test3
108+SELECT * FROM test1;
109+ a | b
110+---+---
111+(0 rows)
112+ 
113+-- DO block inside function
114+CREATE FUNCTION transaction_test4() RETURNS int
115+LANGUAGE plpgsql
116+AS $$
117+BEGIN
118+ EXECUTE 'DO LANGUAGE plpgsql $x$ BEGIN COMMIT; END $x$';
119+ RETURN 1;
120+END;
121+$$;
122+SELECT transaction_test4();
123+ERROR: cannot commit within function
124+CONTEXT: PL/pgSQL function inline_code_block line 1 at COMMIT
125+SQL statement "DO LANGUAGE plpgsql $x$ BEGIN COMMIT; END $x$"
126+PL/pgSQL function transaction_test4() line 3 at EXECUTE statement
127+referenced column: transaction_test4
128+-- proconfig settings currently disallow transaction statements
129+CREATE PROCEDURE transaction_test5()
130+SET work_mem = 555
131+AS
132+BEGIN
133+ COMMIT;
134+END;
135+/
136+CALL transaction_test5();
137+ERROR: invalid transaction termination
138+CONTEXT: PL/pgSQL function transaction_test5() line 3 at COMMIT
139+-- commit inside cursor loop
140+CREATE TABLE test2 (x int);
141+INSERT INTO test2 VALUES (0), (1), (2), (3), (4);
142+TRUNCATE test1;
143+DO LANGUAGE plpgsql $$
144+DECLARE
145+ r RECORD;
146+BEGIN
147+ FOR r IN SELECT * FROM test2 ORDER BY x LOOP
148+ INSERT INTO test1 (a) VALUES (r.x);
149+ COMMIT;
150+ END LOOP;
151+END;
152+$$;
153+SELECT * FROM test1;
154+ a | b
155+---+---
156+ 0 |
157+ 1 |
158+ 2 |
159+ 3 |
160+ 4 |
161+(5 rows)
162+ 
163+-- check that this doesn't leak a holdable portal
164+SELECT * FROM pg_cursors;
165+ name | statement | is_holdable | is_binary | is_scrollable | creation_time
166+------+-----------+-------------+-----------+---------------+---------------
167+(0 rows)
168+ 
169+-- error in cursor loop with commit
170+TRUNCATE test1;
171+DO LANGUAGE plpgsql $$
172+DECLARE
173+ r RECORD;
174+BEGIN
175+ FOR r IN SELECT * FROM test2 ORDER BY x LOOP
176+ INSERT INTO test1 (a) VALUES (12/(r.x-2));
177+ COMMIT;
178+ END LOOP;
179+END;
180+$$;
181+ERROR: division by zero
182+CONTEXT: referenced column: a
183+SQL statement "INSERT INTO test1 (a) VALUES (12/(r.x-2))"
184+PL/pgSQL function inline_code_block line 6 at SQL statement
185+SELECT * FROM test1;
186+ a | b
187+-----+---
188+ -6 |
189+ -12 |
190+(2 rows)
191+ 
192+SELECT * FROM pg_cursors;
193+ name | statement | is_holdable | is_binary | is_scrollable | creation_time
194+------+-----------+-------------+-----------+---------------+---------------
195+(0 rows)
196+ 
197+-- rollback inside cursor loop
198+TRUNCATE test1;
199+DO LANGUAGE plpgsql $$
200+DECLARE
201+ r RECORD;
202+BEGIN
203+ FOR r IN SELECT * FROM test2 ORDER BY x LOOP
204+ INSERT INTO test1 (a) VALUES (r.x);
205+ ROLLBACK;
206+ END LOOP;
207+END;
208+$$;
209+SELECT * FROM test1;
210+ a | b
211+---+---
212+(0 rows)
213+ 
214+SELECT * FROM pg_cursors;
215+ name | statement | is_holdable | is_binary | is_scrollable | creation_time
216+------+-----------+-------------+-----------+---------------+---------------
217+(0 rows)
218+ 
219+-- first commit then rollback inside cursor loop
220+TRUNCATE test1;
221+DO LANGUAGE plpgsql $$
222+DECLARE
223+ r RECORD;
224+BEGIN
225+ FOR r IN SELECT * FROM test2 ORDER BY x LOOP
226+ INSERT INTO test1 (a) VALUES (r.x);
227+ IF r.x % 2 = 0 THEN
228+ COMMIT;
229+ ELSE
230+ ROLLBACK;
231+ END IF;
232+ END LOOP;
233+END;
234+$$;
235+SELECT * FROM test1;
236+ a | b
237+---+---
238+ 0 |
239+ 2 |
240+ 4 |
241+(3 rows)
242+ 
243+SELECT * FROM pg_cursors;
244+ name | statement | is_holdable | is_binary | is_scrollable | creation_time
245+------+-----------+-------------+-----------+---------------+---------------
246+(0 rows)
247+ 
248+-- rollback inside cursor loop
249+TRUNCATE test1;
250+DO LANGUAGE plpgsql $$
251+DECLARE
252+ r RECORD;
253+BEGIN
254+ FOR r IN UPDATE test2 SET x = x * 2 RETURNING x LOOP
255+ INSERT INTO test1 (a) VALUES (r.x);
256+ ROLLBACK;
257+ END LOOP;
258+END;
259+$$;
260+ERROR: cannot perform transaction commands inside a cursor loop that is not read-only
261+CONTEXT: PL/pgSQL function inline_code_block line 7 at ROLLBACK
262+SELECT * FROM test1;
263+ a | b
264+---+---
265+(0 rows)
266+ 
267+SELECT * FROM test2;
268+ x
269+---
270+ 0
271+ 1
272+ 2
273+ 3
274+ 4
275+(5 rows)
276+ 
277+SELECT * FROM pg_cursors;
278+ name | statement | is_holdable | is_binary | is_scrollable | creation_time
279+------+-----------+-------------+-----------+---------------+---------------
280+(0 rows)
281+ 
282+-- commit inside block with exception handler
283+TRUNCATE test1;
284+DO LANGUAGE plpgsql $$
285+BEGIN
286+ BEGIN
287+ INSERT INTO test1 (a) VALUES (1);
288+ COMMIT;
289+ INSERT INTO test1 (a) VALUES (1/0);
290+ COMMIT;
291+ EXCEPTION
292+ WHEN division_by_zero THEN
293+ RAISE NOTICE 'caught division_by_zero';
294+ END;
295+END;
296+$$;
297+NOTICE: caught division_by_zero
298+SELECT * FROM test1;
299+ a | b
300+---+---
301+ 1 |
302+(1 row)
303+ 
304+-- rollback inside block with exception handler
305+TRUNCATE test1;
306+DO LANGUAGE plpgsql $$
307+BEGIN
308+ BEGIN
309+ INSERT INTO test1 (a) VALUES (1);
310+ ROLLBACK;
311+ INSERT INTO test1 (a) VALUES (1/0);
312+ ROLLBACK;
313+ EXCEPTION
314+ WHEN division_by_zero THEN
315+ RAISE NOTICE 'caught division_by_zero';
316+ END;
317+END;
318+$$;
319+NOTICE: caught division_by_zero
320+SELECT * FROM test1;
321+ a | b
322+---+---
323+(0 rows)
324+ 
325+-- COMMIT failures
326+DO LANGUAGE plpgsql $$
327+BEGIN
328+ CREATE TABLE test3 (y int UNIQUE DEFERRABLE INITIALLY DEFERRED);
329+ COMMIT;
330+ INSERT INTO test3 (y) VALUES (1);
331+ COMMIT;
332+ INSERT INTO test3 (y) VALUES (1);
333+ INSERT INTO test3 (y) VALUES (2);
334+ COMMIT;
335+ INSERT INTO test3 (y) VALUES (3); -- won't get here
336+END;
337+$$;
338+NOTICE: CREATE TABLE / UNIQUE will create implicit index "test3_y_key" for table "test3"
339+CONTEXT: SQL statement "CREATE TABLE test3 (y int UNIQUE DEFERRABLE INITIALLY DEFERRED)"
340+PL/pgSQL function inline_code_block line 3 at SQL statement
341+ERROR: duplicate key value violates unique constraint "test3_y_key"
342+DETAIL: Key (y)=(1) already exists.
343+CONTEXT: PL/pgSQL function inline_code_block line 9 at COMMIT
344+SELECT * FROM test3;
345+ y
346+---
347+ 1
348+(1 row)
349+ 
350+DROP TABLE test1;
351+DROP TABLE test2;
352+DROP TABLE test3;
353+--
354+CREATE TABLE test1(id int, name varchar(20));
355+INSERT INTO test1 values(1, 'bbb');
356+CREATE OR REPLACE PROCEDURE PROC_OUT_PARAM_001(P1 OUT INT)
357+AS
358+BEGIN
359+select id into P1 from test1 where name = 'bbb';
360+insert into test1 values(P1, 'dddd');
361+COMMIT;
362+insert into test1 values(P1, 'eee');
363+ROLLBACK;
364+END;
365+/
366+DECLARE
367+V_P1 INT;
368+BEGIN
369+PROC_OUT_PARAM_001(V_P1);
370+END;
371+/
372+SELECT * from test1;
373+ id | name
374+----+------
375+ 1 | bbb
376+ 1 | dddd
377+(2 rows)
378+ 
379+DROP TABLE TEST1;
Msrc/test/regress/parallel_schedule+1-1
@@ -248,7 +248,7 @@ test: subplan_new
248test: select248test: select
249test: col_subplan_base_1 col_subplan_new249test: col_subplan_base_1 col_subplan_new
250test: join250test: join
251-test: select_into select_distinct subselect_part1 subselect_part2 transactions random btree_index select_distinct_on union gs_aggregate arrays hash_index251+test: select_into select_distinct subselect_part1 subselect_part2 transactions transactions_control random btree_index select_distinct_on union gs_aggregate arrays hash_index
252test: aggregates252test: aggregates
253test: portals_p2 window tsearch temp__6 holdable_cursor col_subplan_base_2253test: portals_p2 window tsearch temp__6 holdable_cursor col_subplan_base_2
254 254 
Asrc/test/regress/sql/transactions_control.sql+305-0
@@ -0,0 +1,305 @@
1+CREATE TABLE test1 (a int, b text);
2+ 
3+ 
4+CREATE PROCEDURE transaction_test1()
5+AS
6+BEGIN
7+ FOR i IN 0..9 LOOP
8+ INSERT INTO test1 (a) VALUES (i);
9+ IF i % 2 = 0 THEN
10+ COMMIT;
11+ ELSE
12+ ROLLBACK;
13+ END IF;
14+ END LOOP;
15+END;
16+/
17+ 
18+CALL transaction_test1();
19+ 
20+SELECT * FROM test1;
21+ 
22+ 
23+TRUNCATE test1;
24+ 
25+DO
26+LANGUAGE plpgsql
27+$$
28+BEGIN
29+ FOR i IN 0..9 LOOP
30+ INSERT INTO test1 (a) VALUES (i);
31+ IF i % 2 = 0 THEN
32+ COMMIT;
33+ ELSE
34+ ROLLBACK;
35+ END IF;
36+ END LOOP;
37+END
38+$$;
39+ 
40+SELECT * FROM test1;
41+ 
42+ 
43+-- transaction commands not allowed when called in transaction block
44+START TRANSACTION;
45+CALL transaction_test1();
46+COMMIT;
47+ 
48+START TRANSACTION;
49+DO LANGUAGE plpgsql $$ BEGIN COMMIT; END $$;
50+COMMIT;
51+ 
52+ 
53+TRUNCATE test1;
54+ 
55+-- not allowed in a function
56+CREATE FUNCTION transaction_test2() RETURNS int
57+LANGUAGE plpgsql
58+AS $$
59+BEGIN
60+ FOR i IN 0..9 LOOP
61+ INSERT INTO test1 (a) VALUES (i);
62+ IF i % 2 = 0 THEN
63+ COMMIT;
64+ ELSE
65+ ROLLBACK;
66+ END IF;
67+ END LOOP;
68+ RETURN 1;
69+END
70+$$;
71+ 
72+SELECT transaction_test2();
73+ 
74+SELECT * FROM test1;
75+ 
76+ 
77+-- also not allowed if procedure is called from a function
78+CREATE FUNCTION transaction_test3() RETURNS int
79+LANGUAGE plpgsql
80+AS $$
81+BEGIN
82+ CALL transaction_test1();
83+ RETURN 1;
84+END;
85+$$;
86+ 
87+SELECT transaction_test3();
88+ 
89+SELECT * FROM test1;
90+ 
91+ 
92+-- DO block inside function
93+CREATE FUNCTION transaction_test4() RETURNS int
94+LANGUAGE plpgsql
95+AS $$
96+BEGIN
97+ EXECUTE 'DO LANGUAGE plpgsql $x$ BEGIN COMMIT; END $x$';
98+ RETURN 1;
99+END;
100+$$;
101+ 
102+SELECT transaction_test4();
103+ 
104+ 
105+-- proconfig settings currently disallow transaction statements
106+CREATE PROCEDURE transaction_test5()
107+SET work_mem = 555
108+AS
109+BEGIN
110+ COMMIT;
111+END;
112+/
113+ 
114+CALL transaction_test5();
115+ 
116+ 
117+-- commit inside cursor loop
118+CREATE TABLE test2 (x int);
119+INSERT INTO test2 VALUES (0), (1), (2), (3), (4);
120+ 
121+TRUNCATE test1;
122+ 
123+DO LANGUAGE plpgsql $$
124+DECLARE
125+ r RECORD;
126+BEGIN
127+ FOR r IN SELECT * FROM test2 ORDER BY x LOOP
128+ INSERT INTO test1 (a) VALUES (r.x);
129+ COMMIT;
130+ END LOOP;
131+END;
132+$$;
133+ 
134+SELECT * FROM test1;
135+ 
136+-- check that this doesn't leak a holdable portal
137+SELECT * FROM pg_cursors;
138+ 
139+ 
140+-- error in cursor loop with commit
141+TRUNCATE test1;
142+ 
143+DO LANGUAGE plpgsql $$
144+DECLARE
145+ r RECORD;
146+BEGIN
147+ FOR r IN SELECT * FROM test2 ORDER BY x LOOP
148+ INSERT INTO test1 (a) VALUES (12/(r.x-2));
149+ COMMIT;
150+ END LOOP;
151+END;
152+$$;
153+ 
154+SELECT * FROM test1;
155+ 
156+SELECT * FROM pg_cursors;
157+ 
158+ 
159+-- rollback inside cursor loop
160+TRUNCATE test1;
161+ 
162+DO LANGUAGE plpgsql $$
163+DECLARE
164+ r RECORD;
165+BEGIN
166+ FOR r IN SELECT * FROM test2 ORDER BY x LOOP
167+ INSERT INTO test1 (a) VALUES (r.x);
168+ ROLLBACK;
169+ END LOOP;
170+END;
171+$$;
172+ 
173+SELECT * FROM test1;
174+ 
175+SELECT * FROM pg_cursors;
176+ 
177+ 
178+-- first commit then rollback inside cursor loop
179+TRUNCATE test1;
180+ 
181+DO LANGUAGE plpgsql $$
182+DECLARE
183+ r RECORD;
184+BEGIN
185+ FOR r IN SELECT * FROM test2 ORDER BY x LOOP
186+ INSERT INTO test1 (a) VALUES (r.x);
187+ IF r.x % 2 = 0 THEN
188+ COMMIT;
189+ ELSE
190+ ROLLBACK;
191+ END IF;
192+ END LOOP;
193+END;
194+$$;
195+ 
196+SELECT * FROM test1;
197+ 
198+SELECT * FROM pg_cursors;
199+ 
200+ 
201+-- rollback inside cursor loop
202+TRUNCATE test1;
203+ 
204+DO LANGUAGE plpgsql $$
205+DECLARE
206+ r RECORD;
207+BEGIN
208+ FOR r IN UPDATE test2 SET x = x * 2 RETURNING x LOOP
209+ INSERT INTO test1 (a) VALUES (r.x);
210+ ROLLBACK;
211+ END LOOP;
212+END;
213+$$;
214+ 
215+SELECT * FROM test1;
216+SELECT * FROM test2;
217+ 
218+SELECT * FROM pg_cursors;
219+ 
220+-- commit inside block with exception handler
221+TRUNCATE test1;
222+ 
223+DO LANGUAGE plpgsql $$
224+BEGIN
225+ BEGIN
226+ INSERT INTO test1 (a) VALUES (1);
227+ COMMIT;
228+ INSERT INTO test1 (a) VALUES (1/0);
229+ COMMIT;
230+ EXCEPTION
231+ WHEN division_by_zero THEN
232+ RAISE NOTICE 'caught division_by_zero';
233+ END;
234+END;
235+$$;
236+ 
237+SELECT * FROM test1;
238+ 
239+ 
240+-- rollback inside block with exception handler
241+TRUNCATE test1;
242+ 
243+DO LANGUAGE plpgsql $$
244+BEGIN
245+ BEGIN
246+ INSERT INTO test1 (a) VALUES (1);
247+ ROLLBACK;
248+ INSERT INTO test1 (a) VALUES (1/0);
249+ ROLLBACK;
250+ EXCEPTION
251+ WHEN division_by_zero THEN
252+ RAISE NOTICE 'caught division_by_zero';
253+ END;
254+END;
255+$$;
256+ 
257+SELECT * FROM test1;
258+ 
259+ 
260+-- COMMIT failures
261+DO LANGUAGE plpgsql $$
262+BEGIN
263+ CREATE TABLE test3 (y int UNIQUE DEFERRABLE INITIALLY DEFERRED);
264+ COMMIT;
265+ INSERT INTO test3 (y) VALUES (1);
266+ COMMIT;
267+ INSERT INTO test3 (y) VALUES (1);
268+ INSERT INTO test3 (y) VALUES (2);
269+ COMMIT;
270+ INSERT INTO test3 (y) VALUES (3); -- won't get here
271+END;
272+$$;
273+ 
274+SELECT * FROM test3;
275+ 
276+ 
277+DROP TABLE test1;
278+DROP TABLE test2;
279+DROP TABLE test3;
280+ 
281+--
282+CREATE TABLE test1(id int, name varchar(20));
283+INSERT INTO test1 values(1, 'bbb');
284+ 
285+CREATE OR REPLACE PROCEDURE PROC_OUT_PARAM_001(P1 OUT INT)
286+AS
287+BEGIN
288+select id into P1 from test1 where name = 'bbb';
289+insert into test1 values(P1, 'dddd');
290+COMMIT;
291+insert into test1 values(P1, 'eee');
292+ROLLBACK;
293+END;
294+/
295+ 
296+DECLARE
297+V_P1 INT;
298+BEGIN
299+PROC_OUT_PARAM_001(V_P1);
300+END;
301+/
302+ 
303+SELECT * from test1;
304+ 
305+DROP TABLE TEST1;