Incremental refresh for refresh_continuous_aggregate()

refresh_continuous_aggregate() can now refresh incrementally in batches,
matching the behavior the continuous aggregate policy already supports.
Batching is opt-in for refresh_continuous_aggregate().

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

  - `buckets_per_batch`         (default 0 = non-incremental)
  - `max_batches_per_execution` (default 0 = no limit)
  - `refresh_newest_first`

Usage:

  Without options, it still behaves as previously:

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

  Incremental manual refresh requires `buckets_per_batch` to be set
  larger than 0:

      CALL refresh_continuous_aggregate(
          'cagg', t1, t2,
          options => '{"buckets_per_batch": 10}'::jsonb);

Co-authored-by: Keyur Panchal <keyur@timescale.com>

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

@@ -442,67 +442,19 @@ 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,
-											policy_data.refresh_newest_first);
-	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);
-
-	ListCell *lc;
-	int32 processing_batch = 0;
-	foreach (lc, refresh_window_list)
-	{
-		InternalTimeRange *refresh_window = (InternalTimeRange *) lfirst(lc);
-		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_internal(policy_data.cagg,
+									&policy_data.refresh_window,
+									context,
+									false, /* force */
+									extend_last_bucket);
 
 	if (!policy_data.include_tiered_data_isnull)
 	{
@@ -553,6 +505,7 @@ policy_refresh_cagg_execute(int32 job_id, Jsonb *config)
 			}
 
 			/* Process each chunk in its own transaction */
+			ListCell *lc;
 			foreach (lc, chunkid_lst)
 			{
 				PushActiveSnapshot(GetTransactionSnapshot());

tsl/src/continuous_aggs/common.h

@@ -89,6 +89,10 @@ 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;
 } ContinuousAggRefreshContext;
 
 #define IS_TIME_BUCKET_INFO_TIME_BASED(bucket_function)                                            \

tsl/src/continuous_aggs/create.c

@@ -891,17 +891,16 @@ tsl_process_continuous_agg_viewstmt(Node *node, const char *query_string, void *
 		 * - ts_compute_circumscribed_bucketed_refresh_window_variable()
 		 */
 		refresh_window.start = cagg_get_time_min(cagg);
+		refresh_window.start_isnull = true;
 		refresh_window.end = ts_time_get_noend_or_max(refresh_window.type);
+		refresh_window.end_isnull = true;
 
 		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*/);
+										false /* extend_last_bucket */);
 	}
 
 	return DDL_DONE;