Skip to content

Commit 5c702b7

Browse files
kpan2034kpan2034
committed
Relax lock during CAgg invalidation log processing
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.
1 parent 11b0fd1 commit 5c702b7

10 files changed

Lines changed: 152 additions & 104 deletions

File tree

‎.unreleased/pr_9701‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
Implements: #9701 Relax lock during CAgg invalidation log processing

‎src/hypertable.c‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -445,7 +445,7 @@ hypertable_scan_limit_internal(ScanKeyData *scankey, int num_scankeys, int index
445445
/* update the tuple at this tid. The assumption is that we already hold a
446446
* tuple exclusive lock and no other transaction can modify this tuple
447447
* The sequence of operations for any update is:
448-
* lock the tuple using lock_hypertable_tuple.
448+
* lock the tuple using ts_lock_hypertable_tuple.
449449
* then update the required fields
450450
* call hypertable_update_catalog_tuple to complete the update.
451451
* This ensures correct tuple locking and tuple updates in the presence of
@@ -469,8 +469,8 @@ hypertable_update_catalog_tuple(ItemPointer tid, FormData_hypertable *update)
469469
relation_close(hypertable_rel, NoLock);
470470
}
471471

472-
static bool
473-
lock_hypertable_tuple(int32 htid, ItemPointer tid, FormData_hypertable *form)
472+
TSDLLEXPORT bool
473+
ts_lock_hypertable_tuple(int32 htid, ItemPointer tid, FormData_hypertable *form)
474474
{
475475
bool success = false;
476476
ScanTupLock scantuplock = {
@@ -854,7 +854,7 @@ ts_hypertable_set_name(Hypertable *ht, const char *newname)
854854
FormData_hypertable form;
855855
ItemPointerData tid;
856856
/* lock the tuple entry in the catalog table */
857-
bool found = lock_hypertable_tuple(ht->fd.id, &tid, &form);
857+
bool found = ts_lock_hypertable_tuple(ht->fd.id, &tid, &form);
858858
Ensure(found, "hypertable id %d not found", ht->fd.id);
859859

860860
namestrcpy(&form.table_name, newname);
@@ -868,7 +868,7 @@ ts_hypertable_set_schema(Hypertable *ht, const char *newname)
868868
FormData_hypertable form;
869869
ItemPointerData tid;
870870
/* lock the tuple entry in the catalog table */
871-
bool found = lock_hypertable_tuple(ht->fd.id, &tid, &form);
871+
bool found = ts_lock_hypertable_tuple(ht->fd.id, &tid, &form);
872872
Ensure(found, "hypertable id %d not found", ht->fd.id);
873873

874874
namestrcpy(&form.schema_name, newname);
@@ -882,7 +882,7 @@ ts_hypertable_set_num_dimensions(Hypertable *ht, int16 num_dimensions)
882882
FormData_hypertable form;
883883
ItemPointerData tid;
884884
/* lock the tuple entry in the catalog table */
885-
bool found = lock_hypertable_tuple(ht->fd.id, &tid, &form);
885+
bool found = ts_lock_hypertable_tuple(ht->fd.id, &tid, &form);
886886
Ensure(found, "hypertable id %d not found", ht->fd.id);
887887

888888
Assert(num_dimensions > 0);
@@ -2294,7 +2294,7 @@ ts_hypertable_set_compressed(Hypertable *ht, int32 compressed_hypertable_id)
22942294
FormData_hypertable form;
22952295
ItemPointerData tid;
22962296
/* lock the tuple entry in the catalog table */
2297-
bool found = lock_hypertable_tuple(ht->fd.id, &tid, &form);
2297+
bool found = ts_lock_hypertable_tuple(ht->fd.id, &tid, &form);
22982298
Ensure(found, "hypertable id %d not found", ht->fd.id);
22992299

23002300
Assert(!TS_HYPERTABLE_IS_INTERNAL_COMPRESSION_TABLE(ht));
@@ -2313,7 +2313,7 @@ ts_hypertable_unset_compressed(Hypertable *ht)
23132313
FormData_hypertable form;
23142314
ItemPointerData tid;
23152315
/* lock the tuple entry in the catalog table */
2316-
bool found = lock_hypertable_tuple(ht->fd.id, &tid, &form);
2316+
bool found = ts_lock_hypertable_tuple(ht->fd.id, &tid, &form);
23172317
Ensure(found, "hypertable id %d not found", ht->fd.id);
23182318

23192319
Assert(!TS_HYPERTABLE_IS_INTERNAL_COMPRESSION_TABLE(ht));
@@ -2538,7 +2538,7 @@ ts_hypertable_update_status_osm(Hypertable *ht)
25382538
FormData_hypertable form;
25392539
ItemPointerData tid;
25402540
/* lock the tuple entry in the catalog table */
2541-
bool found = lock_hypertable_tuple(ht->fd.id, &tid, &form);
2541+
bool found = ts_lock_hypertable_tuple(ht->fd.id, &tid, &form);
25422542
Ensure(found, "hypertable id %d not found", ht->fd.id);
25432543

25442544
if (form.status != ht->fd.status)
@@ -2555,7 +2555,7 @@ ts_hypertable_update_chunk_sizing(Hypertable *ht)
25552555
FormData_hypertable form;
25562556
ItemPointerData tid;
25572557
/* lock the tuple entry in the catalog table */
2558-
bool found = lock_hypertable_tuple(ht->fd.id, &tid, &form);
2558+
bool found = ts_lock_hypertable_tuple(ht->fd.id, &tid, &form);
25592559
Ensure(found, "hypertable id %d not found", ht->fd.id);
25602560

25612561
if (OidIsValid(ht->chunk_sizing_func))

‎src/hypertable.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -131,7 +131,9 @@ extern bool ts_hypertable_has_tablespace(const Hypertable *ht, Oid tspc_oid);
131131
extern Tablespace *ts_hypertable_select_tablespace(const Hypertable *ht, const Chunk *chunk);
132132
extern const char *ts_hypertable_select_tablespace_name(const Hypertable *ht, const Chunk *chunk);
133133
extern Tablespace *ts_hypertable_get_tablespace_at_offset_from(int32 hypertable_id,
134+
134135
Oid tablespace_oid, int16 offset);
136+
extern TSDLLEXPORT bool ts_lock_hypertable_tuple(int32 htid, ItemPointer tid, FormData_hypertable *form);
135137
extern TSDLLEXPORT bool ts_hypertable_has_chunks(Oid table_relid, LOCKMODE lockmode);
136138
extern void ts_hypertables_rename_schema_name(const char *old_name, const char *new_name);
137139
extern bool ts_is_partitioning_column(const Hypertable *ht, AttrNumber column_attno);

‎src/ts_catalog/continuous_agg.c‎

Lines changed: 50 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -138,6 +138,56 @@ init_materialization_invalidation_log_scan_by_materialization_id(ScanIterator *i
138138
Int32GetDatum(materialization_id));
139139
}
140140

141+
TSDLLEXPORT bool
142+
ts_lock_continuous_agg_tuple(int32 mat_hypertable_id)
143+
{
144+
bool success = false;
145+
ScanTupLock scantuplock = {
146+
.waitpolicy = LockWaitBlock,
147+
.lockmode = LockTupleExclusive,
148+
};
149+
ScanIterator iterator =
150+
ts_scan_iterator_create(CONTINUOUS_AGG, RowShareLock, CurrentMemoryContext);
151+
init_scan_by_mat_hypertable_id(&iterator, mat_hypertable_id);
152+
iterator.ctx.tuplock = &scantuplock;
153+
iterator.ctx.flags = SCANNER_F_KEEPLOCK;
154+
155+
/* see table_tuple_lock for details about flags that are set in TupleExclusive mode */
156+
scantuplock.lockflags = TUPLE_LOCK_FLAG_LOCK_UPDATE_IN_PROGRESS;
157+
if (!IsolationUsesXactSnapshot())
158+
{
159+
/* in read committed mode, we follow all updates to this tuple */
160+
scantuplock.lockflags |= TUPLE_LOCK_FLAG_FIND_LAST_VERSION;
161+
}
162+
163+
ts_scanner_foreach(&iterator)
164+
{
165+
TupleInfo *ti = ts_scan_iterator_tuple_info(&iterator);
166+
if (ti->lockresult != TM_Ok)
167+
{
168+
if (IsolationUsesXactSnapshot())
169+
{
170+
ereport(ERROR,
171+
(errcode(ERRCODE_T_R_SERIALIZATION_FAILURE),
172+
errmsg("could not serialize access due to concurrent update")));
173+
}
174+
else
175+
{
176+
ereport(ERROR,
177+
(errcode(ERRCODE_INTERNAL_ERROR),
178+
errmsg("unable to lock continuous aggregate catalog tuple, lock result "
179+
"is %d for mat_hypertable_id (%d)",
180+
ti->lockresult,
181+
mat_hypertable_id)));
182+
}
183+
}
184+
success = true;
185+
break;
186+
}
187+
ts_scan_iterator_close(&iterator);
188+
return success;
189+
}
190+
141191
static int32
142192
number_of_continuous_aggs_attached(int32 raw_hypertable_id)
143193
{

‎src/ts_catalog/continuous_agg.h‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -145,6 +145,8 @@ typedef struct ContinuousAggPolicyOffset
145145
const char *name;
146146
} ContinuousAggPolicyOffset;
147147

148+
extern TSDLLEXPORT bool ts_lock_continuous_agg_tuple(int32 mat_hypertable_id);
149+
148150
extern TSDLLEXPORT Oid ts_cagg_permissions_check(Oid cagg_oid, Oid userid);
149151

150152
extern TSDLLEXPORT ContinuousAggInfo ts_continuous_agg_get_all_caggs_info(int32 raw_hypertable_id);

‎tsl/src/continuous_aggs/refresh.c‎

Lines changed: 5 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -701,15 +701,12 @@ process_cagg_invalidations_and_refresh(const ContinuousAgg *cagg,
701701
const ContinuousAggRefreshContext context,
702702
bool bucketing_refresh_window, bool force)
703703
{
704-
Oid hyper_relid = ts_hypertable_id_to_relid(cagg->data.mat_hypertable_id, false);
705-
706-
/* Lock the continuous aggregate's materialized hypertable to protect against
707-
* concurrent invalidation log processing.
708-
*
709-
* This is supposed to be a short transaction and in the future we can consider
710-
* relaxing this lock.
704+
/* Lock the continuous aggregate's catalog table entry to protect against concurrent refreshes
705+
* on the same cagg processing the cagg invalidation logs for that CAgg.
711706
*/
712-
LockRelationOid(hyper_relid, ShareUpdateExclusiveLock);
707+
bool found = ts_lock_continuous_agg_tuple(cagg->data.mat_hypertable_id);
708+
Ensure(found, "continuous aggregate with mat_hypertable_id %d not found",
709+
cagg->data.mat_hypertable_id);
713710
invalidation_process_cagg_log(cagg, refresh_window);
714711

715712
DEBUG_ERROR_INJECTION("cagg_refresh_fail_in_txn2");

‎tsl/test/isolation/expected/cagg_hierarchical_concurrent_refresh.out‎

Lines changed: 33 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -463,7 +463,7 @@ Fri Jan 02 00:00:00 2026 UTC|32.1666666666667|32.1666666666667| 288|
463463
Sat Jan 03 00:00:00 2026 UTC| 31.5| 31.5| 96| 96|t |t
464464

465465

466-
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
466+
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
467467
step L1_refresh_full:
468468
CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04');
469469

@@ -511,6 +511,16 @@ source |lowest |greatest
511511
-------+----------------------------+----------------------------
512512
cagg_6h|Thu Jan 01 00:00:00 2026 UTC|Sat Jan 03 18:00:00 2026 UTC
513513

514+
step lock_L2_source:
515+
BEGIN;
516+
DO $$
517+
BEGIN
518+
PERFORM 1 FROM _timescaledb_catalog.continuous_agg
519+
WHERE user_view_name = 'cagg_1d'
520+
FOR UPDATE;
521+
END;
522+
$$;
523+
514524
step L2_refresh_jan1:
515525
CALL refresh_continuous_aggregate('cagg_1d', '2026-01-01', '2026-01-02');
516526
<waiting ...>
@@ -523,20 +533,6 @@ step insert_ht:
523533
step L1_refresh_full:
524534
CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04');
525535
<waiting ...>
526-
step lock_L2_source:
527-
BEGIN;
528-
DO $$
529-
DECLARE
530-
mat_ht text;
531-
BEGIN
532-
SELECT format('%I.%I', h.schema_name, h.table_name) INTO mat_ht
533-
FROM _timescaledb_catalog.continuous_agg ca
534-
JOIN _timescaledb_catalog.hypertable h ON h.id = ca.mat_hypertable_id
535-
WHERE ca.user_view_name = 'cagg_1d';
536-
EXECUTE format('LOCK TABLE %s IN ACCESS EXCLUSIVE MODE', mat_ht);
537-
END;
538-
$$;
539-
<waiting ...>
540536
step L2b_refresh_jan1_2:
541537
CALL refresh_continuous_aggregate('cagg_1d', '2026-01-01', '2026-01-03');
542538

@@ -549,7 +545,6 @@ debug_waitpoint_release
549545

550546

551547
step L1_refresh_full: <... completed>
552-
step lock_L2_source: <... completed>
553548
step unlock:
554549
ROLLBACK;
555550

@@ -590,7 +585,7 @@ Fri Jan 02 00:00:00 2026 UTC| 31.5|32.1666666666667| 96|
590585
Sat Jan 03 00:00:00 2026 UTC| 31.5| 40.5| 96| 288|f |f
591586

592587

593-
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
588+
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
594589
step L1_refresh_full:
595590
CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04');
596591

@@ -638,6 +633,16 @@ source |lowest |greatest
638633
-------+----------------------------+----------------------------
639634
cagg_6h|Thu Jan 01 00:00:00 2026 UTC|Sat Jan 03 18:00:00 2026 UTC
640635

636+
step lock_L2_source:
637+
BEGIN;
638+
DO $$
639+
BEGIN
640+
PERFORM 1 FROM _timescaledb_catalog.continuous_agg
641+
WHERE user_view_name = 'cagg_1d'
642+
FOR UPDATE;
643+
END;
644+
$$;
645+
641646
step L2b_refresh_jan1_2:
642647
CALL refresh_continuous_aggregate('cagg_1d', '2026-01-01', '2026-01-03');
643648
<waiting ...>
@@ -650,20 +655,6 @@ step insert_ht:
650655
step L1_refresh_full:
651656
CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04');
652657
<waiting ...>
653-
step lock_L2_source:
654-
BEGIN;
655-
DO $$
656-
DECLARE
657-
mat_ht text;
658-
BEGIN
659-
SELECT format('%I.%I', h.schema_name, h.table_name) INTO mat_ht
660-
FROM _timescaledb_catalog.continuous_agg ca
661-
JOIN _timescaledb_catalog.hypertable h ON h.id = ca.mat_hypertable_id
662-
WHERE ca.user_view_name = 'cagg_1d';
663-
EXECUTE format('LOCK TABLE %s IN ACCESS EXCLUSIVE MODE', mat_ht);
664-
END;
665-
$$;
666-
<waiting ...>
667658
step L2_refresh_jan1:
668659
CALL refresh_continuous_aggregate('cagg_1d', '2026-01-01', '2026-01-02');
669660

@@ -676,7 +667,6 @@ debug_waitpoint_release
676667

677668

678669
step L1_refresh_full: <... completed>
679-
step lock_L2_source: <... completed>
680670
step unlock:
681671
ROLLBACK;
682672

@@ -872,7 +862,7 @@ Fri Jan 02 00:00:00 2026 UTC| 31.5| 31.5| 96|
872862
Sat Jan 03 00:00:00 2026 UTC| 40.5| 40.5| 288| 288|t |t
873863

874864

875-
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
865+
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
876866
step L1_refresh_full:
877867
CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04');
878868

@@ -920,6 +910,16 @@ source |lowest |greatest
920910
-------+----------------------------+----------------------------
921911
cagg_6h|Thu Jan 01 00:00:00 2026 UTC|Sat Jan 03 18:00:00 2026 UTC
922912

913+
step lock_L2_source:
914+
BEGIN;
915+
DO $$
916+
BEGIN
917+
PERFORM 1 FROM _timescaledb_catalog.continuous_agg
918+
WHERE user_view_name = 'cagg_1d'
919+
FOR UPDATE;
920+
END;
921+
$$;
922+
923923
step L2_refresh_jan1:
924924
CALL refresh_continuous_aggregate('cagg_1d', '2026-01-01', '2026-01-02');
925925
<waiting ...>
@@ -932,20 +932,6 @@ step insert_ht:
932932
step L1_refresh_full:
933933
CALL refresh_continuous_aggregate('cagg_6h', '2026-01-01', '2026-01-04');
934934
<waiting ...>
935-
step lock_L2_source:
936-
BEGIN;
937-
DO $$
938-
DECLARE
939-
mat_ht text;
940-
BEGIN
941-
SELECT format('%I.%I', h.schema_name, h.table_name) INTO mat_ht
942-
FROM _timescaledb_catalog.continuous_agg ca
943-
JOIN _timescaledb_catalog.hypertable h ON h.id = ca.mat_hypertable_id
944-
WHERE ca.user_view_name = 'cagg_1d';
945-
EXECUTE format('LOCK TABLE %s IN ACCESS EXCLUSIVE MODE', mat_ht);
946-
END;
947-
$$;
948-
<waiting ...>
949935
step L2b_refresh_jan3:
950936
CALL refresh_continuous_aggregate('cagg_1d', '2026-01-03', '2026-01-04');
951937
<waiting ...>
@@ -957,7 +943,6 @@ debug_waitpoint_release
957943

958944

959945
step L1_refresh_full: <... completed>
960-
step lock_L2_source: <... completed>
961946
step unlock:
962947
ROLLBACK;
963948

0 commit comments

Comments
 (0)