SanityCheckPlan error: a constant sort key is lost when a projection reads a duplicate alias #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):

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;

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:

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