apache / apache/datafusion

MIN dynamic filter race across partitions

Open
#25,147 1 comment 0 reactions 1 assignee Claimed by @uddhavdave View on GitHub
bug
Dominant language
Rust
Stars
9.3k
Forks
2.4k
Avg merge
3d 7h
Merged PRs (30d)
344

Description

### 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()`

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.