[Backport 2.19-dev] Support pushdown physical sort operator to speedup SortMergeJoin (#3864)

benchmarks/src/jmh/java/org/opensearch/sql/expression/operator/predicate/MergeArrayAndObjectMapBenchmark.java

@@ -12,7 +12,7 @@
 import java.util.Map;
 import org.openjdk.jmh.annotations.Benchmark;
 import org.opensearch.sql.opensearch.data.type.OpenSearchDataType;
-import org.opensearch.sql.opensearch.request.system.OpenSearchDescribeIndexRequest;
+import org.opensearch.sql.opensearch.util.MergeRules.MergeRuleHelper;
 
 public class MergeArrayAndObjectMapBenchmark {
   private static final List<Map<String, OpenSearchDataType>> candidateMaps = prepareListOfMaps(120);
@@ -21,7 +21,7 @@ public class MergeArrayAndObjectMapBenchmark {
   public void testMerge() {
     Map<String, OpenSearchDataType> finalResult = new HashMap<>();
     for (Map<String, OpenSearchDataType> map : candidateMaps) {
-      OpenSearchDescribeIndexRequest.mergeObjectAndArrayInsideMap(finalResult, map);
+      MergeRuleHelper.merge(finalResult, map);
     }
   }
 

integ-test/src/test/java/org/opensearch/sql/calcite/remote/CalciteExplainIT.java

@@ -48,12 +48,23 @@ public void supportSearchSargPushDown_multiRange() throws IOException {
   // Only for Calcite
   @Test
   public void supportSearchSargPushDown_timeRange() throws IOException {
+    String query =
+        "source=opensearch-sql_test_index_bank"
+            + "| where birthdate >= '2016-12-08 00:00:00.000000000' "
+            + "and birthdate < '2018-11-09 00:00:00.000000000'";
+    var result = explainQueryToString(query);
     String expected = loadExpectedPlan("explain_sarg_filter_push_time_range.json");
-    assertJsonEqualsIgnoreId(
-        expected,
-        explainQueryToString(
-            "source=opensearch-sql_test_index_bank"
-                + "| where birthdate >= '2016-12-08 00:00:00.000000000' "
-                + "and birthdate < '2018-11-09 00:00:00.000000000' "));
+    assertJsonEqualsIgnoreId(expected, result);
+  }
+
+  // Only for Calcite
+  @Test
+  public void supportPushDownSortMergeJoin() throws IOException {
+    String query =
+        "source=opensearch-sql_test_index_bank| join left=l right=r on"
+            + " l.account_number=r.account_number opensearch-sql_test_index_bank";
+    var result = explainQueryToString(query);
+    String expected = loadExpectedPlan("explain_merge_join_sort_push.json");
+    assertJsonEqualsIgnoreId(expected, result);
   }
 }

integ-test/src/test/java/org/opensearch/sql/legacy/AggregationExpressionIT.java

@@ -71,7 +71,7 @@ public void hasGroupKeyAvgOnIntegerShouldPass() {
                 Index.BANK.getName()));
 
     verifySchema(response, schema("gender", null, "text"), schema("AVG(age)", "avg", "double"));
-    verifyDataRows(response, rows("m", 34.25), rows("f", 33.666666666666664d));
+    verifyDataRows(response, rows("M", 34.25), rows("F", 33.666666666666664d));
   }
 
   @Test
@@ -86,7 +86,7 @@ public void hasGroupKeyMaxAddMinShouldPass() {
         response,
         schema("gender", null, "text"),
         schema("MAX(age) + MIN(age)", "addValue", "long"));
-    verifyDataRows(response, rows("m", 60), rows("f", 60));
+    verifyDataRows(response, rows("M", 60), rows("F", 60));
   }
 
   @Test
@@ -98,7 +98,7 @@ public void hasGroupKeyMaxAddLiteralShouldPass() {
                 Index.ACCOUNT.getName()));
 
     verifySchema(response, schema("gender", null, "text"), schema("MAX(age) + 1", "add", "long"));
-    verifyDataRows(response, rows("m", 41), rows("f", 41));
+    verifyDataRows(response, rows("M", 41), rows("F", 41));
   }
 
   @Test
@@ -126,7 +126,7 @@ public void hasGroupKeyLogMaxAddMinShouldPass() {
         response,
         schema("gender", null, "text"),
         schema("Log(MAX(age) + MIN(age))", "logValue", "double"));
-    verifyDataRows(response, rows("m", 4.0943445622221d), rows("f", 4.0943445622221d));
+    verifyDataRows(response, rows("M", 4.0943445622221d), rows("F", 4.0943445622221d));
   }
 
   @Test
@@ -136,7 +136,7 @@ public void AddLiteralOnGroupKeyShouldPass() {
             String.format(
                 "SELECT gender, age+10, max(balance) as `max` "
                     + "FROM %s "
-                    + "WHERE gender = 'm' and age < 22 "
+                    + "WHERE gender = 'M' and age < 22 "
                     + "GROUP BY gender, age "
                     + "ORDER BY age",
                 Index.ACCOUNT.getName()));
@@ -146,7 +146,7 @@ public void AddLiteralOnGroupKeyShouldPass() {
         schema("gender", null, "text"),
         schema("age+10", null, "long"),
         schema("max(balance)", "max", "long"));
-    verifyDataRows(response, rows("m", 30, 49568), rows("m", 31, 49433));
+    verifyDataRows(response, rows("M", 30, 49568), rows("M", 31, 49433));
   }
 
   @Test
@@ -156,7 +156,7 @@ public void logWithAddLiteralOnGroupKeyShouldPass() {
             String.format(
                 "SELECT gender, Log(age+10) as logAge, max(balance) as max "
                     + "FROM %s "
-                    + "WHERE gender = 'm' and age < 22 "
+                    + "WHERE gender = 'M' and age < 22 "
                     + "GROUP BY gender, age "
                     + "ORDER BY age",
                 Index.ACCOUNT.getName()));
@@ -167,7 +167,7 @@ public void logWithAddLiteralOnGroupKeyShouldPass() {
         schema("Log(age+10)", "logAge", "double"),
         schema("max(balance)", "max", "long"));
     verifyDataRows(
-        response, rows("m", 3.4011973816621555d, 49568), rows("m", 3.4339872044851463d, 49433));
+        response, rows("M", 3.4011973816621555d, 49568), rows("M", 3.4339872044851463d, 49433));
   }
 
   @Test
@@ -177,7 +177,7 @@ public void logWithAddLiteralOnGroupKeyAndMaxSubtractLiteralShouldPass() {
             String.format(
                 "SELECT gender, Log(age+10) as logAge, max(balance) - 100 as max "
                     + "FROM %s "
-                    + "WHERE gender = 'm' and age < 22 "
+                    + "WHERE gender = 'M' and age < 22 "
                     + "GROUP BY gender, age "
                     + "ORDER BY age",
                 Index.ACCOUNT.getName()));
@@ -188,7 +188,7 @@ public void logWithAddLiteralOnGroupKeyAndMaxSubtractLiteralShouldPass() {
         schema("Log(age+10)", "logAge", "double"),
         schema("max(balance) - 100", "max", "long"));
     verifyDataRows(
-        response, rows("m", 3.4011973816621555d, 49468), rows("m", 3.4339872044851463d, 49333));
+        response, rows("M", 3.4011973816621555d, 49468), rows("M", 3.4339872044851463d, 49333));
   }
 
   /** The date is in JDBC format. */