fix(EnsureRequirements): remap sort requirement through ProjectionExec on pushdown

[Jeadie]

Which issue does this PR close?

None (Happy to open a tracking issue if preferred).

Rationale for this change

EnforceSorting/EnsureRequirements can produce a physically invalid plan that SanityCheckPlan rejects:

... does not satisfy order requirements: [score@1 DESC NULLS LAST, id@0 ASC]. Child-0 order: []

Root cause. In sort_pushdown.rs::pushdown_requirement_to_children, the fetch-forwarding branch

} else if plan.fetch().is_some()
    && plan.supports_limit_pushdown()
    && plan.maintains_input_order().into_iter().all(|m| m)
{
    ... Ok(Some(vec![Some(parent_required)]))   // forwards the requirement UNCHANGED
}

forwards the parent ordering requirement to the child unchanged. A ProjectionExec reports a fetch() (forwarded from its input) and can renumber/reorder columns, so the requirement, expressed in the projection's output schema, is pushed into the child schema verbatim. For a projection that reorders the sort key (e.g. output [id, score, value] over input [id, value, score]), score@1 (valid above) is pushed down as score@1 (a different column below). The relocated SortExec then advertises an ordering its child cannot provide, and SanityCheckPlan fails.

This reproduces whenever a fetch-bearing global sort sits above a CoalescePartitionsExec over a column-reordering, order-preserving ProjectionExec whose input is already ordered, parallelize_sorts sinks the per-partition SortExec below the projection without remapping the key index.

What changes are included in this PR?

Remap the ordering requirement through the projection's column mapping before pushing it down (new remap_requirement_through_projection helper):

• Each required column at output index i is rewritten to the column the projection produces at that index (projection.expr()[i]).
• If any required column maps to a computed (non-Column) expression, the ordering cannot be expressed below the projection, so pushdown is declined and the sort is kept above the projection.
• Hard/soft-ness and all alternatives of the OrderingRequirements are preserved; unsatisfiable alternatives are dropped.

For non-reordering projections the remap is the identity, so existing plans are unchanged.

Are these changes tested?

Yes.

• Unit tests for the remap helper: reordering remap, softness preservation, multiple hard alternatives, dropping an unsatisfiable alternative while keeping others, and declining when a required column is computed.
• An end-to-end regression test (test_parallelize_sorts_remaps_index_through_reordering_projection) builds the multi-partition reordering-projection plan, runs EnsureRequirements with repartition_sorts enabled, and asserts the result passes SanityCheckPlan. It fails without this fix and passes with it.

Are there any user-facing changes?

No API changes. Plans that previously failed SanityCheckPlan (or produced incorrect orderings) for this shape are now valid; all other plans are unchanged.

[Jeadie]

Worked example

Before EnsureRequirements (still valid)

  SortExec: TopK(fetch=4), expr=[score@1 DESC, a@0 ASC]        out: [a, score, value]
    CoalescePartitionsExec                                     out: [a, score, value]
      ProjectionExec: [a@0 as a, c@2 as score, b@1 as value]   out: [a, score, value]
        SortExec: TopK(fetch=1000), expr=[c@2 DESC]            out: [a, b, c, d, e]
          RepartitionExec: RoundRobinBatch                     out: [a, b, c, d, e]
            DataSourceExec                                     out: [a, b, c, d, e]

Resolving the sort expressions:

• Outer SortExec: its input is CoalescePartitionsExec → [a, score, value]. So score@1 = index 1 = score ✓.
• ProjectionExec: its input is the inner sort → [a, b, c, d, e]. So c@2 = c ✓; output renames it to score at position 1.
• Inner SortExec: input [a, b, c, d, e], c@2 = c ✓.

After EnsureRequirements, without the fix (invalid)

parallelize_sorts replaces CoalescePartitions + the global sort with a SortPreservingMergeExec, and pushes a per-partition SortExec below the projection; copying expr=[score@1, a@0] down unchanged:

  SortPreservingMergeExec: [score@1 DESC, a@0 ASC], fetch=4    out: [a, score, value]
    ProjectionExec: [a@0 as a, c@2 as score, b@1 as value]     out: [a, score, value]
      SortExec: TopK(fetch=4), expr=[score@1 DESC, a@0 ASC]    out: [a, b, c, d, e]   ◀── now BELOW the projection
        SortExec: TopK(fetch=1000), expr=[c@2 DESC]            out: [a, b, c, d, e]
          RepartitionExec: RoundRobinBatch                     out: [a, b, c, d, e]
            DataSourceExec                                     out: [a, b, c, d, e]

The relocated SortExec's input is now [a, b, c, d, e] (not [a, score, value]). So its score@1 resolves against the wrong schema:

• @0 = a ✓
• @1 = b ✗ there is no score at this level; index 1 is b. The sort silently orders by b, not score.

SanityCheckPlan rejects:

• The SortPreservingMergeExec (top) requires its input ordered by [score@1, a@0] over [a, score, value].
• Its child projection derives its output ordering from the relocated sort, which orders by [b@1, a@0]. The projection maps b@1 → value@2, so it can only offer [value@2, a@0] upward.
• Nothing satisfies score@1, and therefore raises does not satisfy order requirements: [score@1 DESC NULLS LAST, a@0 ASC]. Child-0 order: [].

After EnsureRequirements, with the fix (valid)

The requirement is remapped as it crosses the projection (score@1 → c@2), so the relocated sort indexes correctly into [a, b, c, d, e]:

  SortPreservingMergeExec: [score@1 DESC, a@0 ASC], fetch=4    out: [a, score, value]
    ProjectionExec: [a@0 as a, c@2 as score, b@1 as value]     out: [a, score, value]
      SortExec: TopK(fetch=4), expr=[c@2 DESC, a@0 ASC]        out: [a, b, c, d, e]   ◀── remapped @1 -> @2
        SortExec: TopK(fetch=1000), expr=[c@2 DESC]            out: [a, b, c, d, e]
          RepartitionExec: RoundRobinBatch                     out: [a, b, c, d, e]
            DataSourceExec                                     out: [a, b, c, d, e]

Now the relocated sort orders by [c@2, a@0] = [c, a]; the projection maps c@2 → score@1, so its output is ordered by [score@1, a@0] (what the SortPreservingMergeExec needs). Valid.

[alamb]

@zhuqi-lucas could you help review this PR?

[zhuqi-lucas]

LGTM

Nit: this fix completes the TODO for the fetch-bearing branch but the non-fetch case (|| plan.is::() short-circuit in the second pushdown check) is still declining pushdown. The new comment frames that as intentional, but worth either (a) keeping the TODO marker for the non-fetch case, or (b) explicitly noting why non-fetch projection pushdown is not pursued. Otherwise a future reader could mistake "TODO removed" for "feature complete."

[alamb]

Thanks @Jeadie and @zhuqi-lucas

@Jeadie could you please help address @zhuqi-lucas 's comments and then I'll plan to merge this PR?