diff --git a/.unreleased/pr_9701 b/.unreleased/pr_9701 new file mode 100644 index 00000000000..502b2e7a7b3 --- /dev/null +++ b/.unreleased/pr_9701 @@ -0,0 +1 @@ +Implements: #9701 Relax lock during CAgg invalidation log processing diff --git a/src/ts_catalog/continuous_agg.c b/src/ts_catalog/continuous_agg.c index 61dacbfb2bf..f84ab74bae7 100644 --- a/src/ts_catalog/continuous_agg.c +++ b/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) { diff --git a/src/ts_catalog/continuous_agg.h b/src/ts_catalog/continuous_agg.h index ef68f31792a..18a969d2b93 100644 --- a/src/ts_catalog/continuous_agg.h +++ b/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); diff --git a/tsl/src/continuous_aggs/refresh.c b/tsl/src/continuous_aggs/refresh.c index abb771bb067..cd1408d7739 100644 --- a/tsl/src/continuous_aggs/refresh.c +++ b/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"); diff --git a/tsl/test/isolation/expected/cagg_concurrent_refresh.out b/tsl/test/isolation/expected/cagg_concurrent_refresh.out index 40b69d649e7..4a89d815deb 100644 --- a/tsl/test/isolation/expected/cagg_concurrent_refresh.out +++ b/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); - + step R3_refresh: CALL refresh_continuous_aggregate('cond_10', 70, 107); - -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: diff --git a/tsl/test/isolation/expected/cagg_hierarchical_concurrent_refresh.out b/tsl/test/isolation/expected/cagg_hierarchical_concurrent_refresh.out index 440371a3a02..2e0ef6d71fd 100644 --- a/tsl/test/isolation/expected/cagg_hierarchical_concurrent_refresh.out +++ b/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'); @@ -523,20 +533,6 @@ step insert_ht: step L1_refresh_full: CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04'); -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; - $$; - 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'); @@ -650,20 +655,6 @@ step insert_ht: step L1_refresh_full: CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04'); -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; - $$; - 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'); @@ -932,20 +932,6 @@ step insert_ht: step L1_refresh_full: CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04'); -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; - $$; - step L2b_refresh_jan3: CALL refresh_continuous_aggregate('cagg_1d', '2026-01-03', '2026-01-04'); @@ -957,7 +943,6 @@ debug_waitpoint_release step L1_refresh_full: <... completed> -step lock_L2_source: <... completed> step unlock: ROLLBACK; diff --git a/tsl/test/isolation/expected/cagg_multi_iso.out b/tsl/test/isolation/expected/cagg_multi_iso.out index 36c8535466c..195782724a3 100644 --- a/tsl/test/isolation/expected/cagg_multi_iso.out +++ b/tsl/test/isolation/expected/cagg_multi_iso.out @@ -1,6 +1,6 @@ Parsed test spec with 9 sessions -starting permutation: Setup2 LockMat1 Refresh1 Refresh2 UnlockMat1 +starting permutation: Setup2 LockCAggCatalogRow_1 Refresh1 Refresh2 UnlockCAggCatalogRow_1 step Setup2: CREATE MATERIALIZED VIEW continuous_view_1( bkt, cnt) WITH ( timescaledb.continuous, timescaledb.materialized_only = true) @@ -12,62 +12,22 @@ step Setup2: AS SELECT time_bucket('5', time), max(val) FROM ts_continuous_test GROUP BY 1 WITH NO DATA; - CREATE FUNCTION lock_mattable( name text) RETURNS void AS $$ - BEGIN EXECUTE format( 'lock table %s', name); - END; $$ LANGUAGE plpgsql; - -step LockMat1: BEGIN; select lock_mattable(tab) FROM ( SELECT format('%I.%I',materialization_hypertable_schema, materialization_hypertable_name) as tab from timescaledb_information.continuous_aggregates where view_name::text like 'continuous_view_1') q ; - -lock_mattable -------------- - - -step Refresh1: CALL refresh_continuous_aggregate('continuous_view_1', NULL, 30); -step Refresh2: CALL refresh_continuous_aggregate('continuous_view_2', NULL, NULL); -step UnlockMat1: ROLLBACK; -step Refresh1: <... completed> -starting permutation: Setup2 Refresh1 Refresh2 LockMat1 I1 Refresh1 Refresh2 UnlockMat1 Refresh1_sel Refresh2_sel -step Setup2: - CREATE MATERIALIZED VIEW continuous_view_1( bkt, cnt) - WITH ( timescaledb.continuous, timescaledb.materialized_only = true) - AS SELECT time_bucket('5', time), COUNT(val) - FROM ts_continuous_test - GROUP BY 1 WITH NO DATA; - CREATE MATERIALIZED VIEW continuous_view_2(bkt, maxl) - WITH ( timescaledb.continuous, timescaledb.materialized_only = true) - AS SELECT time_bucket('5', time), max(val) - FROM ts_continuous_test - GROUP BY 1 WITH NO DATA; - CREATE FUNCTION lock_mattable( name text) RETURNS void AS $$ - BEGIN EXECUTE format( 'lock table %s', name); + CREATE FUNCTION lock_cagg(mat_hypertable_id integer) RETURNS void AS $$ + BEGIN PERFORM 1 FROM _timescaledb_catalog.continuous_agg ca WHERE ca.mat_hypertable_id = lock_cagg.mat_hypertable_id FOR UPDATE; END; $$ LANGUAGE plpgsql; -step Refresh1: CALL refresh_continuous_aggregate('continuous_view_1', NULL, 30); -step Refresh2: CALL refresh_continuous_aggregate('continuous_view_2', NULL, NULL); -step LockMat1: BEGIN; select lock_mattable(tab) FROM ( SELECT format('%I.%I',materialization_hypertable_schema, materialization_hypertable_name) as tab from timescaledb_information.continuous_aggregates where view_name::text like 'continuous_view_1') q ; - -lock_mattable -------------- - +step LockCAggCatalogRow_1: BEGIN; select lock_cagg(mat_hypertable_id) FROM (SELECT mat_hypertable_id from _timescaledb_catalog.continuous_agg where user_view_name like 'continuous_view_1') q; +lock_cagg +--------- + -step I1: INSERT INTO ts_continuous_test SELECT 0, i*10 FROM (SELECT generate_series(0, 10) AS i) AS i; step Refresh1: CALL refresh_continuous_aggregate('continuous_view_1', NULL, 30); step Refresh2: CALL refresh_continuous_aggregate('continuous_view_2', NULL, NULL); -step UnlockMat1: ROLLBACK; +step UnlockCAggCatalogRow_1: ROLLBACK; step Refresh1: <... completed> -step Refresh1_sel: select * from continuous_view_1 where bkt = 0 or bkt > 30 -bkt|cnt ----+--- - 0| 16 - -step Refresh2_sel: select * from continuous_view_2 where bkt = 0 or bkt > 30 order by bkt; -bkt|maxl ----+---- - 0| 100 - -starting permutation: Setup2 Refresh1 Refresh2 Refresh1_sel Refresh2_sel LockMat1 I2 Refresh1 Refresh2 UnlockMat1 Refresh1_sel Refresh2_sel +starting permutation: Setup2 Refresh1 Refresh2 Refresh1_sel Refresh2_sel LockCAggCatalogRow_1 I1 Refresh1 Refresh2 Refresh2_sel UnlockCAggCatalogRow_1 Refresh1_sel step Setup2: CREATE MATERIALIZED VIEW continuous_view_1( bkt, cnt) WITH ( timescaledb.continuous, timescaledb.materialized_only = true) @@ -79,8 +39,9 @@ step Setup2: AS SELECT time_bucket('5', time), max(val) FROM ts_continuous_test GROUP BY 1 WITH NO DATA; - CREATE FUNCTION lock_mattable( name text) RETURNS void AS $$ - BEGIN EXECUTE format( 'lock table %s', name); + + CREATE FUNCTION lock_cagg(mat_hypertable_id integer) RETURNS void AS $$ + BEGIN PERFORM 1 FROM _timescaledb_catalog.continuous_agg ca WHERE ca.mat_hypertable_id = lock_cagg.mat_hypertable_id FOR UPDATE; END; $$ LANGUAGE plpgsql; step Refresh1: CALL refresh_continuous_aggregate('continuous_view_1', NULL, 30); @@ -95,28 +56,25 @@ bkt|maxl ---+---- 0| 4 -step LockMat1: BEGIN; select lock_mattable(tab) FROM ( SELECT format('%I.%I',materialization_hypertable_schema, materialization_hypertable_name) as tab from timescaledb_information.continuous_aggregates where view_name::text like 'continuous_view_1') q ; +step LockCAggCatalogRow_1: BEGIN; select lock_cagg(mat_hypertable_id) FROM (SELECT mat_hypertable_id from _timescaledb_catalog.continuous_agg where user_view_name like 'continuous_view_1') q; +lock_cagg +--------- + -lock_mattable -------------- - - -step I2: INSERT INTO ts_continuous_test SELECT 40, 1000 ; +step I1: INSERT INTO ts_continuous_test SELECT 0, i*10 FROM (SELECT generate_series(0, 10) AS i) AS i; step Refresh1: CALL refresh_continuous_aggregate('continuous_view_1', NULL, 30); step Refresh2: CALL refresh_continuous_aggregate('continuous_view_2', NULL, NULL); -step UnlockMat1: ROLLBACK; -R1: NOTICE: continuous aggregate "continuous_view_1" is already up-to-date +step Refresh2_sel: select * from continuous_view_2 where bkt = 0 or bkt > 30 order by bkt; +bkt|maxl +---+---- + 0| 100 + +step UnlockCAggCatalogRow_1: ROLLBACK; step Refresh1: <... completed> step Refresh1_sel: select * from continuous_view_1 where bkt = 0 or bkt > 30 bkt|cnt ---+--- - 0| 5 - -step Refresh2_sel: select * from continuous_view_2 where bkt = 0 or bkt > 30 order by bkt; -bkt|maxl ----+---- - 0| 4 - 40|1000 + 0| 16 starting permutation: Setup2 Refresh1 Refresh2 Refresh1_sel Refresh2_sel U1 U2 LInvRow Refresh1 Refresh2 UnlockInvRow Refresh1_sel Refresh2_sel @@ -131,8 +89,9 @@ step Setup2: AS SELECT time_bucket('5', time), max(val) FROM ts_continuous_test GROUP BY 1 WITH NO DATA; - CREATE FUNCTION lock_mattable( name text) RETURNS void AS $$ - BEGIN EXECUTE format( 'lock table %s', name); + + CREATE FUNCTION lock_cagg(mat_hypertable_id integer) RETURNS void AS $$ + BEGIN PERFORM 1 FROM _timescaledb_catalog.continuous_agg ca WHERE ca.mat_hypertable_id = lock_cagg.mat_hypertable_id FOR UPDATE; END; $$ LANGUAGE plpgsql; step Refresh1: CALL refresh_continuous_aggregate('continuous_view_1', NULL, 30); diff --git a/tsl/test/isolation/specs/cagg_concurrent_refresh.spec b/tsl/test/isolation/specs/cagg_concurrent_refresh.spec index 73ba39d1e69..67bbd9a5122 100644 --- a/tsl/test/isolation/specs/cagg_concurrent_refresh.spec +++ b/tsl/test/isolation/specs/cagg_concurrent_refresh.spec @@ -488,8 +488,8 @@ permutation "R1_refresh" "R12_refresh" # TODO: pending materialization ranges not populated yet #permutation "WP_after_enable" "R6_pending_materialization_ranges" "R1_refresh"("WP_after_enable") "K1_cancelpid"("R1_refresh") "R6_pending_materialization_ranges" "WP_after_release" "R1_drop" "R6_pending_materialization_ranges_orphan" -# 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" +# R3 should wait for R1 to finish because they serialize on the catalog tuple lock +permutation "R1_refresh" "R3_refresh" # Concurrent refresh of caggs on non-overlapping ranges should not # block each other in the third transaction (materialization) diff --git a/tsl/test/isolation/specs/cagg_hierarchical_concurrent_refresh.spec b/tsl/test/isolation/specs/cagg_hierarchical_concurrent_refresh.spec index 8e4cd1fff45..bdae3e51c8a 100644 --- a/tsl/test/isolation/specs/cagg_hierarchical_concurrent_refresh.spec +++ b/tsl/test/isolation/specs/cagg_hierarchical_concurrent_refresh.spec @@ -191,9 +191,9 @@ step "L2b_refresh_jan3" } # Session to handle locks. -# It can block L2 refreshes by locking cagg_1d's mat hypertable. -# Unlike waitpoints (which block all refreshes), locking L2's materialization -# table only blocks L2 refreshes while L1 refreshes can run freely. +# It can block L2 refreshes by locking cagg_1d's continuous_agg catalog row. +# Unlike waitpoints (which block all refreshes), locking L2's catalog row +# only blocks L2 refreshes while L1 refreshes can run freely. session "LOCK" setup { @@ -203,14 +203,10 @@ 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); + PERFORM 1 FROM _timescaledb_catalog.continuous_agg + WHERE user_view_name = 'cagg_1d' + FOR UPDATE; END; $$; } @@ -321,10 +317,10 @@ permutation "WP_before_enable" "chk_hyper_invals" "L1b_refresh_jan1_2" "insert_h # Two concurrent refreshes on L2 CAgg with overlapping ranges. One should fail due to overlap. # L1 is refreshed between the two L2 refreshes so that the second L2 refresh encounters an overlapping materialization range and fails. -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" +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" # Same as above, reverse refresh order. -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" +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" # Non-overlapping concurrent refreshes on L1. Both refreshes should succeed. Both are blocked before Txn 3. # Refreshing Jan 3 adds cagg invalidations that are partially overlapping with the one created by refreshing Jan 1. @@ -334,7 +330,7 @@ permutation "WP_before_enable" "chk_hyper_invals" "L1_refresh_jan1" "insert_ht" # Non-overlapping concurrent refresh on L2. Both refreshes should succeed. Both are blocked before Txn 3. # Refreshing Jan 3 adds cagg invalidations that are partially overlapping with the one created by refreshing Jan 1. # Those invalidations are left behind, but both Jan 1 and Jan 3 are refreshed successfully. -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" +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" # L1 (txn3, deleting/inserting mat_inval entries) and L2 (txn2, processing mat_inval entries) should not block each other. # Lock materialization invalidation table to make both refreshes wait before processing entries, then release simultaneously. diff --git a/tsl/test/isolation/specs/cagg_multi_iso.spec b/tsl/test/isolation/specs/cagg_multi_iso.spec index 32a06b16e6f..61f96f1c8f9 100644 --- a/tsl/test/isolation/specs/cagg_multi_iso.spec +++ b/tsl/test/isolation/specs/cagg_multi_iso.spec @@ -14,7 +14,7 @@ setup } teardown { - DROP FUNCTION lock_mattable( text ); + DROP FUNCTION lock_cagg(integer); DROP TABLE ts_continuous_test CASCADE; } @@ -32,14 +32,14 @@ step "Setup2" AS SELECT time_bucket('5', time), max(val) FROM ts_continuous_test GROUP BY 1 WITH NO DATA; - CREATE FUNCTION lock_mattable( name text) RETURNS void AS $$ - BEGIN EXECUTE format( 'lock table %s', name); + + CREATE FUNCTION lock_cagg(mat_hypertable_id integer) RETURNS void AS $$ + BEGIN PERFORM 1 FROM _timescaledb_catalog.continuous_agg ca WHERE ca.mat_hypertable_id = lock_cagg.mat_hypertable_id FOR UPDATE; END; $$ LANGUAGE plpgsql; } session "I" step "I1" { INSERT INTO ts_continuous_test SELECT 0, i*10 FROM (SELECT generate_series(0, 10) AS i) AS i; } -step "I2" { INSERT INTO ts_continuous_test SELECT 40, 1000 ; } session "R1" setup { SET client_min_messages TO NOTICE; } @@ -55,11 +55,10 @@ step "Refresh2" { CALL refresh_continuous_aggregate('continuous_view_2', NULL, N session "R2_sel" step "Refresh2_sel" { select * from continuous_view_2 where bkt = 0 or bkt > 30 order by bkt; } -#locking the materialized table will block refresh1 -session "LM1" -step "LockMat1" { BEGIN; select lock_mattable(tab) FROM ( SELECT format('%I.%I',materialization_hypertable_schema, materialization_hypertable_name) as tab from timescaledb_information.continuous_aggregates where view_name::text like 'continuous_view_1') q ; -} -step "UnlockMat1" { ROLLBACK; } +#locking the catalog row will block refresh1 +session "LC1" +step "LockCAggCatalogRow_1" { BEGIN; select lock_cagg(mat_hypertable_id) FROM (SELECT mat_hypertable_id from _timescaledb_catalog.continuous_agg where user_view_name like 'continuous_view_1') q; } +step "UnlockCAggCatalogRow_1" { ROLLBACK; } #update the hypertable session "Upd" @@ -74,17 +73,15 @@ step "LInvRow" { BEGIN; update _timescaledb_catalog.continuous_aggs_invalidation step "UnlockInvRow" { ROLLBACK; } -#refresh1, refresh2 can run concurrently -permutation "Setup2" "LockMat1" "Refresh1" "Refresh2" "UnlockMat1" - -#refresh1 and refresh2 run concurrently and see the correct invalidation -#test1 - both see the same invalidation -permutation "Setup2" "Refresh1" "Refresh2" "LockMat1" "I1" "Refresh1" "Refresh2" "UnlockMat1" "Refresh1_sel" "Refresh2_sel" +# Test 1: Refreshes on different CAggs can run concurrently. +# Refresh1 is initially blocked by the catalog row lock on continuous_view_1. Refresh2 on continuous_view_2 is not blocked and runs concurrently. +permutation "Setup2" "LockCAggCatalogRow_1" "Refresh1" "Refresh2" "UnlockCAggCatalogRow_1" -##test2 - continuous_view_2 should see results from insert but not the other one. -## Refresh2 will complete first due to LockMat1 and write the invalidation logs out. -permutation "Setup2" "Refresh1" "Refresh2" "Refresh1_sel" "Refresh2_sel" "LockMat1" "I2" "Refresh1" "Refresh2" "UnlockMat1" "Refresh1_sel" "Refresh2_sel" +# Test 2: continuous_view_2 should see results from the insert but not the other one. +# Refresh2 will complete first due to LockCAggCatalogRow_1 and write the invalidation logs out. +# Refresh1 will see these invalidation logs as well +permutation "Setup2" "Refresh1" "Refresh2" "Refresh1_sel" "Refresh2_sel" "LockCAggCatalogRow_1" "I1" "Refresh1" "Refresh2" "Refresh2_sel" "UnlockCAggCatalogRow_1" "Refresh1_sel" -#test3 - both see the updates i.e. the invalidations +# Test 3: Both see the updates i.e. the invalidations ##Refresh1 and Refresh2 are blocked by LockInvRow, when that is unlocked, they should complete serially permutation "Setup2" "Refresh1" "Refresh2" "Refresh1_sel" "Refresh2_sel" "U1" "U2" "LInvRow" "Refresh1" "Refresh2" "UnlockInvRow" "Refresh1_sel" "Refresh2_sel"