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

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-22643-217fd956cf77dfa7ff4611729459110177ef6f87
in repository https://gitbox.apache.org/repos/asf/datafusion.git

commit d070a8b1f1a3c2afe53043cefd11b0af7c818116
Author: Diego Perez Giordán <di...@gmail.com>
AuthorDate: Mon Jun 1 16:37:38 2026 -0300

    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 <an...@nerdnetworks.org>
---
 .../test_files/push_down_filter_regression.slt     | 31 +++++++++++++++-------
 1 file changed, 22 insertions(+), 9 deletions(-)

diff --git a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt b/datafusion/sqllogictest/test_files/push_down_filter_regression.slt
index 923a51afc8..b86bd2c51d 100644
--- a/datafusion/sqllogictest/test_files/push_down_filter_regression.slt
+++ b/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;
 


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org