An error occurred when the sort push down rule pushed sort below join

[haohuaijin]

Describe the bug

datafusion reported error: does not satisfy order requirements: [c@2 ASC NULLS LAST]. Child-0 order: [[b@1 ASC NULLS LAST]] when execution subquery.

To Reproduce

DataFusion CLI v43.0.0
> create table t(a int, b int, c int) as values (1,2,3), (4,5,6);

select * from t where a in (select a from t where b > 1) order by c;
0 row(s) fetched. 
Elapsed 0.033 seconds.

SanityCheckPlan
caused by
Error during planning: Plan: ["SortPreservingMergeExec: [c@2 ASC NULLS LAST]", "  CoalesceBatchesExec: target_batch_size=8192", "    HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(a@0, a@0)]", "      CoalesceBatchesExec: target_batch_size=8192", "        RepartitionExec: partitioning=Hash([a@0], 12), input_partitions=12", "          RepartitionExec: partitioning=RoundRobinBatch(12), input_partitions=1", "            CoalesceBatchesExec: target_batch_size=8192", "              FilterExec: b@1 > 1, projection=[a@0]", "                MemoryExec: partitions=1, partition_sizes=[1]", "      SortExec: expr=[b@1 ASC NULLS LAST], preserve_partitioning=[true]", "        CoalesceBatchesExec: target_batch_size=8192", "          RepartitionExec: partitioning=Hash([a@0], 12), input_partitions=1", "            MemoryExec: partitions=1, partition_sizes=[1]"] does not satisfy order requirements: [c@2 ASC NULLS LAST]. Child-0 order: [[b@1 ASC NULLS LAST]]
> 

the error due to sort push down, after sort push down the sort column change from c to b.

| physical_plan after EnforceDistribution                    | OutputRequirementExec                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
|                                                            |   SortExec: expr=[c@2 ASC NULLS LAST], preserve_partitioning=[false]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                |
|                                                            |     CoalescePartitionsExec                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
|                                                            |       HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(a@0, a@0)]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
|                                                            |         RepartitionExec: partitioning=Hash([a@0], 12), input_partitions=12                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          |
|                                                            |           RepartitionExec: partitioning=RoundRobinBatch(12), input_partitions=1                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
|                                                            |             ProjectionExec: expr=[a@0 as a]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
|                                                            |               FilterExec: b@1 > 1                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                   |
|                                                            |                 MemoryExec: partitions=1, partition_sizes=[1]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       |
|                                                            |         RepartitionExec: partitioning=Hash([a@0], 12), input_partitions=1                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
|                                                            |           MemoryExec: partitions=1, partition_sizes=[1]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
|                                                            |                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
| physical_plan after CombinePartialFinalAggregate           | SAME TEXT AS ABOVE                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  |
| physical_plan after EnforceSorting                         | OutputRequirementExec                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                               |
|                                                            |   SortPreservingMergeExec: [c@2 ASC NULLS LAST]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
|                                                            |     HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(a@0, a@0)]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            |
|                                                            |       RepartitionExec: partitioning=Hash([a@0], 12), input_partitions=12                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            |
|                                                            |         RepartitionExec: partitioning=RoundRobinBatch(12), input_partitions=1                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                       |
|                                                            |           ProjectionExec: expr=[a@0 as a]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
|                                                            |             FilterExec: b@1 > 1                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                     |
|                                                            |               MemoryExec: partitions=1, partition_sizes=[1]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
|                                                            |       SortExec: expr=[b@1 ASC NULLS LAST], preserve_partitioning=[true]                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                             |
|                                                            |         RepartitionExec: partitioning=Hash([a@0], 12), input_partitions=1                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                           |
|                                                            |           MemoryExec: partitions=1, partition_sizes=[1]  

Expected behavior

work in datafusion v42, but reported error in datafusion v43 also error in main branch

Additional context

No response