fix: impl handle_child_pushdown_result for SortExec

[haohuaijin]

Which issue does this PR close?

• Closes filter not pushdown through sort in FilterPushdown #21526.

Rationale for this change

FilterPushdown does not push filters through SortExec because SortExec lacks an implementation of handle_child_pushdown_result. When a FilterExec sits above a plain SortExec (no fetch), the filter can safely be moved below the sort without changing semantics, since sorting preserves all rows.

What changes are included in this PR?

Implemented handle_child_pushdown_result for SortExec in datafusion/physical-plan/src/sorts/sort.rs:

• For plain SortExec (no fetch) in the Pre phase: any filters not absorbed by the child are collected into a new FilterExec inserted between the SortExec and its child.
• For SortExec with fetch (TopK) or non-Pre phases: filters are not absorbed, preserving correct TopK semantics where filtering after limiting would change results.

Are these changes tested?

Added comprehensive tests in datafusion/core/tests/physical_optimizer/filter_pushdown.rs covering:

• Filter pushdown through sort into a scan that supports pushdown
• Filter pushdown through sort when scan does NOT support pushdown (FilterExec inserted between sort and scan)
• Multiple conjunctive filters pushed through sort
• Filter NOT pushed through sort with fetch (TopK)
• Filter with projection pushed through sort
• Filter pushdown preserving the preserve_partitioning flag
• Filter with fetch limit propagated to SortExec (TopK conversion)

and one test case test_filter_pushdown_through_sort_with_projection for use LogicalPlanBuilder to reproduce.
not able to use sql to reproduce due to #15886

Are there any user-facing changes?

[haohuaijin]

the reproduce result in #21526, for this pr

Physical plan:
FilterExec: log@1 LIKE %datafusion%, projection=[_timestamp@0]
  SortExec: expr=[_timestamp@0 DESC NULLS LAST], preserve_partitioning=[true]
    DataSourceExec: partitions=1, partition_sizes=[1]

After FilterPushdown:
ProjectionExec: expr=[_timestamp@0 as _timestamp]
  SortExec: expr=[_timestamp@0 DESC NULLS LAST], preserve_partitioning=[true]
    FilterExec: log@1 LIKE %datafusion%
      DataSourceExec: partitions=1, partition_sizes=[1]

[haohuaijin]

cc @alamb @adriangb do you have time take a look 🙏

[adriangb]

Can we add these as SLT tests instead?

[haohuaijin]

due to

• [DISCUSSION] Sorts being removed from subqueries #15886

the subquery's order by is removed by our sql parser layer, so i can't reproduce use sql

[haohuaijin]

i submit a pr to add the config #21614

[adriangb]

Could you explain why FilterPushdownPhase::Post should not be allowed, maybe with examples of where it would be incorrect?

[haohuaijin]

Good question, my initial though was that the swap of sort and filter is unrelated to dynamic filters, so I only do swap for pre pahse. but today when i reconsider this part, i find one incorrect example. i added this example in 10f1462

if i remove the phase != FilterPushdownPhase::Pre, the result will like below, we insert the dynamic filter under the sort, before the datasource if datasource do not accept dynamic filter,

    insta::assert_snapshot!(
        OptimizationTest::new(
            Arc::clone(&plan),
            FilterPushdown::new_post_optimization(),
            true
        ),
        @r"
    OptimizationTest:
      input:
        - HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)]
        -   DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b], file_type=test, pushdown_supported=true
        -   SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
        -     DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b], file_type=test, pushdown_supported=false
      output:
        Ok:
          - HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(a@0, a@0), (b@1, b@1)]
          -   DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b], file_type=test, pushdown_supported=true
          -   SortExec: expr=[a@0 ASC], preserve_partitioning=[false]
          -     FilterExec: DynamicFilter [ empty ]
          -       DataSourceExec: file_groups={1 group: [[test.parquet]]}, projection=[a, b], file_type=test, pushdown_supported=false
    "
    );

[adriangb]

Swapping the order from FilterExec -> SortExec to SortExec -> FilterExec makes sense to me 👍🏻

[haohuaijin]

Thanks for your reviews @adriangb , i replyed for your comment.