fix: treat typed nulls as missing bounds in aggregate dynamic filter merge

[uddhavdave]

Which issue does this PR close?

• Closes MIN dynamic filter race across partitions #25147.

Rationale for this change

SELECT MIN(col), MAX(col) over a schema-evolved Parquet dataset can return a wrong MIN when datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown is enabled and the query runs with multiple partitions.

When one of the files does not contain the aggregated column, the Partial aggregate for that partition evaluates to a typed null such as Int64(NULL) rather than ScalarValue::Null. The merge of per-partition bounds into the shared dynamic filter bound only short-circuited on ScalarValue::Null. The typed null therefore reached partial_cmp, where None orders before Some(_), and either replaced a valid shared minimum or blocked a later valid minimum from being recorded. The dynamic filter then lost its lower bound and became just col > <max>, which pruned the file holding the true minimum. The result depended on partition scheduling; target_partitions = 1 or disabling the pushdown returned the correct answer.

What changes are included in this PR?

• scalar_cmp_null_short_circuit in datafusion/physical-plan/src/aggregates/aggregate_stream.rs now uses ScalarValue::is_null() so both untyped and typed nulls are treated as "no bound yet" when merging MIN/MAX bounds across partitions.

What is the testing strategy for this PR?

• New unit test scalar_min_max_ignore_typed_nulls in aggregate_stream.rs covers scalar_min/scalar_max with typed nulls, untyped nulls, and regular values on either side. This fails without the fix and is deterministic.
• New sqllogictest case in datafusion/sqllogictest/test_files/push_down_filter_regression.slt that builds a three-file Parquet fixture (one file without the aggregated column) and runs MIN/MAX with target_partitions = 8. The file holding the minimum is named to sort last so it is opened after the other partitions publish their bounds. With the fix reverted locally this reproduced the wrong result in 4 of 5 runs; with the fix it passes consistently. Because the failure depends on scheduling, the unit test is the primary regression guard and the slt case documents the end-to-end scenario.

cargo fmt --all, cargo clippy --all-targets --all-features -- -D warnings, and the push_down_filter_regression sqllogictest all pass locally.

Are there any user-facing changes?

No API changes. Queries affected by the bug now return the correct MIN/MAX.

[kosiew]

@uddhavdave,

Thanks for working on this. The fix looks good overall, and the unit test gives solid deterministic coverage for the typed null merge behavior. I left one non-blocking suggestion about making the end-to-end regression more deterministic.

[kosiew]

The end-to-end regression here depends on partition scheduling. Repeating the query five times helps, but the old implementation could still pass if the missing-column partition does not publish its bound between the high and low partitions. The helper test gives deterministic coverage for the merge semantics, but could we add a focused integration test that controls partition completion or ordering, or otherwise forces this interleaving? That would make sure the SQL/data-source regression cannot become a false green.

[alamb]

I think we could do this as a follow on PR as well. Or perhaps set target partitions to 1?

[uddhavdave]

Thanks @kosiew for pointing this out. Agree, it's better to have a deterministic coverage. I replaced the scheduling-dependent SLT with a Rust integration test that forces the interleaving. verified that it fails deterministically on the old code.

Thank you @alamb for the suggestion. I tried setting target_partitions = 1 but had to remove the CombinePartialFinalAggregate optimizer rule to make sure aggregation mode does not get rewritten to single. The test works well.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 33.33333% with 20 lines in your changes missing coverage. Please review.
✅ Project coverage is 82.51%. Comparing base (ed43e69) to head (6a18f43).
⚠️ Report is 2 commits behind head on main.

Files with missing lines	Patch %	Lines
...n/physical-plan/src/aggregates/aggregate_stream.rs	33.33%	0 Missing and 20 partials ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main   #25149      +/-   ##
==========================================
- Coverage   82.51%   82.51%   -0.01%     
==========================================
  Files        1141     1141              
  Lines      439870   439897      +27     
  Branches   439870   439897      +27     
==========================================
  Hits       362974   362974              
- Misses      54942    54948       +6     
- Partials    21954    21975      +21     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[alamb]

Thanks @uddhavdave and @kosiew

[uddhavdave]

closing this PR since it all of the code was already merged #25579