feat: postpone waiting of sync ddl

Due to the shared-storage architecture, higher requirement for DDL
synchronization is needed. Primary needs to acquire locks among all
replicas before doing acture operation. But it cause too much waiting
on PolarDB PG, some of them are just redundant. For example, primary
acqueries DDL lock and do nothing. This happends really often in
vacuum. And another case is create new table, truncate temp table are
also waiting for the DDL lock, which is unnecessary.

In this commit, postpone the waiting of sync ddl till the actual file
operations, eg, truncate and unlink. This will reduce a lot of
unnecessary waitings. The waiting on the operation of temp table is
also ignored. And thanks to the postpone, the waiting time can also be
collapse. When the waiting begins, it's likely to have all replicas
replayed beyond the point.

Add a new GUC polar_enable_sync_ddl_legacy to enable the old style
synchronous ddl.

Author: qiuyuhang

src/backend/access/transam/xact.c

@@ -2175,6 +2175,8 @@ StartTransaction(void)
 		enable_timeout_after(TRANSACTION_TIMEOUT, TransactionTimeout);
 
 	ShowTransactionState("StartTransaction");
+
+	Assert(polar_ddl_lock_lsn == InvalidXLogRecPtr);
 }
 
 
@@ -2290,6 +2292,9 @@ CommitTransaction(void)
 	if (!is_parallel_worker)
 		PreCommit_CheckForSerializationFailure();
 
+	if (polar_enable_sync_ddl)
+		polar_wait_ddl_lock_for_pending_deletes();
+
 	/* Prevent cancel/die interrupt while cleaning up */
 	HOLD_INTERRUPTS();
 
@@ -2448,6 +2453,8 @@ CommitTransaction(void)
 	XactTopFullTransactionId = InvalidFullTransactionId;
 	nParallelCurrentXids = 0;
 
+	polar_ddl_lock_lsn = InvalidXLogRecPtr;
+
 	/*
 	 * done with commit processing, set current transaction state back to
 	 * default
@@ -2577,6 +2584,9 @@ PrepareTransaction(void)
 				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				 errmsg("cannot PREPARE a transaction that has exported snapshots")));
 
+	if (polar_enable_sync_ddl)
+		polar_wait_ddl_lock_for_pending_deletes();
+
 	/* Prevent cancel/die interrupt while cleaning up */
 	HOLD_INTERRUPTS();
 
@@ -2739,6 +2749,8 @@ PrepareTransaction(void)
 	XactTopFullTransactionId = InvalidFullTransactionId;
 	nParallelCurrentXids = 0;
 
+	polar_ddl_lock_lsn = InvalidXLogRecPtr;
+
 	/*
 	 * done with 1st phase commit processing, set current transaction state
 	 * back to default
@@ -2964,6 +2976,8 @@ AbortTransaction(void)
 		pgstat_report_xact_timestamp(0);
 	}
 
+	polar_ddl_lock_lsn = InvalidXLogRecPtr;
+
 	/*
 	 * State remains TRANS_ABORT until CleanupTransaction().
 	 */

src/backend/access/transam/xlogutils.c

@@ -1280,7 +1280,7 @@ polar_get_recovery_bulk_extend_size(BlockNumber target_block, BlockNumber nblock
 	/* Avoid acceed maximum possible length */
 	bulk_extend_size = Min(MaxBlockNumber - nblocks, bulk_extend_size);
 
-	/* Extend the relation to blockNum + 1 at least */
+	/* Extend the relation to target_block + 1 at least */
 	bulk_extend_size = Max(target_block - nblocks + 1, bulk_extend_size);
 
 	return bulk_extend_size;

src/backend/catalog/storage.c

@@ -28,6 +28,7 @@
 #include "catalog/storage.h"
 #include "catalog/storage_xlog.h"
 #include "miscadmin.h"
+#include "replication/syncrep.h"
 #include "storage/bulk_write.h"
 #include "storage/freespace.h"
 #include "storage/smgr.h"
@@ -342,6 +343,9 @@ RelationTruncate(Relation rel, BlockNumber nblocks)
 		}
 	}
 
+	if (polar_enable_sync_ddl && reln->smgr_rnode.backend == InvalidBackendId)
+		polar_wait_ddl_lock();
+
 	RelationPreTruncate(rel);
 
 	/*

src/backend/commands/dbcommands.c

@@ -2214,14 +2214,6 @@ movedb(const char *dbname, const char *tblspcname)
 	/* Start new transaction for the remaining work; don't need a snapshot */
 	StartTransactionCommand();
 
-	/*
-	 * Remove files from the old tablespace
-	 */
-	if (!rmtree(src_dbpath, true))
-		ereport(WARNING,
-				(errmsg("some useless files may be left behind in old database directory \"%s\"",
-						src_dbpath)));
-
 	/*
 	 * Record the filesystem change in XLOG
 	 */
@@ -2235,11 +2227,21 @@ movedb(const char *dbname, const char *tblspcname)
 		XLogRegisterData((char *) &xlrec, sizeof(xl_dbase_drop_rec));
 		XLogRegisterData((char *) &src_tblspcoid, sizeof(Oid));
 
-		(void) XLogInsert(RM_DBASE_ID,
-						  XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
+		polar_ddl_lock_lsn = XLogInsert(RM_DBASE_ID,
+										XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
+		if (polar_enable_sync_ddl)
+			polar_wait_ddl_lock();
 		POLAR_RECORD_DB_STATE(xlrec.ntablespaces, xlrec.tablespace_ids, xlrec.db_id, POLAR_DB_DROPED);
 	}
 
+	/*
+	 * Remove files from the old tablespace
+	 */
+	if (!rmtree(src_dbpath, true))
+		ereport(WARNING,
+				(errmsg("some useless files may be left behind in old database directory \"%s\"",
+						src_dbpath)));
+
 	/* Now it's safe to release the database lock */
 	UnlockSharedObjectForSession(DatabaseRelationId, db_id, 0,
 								 AccessExclusiveLock);
@@ -3504,7 +3506,6 @@ static void
 polar_sync_dropdb_wal(Oid db_id)
 {
 	xl_dbase_drop_rec xlrec;
-	XLogRecPtr	polar_recptr;
 	Oid		   *tablespace_ids;
 
 	Relation	rel;
@@ -3565,8 +3566,8 @@ polar_sync_dropdb_wal(Oid db_id)
 	XLogRegisterData((char *) &xlrec, MinSizeOfDbaseDropRec);
 	XLogRegisterData((char *) tablespace_ids, ntblspc * sizeof(Oid));
 
-	polar_recptr = XLogInsert(RM_DBASE_ID,
-							  XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
+	polar_ddl_lock_lsn = XLogInsert(RM_DBASE_ID,
+									XLOG_DBASE_DROP | XLR_SPECIAL_REL_UPDATE);
 	POLAR_RECORD_DB_STATE(ntblspc, tablespace_ids, db_id, POLAR_DB_DROPED);
 
 	list_free(ltblspc);
@@ -3575,13 +3576,6 @@ polar_sync_dropdb_wal(Oid db_id)
 	table_endscan(scan);
 	table_close(rel, AccessShareLock);
 
-	/*
-	 * POLAR: synchronous ddl, enable standby lock, wait all replica node
-	 * replay this log
-	 */
 	if (polar_enable_sync_ddl)
-	{
-		XLogFlush(polar_recptr);
-		SyncRepWaitForLSN(polar_recptr, false, true);
-	}
+		polar_wait_ddl_lock();
 }

src/backend/replication/syncrep.c

@@ -88,10 +88,16 @@
 #include "utils/ps_status.h"
 
 /* POLAR */
+#include "access/xlogutils.h"
+#include "catalog/storage.h"
 #include "replication/slot.h"
 
 /* User-settable parameters for sync rep */
 char	   *SyncRepStandbyNames;
+bool		polar_enable_sync_ddl = true;
+bool		polar_enable_sync_ddl_legacy = false;
+
+XLogRecPtr	polar_ddl_lock_lsn = InvalidXLogRecPtr;
 
 #define SyncStandbysDefined() \
 	(SyncRepStandbyNames != NULL && SyncRepStandbyNames[0] != '\0')
@@ -152,7 +158,7 @@ static bool polar_get_ddl_applyptr(XLogRecPtr *ddl_applyptr, bool *replication_s
  * remote_apply, because only commit records provide apply feedback.
  */
 void
-SyncRepWaitForLSN(XLogRecPtr lsn, bool commit, bool polar_force_wait_apply)
+SyncRepWaitForLSN(XLogRecPtr lsn, bool commit, bool sync_ddl)
 {
 	char	   *new_status = NULL;
 	const char *old_status;
@@ -180,7 +186,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit, bool polar_force_wait_apply)
 	if ((!SyncRepRequested() ||
 		 ((((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_status) &
 		  (SYNC_STANDBY_INIT | SYNC_STANDBY_DEFINED)) == SYNC_STANDBY_INIT) &&
-		!polar_force_wait_apply)
+		!sync_ddl)
 		return;
 
 	/* Cap the level for anything other than commit to remote flush only. */
@@ -193,7 +199,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit, bool polar_force_wait_apply)
 	 * POLAR: when enable shared storage, ddl must be in synchronous mode. We
 	 * use streaming replication standby lock for synchronous ddl.
 	 */
-	if (polar_enable_shared_storage_mode && polar_force_wait_apply)
+	if (polar_enable_shared_storage_mode && sync_ddl)
 		mode = POLAR_SYNC_DDL_WAIT_APPLY;
 
 	Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks)));
@@ -214,7 +220,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit, bool polar_force_wait_apply)
 	 * (SYNC_STANDBY_INIT is not set), fall back to a check based on the LSN,
 	 * then do a direct GUC check.
 	 */
-	if (polar_force_wait_apply)
+	if (sync_ddl)
 	{
 		/*
 		 * POLAR: When using synchronous ddl, WalSndCtl's sync_standbys_status
@@ -326,7 +332,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit, bool polar_force_wait_apply)
 		 * We do NOT reset ProcDiePending, so that the process will die after
 		 * the commit is cleaned up.
 		 */
-		if (ProcDiePending && !polar_force_wait_apply)
+		if (ProcDiePending && !sync_ddl)
 		{
 			ereport(WARNING,
 					(errcode(ERRCODE_ADMIN_SHUTDOWN),
@@ -337,7 +343,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit, bool polar_force_wait_apply)
 			break;
 		}
 		/* POLAR: cancel the synchronous ddl. */
-		else if (ProcDiePending && polar_force_wait_apply)
+		else if (ProcDiePending && sync_ddl)
 		{
 			whereToSendOutput = DestNone;
 			SyncRepCancelWait();
@@ -353,7 +359,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit, bool polar_force_wait_apply)
 		 * altogether is not helpful, so we just terminate the wait with a
 		 * suitable warning.
 		 */
-		if (QueryCancelPending && !polar_force_wait_apply)
+		if (QueryCancelPending && !sync_ddl)
 		{
 			QueryCancelPending = false;
 			ereport(WARNING,
@@ -367,7 +373,7 @@ SyncRepWaitForLSN(XLogRecPtr lsn, bool commit, bool polar_force_wait_apply)
 		 * POLAR: cancel the synchronous ddl if a query cancel interrupt
 		 * arrives.
 		 */
-		else if (QueryCancelPending && polar_force_wait_apply)
+		else if (QueryCancelPending && sync_ddl)
 		{
 			QueryCancelPending = false;
 			SyncRepCancelWait();
@@ -1360,3 +1366,33 @@ polar_get_ddl_applyptr(XLogRecPtr *ddl_applyptr, bool *replica_slot_all_active)
 	LWLockRelease(ReplicationSlotControlLock);
 	return replica_slot_exist;
 }
+
+void
+polar_wait_ddl_lock(void)
+{
+	if (InRecovery || !polar_enable_shared_storage_mode)
+		return;
+
+	if (XLogRecPtrIsInvalid(polar_ddl_lock_lsn))
+		return;
+
+	XLogFlush(polar_ddl_lock_lsn);
+	SyncRepWaitForLSN(polar_ddl_lock_lsn, false, true);
+
+	polar_ddl_lock_lsn = InvalidXLogRecPtr;
+}
+
+void
+polar_wait_ddl_lock_for_pending_deletes(void)
+{
+	RelFileNode *rels;
+	int			nrels;
+
+	nrels = smgrGetPendingDeletes(true, &rels);
+
+	if (nrels > 0)
+	{
+		pfree(rels);
+		polar_wait_ddl_lock();
+	}
+}

src/backend/storage/ipc/standby.c

@@ -1391,24 +1391,16 @@ LogAccessExclusiveLock(Oid dbOid, Oid relOid)
 {
 	xl_standby_lock xlrec;
 
-	/* POLAR */
-	XLogRecPtr	polar_sync_recptr;
-
 	xlrec.xid = GetCurrentTransactionId();
 
 	xlrec.dbOid = dbOid;
 	xlrec.relOid = relOid;
 
-	polar_sync_recptr = LogAccessExclusiveLocks(1, &xlrec);
+	polar_ddl_lock_lsn = LogAccessExclusiveLocks(1, &xlrec);
 	MyXactFlags |= XACT_FLAGS_ACQUIREDACCESSEXCLUSIVELOCK;
 
-	/* POLAR: synchronous ddl, wait all replicas to replay this log */
-	if (polar_enable_shared_storage_mode && polar_enable_sync_ddl)
-	{
-		Assert(!XLogRecPtrIsInvalid(polar_sync_recptr));
-		XLogFlush(polar_sync_recptr);
-		SyncRepWaitForLSN(polar_sync_recptr, false, true);
-	}
+	if (polar_enable_sync_ddl_legacy)
+		polar_wait_ddl_lock();
 }
 
 /*

src/backend/utils/misc/guc.c

@@ -864,8 +864,6 @@ bool		polar_enable_track_network_stat;
 bool		polar_enable_track_network_timing;
 bool		polar_enable_alloc_checkinterrupts;
 
-bool		polar_enable_sync_ddl;
-
 /* POLAR end */
 
 bool		polar_disable_escape_inside_gbk_character;
@@ -1591,7 +1589,7 @@ static struct config_bool ConfigureNamesBool[] =
 	},
 
 	{
-		{"polar_enable_sync_ddl", PGC_SIGHUP, REPLICATION_STANDBY,
+		{"polar_enable_sync_ddl", PGC_USERSET, REPLICATION_STANDBY,
 			gettext_noop("Enable synchronous ddl."),
 			NULL,
 			GUC_NO_SHOW_ALL | GUC_NO_RESET_ALL | POLAR_GUC_IS_INVISIBLE | POLAR_GUC_IS_UNCHANGABLE
@@ -1601,6 +1599,17 @@ static struct config_bool ConfigureNamesBool[] =
 		NULL, NULL, NULL
 	},
 
+	{
+		{"polar_enable_sync_ddl_legacy", PGC_USERSET, REPLICATION_STANDBY,
+			gettext_noop("Enable old style synchronous ddl."),
+			NULL,
+			GUC_NO_SHOW_ALL | GUC_NO_RESET_ALL | POLAR_GUC_IS_INVISIBLE | POLAR_GUC_IS_UNCHANGABLE
+		},
+		&polar_enable_sync_ddl_legacy,
+		false,
+		NULL, NULL, NULL
+	},
+
 	{
 		{"polar_enable_alloc_checkinterrupts", PGC_SIGHUP, UNGROUPED,
 			gettext_noop("Enable check interrupt when allocating memory."),

src/include/replication/syncrep.h

@@ -84,9 +84,13 @@ extern PGDLLIMPORT char *syncrep_parse_error_msg;
 
 /* user-settable parameters for synchronous replication */
 extern PGDLLIMPORT char *SyncRepStandbyNames;
+extern bool polar_enable_sync_ddl;
+extern bool polar_enable_sync_ddl_legacy;
+
+extern XLogRecPtr polar_ddl_lock_lsn;
 
 /* called by user backend */
-extern void SyncRepWaitForLSN(XLogRecPtr lsn, bool commit, bool polar_force_wait_apply);
+extern void SyncRepWaitForLSN(XLogRecPtr lsn, bool commit, bool sync_ddl);
 
 /* called at backend exit */
 extern void SyncRepCleanupAtProcExit(void);
@@ -120,5 +124,7 @@ extern void syncrep_scanner_finish(void);
  * POLAR: called by walsender & drop replication slot
  */
 extern bool polar_release_ddl_waiters(void);
+extern void polar_wait_ddl_lock(void);
+extern void polar_wait_ddl_lock_for_pending_deletes(void);
 
 #endif							/* _SYNCREP_H */

src/include/utils/guc.h

@@ -326,7 +326,6 @@ extern bool polar_enable_debug;
 
 extern char *polar_disk_name;
 extern char *polar_storage_cluster_name;
-extern bool polar_enable_sync_ddl;
 
 /* POLAR end */
 

src/test/perl/Makefile

@@ -27,6 +27,7 @@ install: all installdirs
 	$(INSTALL_DATA) $(srcdir)/PostgreSQL/Test/BackgroundPsql.pm '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Test/BackgroundPsql.pm'
 	$(INSTALL_DATA) $(srcdir)/PostgreSQL/Version.pm '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Version.pm'
 	$(INSTALL_DATA) $(srcdir)/PolarDB/DCRegression.pm '$(DESTDIR)$(pgxsdir)/$(subdir)/PolarDB/DCRegression.pm'
+	$(INSTALL_DATA) $(srcdir)/PolarDB/Task.pm '$(DESTDIR)$(pgxsdir)/$(subdir)/PolarDB/Task.pm'
 
 uninstall:
 	rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Test/Utils.pm'
@@ -36,5 +37,6 @@ uninstall:
 	rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Test/BackgroundPsql.pm'
 	rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/PostgreSQL/Version.pm'
 	rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/PolarDB/DCRegression.pm'
+	rm -f '$(DESTDIR)$(pgxsdir)/$(subdir)/PolarDB/Task.pm'
 
 endif