Fix batch cutting for incremental refresh

Currently, incremental refresh uses 30days as the length
for the monthly bucket, which results in batches misaligned with
bucket boundary. Circumscription extends the batch to match
bucket boundary to ensure correctness, but leads to overlapping
between adjacent batches.

This patch fixes the batch cutting to use the exact number of days
in a calendar month.

The patch also makes the following modifications:
-  Generated batches will start right at the latest bucket that overlaps
   both a chunk and invalidations. Before, we generated batches first
   and picked those with overlapping, resulting in batches that started
   earlier than needed. For example, if invalidations is (-inf , +inf)
   and chunks are [10,30) , [50,70). With refresh windows [0,90),
   bucket = 10, and bucket_per_batch = 2, previously we would enumerate
   batches of [0,20), [20,40], [40,60), [60,80), [80,100), then select
   batches [0,20), [20,40), [40,60), [60,80) because these overlap with
   chunk slices and invalidations. With the new logic, we skip forward
   to the next max(dimension slice, invalidations), so we  only generate
   batches [10,30) and [50,70). This explains some test diffs on existing
   tests, because we will have fewer batches in some cases.

- Bug fix: Previously, when the refresh window had null start/end, we
  capped it using the min/max dimension slice, then inscribed the
  window. So if the chunk boundary is not at the bucket boundary, we
  will generate batches based on a window range that is smaller than
  the original window. We restored null at the end so it's still correct
  with respect to the data range being refreshed, but the batches at
  the two ends could be a bucket bigger. Fix by capping to the start/end
  of the bucket that contains the boundary.

Author: pnthao

.unreleased/pr_9890

@@ -0,0 +1 @@
+Fixes: #9890 Fix incremental refresh batch boundaries to align with variable-width buckets and start only where a chunk and an invalidation overlap

src/dimension_slice.c

@@ -514,7 +514,7 @@ ts_dimension_slice_scan_list(int32 dimension_id, int64 coordinate, List **matchi
 										CurrentMemoryContext);
 }
 
-int
+TSDLLEXPORT int
 ts_dimension_slice_scan_iterator_set_range(ScanIterator *it, int32 dimension_id,
 										   StrategyNumber start_strategy, int64 start_value,
 										   StrategyNumber end_strategy, int64 end_value)
@@ -970,7 +970,7 @@ ts_dimension_slice_scan_by_id_and_lock(int32 dimension_slice_id, const ScanTupLo
 	return slice;
 }
 
-ScanIterator
+TSDLLEXPORT ScanIterator
 ts_dimension_slice_scan_iterator_create(const ScanTupLock *tuplock, MemoryContext result_mcxt)
 {
 	ScanIterator it = ts_scan_iterator_create(DIMENSION_SLICE, AccessShareLock, result_mcxt);

src/dimension_slice.h

@@ -98,15 +98,15 @@ extern TSDLLEXPORT List *ts_dimension_slice_get_chunkids_to_compress(
 	StrategyNumber end_strategy, int64 end_value, bool compress, bool recompress, int32 numchunks);
 
 extern DimensionSlice *ts_dimension_slice_from_tuple(TupleInfo *ti);
-extern ScanIterator ts_dimension_slice_scan_iterator_create(const ScanTupLock *tuplock,
-															MemoryContext result_mcxt);
+extern TSDLLEXPORT ScanIterator ts_dimension_slice_scan_iterator_create(const ScanTupLock *tuplock,
+																		MemoryContext result_mcxt);
 extern void ts_dimension_slice_scan_iterator_set_slice_id(ScanIterator *it, int32 slice_id);
 extern DimensionSlice *ts_dimension_slice_scan_iterator_get_by_id(ScanIterator *it, int32 slice_id);
 
-extern int ts_dimension_slice_scan_iterator_set_range(ScanIterator *it, int32 dimension_id,
-													  StrategyNumber start_strategy,
-													  int64 start_value,
-													  StrategyNumber end_strategy, int64 end_value);
+extern TSDLLEXPORT int
+ts_dimension_slice_scan_iterator_set_range(ScanIterator *it, int32 dimension_id,
+										   StrategyNumber start_strategy, int64 start_value,
+										   StrategyNumber end_strategy, int64 end_value);
 
 extern bool ts_osm_chunk_range_overlaps(int32 osm_dimension_slice_id, int32 dimension_id,
 										int64 range_start, int64 range_end);

tsl/src/bgw_policy/job.c

@@ -445,11 +445,9 @@ policy_refresh_cagg_execute(int32 job_id, Jsonb *config)
 	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);
+	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);
@@ -461,11 +459,20 @@ policy_refresh_cagg_execute(int32 job_id, Jsonb *config)
 
 	context.number_of_batches = list_length(refresh_window_list);
 
-	ListCell *lc;
+	/*
+	 * 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;
-	foreach (lc, refresh_window_list)
-	{
-		InternalTimeRange *refresh_window = (InternalTimeRange *) lfirst(lc);
+	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),
@@ -553,6 +560,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());