Relax lock during CAgg invalidation log processing (#9701)

When processing the continuous aggregate invalidation log, we currently
take a ShareUpdateExclusiveLock on the materialized hypertable to
prevent logs being cut by concurrent refreshes on the same continuous
aggregate. However, this can also block other concurrent operations.

We can instead take a RowExclusiveLock on the catalog table entry for
the continuous aggregate. This has the same effect as earlier -
serializing cagg log processing, without the need for the heavy lock
that was taken previously.

Author: kpan2034

.unreleased/pr_9701

@@ -0,0 +1 @@
+Implements: #9701 Relax lock during CAgg invalidation log processing

src/ts_catalog/continuous_agg.c

@@ -138,6 +138,56 @@ init_materialization_invalidation_log_scan_by_materialization_id(ScanIterator *i
 		Int32GetDatum(materialization_id));
 }
 
+TSDLLEXPORT bool
+ts_lock_continuous_agg_tuple(int32 mat_hypertable_id)
+{
+	bool success = false;
+	ScanTupLock scantuplock = {
+		.waitpolicy = LockWaitBlock,
+		.lockmode = LockTupleExclusive,
+	};
+	ScanIterator iterator =
+		ts_scan_iterator_create(CONTINUOUS_AGG, RowShareLock, CurrentMemoryContext);
+	init_scan_by_mat_hypertable_id(&iterator, mat_hypertable_id);
+	iterator.ctx.tuplock = &scantuplock;
+	iterator.ctx.flags = SCANNER_F_KEEPLOCK;
+
+	/* see table_tuple_lock for details about flags that are set in TupleExclusive mode */
+	scantuplock.lockflags = TUPLE_LOCK_FLAG_LOCK_UPDATE_IN_PROGRESS;
+	if (!IsolationUsesXactSnapshot())
+	{
+		/* in read committed mode, we follow all updates to this tuple */
+		scantuplock.lockflags |= TUPLE_LOCK_FLAG_FIND_LAST_VERSION;
+	}
+
+	ts_scanner_foreach(&iterator)
+	{
+		TupleInfo *ti = ts_scan_iterator_tuple_info(&iterator);
+		if (ti->lockresult != TM_Ok)
+		{
+			if (IsolationUsesXactSnapshot())
+			{
+				ereport(ERROR,
+						(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
+						 errmsg("could not serialize access due to concurrent update")));
+			}
+			else
+			{
+				ereport(ERROR,
+						(errcode(ERRCODE_INTERNAL_ERROR),
+						 errmsg("unable to lock continuous aggregate catalog tuple, lock result "
+								"is %d for mat_hypertable_id (%d)",
+								ti->lockresult,
+								mat_hypertable_id)));
+			}
+		}
+		success = true;
+		break;
+	}
+	ts_scan_iterator_close(&iterator);
+	return success;
+}
+
 static int32
 number_of_continuous_aggs_attached(int32 raw_hypertable_id)
 {

src/ts_catalog/continuous_agg.h

@@ -145,6 +145,8 @@ typedef struct ContinuousAggPolicyOffset
 	const char *name;
 } ContinuousAggPolicyOffset;
 
+extern TSDLLEXPORT bool ts_lock_continuous_agg_tuple(int32 mat_hypertable_id);
+
 extern TSDLLEXPORT Oid ts_cagg_permissions_check(Oid cagg_oid, Oid userid);
 
 extern TSDLLEXPORT ContinuousAggInfo ts_continuous_agg_get_all_caggs_info(int32 raw_hypertable_id);

tsl/src/continuous_aggs/refresh.c

@@ -703,15 +703,13 @@ process_cagg_invalidations_and_refresh(const ContinuousAgg *cagg,
 									   const ContinuousAggRefreshContext context,
 									   bool bucketing_refresh_window, bool force)
 {
-	Oid hyper_relid = ts_hypertable_id_to_relid(cagg->data.mat_hypertable_id, false);
-
-	/* Lock the continuous aggregate's materialized hypertable to protect against
-	 * concurrent invalidation log processing.
-	 *
-	 * This is supposed to be a short transaction and in the future we can consider
-	 * relaxing this lock.
+	/* Lock the continuous aggregate's catalog table entry to protect against concurrent refreshes
+	 * on the same cagg processing the cagg invalidation logs for that CAgg.
 	 */
-	LockRelationOid(hyper_relid, ShareUpdateExclusiveLock);
+	bool found = ts_lock_continuous_agg_tuple(cagg->data.mat_hypertable_id);
+	Ensure(found,
+		   "continuous aggregate with mat_hypertable_id %d not found",
+		   cagg->data.mat_hypertable_id);
 	invalidation_process_cagg_log(cagg, refresh_window);
 
 	DEBUG_ERROR_INJECTION("cagg_refresh_fail_in_txn2");

tsl/test/isolation/expected/cagg_concurrent_refresh.out

@@ -258,34 +258,18 @@ step R12_refresh:
     CALL refresh_continuous_aggregate('cond2_10', 25, 70);
 
 
-starting permutation: WP_before_enable R1_refresh R3_refresh WP_before_release
+starting permutation: R1_refresh R3_refresh
 R5: LOG:  statement: 
     SET SESSION lock_timeout = '500ms';
     SET SESSION deadlock_timeout = '500ms';
     SET SESSION client_min_messages = 'DEBUG1';
 
-step WP_before_enable: 
-    SELECT debug_waitpoint_enable('before_process_cagg_invalidations_for_refresh_lock');
-
-debug_waitpoint_enable
-----------------------
-                      
-
 step R1_refresh: 
     CALL refresh_continuous_aggregate('cond_10', 25, 70);
- <waiting ...>
+
 step R3_refresh: 
     CALL refresh_continuous_aggregate('cond_10', 70, 107);
- <waiting ...>
-step WP_before_release: 
-    SELECT debug_waitpoint_release('before_process_cagg_invalidations_for_refresh_lock');
-
-debug_waitpoint_release
------------------------
-                       
 
-step R1_refresh: <... completed>
-step R3_refresh: <... completed>
 
 starting permutation: WP_after_materialization_enable R1_refresh WP_after_materialization_release R3_refresh
 R5: LOG:  statement: 

tsl/test/isolation/expected/cagg_hierarchical_concurrent_refresh.out

@@ -463,7 +463,7 @@ Fri Jan 02 00:00:00 2026 UTC|32.1666666666667|32.1666666666667|        288|
 Sat Jan 03 00:00:00 2026 UTC|            31.5|            31.5|         96|           96|t        |t          
 
 
-starting permutation: L1_refresh_full chk_1d_consistency WP_before_enable chk_hyper_invals L2_refresh_jan1 insert_ht L1_refresh_full lock_L2_source L2b_refresh_jan1_2 WP_before_release unlock chk_cagg_1d chk_1d_consistency
+starting permutation: L1_refresh_full chk_1d_consistency WP_before_enable chk_hyper_invals lock_L2_source L2_refresh_jan1 insert_ht L1_refresh_full L2b_refresh_jan1_2 WP_before_release unlock chk_cagg_1d chk_1d_consistency
 step L1_refresh_full: 
     CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04');
 
@@ -511,6 +511,16 @@ source |lowest                      |greatest
 -------+----------------------------+----------------------------
 cagg_6h|Thu Jan 01 00:00:00 2026 UTC|Sat Jan 03 18:00:00 2026 UTC
 
+step lock_L2_source: 
+    BEGIN;
+    DO $$
+    BEGIN
+        PERFORM 1 FROM _timescaledb_catalog.continuous_agg
+        WHERE user_view_name = 'cagg_1d'
+        FOR UPDATE;
+    END;
+    $$;
+
 step L2_refresh_jan1: 
     CALL refresh_continuous_aggregate('cagg_1d', '2026-01-01', '2026-01-02');
  <waiting ...>
@@ -523,20 +533,6 @@ step insert_ht:
 step L1_refresh_full: 
     CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04');
  <waiting ...>
-step lock_L2_source: 
-    BEGIN;
-    DO $$
-    DECLARE
-        mat_ht text;
-    BEGIN
-        SELECT format('%I.%I', h.schema_name, h.table_name) INTO mat_ht
-        FROM _timescaledb_catalog.continuous_agg ca
-        JOIN _timescaledb_catalog.hypertable h ON h.id = ca.mat_hypertable_id
-        WHERE ca.user_view_name = 'cagg_1d';
-        EXECUTE format('LOCK TABLE %s IN ACCESS EXCLUSIVE MODE', mat_ht);
-    END;
-    $$;
- <waiting ...>
 step L2b_refresh_jan1_2: 
     CALL refresh_continuous_aggregate('cagg_1d', '2026-01-01', '2026-01-03');
 
@@ -549,7 +545,6 @@ debug_waitpoint_release
 
 
 step L1_refresh_full: <... completed>
-step lock_L2_source: <... completed>
 step unlock: 
     ROLLBACK;
 
@@ -590,7 +585,7 @@ Fri Jan 02 00:00:00 2026 UTC|            31.5|32.1666666666667|         96|
 Sat Jan 03 00:00:00 2026 UTC|            31.5|            40.5|         96|          288|f        |f          
 
 
-starting permutation: L1_refresh_full chk_1d_consistency WP_before_enable chk_hyper_invals L2b_refresh_jan1_2 insert_ht L1_refresh_full lock_L2_source L2_refresh_jan1 WP_before_release unlock chk_cagg_1d chk_1d_consistency
+starting permutation: L1_refresh_full chk_1d_consistency WP_before_enable chk_hyper_invals lock_L2_source L2b_refresh_jan1_2 insert_ht L1_refresh_full L2_refresh_jan1 WP_before_release unlock chk_cagg_1d chk_1d_consistency
 step L1_refresh_full: 
     CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04');
 
@@ -638,6 +633,16 @@ source |lowest                      |greatest
 -------+----------------------------+----------------------------
 cagg_6h|Thu Jan 01 00:00:00 2026 UTC|Sat Jan 03 18:00:00 2026 UTC
 
+step lock_L2_source: 
+    BEGIN;
+    DO $$
+    BEGIN
+        PERFORM 1 FROM _timescaledb_catalog.continuous_agg
+        WHERE user_view_name = 'cagg_1d'
+        FOR UPDATE;
+    END;
+    $$;
+
 step L2b_refresh_jan1_2: 
     CALL refresh_continuous_aggregate('cagg_1d', '2026-01-01', '2026-01-03');
  <waiting ...>
@@ -650,20 +655,6 @@ step insert_ht:
 step L1_refresh_full: 
     CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04');
  <waiting ...>
-step lock_L2_source: 
-    BEGIN;
-    DO $$
-    DECLARE
-        mat_ht text;
-    BEGIN
-        SELECT format('%I.%I', h.schema_name, h.table_name) INTO mat_ht
-        FROM _timescaledb_catalog.continuous_agg ca
-        JOIN _timescaledb_catalog.hypertable h ON h.id = ca.mat_hypertable_id
-        WHERE ca.user_view_name = 'cagg_1d';
-        EXECUTE format('LOCK TABLE %s IN ACCESS EXCLUSIVE MODE', mat_ht);
-    END;
-    $$;
- <waiting ...>
 step L2_refresh_jan1: 
     CALL refresh_continuous_aggregate('cagg_1d', '2026-01-01', '2026-01-02');
 
@@ -676,7 +667,6 @@ debug_waitpoint_release
 
 
 step L1_refresh_full: <... completed>
-step lock_L2_source: <... completed>
 step unlock: 
     ROLLBACK;
 
@@ -872,7 +862,7 @@ Fri Jan 02 00:00:00 2026 UTC|            31.5|            31.5|         96|
 Sat Jan 03 00:00:00 2026 UTC|            40.5|            40.5|        288|          288|t        |t          
 
 
-starting permutation: L1_refresh_full chk_1d_consistency WP_before_enable chk_hyper_invals L2_refresh_jan1 insert_ht L1_refresh_full lock_L2_source L2b_refresh_jan3 WP_before_release unlock chk_cagg_1d chk_1d_consistency
+starting permutation: L1_refresh_full chk_1d_consistency WP_before_enable chk_hyper_invals lock_L2_source L2_refresh_jan1 insert_ht L1_refresh_full L2b_refresh_jan3 WP_before_release unlock chk_cagg_1d chk_1d_consistency
 step L1_refresh_full: 
     CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04');
 
@@ -920,6 +910,16 @@ source |lowest                      |greatest
 -------+----------------------------+----------------------------
 cagg_6h|Thu Jan 01 00:00:00 2026 UTC|Sat Jan 03 18:00:00 2026 UTC
 
+step lock_L2_source: 
+    BEGIN;
+    DO $$
+    BEGIN
+        PERFORM 1 FROM _timescaledb_catalog.continuous_agg
+        WHERE user_view_name = 'cagg_1d'
+        FOR UPDATE;
+    END;
+    $$;
+
 step L2_refresh_jan1: 
     CALL refresh_continuous_aggregate('cagg_1d', '2026-01-01', '2026-01-02');
  <waiting ...>
@@ -932,20 +932,6 @@ step insert_ht:
 step L1_refresh_full: 
     CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04');
  <waiting ...>
-step lock_L2_source: 
-    BEGIN;
-    DO $$
-    DECLARE
-        mat_ht text;
-    BEGIN
-        SELECT format('%I.%I', h.schema_name, h.table_name) INTO mat_ht
-        FROM _timescaledb_catalog.continuous_agg ca
-        JOIN _timescaledb_catalog.hypertable h ON h.id = ca.mat_hypertable_id
-        WHERE ca.user_view_name = 'cagg_1d';
-        EXECUTE format('LOCK TABLE %s IN ACCESS EXCLUSIVE MODE', mat_ht);
-    END;
-    $$;
- <waiting ...>
 step L2b_refresh_jan3: 
     CALL refresh_continuous_aggregate('cagg_1d', '2026-01-03', '2026-01-04');
  <waiting ...>
@@ -957,7 +943,6 @@ debug_waitpoint_release
 
 
 step L1_refresh_full: <... completed>
-step lock_L2_source: <... completed>
 step unlock: 
     ROLLBACK;