cetra3 opened a new issue, #26391:
URL: https://github.com/apache/datafusion/issues/26391
### Describe the bug
A query with a filter and a sort on an expression over a struct field fails
planning with a `SanityCheckPlan` error.
`EnsureRequirements` removes the sort key because the filter makes it a
constant. After `ProjectionPushdown` merges two projections, the plan no longer
knows that the key is a constant, so the sanity check rejects the plan.
The cause is in `EquivalenceGroup::project`. A projection can name one
expression twice, under two aliases. A constant is then recorded on the first
alias only. An expression that reads the second alias does not match the
constant, although the planner knows that the two aliases are equal.
Leaf expression pushdown
(`datafusion.optimizer.enable_leaf_expression_pushdown`, `true` by default)
creates such a projection when a filter and a projection read the same struct
field (#25447).
### To Reproduce
In `datafusion-cli` at `main` (`e732212ca`):
```sql
set datafusion.execution.target_partitions = 4;
set datafusion.execution.parquet.pushdown_filters = true;
copy (select * from (values ({k: 'a'}, 1), ({k: 'b'}, 2)) as v(s, id))
to '/tmp/df_const_repro.parquet';
create external table t stored as parquet location
'/tmp/df_const_repro.parquet';
select upper(s['k']) as c, synth
from (select s, id + 1 as synth from t) as v
where upper(s['k']) = 'A'
order by synth, c;
```
```text
SanityCheckPlan
caused by
Error during planning: Plan: ["SortPreservingMergeExec: [synth@1 ASC NULLS
LAST, c@0 ASC NULLS LAST]", " SortExec: expr=[synth@1 ASC NULLS LAST],
preserve_partitioning=[true]", " ProjectionExec:
expr=[upper(__datafusion_extracted_2@2) as c, id@1 + 1 as synth]", "
RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1", "
DataSourceExec: file_groups={1 group: [[tmp/df_const_repro.parquet]]},
projection=[get_field(s@0, k) as __datafusion_extracted_1, id, get_field(s@0,
k) as __datafusion_extracted_2], file_type=parquet,
predicate=upper(get_field(s@0, k)) = A"] does not satisfy order requirements:
[synth@1 ASC NULLS LAST, c@0 ASC NULLS LAST]. Child-0 order: [[synth@1 ASC
NULLS LAST]]
```
`pushdown_filters = true` is required. `target_partitions` must be 2 or
more; the default (the number of CPU cores) also fails on a machine with 2 or
more cores.
With `set datafusion.optimizer.enable_leaf_expression_pushdown = false;`,
the query returns `A | 2`.
### Expected behavior
The query returns one row, `A | 2`.
### Additional context
The relevant part of the plan after `EnsureRequirements`:
```text
SortExec: expr=[synth@1 ASC NULLS LAST]
ProjectionExec: expr=[upper(__datafusion_extracted_2@1) as c, synth@0 as
synth]
ProjectionExec: expr=[id@1 + 1 as synth, __datafusion_extracted_2@2 as
__datafusion_extracted_2]
RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
ProjectionExec: expr=[get_field(s@0, k) as __datafusion_extracted_1,
id@1 as id, get_field(s@0, k) as __datafusion_extracted_2]
DataSourceExec: ..., predicate=upper(get_field(s@0, k)) = A
```
1. The scan predicate gives the equivalence class `{upper(get_field(s, k)),
'A'}`.
2. The bottom projection maps `get_field(s, k)` to both aliases. It records
`{__datafusion_extracted_1, __datafusion_extracted_2}` as one class, and it
rewrites the constant with the first alias only:
`{upper(__datafusion_extracted_1), 'A'}`.
3. The middle projection does not output `__datafusion_extracted_1`, so the
constant is rewritten through the class as `upper(__datafusion_extracted_2)`.
The top projection then matches it, `c` is a constant, and `EnsureRequirements`
removes `c` from the `SortExec` keys. It keeps `c` in the
`SortPreservingMergeExec`.
4. `ProjectionPushdown` merges the middle projection into the top one. Now
`upper(__datafusion_extracted_2)` is computed directly over the scan, whose
constant is `upper(__datafusion_extracted_1)`. The match fails, `c` is no
longer a constant, and `SanityCheckPlan` rejects the `SortPreservingMergeExec`.
In `EquivalenceGroup::project`
(`datafusion/physical-expr/src/equivalence/class.rs`), two lookups use the
source expression as written, not normalized through the equivalence classes:
- `self.is_expr_constant(source)`
- `self.get_equivalence_class(k)` in `augment_projection_mapping`
Normalizing the source before both lookups fixes the query. `project`
already computes `normalized_expr` for each mapping entry.
A fix for #25447 would remove this trigger, but the equivalence code should
handle a projection that names one expression twice, whatever produces it.
Related: #25459.
### Willingness to contribute
I plan to contribute a fix for this bug independently
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
