Describe the bug
When aggregate dynamic-filter pushdown is enabled with multiple execution partitions, DataFusion can return an incorrect MIN for a schema-evolved Parquet dataset where one file does not contain the aggregated column.
The missing column is represented as a typed null, such as ScalarValue::Int64(None). However, scalar_cmp_null_short_circuit recognizes only ScalarValue::Null. The typed null therefore reaches partial_cmp, where None compares as less than Some(value). During scalar_min, this can replace a valid shared minimum with the typed null.
The resulting dynamic filter loses its lower-bound predicate and may incorrectly prune Parquet files containing the true minimum. The result depends on partition scheduling. The query returns the correct result with target_partitions = 1 or when aggregate dynamic-filter pushdown is disabled.
To Reproduce
since its a race condition and depends on partition scheduling I used claude to write this small script for reproducibility.
#!/usr/bin/env bash
set -euo pipefail
CLI="${CLI:-datafusion-cli}"
REPRO_DIR="$(mktemp -d /tmp/df-minmax-repro.XXXXXX)"
# Create two files containing latency_ms and one file without it.
"$CLI" -q --format csv \
-c "COPY (
SELECT * FROM (VALUES (100), (101), (102), (103), (104))
AS t(latency_ms)
) TO '$REPRO_DIR/01_low.parquet' STORED AS PARQUET" \
-c "COPY (
SELECT * FROM (VALUES ('h1'), ('h1'), ('h1'), ('h1'), ('h1'))
AS t(host)
) TO '$REPRO_DIR/02_missing.parquet' STORED AS PARQUET" \
-c "COPY (
SELECT * FROM (VALUES (200), (201), (202), (203), (204))
AS t(latency_ms)
) TO '$REPRO_DIR/03_high.parquet' STORED AS PARQUET"
QUERY_ARGS=()
for _ in {1..100}; do
QUERY_ARGS+=(
-c "SELECT MIN(latency_ms), MAX(latency_ms) FROM repro"
)
done
OUTPUT=$(
"$CLI" -q --format csv \
-c "SET datafusion.execution.target_partitions = 8" \
-c "SET datafusion.optimizer.enable_aggregate_dynamic_filter_pushdown = true" \
-c "CREATE EXTERNAL TABLE repro
STORED AS PARQUET LOCATION '$REPRO_DIR'" \
"${QUERY_ARGS[@]}"
)
printf '%s\n' "$OUTPUT" | awk -F, '
$0 == "100,204" { correct++ }
$0 == "200,204" { wrong++ }
END {
printf "correct=%d wrong=%d\n", correct, wrong
exit(wrong > 0 ? 1 : 0)
}
'
Expected behavior
Expected correct result: 100,204
Additional context
The bug is that the dynamic-filter merge recognizes only ScalarValue::Null, instead of identifying a broader set of typed nulls as well by using ScalarValue::is_null()
Describe the bug
When aggregate dynamic-filter pushdown is enabled with multiple execution partitions, DataFusion can return an incorrect MIN for a schema-evolved Parquet dataset where one file does not contain the aggregated column.
The missing column is represented as a typed null, such as ScalarValue::Int64(None). However, scalar_cmp_null_short_circuit recognizes only ScalarValue::Null. The typed null therefore reaches partial_cmp, where None compares as less than Some(value). During scalar_min, this can replace a valid shared minimum with the typed null.
The resulting dynamic filter loses its lower-bound predicate and may incorrectly prune Parquet files containing the true minimum. The result depends on partition scheduling. The query returns the correct result with target_partitions = 1 or when aggregate dynamic-filter pushdown is disabled.
To Reproduce
since its a race condition and depends on partition scheduling I used claude to write this small script for reproducibility.
Expected behavior
Expected correct result: 100,204
Additional context
The bug is that the dynamic-filter merge recognizes only
ScalarValue::Null, instead of identifying a broader set of typed nulls as well by usingScalarValue::is_null()