fix(optimizer): Fix local exchange for merge join (#26590)

## Description
Current code will try to add a round robin local exchange below the
merge join node, which will break the sorted property of the input. In
this PR, we fixed it.

## Motivation and Context
Bug fix

## Impact
Bug fix

## Test Plan
Unit test

## Contributor checklist

- [ ] Please make sure your submission complies with our [contributing
guide](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md),
in particular [code
style](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#code-style)
and [commit
standards](https://github.com/prestodb/presto/blob/master/CONTRIBUTING.md#commit-standards).
- [ ] PR description addresses the issue accurately and concisely. If
the change is non-trivial, a GitHub Issue is referenced.
- [ ] Documented new properties (with its default value), SQL syntax,
functions, or other functionality.
- [ ] If release notes are required, they follow the [release notes
guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines).
- [ ] Adequate tests were added if applicable.
- [ ] CI passed.
- [ ] If adding new dependencies, verified they have an [OpenSSF
Scorecard](https://securityscorecards.dev/#the-checks) score of 5.0 or
higher (or obtained explicit TSC approval for lower scores).

## Release Notes
Please follow [release notes
guidelines](https://github.com/prestodb/presto/wiki/Release-Notes-Guidelines)
and fill in the release notes below.


```
== NO RELEASE NOTE ==
```

Author: feilong-liu

presto-hive/src/test/java/com/facebook/presto/hive/TestMergeJoinPlanPrestoOnSpark.java

@@ -17,6 +17,7 @@
 import com.facebook.presto.spi.plan.JoinType;
 import com.facebook.presto.sql.analyzer.FeaturesConfig;
 import com.facebook.presto.sql.planner.assertions.PlanMatchPattern;
+import com.facebook.presto.sql.planner.plan.ExchangeNode;
 import com.facebook.presto.testing.QueryRunner;
 import com.facebook.presto.tests.AbstractTestQueryFramework;
 import com.google.common.collect.ImmutableList;
@@ -37,10 +38,13 @@
 import static com.facebook.presto.spi.plan.JoinType.RIGHT;
 import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.anyTree;
 import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.equiJoinClause;
+import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.exchange;
 import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.join;
 import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.mergeJoin;
 import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.sort;
 import static com.facebook.presto.sql.planner.assertions.PlanMatchPattern.tableScan;
+import static com.facebook.presto.sql.tree.SortItem.NullOrdering.FIRST;
+import static com.facebook.presto.sql.tree.SortItem.Ordering.ASCENDING;
 import static io.airlift.tpch.TpchTable.CUSTOMER;
 import static io.airlift.tpch.TpchTable.LINE_ITEM;
 import static io.airlift.tpch.TpchTable.NATION;
@@ -179,7 +183,8 @@ public void testDifferentBucketedByKey()
                     anyTree(
                             mergeJoin(INNER, ImmutableList.of(equiJoinClause("custkey_l", "custkey_r")),
                                     Optional.empty(),
-                                    anyTree(sort(anyTree(tableScan("test_join_customer2", ImmutableMap.of("custkey_l", "custkey"))))),
+                                    exchange(ExchangeNode.Scope.LOCAL, ExchangeNode.Type.GATHER, ImmutableList.of(sort("custkey_l", ASCENDING, FIRST)),
+                                            sort(anyTree(tableScan("test_join_customer2", ImmutableMap.of("custkey_l", "custkey"))))),
                                     tableScan("test_join_order2", ImmutableMap.of("custkey_r", "custkey")))));
         }
         finally {
@@ -211,7 +216,8 @@ public void testDifferentSortByKey()
                     anyTree(
                             mergeJoin(INNER, ImmutableList.of(equiJoinClause("custkey_l", "custkey_r")),
                                     Optional.empty(),
-                                    anyTree(sort(tableScan("test_join_customer3", ImmutableMap.of("custkey_l", "custkey")))),
+                                    exchange(ExchangeNode.Scope.LOCAL, ExchangeNode.Type.GATHER, ImmutableList.of(sort("custkey_l", ASCENDING, FIRST)),
+                                            sort(tableScan("test_join_customer3", ImmutableMap.of("custkey_l", "custkey")))),
                                     tableScan("test_join_order3", ImmutableMap.of("custkey_r", "custkey")))));
         }
         finally {

presto-main-base/src/main/java/com/facebook/presto/sql/planner/optimizations/AddLocalExchanges.java

@@ -79,7 +79,6 @@
 import static com.facebook.presto.SystemSessionProperties.isQuickDistinctLimitEnabled;
 import static com.facebook.presto.SystemSessionProperties.isSegmentedAggregationEnabled;
 import static com.facebook.presto.SystemSessionProperties.isSpillEnabled;
-import static com.facebook.presto.SystemSessionProperties.preferSortMergeJoin;
 import static com.facebook.presto.common.type.BigintType.BIGINT;
 import static com.facebook.presto.common.type.VarbinaryType.VARBINARY;
 import static com.facebook.presto.operator.aggregation.AggregationUtils.hasSingleNodeExecutionPreference;
@@ -893,12 +892,11 @@ public PlanWithProperties visitSpatialJoin(SpatialJoinNode node, StreamPreferred
         @Override
         public PlanWithProperties visitMergeJoin(MergeJoinNode node, StreamPreferredProperties parentPreferences)
         {
-            if (preferSortMergeJoin(session)) {
-                PlanWithProperties probe = planAndEnforce(node.getLeft(), singleStream(), singleStream());
-                PlanWithProperties build = planAndEnforce(node.getRight(), singleStream(), singleStream());
-                return rebaseAndDeriveProperties(node, ImmutableList.of(probe, build));
-            }
-            return super.visitMergeJoin(node, parentPreferences);
+            // The optimizer rule MergeJoinForSortedInputOptimizer and SortMergeJoinOptimizer which add the merge join node is responsible to ensure the input of the merge join is sorted.
+            // Here we use `any().withOrderSensitivity()` meaning respect the input distribution of the input and keep the input order.
+            PlanWithProperties probe = planAndEnforce(node.getLeft(), any().withOrderSensitivity(), any().withOrderSensitivity());
+            PlanWithProperties build = planAndEnforce(node.getRight(), any().withOrderSensitivity(), any().withOrderSensitivity());
+            return rebaseAndDeriveProperties(node, ImmutableList.of(probe, build));
         }
 
         @Override

presto-main-base/src/test/java/com/facebook/presto/sql/planner/TestLogicalPlanner.java

@@ -558,8 +558,7 @@ public void testSortMergeJoin()
                 preferSortMergeJoin,
                 anyTree(
                         mergeJoin(INNER, ImmutableList.of(equiJoinClause("ORDERS_OK", "LINEITEM_PK")), Optional.empty(),
-                                exchange(LOCAL, GATHER, ImmutableList.of(),
-                                        tableScan("orders", ImmutableMap.of("ORDERS_OK", "orderkey"))),
+                                        tableScan("orders", ImmutableMap.of("ORDERS_OK", "orderkey")),
                                 sort(
                                         ImmutableList.of(sort("LINEITEM_PK", ASCENDING, FIRST)),
                                         exchange(LOCAL, GATHER, ImmutableList.of(),
@@ -574,18 +573,15 @@ public void testSortMergeJoin()
                                         ImmutableList.of(sort("ORDERS_CK", ASCENDING, FIRST)),
                                         exchange(LOCAL, GATHER, ImmutableList.of(),
                                                 tableScan("orders", ImmutableMap.of("ORDERS_CK", "custkey")))),
-                                exchange(LOCAL, GATHER, ImmutableList.of(),
-                                        tableScan("lineitem", ImmutableMap.of("LINEITEM_OK", "orderkey"))))));
+                                        tableScan("lineitem", ImmutableMap.of("LINEITEM_OK", "orderkey")))));
 
         // Both sides are sorted.
         assertPlan("SELECT o.orderkey FROM orders o INNER JOIN lineitem l ON o.orderkey = l.orderkey",
                 preferSortMergeJoin,
                 anyTree(
                         mergeJoin(INNER, ImmutableList.of(equiJoinClause("ORDERS_OK", "LINEITEM_OK")), Optional.empty(),
-                                exchange(LOCAL, GATHER, ImmutableList.of(),
-                                        tableScan("orders", ImmutableMap.of("ORDERS_OK", "orderkey"))),
-                                exchange(LOCAL, GATHER, ImmutableList.of(),
-                                        tableScan("lineitem", ImmutableMap.of("LINEITEM_OK", "orderkey"))))));
+                                        tableScan("orders", ImmutableMap.of("ORDERS_OK", "orderkey")),
+                                        tableScan("lineitem", ImmutableMap.of("LINEITEM_OK", "orderkey")))));
     }
 
     @Test