test: make push_down_filter_regression dynamic filter content deterministic (#22621) (#22643)

## Which issue does this PR close?

- Closes #22621.

## Rationale for this change

`push_down_filter_regression.slt` (added in #22150) asserts the exact
`DynamicFilter` content rendered by `EXPLAIN ANALYZE` on the `agg_dyn_*`
fixtures. That content is **not** deterministic: the filter threshold
tightens
as each `AggregateExec(mode=Partial)` publishes its running `min`/`max`,
and the
`EXPLAIN ANALYZE` snapshot can be taken while the filter is still
converging.

For `agg_dyn_single`, `file_0` holds the global `min` (1) and `file_1` a
larger
partial `min` (3). If the snapshot lands after `file_1` publishes `3`
but before
`file_0` publishes `1`, the filter reads `a < 3` instead of the final `a
< 1` —
exactly the intermittent CI failure reported in #22621. The fixture's
comment
incorrectly claimed the filter *content* was deterministic and only the
pruning
*counts* raced.

## What changes are included in this PR?

Make the filter content independent of publish order by giving **every
file the
same per-file min/max**, so any snapshot equals the fully converged
filter:

- `agg_dyn_single` — both files `(1), (8)` → each file `min=1, max=8`.
- `agg_dyn_two_col` — each file `min(a)=1, max(b)=9`.
- `agg_dyn_mixed` — each file `min(a)=1, max(a)=8, max(b)=12`.

`agg_dyn_two_col` and `agg_dyn_mixed` were not in the reported failure
but shared
the same latent race (differing per-file extremes), so they are fixed
too.
`agg_dyn_nulls` is left untouched — its filter is always `true` and
never races.

The expected plan text is **unchanged**; only the input data and the
misleading
comments are modified. The alternative of forcing a single partition was
rejected: dynamic aggregate filters are only emitted in `Partial+Final`
mode
(`target_partitions >= 2`), so a single partition would emit no filter
at all.

## Are these changes tested?

Yes — the modified `push_down_filter_regression.slt` itself is the test.
It
passes, and because the asserted filter content no longer depends on
partition
scheduling, it is stable across runs (verified by running it repeatedly
locally).

## Are there any user-facing changes?

No. Test-only change.

Co-authored-by: Andrew Lamb <andrew@nerdnetworks.org>

Author: diegoQuinas

datafusion/sqllogictest/test_files/push_down_filter_regression.slt

@@ -275,17 +275,25 @@ drop table agg_dyn_e2e;
 statement ok
 set datafusion.execution.target_partitions = 2;
 
-# --- single-column fixture ([5, 1, 3, 8]) split across 2 files ---
+# --- single-column fixture ([1, 8, 1, 8]) split across 2 files ---
+#
+# Every file shares the same per-file min (1) and max (8). This makes the
+# DynamicFilter content deterministic under parallel execution: no matter the
+# order in which the Partial aggregates publish their bounds, every partition
+# contributes the same min/max, so any snapshot taken by `EXPLAIN ANALYZE`
+# equals the fully converged filter. Using files with differing per-file
+# extremes (e.g. min 1 vs 3) makes the snapshot race-dependent, which is what
+# caused the flakiness reported in #22621.
 
 statement ok
 COPY (
-  SELECT * FROM (VALUES (5), (1)) AS v(a)
+  SELECT * FROM (VALUES (1), (8)) AS v(a)
 ) TO 'test_files/scratch/push_down_filter_regression/agg_dyn_single/file_0.parquet'
 STORED AS PARQUET;
 
 statement ok
 COPY (
-  SELECT * FROM (VALUES (3), (8)) AS v(a)
+  SELECT * FROM (VALUES (1), (8)) AS v(a)
 ) TO 'test_files/scratch/push_down_filter_regression/agg_dyn_single/file_1.parquet'
 STORED AS PARQUET;
 
@@ -296,10 +304,11 @@ LOCATION 'test_files/scratch/push_down_filter_regression/agg_dyn_single/';
 
 # Use `analyze_level = summary` + `analyze_categories = 'none'` so metrics
 # render empty; we only care that the `predicate=DynamicFilter [ ... ]` text
-# matches. Pruning metrics here are subject to a parallel-execution race
+# matches. The pruning *counts* are still subject to a parallel-execution race
 # (the order in which Partial aggregates publish filter updates vs. when the
-# scan reads each partition), so the filter *content* is deterministic but
-# the pruning counts are not.
+# scan reads each partition), which is why metrics are suppressed. The filter
+# *content* is kept deterministic by giving every file the same per-file
+# min/max (see the fixture comment above).
 statement ok
 set datafusion.explain.analyze_level = summary;
 
@@ -350,16 +359,18 @@ statement ok
 drop table agg_dyn_single;
 
 # --- two-column fixture: MIN(a) + MAX(b) across columns ---
+# Every file shares the same per-file min(a)=1 and max(b)=9 so the DynamicFilter
+# content is deterministic regardless of publish order (see #22621).
 
 statement ok
 COPY (
-  SELECT * FROM (VALUES (5, 7), (1, 2)) AS v(a, b)
+  SELECT * FROM (VALUES (1, 5), (4, 9)) AS v(a, b)
 ) TO 'test_files/scratch/push_down_filter_regression/agg_dyn_two_col/file_0.parquet'
 STORED AS PARQUET;
 
 statement ok
 COPY (
-  SELECT * FROM (VALUES (3, 4), (8, 9)) AS v(a, b)
+  SELECT * FROM (VALUES (1, 6), (2, 9)) AS v(a, b)
 ) TO 'test_files/scratch/push_down_filter_regression/agg_dyn_two_col/file_1.parquet'
 STORED AS PARQUET;
 
@@ -384,10 +395,12 @@ drop table agg_dyn_two_col;
 # --- mixed expressions: MIN(a), MAX(a), MAX(b), MIN(c+1) ---
 # Supported aggregates (MIN(a), MAX(a), MAX(b)) should drive a filter;
 # MIN(c+1) is unsupported and must not contribute.
+# Every file shares the same per-file min(a)=1, max(a)=8 and max(b)=12 so the
+# DynamicFilter content is deterministic regardless of publish order (see #22621).
 
 statement ok
 COPY (
-  SELECT * FROM (VALUES (5, 10, 100), (1, 4, 70)) AS v(a, b, c)
+  SELECT * FROM (VALUES (1, 12, 100), (8, 4, 70)) AS v(a, b, c)
 ) TO 'test_files/scratch/push_down_filter_regression/agg_dyn_mixed/file_0.parquet'
 STORED AS PARQUET;