Implement incremental refresh for manual refreshes

Author: kpan2034

tsl/src/bgw_policy/continuous_aggregate_api.c

@@ -27,17 +27,14 @@
 #include "policy_config.h"
 #include "policy_utils.h"
 #include "time_utils.h"
+#include "continuous_aggs/refresh.h"
 #include "ts_catalog/continuous_agg.h"
 
 /* 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,21 @@ 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,
+									policy_data.refresh_window.start_isnull,
+									policy_data.refresh_window.end_isnull,
+									false, /* force */
+									extend_last_bucket);
 
 	if (!policy_data.include_tiered_data_isnull)
 	{
@@ -553,6 +507,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

@@ -899,9 +899,8 @@ tsl_process_continuous_agg_viewstmt(Node *node, const char *query_string, void *
 										context,
 										true,  /* start_isnull */
 										true,  /* end_isnull */
-										true,  /* bucketing_refresh_window */
 										false, /* force */
-										false /*extend_last_bucket*/);
+										false  /* extend_last_bucket */);
 	}
 
 	return DDL_DONE;