Update trendline explain IT

Additionally correct merge collation method

Signed-off-by: Yuanchun Shen <yuanchu@amazon.com>

Author: yuancu

integ-test/src/test/resources/expectedOutput/calcite/explain_sort_push.json

@@ -1,6 +1,6 @@
 {
   "calcite": {
     "logical": "LogicalSort(sort0=[$0], dir0=[ASC])\n  LogicalProject(age=[$8])\n    LogicalFilter(condition=[>($8, 30)])\n      LogicalSort(sort0=[$8], dir0=[ASC])\n        CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
-    "physical":"EnumerableCalc(expr#0..16=[{inputs}], age=[$t8])\n  CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[SORT->[{\n  \"age\" : {\n    \"order\" : \"asc\"\n  }\n}], FILTER->>($8, 30)], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"query\":{\"range\":{\"age\":{\"from\":30,\"to\":null,\"include_lower\":false,\"include_upper\":true,\"boost\":1.0}}},\"sort\":[{\"age\":{\"order\":\"asc\"}}]}, requestedTotalSize=10000, pageSize=null, startFrom=0)])\n"
+    "physical": "CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[PROJECT->[age], SORT->[{\n  \"age\" : {\n    \"order\" : \"asc\"\n  }\n}], FILTER->>($0, 30)], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"timeout\":\"1m\",\"query\":{\"range\":{\"age\":{\"from\":30,\"to\":null,\"include_lower\":false,\"include_upper\":true,\"boost\":1.0}}},\"_source\":{\"includes\":[\"age\"],\"excludes\":[]},\"sort\":[{\"age\":{\"order\":\"asc\"}}]}, requestedTotalSize=2147483647, pageSize=null, startFrom=0)])\n"
   }
-}
+}

integ-test/src/test/resources/expectedOutput/calcite/explain_trendline_sort_push.json

@@ -1,6 +1,6 @@
 {
   "calcite": {
     "logical": "LogicalProject(ageTrend=[CASE(>(COUNT() OVER (ROWS 1 PRECEDING), 1), /(SUM($8) OVER (ROWS 1 PRECEDING), CAST(COUNT($8) OVER (ROWS 1 PRECEDING)):DOUBLE NOT NULL), null:NULL)])\n  LogicalFilter(condition=[IS NOT NULL($8)])\n    LogicalSort(sort0=[$8], dir0=[ASC])\n      LogicalSort(fetch=[5])\n        CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]])\n",
-    "physical": "EnumerableCalc(expr#0..3=[{inputs}], expr#4=[1], expr#5=[>($t1, $t4)], expr#6=[CAST($t3):DOUBLE NOT NULL], expr#7=[/($t2, $t6)], expr#8=[null:NULL], expr#9=[CASE($t5, $t7, $t8)], ageTrend=[$t9])\n  EnumerableWindow(window#0=[window(rows between $1 PRECEDING and CURRENT ROW aggs [COUNT(), $SUM0($0), COUNT($0)])])\n    EnumerableCalc(expr#0=[{inputs}], expr#1=[IS NOT NULL($t0)], age=[$t0], $condition=[$t1])\n      EnumerableSort(sort0=[$0], dir0=[ASC])\n        CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[LIMIT->5, PROJECT->[age]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":5,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"age\"],\"excludes\":[]}}, requestedTotalSize=5, pageSize=null, startFrom=0)])\n"
+    "physical": "EnumerableCalc(expr#0..3=[{inputs}], expr#4=[1], expr#5=[>($t1, $t4)], expr#6=[CAST($t3):DOUBLE NOT NULL], expr#7=[/($t2, $t6)], expr#8=[null:NULL], expr#9=[CASE($t5, $t7, $t8)], ageTrend=[$t9])\n  EnumerableWindow(window#0=[window(rows between $1 PRECEDING and CURRENT ROW aggs [COUNT(), $SUM0($0), COUNT($0)])])\n    EnumerableCalc(expr#0=[{inputs}], expr#1=[IS NOT NULL($t0)], age=[$t0], $condition=[$t1])\n      CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_account]], PushDownContext=[[LIMIT->5, PROJECT->[age], SORT->[{\n  \"age\" : {\n    \"order\" : \"asc\"\n  }\n}]], OpenSearchRequestBuilder(sourceBuilder={\"from\":0,\"size\":5,\"timeout\":\"1m\",\"_source\":{\"includes\":[\"age\"],\"excludes\":[]},\"sort\":[{\"age\":{\"order\":\"asc\"}}]}, requestedTotalSize=5, pageSize=null, startFrom=0)])\n"
   }
-}
+}

opensearch/src/main/java/org/opensearch/sql/opensearch/storage/scan/CalciteLogicalIndexScan.java

@@ -211,11 +211,11 @@ public CalciteLogicalIndexScan pushDownSort(List<RelFieldCollation> collations)
           mergeCollations(existingFieldCollations, collations);
 
       // Propagate the sort to the new scan
-      RelTraitSet newTraitSet = getTraitSet().plus(RelCollations.of(mergedCollations));
+      RelTraitSet traitsWithCollations = getTraitSet().plus(RelCollations.of(mergedCollations));
       CalciteLogicalIndexScan newScan =
           new CalciteLogicalIndexScan(
               getCluster(),
-              newTraitSet,
+              traitsWithCollations,
               hints,
               table,
               osIndex,
@@ -267,8 +267,12 @@ public CalciteLogicalIndexScan pushDownSort(List<RelFieldCollation> collations)
    */
   private static List<RelFieldCollation> mergeCollations(
       List<RelFieldCollation> existingCollations, List<RelFieldCollation> newCollations) {
-    List<RelFieldCollation> concatenatedCollations = new ArrayList<>(existingCollations);
-    concatenatedCollations.addAll(newCollations);
+    // We add new collations first, then existing collations
+    // Consider `sort a | sort b`, the second sort should take precedence
+    List<RelFieldCollation> concatenatedCollations = new ArrayList<>(newCollations);
+    concatenatedCollations.addAll(existingCollations);
+    // Within the same group of collations, the first occurrence of each field index
+    // should take precedence, so we need to remove duplicates while preserving the order
     LinkedList<RelFieldCollation> mergedCollations = new LinkedList<>();
     for (RelFieldCollation collation : concatenatedCollations) {
       // If the collation is already in the merged list, remove it from the list before adding