Bucket options for refresh_continuous_aggregate()

Author: melihmutlu

tsl/src/bgw_policy/job.c

@@ -453,8 +453,6 @@ policy_refresh_cagg_execute(int32 job_id, Jsonb *config)
 	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);
 

tsl/src/continuous_aggs/create.c

@@ -891,14 +891,14 @@ 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 */
 										false, /* force */
 										false  /* extend_last_bucket */);
 	}

tsl/src/continuous_aggs/refresh.c

@@ -18,6 +18,7 @@
 #include <utils/lsyscache.h>
 #include <utils/snapmgr.h>
 
+#include "bgw_policy/policies_v2.h"
 #include "debug_point.h"
 #include "dimension.h"
 #include "dimension_slice.h"
@@ -666,16 +667,60 @@ continuous_agg_refresh(PG_FUNCTION_ARGS)
 		refresh_window.end_isnull = true;
 	}
 
+	/*
+	 * For manual refreshes batching is opt-in: callers must set
+	 * buckets_per_batch in the options JSONB to trigger incremental refresh.
+	 * Default 0 = atomic single-pass refresh (keeps the previous behaviour).
+	 */
+	int32 buckets_per_batch = 0;
+	int32 max_batches_per_execution = 0;
+	bool refresh_newest_first = DEFAULT_REFRESH_NEWEST_FIRST;
+
+	if (!PG_ARGISNULL(4))
+	{
+		Jsonb *options = PG_GETARG_JSONB_P(4);
+		bool found;
+
+		int32 v = ts_jsonb_get_int32_field(options, POL_REFRESH_CONF_KEY_BUCKETS_PER_BATCH, &found);
+		if (found)
+			buckets_per_batch = v;
+
+		v = ts_jsonb_get_int32_field(options,
+									 POL_REFRESH_CONF_KEY_MAX_BATCHES_PER_EXECUTION,
+									 &found);
+		if (found)
+			max_batches_per_execution = v;
+
+		bool b =
+			ts_jsonb_get_bool_field(options, POL_REFRESH_CONF_KEY_REFRESH_NEWEST_FIRST, &found);
+		if (found)
+			refresh_newest_first = b;
+	}
+
+	if (buckets_per_batch < 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("invalid buckets per batch"),
+				 errdetail("buckets_per_batch: %d", buckets_per_batch),
+				 errhint("The buckets per batch should be greater than or equal to zero.")));
+
+	if (max_batches_per_execution < 0)
+		ereport(ERROR,
+				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
+				 errmsg("invalid max batches per execution"),
+				 errdetail("max_batches_per_execution: %d", max_batches_per_execution),
+				 errhint(
+					 "The max batches per execution should be greater than or equal to zero.")));
+
 	ContinuousAggRefreshContext context = {
 		.callctx = CAGG_REFRESH_WINDOW,
-		.buckets_per_batch = DEFAULT_BUCKETS_PER_BATCH,
-		.refresh_newest_first = DEFAULT_REFRESH_NEWEST_FIRST,
+		.buckets_per_batch = buckets_per_batch,
+		.max_batches_per_execution = max_batches_per_execution,
+		.refresh_newest_first = refresh_newest_first,
 	};
 	continuous_agg_refresh_internal(cagg,
 									&refresh_window,
 									context,
-									PG_ARGISNULL(1),
-									PG_ARGISNULL(2),
 									force,
 									false /*extend_last_bucket*/);
 
@@ -1082,8 +1127,8 @@ continuous_agg_refresh_single_window(const ContinuousAgg *cagg_arg,
 void
 continuous_agg_refresh_internal(const ContinuousAgg *cagg,
 								const InternalTimeRange *refresh_window,
-								ContinuousAggRefreshContext context, const bool start_isnull,
-								const bool end_isnull, bool force, bool extend_last_bucket)
+								ContinuousAggRefreshContext context, bool force,
+								bool extend_last_bucket)
 {
 	/* Like regular materialized views, require owner to refresh. */
 	if (!object_ownercheck(RelationRelationId, cagg->relid, GetUserId()))
@@ -1140,11 +1185,12 @@ continuous_agg_refresh_internal(const ContinuousAgg *cagg,
 			(context.refresh_newest_first ? processing_batch == 1 :
 											processing_batch == context.number_of_batches);
 
-		elog(DEBUG1,
-			 "refreshing continuous aggregate \"%s\" in window [ %s, %s ]",
-			 NameStr(cagg->data.user_view_name),
-			 ts_internal_to_time_string(batch_window->start, batch_window->type),
-			 ts_internal_to_time_string(batch_window->end, batch_window->type));
+		if (context.number_of_batches > 1)
+			elog(DEBUG1,
+				 "refreshing continuous aggregate \"%s\" in window [ %s, %s ]",
+				 NameStr(cagg->data.user_view_name),
+				 ts_internal_to_time_string(batch_window->start, batch_window->type),
+				 ts_internal_to_time_string(batch_window->end, batch_window->type));
 
 		bool refreshed = continuous_agg_refresh_single_window(cagg,
 															  batch_window,

tsl/src/continuous_aggs/refresh.h

@@ -21,9 +21,8 @@ Hypertable *cagg_get_hypertable_or_fail(int32 hypertable_id);
 extern Datum continuous_agg_refresh(PG_FUNCTION_ARGS);
 extern void continuous_agg_refresh_internal(const ContinuousAgg *cagg,
 											const InternalTimeRange *refresh_window,
-											ContinuousAggRefreshContext context,
-											const bool start_isnull, const bool end_isnull,
-											bool force, bool extend_last_bucket);
+											ContinuousAggRefreshContext context, bool force,
+											bool extend_last_bucket);
 extern List *continuous_agg_split_refresh_window(const ContinuousAgg *cagg,
 												 const InternalTimeRange *original_refresh_window,
 												 int32 buckets_per_batch,