Use continuous aggregate catalog table for tuple lock instead

Author: kpan2034

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

@@ -643,21 +643,12 @@ process_cagg_invalidations_and_refresh(const ContinuousAgg *cagg,
 									   const ContinuousAggRefreshContext context,
 									   bool bucketing_refresh_window, bool force)
 {
-	/* Lock the materialized hypertable's catalog table entry to protect against concurrent refreshes on the same cagg from 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.
 	 */
-	Cache *hcache = ts_hypertable_cache_pin();
-	Hypertable *mat_ht =
-		ts_hypertable_cache_get_entry_by_id(hcache, cagg->data.mat_hypertable_id);
-	ts_cache_release(&hcache);
-
-	FormData_hypertable form;
-	ItemPointerData tid;
-	/* lock the tuple entry in the catalog table */
-	bool found = ts_lock_hypertable_tuple(mat_ht->fd.id, &tid, &form);
-	Ensure(found, "hypertable id %d not found", mat_ht->fd.id);
+	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");