code formatting

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,13 +21,13 @@
 #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"
 #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 */

tsl/src/continuous_aggs/create.c

@@ -900,7 +900,7 @@ tsl_process_continuous_agg_viewstmt(Node *node, const char *query_string, void *
 										&refresh_window,
 										context,
 										false, /* force */
-										false  /* extend_last_bucket */);
+										false /* extend_last_bucket */);
 	}
 
 	return DDL_DONE;

tsl/src/continuous_aggs/refresh.c

@@ -683,34 +683,44 @@ continuous_agg_refresh(PG_FUNCTION_ARGS)
 
 		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,
@@ -1125,8 +1135,7 @@ continuous_agg_refresh_single_window(const ContinuousAgg *cagg_arg,
  * own independent SPI session and commits independently (non-atomic).
  */
 void
-continuous_agg_refresh_internal(const ContinuousAgg *cagg,
-								const InternalTimeRange *refresh_window,
+continuous_agg_refresh_internal(const ContinuousAgg *cagg, const InternalTimeRange *refresh_window,
 								ContinuousAggRefreshContext context, bool force,
 								bool extend_last_bucket)
 {
@@ -1144,12 +1153,11 @@ continuous_agg_refresh_internal(const ContinuousAgg *cagg,
 
 	if (context.buckets_per_batch > 0)
 	{
-		refresh_window_list =
-			continuous_agg_split_refresh_window(cagg,
-												refresh_window,
-												context.buckets_per_batch,
-												context.refresh_newest_first,
-												force);
+		refresh_window_list = continuous_agg_split_refresh_window(cagg,
+																  refresh_window,
+																  context.buckets_per_batch,
+																  context.refresh_newest_first,
+																  force);
 	}
 
 	if (refresh_window_list == NIL)
@@ -1162,7 +1170,9 @@ continuous_agg_refresh_internal(const ContinuousAgg *cagg,
 		/* Batched windows are already bucket-aligned by the split function */
 		bucketing_refresh_window = false;
 		if (context.callctx == CAGG_REFRESH_POLICY)
+		{
 			context.callctx = CAGG_REFRESH_POLICY_BATCHED;
+		}
 	}
 
 	context.number_of_batches = list_length(refresh_window_list);
@@ -1186,11 +1196,13 @@ continuous_agg_refresh_internal(const ContinuousAgg *cagg,
 											processing_batch == context.number_of_batches);
 
 		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,
@@ -1356,14 +1368,12 @@ continuous_agg_split_refresh_window(const ContinuousAgg *cagg,
 	 * 5. If the batch overlaps with both then it's a valid batch to be processed
 	 * 6. If the batch overlaps with only one of them then it's not a valid batch to be processed
 	 * 7. If the batch does not overlap with any of them then it's not a valid batch to be processed
-	 */
-	/*
+	 *
 	 * When force is true, skip the invalidation log filter since forced refresh
 	 * synthesizes invalidations per-window and there may be no existing log entries.
 	 * Only filter on dimension slices (chunks with data).
 	 */
-	const char *query_str_template = force ?
-		" \
+	const char *query_str_template = force ? " \
 		WITH dimension_slices AS ( \
 			SELECT \
 				range_start AS start, \
@@ -1392,9 +1402,8 @@ continuous_agg_split_refresh_window(const ContinuousAgg *cagg,
 					pg_catalog.int8range(dimension_slices.start, dimension_slices.end) \
 			) \
 		ORDER BY \
-			refresh_start %s;"
-		:
-		" \
+			refresh_start %s;" :
+											 " \
 		WITH dimension_slices AS ( \
 			SELECT \
 				range_start AS start, \

tsl/src/continuous_aggs/refresh.h

@@ -25,8 +25,8 @@ extern void continuous_agg_refresh_internal(const ContinuousAgg *cagg,
 											bool extend_last_bucket);
 extern List *continuous_agg_split_refresh_window(const ContinuousAgg *cagg,
 												 const InternalTimeRange *original_refresh_window,
-												 int32 buckets_per_batch,
-												 bool refresh_newest_first, bool force);
+												 int32 buckets_per_batch, bool refresh_newest_first,
+												 bool force);
 InternalTimeRange
 compute_circumscribed_bucketed_refresh_window(const InternalTimeRange *const refresh_window,
 											  const ContinuousAggBucketFunction *bucket_function);