Incremental manual refresh support

refresh_continuous_aggregate() can now refresh incrementally in batches,
matching the behavior the continuous aggregate policy already supports.
Forced refresh is also incremental by default.

Options available for incremental refresh with policies can be set for
manual refresh via JSONB options:

  - `buckets_per_batch`         (default 10, as policies)
  - `max_batches_per_execution` (default 0 = no limit)
  - `refresh_newest_first`

Usage:

  Without options, it behaves similar to a refresh policy with default
  values:

      CALL refresh_continuous_aggregate('cagg', t1, t2);

Author: melihmutlu

.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;