Relax lock when processing CAgg invalidation logs

fabriziomello · fabriziomello · commit 26a2c50ebbd3 · 2025-09-19T15:20:20.000-03:00
In timescale#8515 we made some improvements on Concurrent CAgg Refreshes but an oversight happened keeping the `ExclusiveLock` in the materialization hypertable during the CAgg invalidation logs processing ending up with concurrent refreshes on non-overlaping ranges waiting for each other. Fixed it by relaxing the `ExclusiveLock` to `ShareUpdateExclusiveLock` since we should only guarantee that the CAgg invalidation logs processing transaction be executed serially but the next transaction (materialization phase) can't be blocked by this or block other concurrent refreshes.
diff --git a/tsl/src/continuous_aggs/refresh.c b/tsl/src/continuous_aggs/refresh.c
@@ -697,12 +697,15 @@ process_cagg_invalidations_and_refresh(const ContinuousAgg *cagg,
 	 * This is supposed to be a short transaction and in the future we can consider
 	 * relaxing this lock.
 	 */
-	LockRelationOid(hyper_relid, ExclusiveLock);
+	LockRelationOid(hyper_relid, ShareUpdateExclusiveLock);
 	invalidations = invalidation_process_cagg_log(cagg,
 												  refresh_window,
 												  ts_guc_cagg_max_individual_materializations,
 												  context,
 												  force);
+
+	DEBUG_WAITPOINT("before_process_cagg_invalidations_for_refresh_lock");
+
 	SPI_commit_and_chain();
 
 	DEBUG_WAITPOINT("after_process_cagg_invalidations_for_refresh_lock");
@@ -957,6 +960,8 @@ continuous_agg_refresh_internal(const ContinuousAgg *cagg,
 	if (!refreshed && !has_pending_materializations)
 		emit_up_to_date_notice(cagg, context);
 
+	DEBUG_WAITPOINT("after_process_cagg_materializations");
+
 	/* Restore search_path */
 	AtEOXact_GUC(false, save_nestlevel);
 
diff --git a/tsl/test/isolation/expected/cagg_concurrent_refresh.out b/tsl/test/isolation/expected/cagg_concurrent_refresh.out
@@ -1,4 +1,4 @@
-Parsed test spec with 14 sessions
+Parsed test spec with 16 sessions
 
 starting permutation: R1_refresh S1_select R3_refresh S1_select L2_read_unlock_threshold_table L3_unlock_cagg_table L1_unlock_threshold_table
 step R1_refresh: 
@@ -544,7 +544,7 @@ step R12_refresh:
     CALL refresh_continuous_aggregate('cond2_10', 25, 70);
 
 
-starting permutation: WP_enable R1_refresh R6_pending_materialization_ranges R5_refresh R6_pending_materialization_ranges WP_release R6_pending_materialization_ranges S1_select
+starting permutation: WP_after_enable R1_refresh R6_pending_materialization_ranges R5_refresh R6_pending_materialization_ranges WP_after_release R6_pending_materialization_ranges S1_select
 R5: LOG:  statement: 
     SET SESSION lock_timeout = '500ms';
     SET SESSION deadlock_timeout = '500ms';
@@ -553,7 +553,7 @@ R5: LOG:  statement:
 L1: WARNING:  there is already a transaction in progress
 L2: WARNING:  there is already a transaction in progress
 L3: WARNING:  there is already a transaction in progress
-step WP_enable: 
+step WP_after_enable: 
     SELECT debug_waitpoint_enable('after_process_cagg_invalidations_for_refresh_lock');
 
 debug_waitpoint_enable
@@ -587,7 +587,7 @@ cond_10       |                   30|                     70
 cond_10       |                   70|                    100
 (2 rows)
 
-step WP_release: 
+step WP_after_release: 
     SELECT debug_waitpoint_release('after_process_cagg_invalidations_for_refresh_lock');
 
 debug_waitpoint_release
@@ -645,7 +645,7 @@ conditions2|-2147483648
 (2 rows)
 
 
-starting permutation: WP_enable R6_pending_materialization_ranges R1_refresh R3_refresh K1_killpid R6_pending_materialization_ranges WP_release R13_refresh R6_pending_materialization_ranges
+starting permutation: WP_after_enable R6_pending_materialization_ranges R1_refresh R3_refresh K1_cancelpid R6_pending_materialization_ranges WP_after_release R13_refresh R6_pending_materialization_ranges
 R5: LOG:  statement: 
     SET SESSION lock_timeout = '500ms';
     SET SESSION deadlock_timeout = '500ms';
@@ -654,7 +654,7 @@ R5: LOG:  statement:
 L1: WARNING:  there is already a transaction in progress
 L2: WARNING:  there is already a transaction in progress
 L3: WARNING:  there is already a transaction in progress
-step WP_enable: 
+step WP_after_enable: 
     SELECT debug_waitpoint_enable('after_process_cagg_invalidations_for_refresh_lock');
 
 debug_waitpoint_enable
@@ -675,15 +675,11 @@ step R1_refresh:
 step R3_refresh: 
     CALL refresh_continuous_aggregate('cond_10', 70, 107);
  <waiting ...>
-step K1_killpid: 
-    CALL killpids();
+step K1_cancelpid: 
+    CALL cancelpids();
  <waiting ...>
 step R1_refresh: <... completed>
-FATAL:  terminating connection due to administrator command
-server closed the connection unexpectedly
-	This probably means the server terminated abnormally
-	before or while processing the request.
-
+ERROR:  canceling statement due to user request
 step R6_pending_materialization_ranges: 
     SELECT * FROM pending_materialization_ranges WHERE user_view_name = 'cond_10';
 
@@ -693,8 +689,8 @@ cond_10       |                   30|                     70
 cond_10       |                   70|                    100
 (2 rows)
 
-step K1_killpid: <... completed>
-step WP_release: 
+step K1_cancelpid: <... completed>
+step WP_after_release: 
     SELECT debug_waitpoint_release('after_process_cagg_invalidations_for_refresh_lock');
 
 debug_waitpoint_release
@@ -713,3 +709,70 @@ user_view_name|lowest_modified_value|greatest_modified_value
 --------------+---------------------+-----------------------
 (0 rows)
 
+
+starting permutation: WP_before_enable R1_refresh R3_refresh WP_before_release
+R5: LOG:  statement: 
+    SET SESSION lock_timeout = '500ms';
+    SET SESSION deadlock_timeout = '500ms';
+    SET SESSION client_min_messages = 'DEBUG1';
+
+L1: WARNING:  there is already a transaction in progress
+L2: WARNING:  there is already a transaction in progress
+L3: WARNING:  there is already a transaction in progress
+step WP_before_enable: 
+    SELECT debug_waitpoint_enable('before_process_cagg_invalidations_for_refresh_lock');
+
+debug_waitpoint_enable
+----------------------
+                      
+(1 row)
+
+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
+-----------------------
+                       
+(1 row)
+
+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: 
+    SET SESSION lock_timeout = '500ms';
+    SET SESSION deadlock_timeout = '500ms';
+    SET SESSION client_min_messages = 'DEBUG1';
+
+L1: WARNING:  there is already a transaction in progress
+L2: WARNING:  there is already a transaction in progress
+L3: WARNING:  there is already a transaction in progress
+step WP_after_materialization_enable: 
+    SELECT debug_waitpoint_enable('after_process_cagg_materializations');
+
+debug_waitpoint_enable
+----------------------
+                      
+(1 row)
+
+step R1_refresh: 
+    CALL refresh_continuous_aggregate('cond_10', 25, 70);
+ <waiting ...>
+step WP_after_materialization_release: 
+    SELECT debug_waitpoint_release('after_process_cagg_materializations');
+
+debug_waitpoint_release
+-----------------------
+                       
+(1 row)
+
+step R1_refresh: <... completed>
+step R3_refresh: 
+    CALL refresh_continuous_aggregate('cond_10', 70, 107);
+
diff --git a/tsl/test/isolation/specs/cagg_concurrent_refresh.spec b/tsl/test/isolation/specs/cagg_concurrent_refresh.spec
@@ -113,21 +113,21 @@ setup
       FROM _timescaledb_catalog.continuous_agg
       WHERE user_view_name = cagg
       INTO mattable;
-      EXECUTE format('LOCK table %s IN ROW EXCLUSIVE MODE', mattable);
+      EXECUTE format('LOCK table %s IN SHARE UPDATE EXCLUSIVE MODE', mattable);
     END; $$ LANGUAGE plpgsql;
 
-    CREATE TABLE killpid (
+    CREATE TABLE cancelpid (
         pid INTEGER NOT NULL PRIMARY KEY
     );
-    CREATE OR REPLACE PROCEDURE killpids() AS 
+    CREATE OR REPLACE PROCEDURE cancelpids() AS 
     $$
     BEGIN
-        PERFORM pg_terminate_backend(pid) FROM killpid;
-        WHILE EXISTS (SELECT FROM pg_stat_activity WHERE pid IN (SELECT pid FROM killpid))
+        PERFORM pg_cancel_backend(pid) FROM cancelpid;
+        WHILE EXISTS (SELECT FROM pg_stat_activity WHERE pid IN (SELECT pid FROM cancelpid) AND state = 'active')
         LOOP
             PERFORM pg_sleep(0.01);
         END LOOP;
-        DELETE FROM killpid;
+        DELETE FROM cancelpid;
     END;
     $$ LANGUAGE plpgsql;
 
@@ -176,27 +176,47 @@ setup
 teardown {
     DROP TABLE conditions CASCADE;
     DROP TABLE conditions2 CASCADE;
-    DROP TABLE killpid;
+    DROP TABLE cancelpid;
 }
 
 # Waitpoint for cagg invalidation logs
-session "WP"
-step "WP_enable"
+session "WP_after"
+step "WP_after_enable"
 {
     SELECT debug_waitpoint_enable('after_process_cagg_invalidations_for_refresh_lock');
 }
-step "WP_release"
+step "WP_after_release"
 {
     SELECT debug_waitpoint_release('after_process_cagg_invalidations_for_refresh_lock');
 }
 
+session "WP_before"
+step "WP_before_enable"
+{
+    SELECT debug_waitpoint_enable('before_process_cagg_invalidations_for_refresh_lock');
+}
+step "WP_before_release"
+{
+    SELECT debug_waitpoint_release('before_process_cagg_invalidations_for_refresh_lock');
+}
+
+session "WP_after_materialization"
+step "WP_after_materialization_enable"
+{
+    SELECT debug_waitpoint_enable('after_process_cagg_materializations');
+}
+step "WP_after_materialization_release"
+{
+    SELECT debug_waitpoint_release('after_process_cagg_materializations');
+}
+
 # Session to refresh the cond_10 continuous aggregate
 session "R1"
 setup
 {
     SET SESSION lock_timeout = '500ms';
     SET SESSION deadlock_timeout = '500ms';
-    INSERT INTO killpid VALUES (pg_backend_pid())
+    INSERT INTO cancelpid VALUES (pg_backend_pid())
     ON CONFLICT (pid) DO NOTHING;
 }
 step "R1_refresh"
@@ -364,9 +384,9 @@ step "S1_select"
 }
 
 session "K1"
-step "K1_killpid"
+step "K1_cancelpid"
 {
-    CALL killpids();
+    CALL cancelpids();
 }
 
 ####################################################################
@@ -396,7 +416,7 @@ permutation "R3_refresh" "L2_read_lock_threshold_table" "R1_refresh" "L2_read_un
 ##################################################################
 #
 # Tests for concurrent refreshes of continuous aggregates (second
-# transaction of a refresh).
+# and third transactions of a refresh).
 #
 ##################################################################
 
@@ -406,8 +426,8 @@ permutation "L3_lock_cagg_table" "R1_refresh" "L3_unlock_cagg_table" "S1_select"
 # R1 and R2 queued to refresh
 permutation "L3_lock_cagg_table" "R1_refresh" "R2_refresh" "L3_unlock_cagg_table" "S1_select" "L1_unlock_threshold_table" "L2_read_unlock_threshold_table"
 
-# R1 and R3 don't have overlapping refresh windows, but should skip
-# locks and process the materialization.
+# R1 and R3 don't have overlapping refresh windows, but should serialize
+# anyway cause we're locking the cagg hypertable
 permutation "L3_lock_cagg_table" "R1_refresh" "R3_refresh" "L3_unlock_cagg_table" "S1_select" "L1_unlock_threshold_table" "L2_read_unlock_threshold_table"
 
 # Concurrent refreshing across two different aggregates on same
@@ -420,8 +440,15 @@ permutation "R1_refresh" "R12_refresh"
 
 # CAgg invalidation logs processing in a separated transaction and the materialization
 # transaction can be executed concurrently
-permutation "WP_enable" "R1_refresh"("WP_enable") "R6_pending_materialization_ranges" "R5_refresh"("WP_enable") "R6_pending_materialization_ranges" "WP_release" "R6_pending_materialization_ranges" "S1_select"
+permutation "WP_after_enable" "R1_refresh"("WP_after_enable") "R6_pending_materialization_ranges" "R5_refresh"("WP_after_enable") "R6_pending_materialization_ranges" "WP_after_release" "R6_pending_materialization_ranges" "S1_select"
 
 # CAgg materialization phase (third trasaction of the refresh procedure) terminated by another session and then
 # refreshing again and make sure the pending ranges will be processed
-permutation "WP_enable" "R6_pending_materialization_ranges" "R1_refresh"("WP_enable") "R3_refresh"("WP_enable") "K1_killpid"("R1_refresh") "R6_pending_materialization_ranges" "WP_release" "R13_refresh"("K1_killpid") "R6_pending_materialization_ranges"
+permutation "WP_after_enable" "R6_pending_materialization_ranges" "R1_refresh"("WP_after_enable") "R3_refresh"("WP_after_enable") "K1_cancelpid"("R1_refresh") "R6_pending_materialization_ranges" "WP_after_release" "R13_refresh"("K1_cancelpid") "R6_pending_materialization_ranges"
+
+# R3 should wait for R1 to finish because there are cagg invalidation rows locked
+permutation "WP_before_enable" "R1_refresh"("WP_before_enable") "R3_refresh" "WP_before_release"
+
+# Concurrent refresh of caggs on non-overlapping ranges should not
+# block each other in the third transaction (materialization)
+permutation "WP_after_materialization_enable" "R1_refresh"("WP_after_materialization_enable") "WP_after_materialization_release" "R3_refresh" 
