Merge pull request #1574 from citusdata/auto_2pc_recovery

Auto-recover 2PCs, enable 2PC by default

src/backend/distributed/executor/multi_utility.c

@@ -156,7 +156,6 @@ static bool IsAlterTableRenameStmt(RenameStmt *renameStmt);
 static bool AlterInvolvesPartitionColumn(AlterTableStmt *alterTableStatement,
 										 AlterTableCmd *command);
 static void ExecuteDistributedDDLJob(DDLJob *ddlJob);
-static void ShowNoticeIfNotUsing2PC(void);
 static List * DDLTaskList(Oid relationId, const char *commandString);
 static List * CreateIndexTaskList(Oid relationId, IndexStmt *indexStmt);
 static List * DropIndexTaskList(Oid relationId, Oid indexId, DropStmt *dropStmt);
@@ -169,9 +168,6 @@ static List * CopyGetAttnums(TupleDesc tupDesc, Relation rel, List *attnamelist)
 static void PostProcessUtility(Node *parsetree);
 
 
-static bool warnedUserAbout2PC = false;
-
-
 /*
  * multi_ProcessUtility9x is the 9.x-compatible wrapper for Citus' main utility
  * hook. It simply adapts the old-style hook to call into the new-style (10+)
@@ -2876,8 +2872,6 @@ ExecuteDistributedDDLJob(DDLJob *ddlJob)
 
 	if (!ddlJob->concurrentIndexCmd)
 	{
-		ShowNoticeIfNotUsing2PC();
-
 		if (shouldSyncMetadata)
 		{
 			SendCommandToWorkers(WORKERS_WITH_METADATA, DISABLE_DDL_PROPAGATION);
@@ -2919,25 +2913,6 @@ ExecuteDistributedDDLJob(DDLJob *ddlJob)
 }
 
 
-/*
- * ShowNoticeIfNotUsing2PC shows a notice message about using 2PC by setting
- * citus.multi_shard_commit_protocol to 2PC. The notice message is shown only once in a
- * session
- */
-static void
-ShowNoticeIfNotUsing2PC(void)
-{
-	if (MultiShardCommitProtocol != COMMIT_PROTOCOL_2PC && !warnedUserAbout2PC)
-	{
-		ereport(NOTICE, (errmsg("using one-phase commit for distributed DDL commands"),
-						 errhint("You can enable two-phase commit for extra safety with: "
-								 "SET citus.multi_shard_commit_protocol TO '2pc'")));
-
-		warnedUserAbout2PC = true;
-	}
-}
-
-
 /*
  * DDLTaskList builds a list of tasks to execute a DDL command on a
  * given list of shards.

src/backend/distributed/shared_library_init.c

@@ -43,6 +43,7 @@
 #include "distributed/statistics_collection.h"
 #include "distributed/task_tracker.h"
 #include "distributed/transaction_management.h"
+#include "distributed/transaction_recovery.h"
 #include "distributed/worker_manager.h"
 #include "distributed/worker_protocol.h"
 #include "postmaster/postmaster.h"
@@ -457,6 +458,20 @@ RegisterCitusConfigVariables(void)
 		0,
 		ErrorIfNotASuitableDeadlockFactor, NULL, NULL);
 
+	DefineCustomIntVariable(
+		"citus.recover_2pc_interval",
+		gettext_noop("Sets the time to wait between recovering 2PCs."),
+		gettext_noop("2PC transaction recovery needs to run every so often "
+					 "to clean up records in pg_dist_transaction and "
+					 "potentially roll failed 2PCs forward. This setting "
+					 "determines how often recovery should run, "
+					 "use -1 to disable."),
+		&Recover2PCInterval,
+		60000, -1, 7 * 24 * 3600 * 1000,
+		PGC_SIGHUP,
+		GUC_UNIT_MS,
+		NULL, NULL, NULL);
+
 	DefineCustomBoolVariable(
 		"citus.enable_deadlock_prevention",
 		gettext_noop("Prevents transactions from expanding to multiple nodes"),
@@ -670,13 +685,11 @@ RegisterCitusConfigVariables(void)
 		"citus.multi_shard_commit_protocol",
 		gettext_noop("Sets the commit protocol for commands modifying multiple shards."),
 		gettext_noop("When a failure occurs during commands that modify multiple "
-					 "shards (currently, only COPY on distributed tables modifies more "
+					 "shards, two-phase commit is required to ensure data is never lost "
-					 "than one shard), two-phase commit is required to ensure data is "
+					 "and this is the default. However, changing to 1pc may give small "
-					 "never lost. Change this setting to '2pc' from its default '1pc' to "
+					 "performance benefits."),
-					 "enable 2 PC. You must also set max_prepared_transactions on the "
-					 "worker nodes. Recovery from failed 2PCs is currently manual."),
 		&MultiShardCommitProtocol,
-		COMMIT_PROTOCOL_1PC,
+		COMMIT_PROTOCOL_2PC,
 		multi_shard_commit_protocol_options,
 		PGC_USERSET,
 		0,

src/backend/distributed/transaction/transaction_management.c

@@ -33,7 +33,7 @@
 CoordinatedTransactionState CurrentCoordinatedTransactionState = COORD_TRANS_NONE;
 
 /* GUC, the commit protocol to use for commands affecting more than one connection */
-int MultiShardCommitProtocol = COMMIT_PROTOCOL_1PC;
+int MultiShardCommitProtocol = COMMIT_PROTOCOL_2PC;
 int SavedMultiShardCommitProtocol = COMMIT_PROTOCOL_BARE;
 
 /* state needed to keep track of operations used during a transaction */

src/backend/distributed/transaction/transaction_recovery.c

@@ -46,7 +46,6 @@ PG_FUNCTION_INFO_V1(recover_prepared_transactions);
 
 
 /* Local functions forward declarations */
-static int RecoverPreparedTransactions(void);
 static int RecoverWorkerTransactions(WorkerNode *workerNode);
 static List * PendingWorkerTransactionList(MultiConnection *connection);
 static bool IsTransactionInProgress(HTAB *activeTransactionNumberSet,
@@ -66,7 +65,7 @@ recover_prepared_transactions(PG_FUNCTION_ARGS)
 
 	CheckCitusVersion(ERROR);
 
-	recoveredTransactionCount = RecoverPreparedTransactions();
+	recoveredTransactionCount = RecoverTwoPhaseCommits();
 
 	PG_RETURN_INT32(recoveredTransactionCount);
 }
@@ -109,11 +108,11 @@ LogTransactionRecord(int groupId, char *transactionName)
 
 
 /*
- * RecoverPreparedTransactions recovers any pending prepared
+ * RecoverTwoPhaseCommits recovers any pending prepared
  * transactions started by this node on other nodes.
  */
-static int
+int
-RecoverPreparedTransactions(void)
+RecoverTwoPhaseCommits(void)
 {
 	List *workerList = NIL;
 	ListCell *workerNodeCell = NULL;

src/backend/distributed/utils/listutils.c

@@ -121,7 +121,7 @@ ListToHashSet(List *itemList, Size keySize, bool isStringList)
 	HASHCTL info;
 	HTAB *itemSet = NULL;
 	ListCell *itemCell = NULL;
-	int flags = HASH_ELEM;
+	int flags = HASH_ELEM | HASH_CONTEXT;
 
 	/* allocate sufficient capacity for O(1) expected look-up time */
 	int capacity = (int) (list_length(itemList) / 0.75) + 1;
@@ -130,6 +130,7 @@ ListToHashSet(List *itemList, Size keySize, bool isStringList)
 	memset(&info, 0, sizeof(info));
 	info.keysize = keySize;
 	info.entrysize = keySize;
+	info.hcxt = CurrentMemoryContext;
 
 	if (!isStringList)
 	{

src/backend/distributed/utils/maintenanced.c

@@ -22,8 +22,10 @@
 #include "pgstat.h"
 
 #include "access/xact.h"
+#include "access/xlog.h"
 #include "catalog/pg_extension.h"
 #include "citus_version.h"
+#include "catalog/pg_namespace.h"
 #include "commands/extension.h"
 #include "libpq/pqsignal.h"
 #include "catalog/namespace.h"
@@ -32,8 +34,10 @@
 #include "distributed/master_protocol.h"
 #include "distributed/metadata_cache.h"
 #include "distributed/statistics_collection.h"
+#include "distributed/transaction_recovery.h"
 #include "nodes/makefuncs.h"
 #include "postmaster/bgworker.h"
+#include "nodes/makefuncs.h"
 #include "storage/ipc.h"
 #include "storage/proc.h"
 #include "storage/latch.h"
@@ -41,6 +45,7 @@
 #include "storage/lwlock.h"
 #include "tcop/tcopprot.h"
 #include "utils/memutils.h"
+#include "utils/lsyscache.h"
 
 
 /*
@@ -80,6 +85,7 @@ typedef struct MaintenanceDaemonDBData
 
 /* config variable for distributed deadlock detection timeout */
 double DistributedDeadlockDetectionTimeoutFactor = 2.0;
+int Recover2PCInterval = 60000;
 
 static shmem_startup_hook_type prev_shmem_startup_hook = NULL;
 static MaintenanceDaemonControlData *MaintenanceDaemonControl = NULL;
@@ -221,6 +227,7 @@ CitusMaintenanceDaemonMain(Datum main_arg)
 		TimestampTzPlusMilliseconds(GetCurrentTimestamp(), 60 * 1000);
 	bool retryStatsCollection USED_WITH_LIBCURL_ONLY = false;
 	ErrorContextCallback errorCallback;
+	TimestampTz lastRecoveryTime = 0;
 
 	/*
 	 * Look up this worker's configuration.
@@ -361,9 +368,54 @@ CitusMaintenanceDaemonMain(Datum main_arg)
 		}
 #endif
 
+		/*
+		 * If enabled, run 2PC recovery on primary nodes (where !RecoveryInProgress()),
+		 * since we'll write to the pg_dist_transaction log.
+		 */
+		if (Recover2PCInterval > 0 && !RecoveryInProgress() &&
+			TimestampDifferenceExceeds(lastRecoveryTime, GetCurrentTimestamp(),
+									   Recover2PCInterval))
+		{
+			int recoveredTransactionCount = 0;
+
+			InvalidateMetadataSystemCache();
+			StartTransactionCommand();
+
+			if (!LockCitusExtension())
+			{
+				ereport(DEBUG1, (errmsg("could not lock the citus extension, "
+										"skipping 2PC recovery")));
+			}
+			else if (CheckCitusVersion(DEBUG1) && CitusHasBeenLoaded())
+			{
+				/*
+				 * Record last recovery time at start to ensure we run once per
+				 * Recover2PCInterval even if RecoverTwoPhaseCommits takes some time.
+				 */
+				lastRecoveryTime = GetCurrentTimestamp();
+
+				recoveredTransactionCount = RecoverTwoPhaseCommits();
+			}
+
+			CommitTransactionCommand();
+
+			if (recoveredTransactionCount > 0)
+			{
+				ereport(LOG, (errmsg("maintenance daemon recovered %d distributed "
+									 "transactions",
+									 recoveredTransactionCount)));
+			}
+
+			/* make sure we don't wait too long */
+			timeout = Min(timeout, Recover2PCInterval);
+		}
+
 		/* the config value -1 disables the distributed deadlock detection  */
 		if (DistributedDeadlockDetectionTimeoutFactor != -1.0)
 		{
+			double deadlockTimeout =
+				DistributedDeadlockDetectionTimeoutFactor * (double) DeadlockTimeout;
+
 			InvalidateMetadataSystemCache();
 			StartTransactionCommand();
 
@@ -397,13 +449,13 @@ CitusMaintenanceDaemonMain(Datum main_arg)
 			 * citus.distributed_deadlock_detection_factor 2), we'd be able to cancel
 			 * ~10 distributed deadlocks per second.
 			 */
-			timeout =
-				DistributedDeadlockDetectionTimeoutFactor * (double) DeadlockTimeout;
-
 			if (foundDeadlock)
 			{
-				timeout = timeout / 20.0;
+				deadlockTimeout = deadlockTimeout / 20.0;
 			}
+
+			/* make sure we don't wait too long */
+			timeout = Min(timeout, deadlockTimeout);
 		}
 
 		/*

src/include/distributed/transaction_recovery.h

@@ -12,8 +12,13 @@
 #define TRANSACTION_RECOVERY_H
 
 
+/* GUC to configure interval for 2PC auto-recovery */
+extern int Recover2PCInterval;
+
+
 /* Functions declarations for worker transactions */
 extern void LogTransactionRecord(int groupId, char *transactionName);
+extern int RecoverTwoPhaseCommits(void);
 
 
 #endif /* TRANSACTION_RECOVERY_H */

src/test/regress/expected/multi_alter_table_add_constraints.out

@@ -24,8 +24,6 @@ ALTER TABLE products ADD CONSTRAINT p_key PRIMARY KEY(name);
 ERROR:  cannot create constraint on "products"
 DETAIL:  Distributed relations cannot have UNIQUE, EXCLUDE, or PRIMARY KEY constraints that do not include the partition column (with an equality operator if EXCLUDE).
 ALTER TABLE products ADD CONSTRAINT p_key PRIMARY KEY(product_no);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 INSERT INTO products VALUES(1, 'product_1', 1);
 -- Should error out, since we are trying to add a new row having a value on p_key column
 -- conflicting with the existing row.
@@ -469,8 +467,6 @@ SELECT "Constraint", "Definition" FROM table_checks WHERE relid='public.products
 BEGIN;
 -- Add constraints (which will be rollbacked)
 ALTER TABLE products ADD CONSTRAINT unn_pno UNIQUE(product_no);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ALTER TABLE products ADD CONSTRAINT check_price CHECK(price > discounted_price);
 ALTER TABLE products ADD CONSTRAINT p_key_product PRIMARY KEY(product_no);
 ROLLBACK;
@@ -533,8 +529,6 @@ SELECT create_distributed_table('alter_add_prim_key', 'x');
 
 CREATE UNIQUE INDEX CONCURRENTLY alter_pk_idx ON alter_add_prim_key(x);
 ALTER TABLE alter_add_prim_key ADD CONSTRAINT alter_pk_idx PRIMARY KEY USING INDEX alter_pk_idx;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 SELECT (run_command_on_workers($$
     SELECT  
         kc.constraint_name 

src/test/regress/expected/multi_create_table.out

@@ -32,8 +32,6 @@ HINT:  Consider using hash partitioning.
 (1 row)
 
 CREATE INDEX lineitem_time_index ON lineitem (l_shipdate);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 CREATE TABLE orders (
 	o_orderkey bigint not null,
 	o_custkey integer not null,
@@ -498,8 +496,6 @@ NOTICE:  Copying data from local table...
 (1 row)
 
 CREATE INDEX data_load_test_idx ON data_load_test (col2);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 DROP TABLE data_load_test;
 END;
 -- popping in and out of existence in the same transaction works
@@ -651,8 +647,6 @@ SELECT create_distributed_table('rollback_table','id');
 
 \copy rollback_table from stdin delimiter ','
 CREATE INDEX rollback_index ON rollback_table(id);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 COMMIT;
 -- Check the table is created 
 SELECT count(*) FROM rollback_table;

src/test/regress/expected/multi_deparse_shard_query.out

@@ -410,8 +410,6 @@ INFO:  query: INSERT INTO public.raw_events_1 (tenant_id, value_4, value_6, valu
 
 -- test dropped table as well
 ALTER TABLE raw_events_1 DROP COLUMN value_5;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 SELECT deparse_shard_query_test('
 INSERT INTO raw_events_1(tenant_id, value_7, value_4)
 SELECT

src/test/regress/expected/multi_dropped_column_aliases.out

@@ -15,8 +15,6 @@ SELECT * FROM customer LIMIT 2;
 (2 rows)
 
 ALTER TABLE customer ADD COLUMN new_column1 INTEGER;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ALTER TABLE customer ADD COLUMN new_column2 INTEGER;
 SELECT count(*) FROM customer;
  count 

src/test/regress/expected/multi_explain.out

@@ -1057,8 +1057,6 @@ CREATE TABLE explain_table(id int);
 SELECT create_distributed_table('explain_table', 'id');
 
 ALTER TABLE explain_table ADD COLUMN value int;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ROLLBACK;
 -- test explain with local INSERT ... SELECT
 EXPLAIN (COSTS OFF)

src/test/regress/expected/multi_explain_0.out

@@ -1057,8 +1057,6 @@ CREATE TABLE explain_table(id int);
 SELECT create_distributed_table('explain_table', 'id');
 
 ALTER TABLE explain_table ADD COLUMN value int;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ROLLBACK;
 -- test explain with local INSERT ... SELECT
 EXPLAIN (COSTS OFF)

src/test/regress/expected/multi_foreign_key.out

@@ -162,8 +162,6 @@ INSERT INTO referenced_table VALUES(3, 3);
 INSERT INTO referencing_table VALUES(3, 3);
 BEGIN;
 ALTER TABLE referencing_table ADD COLUMN x int DEFAULT 0;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 SELECT master_modify_multiple_shards('DELETE FROM referenced_table');
  master_modify_multiple_shards 
 -------------------------------

src/test/regress/expected/multi_function_evaluation.out

@@ -38,8 +38,6 @@ SELECT * FROM example;
 
 -- non-immutable functions inside CASE/COALESCE aren't allowed
 ALTER TABLE example DROP value;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ALTER TABLE example ADD value timestamp;
 -- this is allowed because there are no mutable funcs in the CASE
 UPDATE example SET value = (CASE WHEN value > timestamp '12-12-1991' THEN timestamp '12-12-1991' ELSE value + interval '1 hour' END) WHERE key = 1;

src/test/regress/expected/multi_index_statements.out

@@ -64,8 +64,6 @@ SELECT master_create_empty_shard('index_test_append');
 --
 -- Verify that we can create different types of indexes
 CREATE INDEX lineitem_orderkey_index ON lineitem (l_orderkey);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 CREATE INDEX lineitem_partkey_desc_index ON lineitem (l_partkey DESC);
 CREATE INDEX lineitem_partial_index ON lineitem (l_shipdate)
 	WHERE l_shipdate < '1995-01-01';
@@ -210,8 +208,6 @@ ERROR:  cannot drop multiple distributed objects in a single command
 HINT:  Try dropping each object in a separate DROP command.
 -- Verify that we can succesfully drop indexes
 DROP INDEX lineitem_orderkey_index;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 DROP INDEX lineitem_orderkey_index_new;
 DROP INDEX lineitem_partkey_desc_index;
 DROP INDEX lineitem_partial_index;

src/test/regress/expected/multi_insert_select.out

@@ -1653,8 +1653,6 @@ DETAIL:  Limit in subquery is currently unsupported
 -- connections for all co-located placements.
 BEGIN;
 ALTER TABLE raw_events_second DROP COLUMN value_4;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 INSERT INTO raw_events_first SELECT * FROM raw_events_second; 
 ROLLBACK;
 -- Alterating a table and selecting from it using a single-shard statement

src/test/regress/expected/multi_join_order_additional.out

@@ -43,8 +43,6 @@ SELECT master_create_worker_shards('lineitem_hash', 2, 1);
 
 CREATE INDEX lineitem_hash_time_index ON lineitem_hash (l_shipdate);
 DEBUG:  building index "lineitem_hash_time_index" on table "lineitem_hash"
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 CREATE TABLE orders_hash (
 	o_orderkey bigint not null,
 	o_custkey integer not null,

src/test/regress/expected/multi_metadata_sync.out

@@ -71,8 +71,6 @@ SELECT unnest(master_metadata_snapshot());
 
 -- Show that CREATE INDEX commands are included in the metadata snapshot
 CREATE INDEX mx_index ON mx_test_table(col_2);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 SELECT unnest(master_metadata_snapshot());
                                                                                                                                                                                                                                                                                                                                                              unnest                                                                                                                                                                                                                                                                                                                                                             
 --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
@@ -1286,8 +1284,6 @@ SELECT shardid AS ref_table_shardid FROM pg_dist_shard WHERE logicalrelid='mx_re
 -- Check that DDL commands are propagated to reference tables on workers
 \c - - - :master_port
 ALTER TABLE mx_ref ADD COLUMN col_3 NUMERIC DEFAULT 0;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 CREATE INDEX mx_ref_index ON mx_ref(col_1);
 SELECT "Column", "Type", "Modifiers" FROM table_desc WHERE relid='mx_ref'::regclass;
  Column |  Type   | Modifiers 

src/test/regress/expected/multi_modifications.out

@@ -460,8 +460,6 @@ UPDATE limit_orders SET symbol = UPPER(symbol) WHERE id = 246 RETURNING id, LOWE
 (1 row)
 
 ALTER TABLE limit_orders ADD COLUMN array_of_values integer[];
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 -- updates referencing STABLE functions are allowed
 UPDATE limit_orders SET placed_at = LEAST(placed_at, now()::timestamp) WHERE id = 246;
 -- so are binary operators
@@ -505,8 +503,6 @@ SELECT array_of_values FROM limit_orders WHERE id = 246;
 (1 row)
 
 ALTER TABLE limit_orders DROP array_of_values;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 -- even in RETURNING
 UPDATE limit_orders SET placed_at = placed_at WHERE id = 246 RETURNING NOW();
 ERROR:  non-IMMUTABLE functions are not allowed in the RETURNING clause

src/test/regress/expected/multi_modifying_xacts.out

@@ -38,8 +38,6 @@ SELECT master_create_worker_shards('labs', 1, 1);
 
 -- might be confusing to have two people in the same lab with the same name
 CREATE UNIQUE INDEX avoid_name_confusion_idx ON researchers (lab_id, name);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 -- add some data
 INSERT INTO researchers VALUES (1, 1, 'Donald Knuth');
 INSERT INTO researchers VALUES (2, 1, 'Niklaus Wirth');
@@ -371,6 +369,7 @@ ORDER BY nodeport, shardid;
 \set VERBOSITY terse
 -- deferred check should abort the transaction
 BEGIN;
+SET LOCAL citus.multi_shard_commit_protocol TO '1pc';
 DELETE FROM researchers WHERE lab_id = 6;
 \copy researchers FROM STDIN delimiter ','
 \copy researchers FROM STDIN delimiter ','
@@ -1436,7 +1435,6 @@ SELECT id FROM users WHERE id = 6;
 (1 row)
 
 ALTER TABLE items ADD COLUMN last_update timestamptz;
-NOTICE:  using one-phase commit for distributed DDL commands
 ERROR:  cannot perform a parallel DDL command because multiple placements have been accessed over the same connection
 END;
 -- but the other way around is fine

src/test/regress/expected/multi_mx_create_table.out

@@ -269,8 +269,6 @@ SELECT create_distributed_table('lineitem_mx', 'l_orderkey');
 (1 row)
 
 CREATE INDEX lineitem_mx_time_index ON lineitem_mx (l_shipdate);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 CREATE TABLE orders_mx (
     o_orderkey bigint not null,
     o_custkey integer not null,

src/test/regress/expected/multi_mx_ddl.out

@@ -14,8 +14,6 @@ SELECT * FROM mx_ddl_table ORDER BY key;
 
 -- CREATE INDEX
 CREATE INDEX ddl_test_index ON mx_ddl_table(value);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 CREATE INDEX CONCURRENTLY ddl_test_concurrent_index ON mx_ddl_table(value);
 -- ADD COLUMN
 ALTER TABLE mx_ddl_table ADD COLUMN version INTEGER;
@@ -143,8 +141,6 @@ INSERT INTO mx_ddl_table VALUES (38, 78);
 \c - - - :master_port
 -- SET DATA TYPE
 ALTER TABLE mx_ddl_table ALTER COLUMN version SET DATA TYPE double precision;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 INSERT INTO mx_ddl_table VALUES (78, 83, 2.1);
 \c - - - :worker_1_port
 SELECT * FROM mx_ddl_table ORDER BY key;
@@ -167,8 +163,6 @@ SELECT * FROM mx_ddl_table ORDER BY key;
 \c - - - :master_port
 -- DROP INDEX
 DROP INDEX ddl_test_index;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 DROP INDEX CONCURRENTLY ddl_test_concurrent_index;
 -- DROP DEFAULT
 ALTER TABLE mx_ddl_table ALTER COLUMN version DROP DEFAULT;
@@ -249,8 +243,6 @@ SELECT "Column", "Type", "Modifiers" FROM table_desc WHERE relid='mx_ddl_table_1
 SET client_min_messages TO debug2;
 CREATE INDEX ddl_test_index ON mx_ddl_table(value);
 DEBUG:  building index "ddl_test_index" on table "mx_ddl_table"
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 RESET client_min_messages;
 DROP INDEX ddl_test_index;
 -- show that sequences owned by mx tables result in unique values
@@ -279,6 +271,4 @@ SELECT :worker_1_lastval = :worker_2_lastval;
 
 -- the type of sequences can't be changed
 ALTER TABLE mx_sequence ALTER value TYPE BIGINT;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ALTER TABLE mx_sequence ALTER value TYPE INT;

src/test/regress/expected/multi_mx_metadata.out

@@ -1,12 +1,15 @@
 -- Test creation of mx tables and metadata syncing
--- get rid of the previously created entries in pg_dist_transaction
+-- Temporarily disable automatic 2PC recovery
--- for the sake of getting consistent results in this test file
+ALTER SYSTEM SET citus.recover_2pc_interval TO -1;
-SELECT recover_prepared_transactions();
+SELECT pg_reload_conf();
- recover_prepared_transactions 
+ pg_reload_conf 
--------------------------------
+----------------
-                             0
+ t
 (1 row)
 
+-- get rid of the previously created entries in pg_dist_transaction
+-- for the sake of getting consistent results in this test file
+TRUNCATE pg_dist_transaction;
 CREATE TABLE distributed_mx_table (
     key text primary key,
     value jsonb
@@ -25,7 +28,7 @@ SELECT create_distributed_table('distributed_mx_table', 'key');
 SELECT count(*) FROM pg_dist_transaction;
  count 
 -------
-     5
+     2
 (1 row)
 
 -- Confirm that the metadata transactions have been committed
@@ -39,7 +42,7 @@ SELECT recover_prepared_transactions();
 SELECT count(*) FROM pg_dist_transaction;
  count 
 -------
-     3
+     0
 (1 row)
 
 \c - - - :worker_1_port
@@ -267,7 +270,7 @@ SELECT recover_prepared_transactions();
 SELECT count(*) FROM pg_dist_transaction;
  count 
 -------
-     3
+     0
 (1 row)
 
 -- Confirm that transactions were correctly rolled forward
@@ -284,3 +287,12 @@ SELECT count(*) FROM pg_tables WHERE tablename = 'should_commit';
      1
 (1 row)
 
+-- Resume ordinary recovery
+\c - - - :master_port
+ALTER SYSTEM RESET citus.recover_2pc_interval;
+SELECT pg_reload_conf();
+ pg_reload_conf 
+----------------
+ t
+(1 row)
+

src/test/regress/expected/multi_mx_modifications.out

@@ -268,8 +268,6 @@ UPDATE limit_orders_mx SET symbol = UPPER(symbol) WHERE id = 246 RETURNING id, L
 -- connect coordinator to run the DDL
 \c - - - :master_port
 ALTER TABLE limit_orders_mx ADD COLUMN array_of_values integer[];
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 -- connect back to the other node
 \c - - - :worker_2_port
 -- updates referencing STABLE functions are allowed
@@ -310,8 +308,6 @@ SELECT array_of_values FROM limit_orders_mx WHERE id = 246;
 -- connect coordinator to run the DDL
 \c - - - :master_port
 ALTER TABLE limit_orders_mx DROP array_of_values;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 -- connect back to the other node
 \c - - - :worker_2_port
 -- even in RETURNING

src/test/regress/expected/multi_mx_transaction_recovery.out

@@ -10,6 +10,14 @@ SELECT create_distributed_table('test_recovery', 'x');
 (1 row)
 
 \c - - - :worker_1_port
+-- Disable auto-recovery for the initial tests
+ALTER SYSTEM SET citus.recover_2pc_interval TO -1;
+SELECT pg_reload_conf();
+ pg_reload_conf 
+----------------
+ t
+(1 row)
+
 SET citus.multi_shard_commit_protocol TO '2pc';
 -- Ensure pg_dist_transaction is empty for test
 SELECT recover_prepared_transactions();
@@ -116,10 +124,31 @@ SELECT count(*) FROM pg_dist_transaction;
      3
 (1 row)
 
-SELECT recover_prepared_transactions();
+-- Test whether auto-recovery runs
- recover_prepared_transactions 
+ALTER SYSTEM SET citus.recover_2pc_interval TO 10;
--------------------------------
+SELECT pg_reload_conf();
-                             0
+ pg_reload_conf 
+----------------
+ t
+(1 row)
+
+SELECT pg_sleep(0.2);
+ pg_sleep 
+----------
+ 
+(1 row)
+
+SELECT count(*) FROM pg_dist_transaction;
+ count 
+-------
+     0
+(1 row)
+
+ALTER SYSTEM RESET citus.recover_2pc_interval;
+SELECT pg_reload_conf();
+ pg_reload_conf 
+----------------
+ t
 (1 row)
 
 DROP TABLE table_should_commit;

src/test/regress/expected/multi_name_lengths.out

@@ -74,8 +74,6 @@ SELECT master_create_worker_shards('name_lengths', '2', '2');
 -- Verify that we CAN add columns with "too-long names", because
 -- the columns' names are not extended in the corresponding shard tables.
 ALTER TABLE name_lengths ADD COLUMN float_col_12345678901234567890123456789012345678901234567890 FLOAT;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ALTER TABLE name_lengths ADD COLUMN date_col_12345678901234567890123456789012345678901234567890 DATE;
 ALTER TABLE name_lengths ADD COLUMN int_col_12345678901234567890123456789012345678901234567890 INTEGER DEFAULT 1;
 -- Placeholders for unsupported ALTER TABLE to add constraints with implicit names that are likely too long
@@ -105,8 +103,6 @@ ALTER TABLE name_lengths ADD CONSTRAINT nl_exclude_12345678901234567890123456789
 ERROR:  cannot create constraint on "name_lengths"
 DETAIL:  Distributed relations cannot have UNIQUE, EXCLUDE, or PRIMARY KEY constraints that do not include the partition column (with an equality operator if EXCLUDE).
 ALTER TABLE name_lengths ADD CONSTRAINT nl_checky_12345678901234567890123456789012345678901234567890 CHECK (date_col_12345678901234567890123456789012345678901234567890 >= '2014-01-01'::date);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 \c - - - :worker_1_port
 SELECT "Constraint", "Definition" FROM table_checks WHERE relid='public.name_lengths_225002'::regclass;
                            Constraint                            |                                        Definition                                         
@@ -122,8 +118,6 @@ ALTER TABLE name_lengths RENAME CONSTRAINT unique_123456789012345678901234567890
 ERROR:  renaming constraints belonging to distributed tables is currently unsupported
 -- Verify that CREATE INDEX on already distributed table has proper shard names.
 CREATE INDEX tmp_idx_12345678901234567890123456789012345678901234567890 ON name_lengths(col2);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 \c - - - :worker_1_port
 \d tmp_idx_*
 Index "public.tmp_idx_123456789012345678901234567890123456789_5e470afa_225002"
@@ -143,8 +137,6 @@ btree, for table "public.name_lengths_225003"
 -- by the parser/rewriter before further processing, just as in Postgres.
 CREATE INDEX tmp_idx_123456789012345678901234567890123456789012345678901234567890 ON name_lengths(col2);
 NOTICE:  identifier "tmp_idx_123456789012345678901234567890123456789012345678901234567890" will be truncated to "tmp_idx_1234567890123456789012345678901234567890123456789012345"
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 \c - - - :worker_1_port
 \d tmp_idx_*
 Index "public.tmp_idx_123456789012345678901234567890123456789_599636aa_225002"

src/test/regress/expected/multi_partitioning.out

@@ -157,8 +157,6 @@ SELECT create_distributed_table('partitioning_test_2013', 'id');
 INSERT INTO partitioning_test_2013 VALUES (7, '2013-06-06');
 INSERT INTO partitioning_test_2013 VALUES (8, '2013-07-07');
 ALTER TABLE partitioning_test ATTACH PARTITION partitioning_test_2013 FOR VALUES FROM ('2013-01-01') TO ('2014-01-01');
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 -- see the data is loaded to shards
 SELECT * FROM partitioning_test ORDER BY 1;
  id |    time    

src/test/regress/expected/multi_reference_table.out

@@ -1317,8 +1317,6 @@ SELECT create_reference_table('reference_table_ddl');
 
 -- CREATE & DROP index and check the workers
 CREATE INDEX reference_index_1 ON reference_table_ddl(value_1);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 CREATE INDEX reference_index_2 ON reference_table_ddl(value_2, value_3);
 -- should be able to create/drop UNIQUE index on a reference table
 CREATE UNIQUE INDEX reference_index_3 ON reference_table_ddl(value_1);
@@ -1367,8 +1365,6 @@ btree, for table "public.reference_table_ddl_1250019"
 
 \c - - - :master_port
 DROP INDEX reference_index_2;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 \c - - - :worker_1_port
 SELECT "Column", "Type", "Modifiers" FROM table_desc WHERE relid='public.reference_table_ddl_1250019'::regclass;
  Column  |            Type             |  Modifiers   
@@ -1620,8 +1616,6 @@ ROLLBACK;
 -- DDL+DML is allowed
 BEGIN;
 ALTER TABLE reference_table_test ADD COLUMN value_dummy INT;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 INSERT INTO reference_table_test VALUES (2, 2.0, '2', '2016-12-02');
 ROLLBACK;
 -- clean up tables

src/test/regress/expected/multi_remove_node_reference_table.out

@@ -654,8 +654,6 @@ WHERE
 \c - - - :master_port
 BEGIN;
 ALTER TABLE remove_node_reference_table ADD column2 int;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 SELECT master_remove_node('localhost', :worker_2_port);
  master_remove_node 
 --------------------

src/test/regress/expected/multi_repair_shards.out

@@ -43,8 +43,6 @@ UPDATE pg_dist_placement SET shardstate = 3 WHERE shardid = :newshardid
 -- cannot repair a shard after a modification (transaction still open during repair)
 BEGIN;
 ALTER TABLE customer_engagements ADD COLUMN value float;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 SELECT master_copy_shard_placement(:newshardid, 'localhost', :worker_1_port, 'localhost', :worker_2_port);
 ERROR:  cannot open new connections after the first modification command within a transaction
 ROLLBACK;

src/test/regress/expected/multi_replicate_reference_table.out

@@ -510,8 +510,6 @@ SELECT create_reference_table('replicate_reference_table_ddl');
 
 BEGIN;
 ALTER TABLE replicate_reference_table_ddl ADD column2 int;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 SELECT 1 FROM master_add_node('localhost', :worker_2_port);
 NOTICE:  Replicating reference table "replicate_reference_table_ddl" to the node localhost:57638
 ERROR:  cannot open new connections after the first modification command within a transaction

src/test/regress/expected/multi_schema_support.out

@@ -583,8 +583,6 @@ SELECT * FROM nation_hash_composite_types WHERE test_col = '(a,a)'::new_composit
 -- test ALTER TABLE ADD/DROP queries with schemas
 SET search_path TO public;
 ALTER TABLE test_schema_support.nation_hash ADD COLUMN new_col INT;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 -- verify column is added
 SELECT "Column", "Type", "Modifiers" FROM table_desc WHERE relid='test_schema_support.nation_hash'::regclass;
    Column    |          Type          | Modifiers 
@@ -610,8 +608,6 @@ SELECT "Column", "Type", "Modifiers" FROM table_desc WHERE relid='test_schema_su
 \c - - - :master_port
 ALTER TABLE test_schema_support.nation_hash DROP COLUMN IF EXISTS non_existent_column;
 NOTICE:  column "non_existent_column" of relation "nation_hash" does not exist, skipping
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ALTER TABLE test_schema_support.nation_hash DROP COLUMN IF EXISTS new_col;
 -- verify column is dropped
 SELECT "Column", "Type", "Modifiers" FROM table_desc WHERE relid='test_schema_support.nation_hash'::regclass;
@@ -637,8 +633,6 @@ SELECT "Column", "Type", "Modifiers" FROM table_desc WHERE relid='test_schema_su
 --test with search_path is set
 SET search_path TO test_schema_support;
 ALTER TABLE nation_hash ADD COLUMN new_col INT;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 -- verify column is added
 SELECT "Column", "Type", "Modifiers" FROM public.table_desc WHERE relid='test_schema_support.nation_hash'::regclass;
    Column    |          Type          | Modifiers 
@@ -665,8 +659,6 @@ SELECT "Column", "Type", "Modifiers" FROM public.table_desc WHERE relid='test_sc
 SET search_path TO test_schema_support;
 ALTER TABLE nation_hash DROP COLUMN IF EXISTS non_existent_column;
 NOTICE:  column "non_existent_column" of relation "nation_hash" does not exist, skipping
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ALTER TABLE nation_hash DROP COLUMN IF EXISTS new_col;
 -- verify column is dropped
 SELECT "Column", "Type", "Modifiers" FROM public.table_desc WHERE relid='test_schema_support.nation_hash'::regclass;
@@ -693,8 +685,6 @@ SELECT "Column", "Type", "Modifiers" FROM public.table_desc WHERE relid='test_sc
 SET search_path TO public;
 -- CREATE index
 CREATE INDEX index1 ON test_schema_support.nation_hash(n_name);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 --verify INDEX is created
 \d test_schema_support.index1
  Index "test_schema_support.index1"
@@ -714,8 +704,6 @@ btree, for table "test_schema_support.nation_hash_1190003"
 \c - - - :master_port
 -- DROP index
 DROP INDEX test_schema_support.index1;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 --verify INDEX is dropped
 \d test_schema_support.index1
 \c - - - :worker_1_port
@@ -725,8 +713,6 @@ HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_sh
 SET search_path TO test_schema_support;
 -- CREATE index
 CREATE INDEX index1 ON nation_hash(n_name);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 --verify INDEX is created
 \d test_schema_support.index1
  Index "test_schema_support.index1"
@@ -747,8 +733,6 @@ btree, for table "test_schema_support.nation_hash_1190003"
 -- DROP index
 SET search_path TO test_schema_support;
 DROP INDEX index1;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 --verify INDEX is dropped
 \d test_schema_support.index1
 \c - - - :worker_1_port

src/test/regress/expected/multi_size_queries.out

@@ -50,8 +50,6 @@ SELECT citus_total_relation_size('customer_copy_hash');
 (1 row)
 
 CREATE INDEX index_1 on customer_copy_hash(c_custkey);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 VACUUM (FULL) customer_copy_hash;
 -- Tests on distributed table with index.
 SELECT citus_table_size('customer_copy_hash');

src/test/regress/expected/multi_transaction_recovery.out

@@ -1,17 +1,20 @@
-SET citus.next_shard_id TO 1220000;
 -- Tests for prepared transaction recovery
--- Ensure pg_dist_transaction is empty for test
+SET citus.next_shard_id TO 1220000;
+-- Disable auto-recovery for the initial tests
+ALTER SYSTEM SET citus.recover_2pc_interval TO -1;
+SELECT pg_reload_conf();
+ pg_reload_conf 
+----------------
+ t
+(1 row)
+
+-- Ensure pg_dist_transaction is empty
 SELECT recover_prepared_transactions();
  recover_prepared_transactions 
 -------------------------------
                              0
 (1 row)
 
-SELECT * FROM pg_dist_transaction;
- groupid | gid 
----------+-----
-(0 rows)
-
 -- Create some "fake" prepared transactions to recover
 \c - - - :worker_1_port
 BEGIN;
@@ -99,7 +102,7 @@ SELECT count(*) FROM pg_dist_transaction;
      0
 (1 row)
 
--- Committed DDL commands should write 4 transaction recovery records
+-- Aborted DDL commands should not write transaction recovery records
 BEGIN;
 ALTER TABLE test_recovery ADD COLUMN y text;
 ROLLBACK;
@@ -109,6 +112,7 @@ SELECT count(*) FROM pg_dist_transaction;
      0
 (1 row)
 
+-- Committed DDL commands should write 4 transaction recovery records
 ALTER TABLE test_recovery ADD COLUMN y text;
 SELECT count(*) FROM pg_dist_transaction;
  count 
@@ -167,7 +171,7 @@ SELECT count(*) FROM pg_dist_transaction;
      0
 (1 row)
 
--- Committed INSERT..SELECT should write 4 transaction recovery records
+-- Aborted INSERT..SELECT should not write transaction recovery records
 BEGIN;
 INSERT INTO test_recovery SELECT x, 'earth' FROM test_recovery;
 ROLLBACK;
@@ -177,6 +181,7 @@ SELECT count(*) FROM pg_dist_transaction;
      0
 (1 row)
 
+-- Committed INSERT..SELECT should write 4 transaction recovery records
 INSERT INTO test_recovery SELECT x, 'earth' FROM test_recovery;
 SELECT count(*) FROM pg_dist_transaction;
  count 
@@ -218,5 +223,52 @@ SELECT recover_prepared_transactions();
                              0
 (1 row)
 
+-- Create a single-replica table to enable 2PC in multi-statement transactions
+CREATE TABLE test_recovery_single (LIKE test_recovery);
+SELECT create_distributed_table('test_recovery_single', 'x');
+ create_distributed_table 
+--------------------------
+ 
+(1 row)
+
+-- Multi-statement transactions should write 2 transaction recovery records
+BEGIN;
+INSERT INTO test_recovery_single VALUES ('hello-0');
+INSERT INTO test_recovery_single VALUES ('hello-2');
+COMMIT;
+SELECT count(*) FROM pg_dist_transaction;
+ count 
+-------
+     2
+(1 row)
+
+-- Test whether auto-recovery runs
+ALTER SYSTEM SET citus.recover_2pc_interval TO 10;
+SELECT pg_reload_conf();
+ pg_reload_conf 
+----------------
+ t
+(1 row)
+
+SELECT pg_sleep(0.1);
+ pg_sleep 
+----------
+ 
+(1 row)
+
+SELECT count(*) FROM pg_dist_transaction;
+ count 
+-------
+     0
+(1 row)
+
+ALTER SYSTEM RESET citus.recover_2pc_interval;
+SELECT pg_reload_conf();
+ pg_reload_conf 
+----------------
+ t
+(1 row)
+
 DROP TABLE test_recovery_ref;
 DROP TABLE test_recovery;
+DROP TABLE test_recovery_single;

src/test/regress/expected/multi_unsupported_worker_operations.out

@@ -140,8 +140,6 @@ SELECT * FROM mx_ref_table ORDER BY col_1;
 \c - - - :master_port
 DROP TABLE mx_ref_table;
 CREATE UNIQUE INDEX mx_test_uniq_index ON mx_table(col_1);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 \c - - - :worker_1_port
 -- DDL commands
 SELECT "Column", "Type", "Modifiers" FROM table_desc WHERE relid='public.mx_table'::regclass;
@@ -228,8 +226,6 @@ SELECT count(1) FROM pg_dist_node WHERE nodename='localhost' AND nodeport=5432;
 -- master_remove_node
 \c - - - :master_port
 DROP INDEX mx_test_uniq_index;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 SELECT 1 FROM master_add_node('localhost', 5432);
  ?column? 
 ----------

src/test/regress/expected/multi_upsert.out

@@ -241,8 +241,6 @@ SELECT master_create_worker_shards('dropcol_distributed', 4, 1);
 INSERT INTO dropcol_distributed AS dropcol (key, keep1, keep2) VALUES (1, '5', 5) ON CONFLICT(key)
 	DO UPDATE SET keep1 = dropcol.keep1;
 ALTER TABLE dropcol_distributed DROP COLUMN drop2;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 INSERT INTO dropcol_distributed (key, keep1, keep2) VALUES (1, '5', 5) ON CONFLICT(key)
 	DO UPDATE SET keep1 = dropcol_distributed.keep1;
 ALTER TABLE dropcol_distributed DROP COLUMN keep2;

src/test/regress/input/multi_alter_table_statements.source

@@ -300,7 +300,7 @@ COMMIT;
 SELECT indexname, tablename FROM pg_indexes WHERE tablename = 'single_shard_items' ORDER BY 1;
 
 -- Now try with 2pc off
-RESET citus.multi_shard_commit_protocol;
+SET citus.multi_shard_commit_protocol TO '1pc';
 BEGIN;
 CREATE INDEX single_index_2 ON single_shard_items(id);
 CREATE INDEX single_index_3 ON single_shard_items(name);

src/test/regress/output/multi_alter_table_statements.source

@@ -32,8 +32,6 @@ SELECT master_create_distributed_table('lineitem_alter', 'l_orderkey', 'append')
 \copy lineitem_alter FROM '@abs_srcdir@/data/lineitem.1.data' with delimiter '|'
 -- Verify that we can add columns
 ALTER TABLE lineitem_alter ADD COLUMN float_column FLOAT;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ALTER TABLE lineitem_alter ADD COLUMN date_column DATE;
 ALTER TABLE lineitem_alter ADD COLUMN int_column1 INTEGER DEFAULT 1;
 ALTER TABLE lineitem_alter ADD COLUMN int_column2 INTEGER DEFAULT 2;
@@ -117,8 +115,6 @@ SELECT int_column1, count(*) FROM lineitem_alter GROUP BY int_column1;
 
 -- Verify that SET|DROP DEFAULT works
 ALTER TABLE lineitem_alter ALTER COLUMN float_column SET DEFAULT 1;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ALTER TABLE lineitem_alter ALTER COLUMN int_column1 DROP DEFAULT;
 -- \copy to verify that default values take effect
 \copy lineitem_alter (l_orderkey, l_partkey, l_suppkey, l_linenumber, l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate, l_commitdate, l_receiptdate, l_shipinstruct, l_shipmode, l_comment) FROM '@abs_srcdir@/data/lineitem.1.data' with delimiter '|'
@@ -526,8 +522,6 @@ ALTER TABLE lineitem_alter_220000 ADD COLUMN first integer;
 -- and try to add it in a multi-statement block, which fails
 BEGIN;
 CREATE INDEX temp_index_2 ON lineitem_alter(l_orderkey);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ALTER TABLE lineitem_alter ADD COLUMN first integer;
 ERROR:  column "first" of relation "lineitem_alter_220000" already exists
 CONTEXT:  while executing command on localhost:57638
@@ -657,11 +651,9 @@ SELECT indexname, tablename FROM pg_indexes WHERE tablename = 'single_shard_item
 (0 rows)
 
 -- Now try with 2pc off
-RESET citus.multi_shard_commit_protocol;
+SET citus.multi_shard_commit_protocol TO '1pc';
 BEGIN;
 CREATE INDEX single_index_2 ON single_shard_items(id);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 CREATE INDEX single_index_3 ON single_shard_items(name);
 COMMIT;
 WARNING:  duplicate key value violates unique constraint "ddl_commands_command_key"
@@ -684,8 +676,6 @@ DROP TABLE ddl_commands;
 -- Distributed SELECTs cannot appear after ALTER
 BEGIN;
 CREATE INDEX temp_index_2 ON lineitem_alter(l_orderkey);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 SELECT count(*) FROM lineitem_alter;
 ERROR:  cannot open new connections after the first modification command within a transaction
 COMMIT;
@@ -863,8 +853,6 @@ SELECT create_distributed_table('sequence_deadlock_test', 'a');
 
 BEGIN;
 ALTER TABLE sequence_deadlock_test ADD COLUMN c int;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 DROP SEQUENCE sequence_deadlock_test_b_seq CASCADE;
 NOTICE:  drop cascades to default for table sequence_deadlock_test column b
 END;
@@ -902,8 +890,6 @@ SELECT value, count(*) FROM trigger_table GROUP BY value ORDER BY value;
 (1 row)
 
 ALTER TABLE trigger_table DISABLE TRIGGER ALL;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 INSERT INTO trigger_table VALUES (1, 'trigger disabled');
 SELECT value, count(*) FROM trigger_table GROUP BY value ORDER BY value;
       value       | count 
@@ -961,8 +947,6 @@ SELECT create_distributed_table('test_table_1','id');
 (1 row)
 
 ALTER TABLE test_table_1 ADD CONSTRAINT u_key UNIQUE(id);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 DROP TABLE test_table_1;
 END;
 -- There should be no test_table_1 shard on workers

src/test/regress/output/multi_behavioral_analytics_create_table.source

@@ -76,8 +76,6 @@ COPY users_table FROM '@abs_srcdir@/data/users_table.data' WITH CSV;
 COPY events_table FROM '@abs_srcdir@/data/events_table.data' WITH CSV;
 -- create indexes for
 CREATE INDEX is_index1 ON users_table(user_id);
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 CREATE INDEX is_index2 ON events_table(user_id);
 CREATE INDEX is_index3 ON users_table(value_1);
 CREATE INDEX is_index4 ON events_table(event_type);

src/test/regress/output/multi_copy.source

@@ -174,8 +174,6 @@ SELECT count(*) FROM customer_with_default where c_time IS NOT NULL;
 
 -- Add columns to the table and perform a COPY
 ALTER TABLE customer_copy_hash ADD COLUMN extra1 INT DEFAULT 0;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 ALTER TABLE customer_copy_hash ADD COLUMN extra2 INT DEFAULT 0;
 COPY customer_copy_hash (c_custkey, c_name, extra1, extra2) FROM STDIN CSV;
 SELECT * FROM customer_copy_hash WHERE extra1 = 1;
@@ -1052,8 +1050,6 @@ SELECT create_distributed_table('drop_copy_test_table','col3');
 (1 row)
 
 ALTER TABLE drop_copy_test_table drop column col1;
-NOTICE:  using one-phase commit for distributed DDL commands
-HINT:  You can enable two-phase commit for extra safety with: SET citus.multi_shard_commit_protocol TO '2pc'
 COPY drop_copy_test_table (col2,col3,col4) from STDIN with CSV;
 SELECT * FROM drop_copy_test_table WHERE col3 = 1;
  col2 | col3 | col4 

src/test/regress/sql/multi_modifying_xacts.sql

@@ -295,6 +295,7 @@ ORDER BY nodeport, shardid;
 \set VERBOSITY terse
 -- deferred check should abort the transaction
 BEGIN;
+SET LOCAL citus.multi_shard_commit_protocol TO '1pc';
 DELETE FROM researchers WHERE lab_id = 6;
 \copy researchers FROM STDIN delimiter ','
 31, 6, 'Bjarne Stroustrup'

src/test/regress/sql/multi_mx_metadata.sql

@@ -1,8 +1,12 @@
 -- Test creation of mx tables and metadata syncing
 
+-- Temporarily disable automatic 2PC recovery
+ALTER SYSTEM SET citus.recover_2pc_interval TO -1;
+SELECT pg_reload_conf();
+
 -- get rid of the previously created entries in pg_dist_transaction
 -- for the sake of getting consistent results in this test file
-SELECT recover_prepared_transactions();
+TRUNCATE pg_dist_transaction;
 
 CREATE TABLE distributed_mx_table (
     key text primary key,
@@ -14,6 +18,7 @@ SET citus.shard_replication_factor TO 1;
 SET citus.replication_model TO streaming;
 
 SET citus.shard_count TO 4;
+
 SELECT create_distributed_table('distributed_mx_table', 'key');
 
 -- Verify that we've logged commit records
@@ -165,3 +170,8 @@ SELECT count(*) FROM pg_dist_transaction;
 \c - - - :worker_1_port
 SELECT count(*) FROM pg_tables WHERE tablename = 'should_abort';
 SELECT count(*) FROM pg_tables WHERE tablename = 'should_commit';
+
+-- Resume ordinary recovery
+\c - - - :master_port
+ALTER SYSTEM RESET citus.recover_2pc_interval;
+SELECT pg_reload_conf();

src/test/regress/sql/multi_mx_transaction_recovery.sql

@@ -9,6 +9,10 @@ SELECT create_distributed_table('test_recovery', 'x');
 
 \c - - - :worker_1_port
 
+-- Disable auto-recovery for the initial tests
+ALTER SYSTEM SET citus.recover_2pc_interval TO -1;
+SELECT pg_reload_conf();
+
 SET citus.multi_shard_commit_protocol TO '2pc';
 
 -- Ensure pg_dist_transaction is empty for test
@@ -68,7 +72,14 @@ world-1
 \.
 
 SELECT count(*) FROM pg_dist_transaction;
-SELECT recover_prepared_transactions();
+
+-- Test whether auto-recovery runs
+ALTER SYSTEM SET citus.recover_2pc_interval TO 10;
+SELECT pg_reload_conf();
+SELECT pg_sleep(0.2);
+SELECT count(*) FROM pg_dist_transaction;
+ALTER SYSTEM RESET citus.recover_2pc_interval;
+SELECT pg_reload_conf();
 
 DROP TABLE table_should_commit;
 

src/test/regress/sql/multi_transaction_recovery.sql

@@ -1,12 +1,13 @@
+-- Tests for prepared transaction recovery
 SET citus.next_shard_id TO 1220000;
 
--- Tests for prepared transaction recovery
+-- Disable auto-recovery for the initial tests
+ALTER SYSTEM SET citus.recover_2pc_interval TO -1;
+SELECT pg_reload_conf();
 
--- Ensure pg_dist_transaction is empty for test
+-- Ensure pg_dist_transaction is empty
 SELECT recover_prepared_transactions();
 
-SELECT * FROM pg_dist_transaction;
-
 -- Create some "fake" prepared transactions to recover
 \c - - - :worker_1_port
 
@@ -56,12 +57,13 @@ SELECT recover_prepared_transactions();
 INSERT INTO test_recovery VALUES ('hello');
 SELECT count(*) FROM pg_dist_transaction;
 
--- Committed DDL commands should write 4 transaction recovery records
+-- Aborted DDL commands should not write transaction recovery records
 BEGIN;
 ALTER TABLE test_recovery ADD COLUMN y text;
 ROLLBACK;
 SELECT count(*) FROM pg_dist_transaction;
 
+-- Committed DDL commands should write 4 transaction recovery records
 ALTER TABLE test_recovery ADD COLUMN y text;
 
 SELECT count(*) FROM pg_dist_transaction;
@@ -80,12 +82,13 @@ SELECT count(*) FROM pg_dist_transaction;
 SELECT recover_prepared_transactions();
 SELECT count(*) FROM pg_dist_transaction;
 
--- Committed INSERT..SELECT should write 4 transaction recovery records
+-- Aborted INSERT..SELECT should not write transaction recovery records
 BEGIN;
 INSERT INTO test_recovery SELECT x, 'earth' FROM test_recovery;
 ROLLBACK;
 SELECT count(*) FROM pg_dist_transaction;
 
+-- Committed INSERT..SELECT should write 4 transaction recovery records
 INSERT INTO test_recovery SELECT x, 'earth' FROM test_recovery;
 
 SELECT count(*) FROM pg_dist_transaction;
@@ -106,5 +109,26 @@ hello-1
 SELECT count(*) FROM pg_dist_transaction;
 SELECT recover_prepared_transactions();
 
+-- Create a single-replica table to enable 2PC in multi-statement transactions
+CREATE TABLE test_recovery_single (LIKE test_recovery);
+SELECT create_distributed_table('test_recovery_single', 'x');
+
+-- Multi-statement transactions should write 2 transaction recovery records
+BEGIN;
+INSERT INTO test_recovery_single VALUES ('hello-0');
+INSERT INTO test_recovery_single VALUES ('hello-2');
+COMMIT;
+SELECT count(*) FROM pg_dist_transaction;
+
+-- Test whether auto-recovery runs
+ALTER SYSTEM SET citus.recover_2pc_interval TO 10;
+SELECT pg_reload_conf();
+SELECT pg_sleep(0.1);
+SELECT count(*) FROM pg_dist_transaction;
+
+ALTER SYSTEM RESET citus.recover_2pc_interval;
+SELECT pg_reload_conf();
+
 DROP TABLE test_recovery_ref;
 DROP TABLE test_recovery;
+DROP TABLE test_recovery_single;