Incremental refresh for refresh_continuous_aggregate()

.unreleased/pr_9903

@@ -0,0 +1 @@
+Implements: #9903 Incremental refresh for refresh_continuous_aggregate()

tsl/src/bgw_policy/continuous_aggregate_api.c

@@ -21,6 +21,7 @@
 #include "bgw_policy/job_api.h"
 #include "bgw_policy/policies_v2.h"
 #include "bgw_policy/policy_utils.h"
+#include "continuous_aggs/refresh.h"
 #include "dimension.h"
 #include "guc.h"
 #include "jsonb_utils.h"
@@ -32,12 +33,8 @@
 /* Default max runtime for a continuous aggregate jobs is unlimited for now */
 #define DEFAULT_MAX_RUNTIME                                                                        \
 	DatumGetIntervalP(DirectFunctionCall3(interval_in, CStringGetDatum("0"), InvalidOid, -1))
-/* Default buckets per batch is 1, which means that the job will refresh 1 bucket at a time */
-#define DEFAULT_BUCKETS_PER_BATCH 10
 /* Default max batches per execution is 0, which means no limit */
 #define DEFAULT_MAX_BATCHES_PER_EXECUTION 0
-/* Default refresh newest first is true, which means from newest data to the oldest */
-#define DEFAULT_REFRESH_NEWEST_FIRST true
 /* Default compress after refresh is false, which means compression does not run after refresh */
 #define DEFAULT_COMPRESS_AFTER_REFRESH false
 

tsl/src/bgw_policy/job.c

@@ -422,74 +422,18 @@ policy_refresh_cagg_execute(int32 job_id, Jsonb *config)
 						PGC_S_SESSION);
 	}
 
-	ContinuousAggRefreshContext context = { .callctx = CAGG_REFRESH_POLICY, .job_id = job_id };
-
-	/* Try to split window range into a list of ranges */
-	List *refresh_window_list = continuous_agg_split_refresh_window(policy_data.cagg,
-																	&policy_data.refresh_window,
-																	policy_data.buckets_per_batch);
-	if (refresh_window_list == NIL)
-	{
-		refresh_window_list = lappend(refresh_window_list, &policy_data.refresh_window);
-	}
-	else
-	{
-		context.callctx = CAGG_REFRESH_POLICY_BATCHED;
-	}
-
-	context.number_of_batches = list_length(refresh_window_list);
-
-	/*
-	 * The list is always built oldest-first. When refresh_newest_first is true we
-	 * iterate from the last element down to the first using index-based access so
-	 * that no reversal copy of the list is needed.
-	 */
-	int32 processing_batch = 0;
-	int32 nbatches = list_length(refresh_window_list);
-	int32 batch_start = policy_data.refresh_newest_first ? nbatches - 1 : 0;
-	int32 batch_end = policy_data.refresh_newest_first ? -1 : nbatches;
-	int32 batch_step = policy_data.refresh_newest_first ? -1 : 1;
-	for (int32 batch_idx = batch_start; batch_idx != batch_end; batch_idx += batch_step)
-	{
-		InternalTimeRange *refresh_window =
-			(InternalTimeRange *) list_nth(refresh_window_list, batch_idx);
-		elog(DEBUG1,
-			 "refreshing continuous aggregate \"%s\" from %s to %s",
-			 NameStr(policy_data.cagg->data.user_view_name),
-			 ts_internal_to_time_string(refresh_window->start, refresh_window->type),
-			 ts_internal_to_time_string(refresh_window->end, refresh_window->type));
-
-		context.processing_batch = ++processing_batch;
-
-		/* extend_last_bucket must only apply to the boundary batch — the one
-		 * whose window abuts the adjacent policy.  For newest-first ordering
-		 * that is batch 1; for oldest-first it is the final batch.
-		 * In non-batched mode (single batch) the one batch is always the boundary. */
-		bool apply_extend =
-			extend_last_bucket &&
-			(policy_data.refresh_newest_first ? processing_batch == 1 :
-												processing_batch == context.number_of_batches);
-
-		continuous_agg_refresh_internal(policy_data.cagg,
-										refresh_window,
-										context,
-										refresh_window->start_isnull,
-										refresh_window->end_isnull,
-										(context.callctx != CAGG_REFRESH_POLICY_BATCHED),
-										false, /* force */
-										apply_extend);
-		DEBUG_ERROR_INJECTION(psprintf("cagg_policy_batch_%d_after_refresh", processing_batch));
-		if (processing_batch >= policy_data.max_batches_per_execution &&
-			processing_batch < context.number_of_batches &&
-			policy_data.max_batches_per_execution > 0)
-		{
-			elog(LOG,
-				 "reached maximum number of batches per execution (%d), batches not processed (%d)",
-				 policy_data.max_batches_per_execution,
-				 context.number_of_batches - processing_batch);
-			break;
-		}
-	}
+	ContinuousAggRefreshContext context = {
+		.callctx = CAGG_REFRESH_POLICY,
+		.job_id = job_id,
+		.buckets_per_batch = policy_data.buckets_per_batch,
+		.max_batches_per_execution = policy_data.max_batches_per_execution,
+		.refresh_newest_first = policy_data.refresh_newest_first,
+	};
+
+	continuous_agg_refresh_batched(policy_data.cagg,
+								   &policy_data.refresh_window,
+								   context,
+								   extend_last_bucket);
 
 	if (!policy_data.include_tiered_data_isnull)
 	{

tsl/src/continuous_aggs/common.c

@@ -2102,3 +2102,21 @@ cagg_find_groupingcols(ContinuousAgg *agg, Hypertable *mat_ht)
 	}
 	return retlist;
 }
+
+void
+emit_up_to_date_notice(const ContinuousAgg *cagg, const ContinuousAggRefreshContext context)
+{
+	switch (context.callctx)
+	{
+		case CAGG_REFRESH_WINDOW:
+		case CAGG_REFRESH_CREATION:
+		case CAGG_REFRESH_WINDOW_BATCHED:
+			elog(NOTICE,
+				 "continuous aggregate \"%s\" is already up-to-date",
+				 NameStr(cagg->data.user_view_name));
+			break;
+		case CAGG_REFRESH_POLICY:
+		case CAGG_REFRESH_POLICY_BATCHED:
+			break;
+	}
+}

tsl/src/continuous_aggs/common.h

@@ -79,6 +79,7 @@ typedef enum ContinuousAggRefreshCallContext
 {
 	CAGG_REFRESH_CREATION,
 	CAGG_REFRESH_WINDOW,
+	CAGG_REFRESH_WINDOW_BATCHED,
 	CAGG_REFRESH_POLICY,
 	CAGG_REFRESH_POLICY_BATCHED
 } ContinuousAggRefreshCallContext;
@@ -89,6 +90,11 @@ typedef struct ContinuousAggRefreshContext
 	int32 job_id;
 	int32 processing_batch;
 	int32 number_of_batches;
+	/* Batch configuration */
+	int32 buckets_per_batch;		 /* 0 = disabled */
+	int32 max_batches_per_execution; /* 0 = no limit */
+	bool refresh_newest_first;
+	bool force; /* re-materialize the whole window, ignoring invalidations */
 } ContinuousAggRefreshContext;
 
 #define IS_TIME_BUCKET_INFO_TIME_BASED(bucket_function)                                            \
@@ -173,3 +179,5 @@ extern bool caggtimebucket_validate_common(ContinuousAggBucketFunction *bf, List
 										   List *targetList, List *rtable, int ht_partcolno,
 										   StringInfo msg, bool is_cagg_create,
 										   const bool for_rewrites);
+extern void emit_up_to_date_notice(const ContinuousAgg *cagg,
+								   const ContinuousAggRefreshContext context);

tsl/src/continuous_aggs/create.c

@@ -857,8 +857,11 @@ tsl_process_continuous_agg_viewstmt(Node *node, const char *query_string, void *
 
 	if (!stmt->into->skipData)
 	{
+		bool refreshed = false;
 		InternalTimeRange refresh_window = {
 			.type = InvalidOid,
+			.start_isnull = true,
+			.end_isnull = true,
 		};
 
 		/*
@@ -887,14 +890,15 @@ tsl_process_continuous_agg_viewstmt(Node *node, const char *query_string, void *
 		refresh_window.end = ts_time_get_noend_or_max(refresh_window.type);
 
 		ContinuousAggRefreshContext context = { .callctx = CAGG_REFRESH_CREATION };
-		continuous_agg_refresh_internal(cagg,
-										&refresh_window,
-										context,
-										true,  /* start_isnull */
-										true,  /* end_isnull */
-										true,  /* bucketing_refresh_window */
-										false, /* force */
-										false /*extend_last_bucket*/);
+		refreshed = continuous_agg_refresh_internal(cagg,
+													&refresh_window,
+													context,
+													true, /* bucketing_refresh_window */
+													false /*extend_last_bucket*/);
+		if (!refreshed)
+		{
+			emit_up_to_date_notice(cagg, context);
+		}
 	}
 
 	return DDL_DONE;