Implement ppl relation subquery command with Calcite (#3378)

Signed-off-by: Lantao Jin <ltjin@amazon.com>

Author: LantaoJin

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

@@ -22,10 +22,4 @@ public void init() throws IOException {
   @Ignore
   @Override
   public void testSortIpField() throws IOException {}
-
-  // TODO: Fix incorrect results for NULL values, addressed by issue:
-  // https://github.com/opensearch-project/sql/issues/3375
-  @Ignore
-  @Override
-  public void testSortWithNullValue() throws IOException {}
 }

integ-test/src/test/java/org/opensearch/sql/calcite/standalone/CalcitePPLJoinIT.java

@@ -18,6 +18,7 @@
 import org.junit.Ignore;
 import org.junit.Test;
 import org.opensearch.client.Request;
+import org.opensearch.sql.legacy.TestsConstants;
 
 public class CalcitePPLJoinIT extends CalcitePPLIntegTestCase {
 
@@ -29,22 +30,22 @@ public void init() throws IOException {
     loadIndex(Index.OCCUPATION);
     loadIndex(Index.HOBBIES);
     Request request1 =
-        new Request("PUT", "/opensearch-sql_test_index_state_country/_doc/5?refresh=true");
+        new Request("PUT", "/" + TestsConstants.TEST_INDEX_STATE_COUNTRY + "/_doc/5?refresh=true");
     request1.setJsonEntity(
         "{\"name\":\"Jim\",\"age\":27,\"state\":\"B.C\",\"country\":\"Canada\",\"year\":2023,\"month\":4}");
     client().performRequest(request1);
     Request request2 =
-        new Request("PUT", "/opensearch-sql_test_index_state_country/_doc/6?refresh=true");
+        new Request("PUT", "/" + TestsConstants.TEST_INDEX_STATE_COUNTRY + "/_doc/6?refresh=true");
     request2.setJsonEntity(
         "{\"name\":\"Peter\",\"age\":57,\"state\":\"B.C\",\"country\":\"Canada\",\"year\":2023,\"month\":4}");
     client().performRequest(request2);
     Request request3 =
-        new Request("PUT", "/opensearch-sql_test_index_state_country/_doc/7?refresh=true");
+        new Request("PUT", "/" + TestsConstants.TEST_INDEX_STATE_COUNTRY + "/_doc/7?refresh=true");
     request3.setJsonEntity(
         "{\"name\":\"Rick\",\"age\":70,\"state\":\"B.C\",\"country\":\"Canada\",\"year\":2023,\"month\":4}");
     client().performRequest(request3);
     Request request4 =
-        new Request("PUT", "/opensearch-sql_test_index_state_country/_doc/8?refresh=true");
+        new Request("PUT", "/" + TestsConstants.TEST_INDEX_STATE_COUNTRY + "/_doc/8?refresh=true");
     request4.setJsonEntity(
         "{\"name\":\"David\",\"age\":40,\"state\":\"Washington\",\"country\":\"USA\",\"year\":2023,\"month\":4}");
     client().performRequest(request4);
@@ -213,9 +214,9 @@ public void testComplexLeftJoin() {
         actual,
         rows("Jane", 20, "Quebec", "Canada", "Scientist", "Canada", 90000),
         rows("John", 25, "Ontario", "Canada", "Doctor", "Canada", 120000),
-        rows("Jim", 27, "B.C", "Canada", null, null, 0),
-        rows("Peter", 57, "B.C", "Canada", null, null, 0),
-        rows("Rick", 70, "B.C", "Canada", null, null, 0));
+        rows("Jim", 27, "B.C", "Canada", null, null, null),
+        rows("Peter", 57, "B.C", "Canada", null, null, null),
+        rows("Rick", 70, "B.C", "Canada", null, null, null));
   }
 
   @Test
@@ -240,10 +241,10 @@ public void testComplexRightJoin() {
         actual,
         rows("Jane", 20, "Quebec", "Canada", "Scientist", "Canada", 90000),
         rows("John", 25, "Ontario", "Canada", "Doctor", "Canada", 120000),
-        rows(null, 0, null, null, "Engineer", "England", 100000),
-        rows(null, 0, null, null, "Artist", "USA", 70000),
-        rows(null, 0, null, null, "Doctor", "USA", 120000),
-        rows(null, 0, null, null, "Unemployed", "Canada", 0));
+        rows(null, null, null, null, "Engineer", "England", 100000),
+        rows(null, null, null, null, "Artist", "USA", 70000),
+        rows(null, null, null, null, "Doctor", "USA", 120000),
+        rows(null, null, null, null, "Unemployed", "Canada", 0));
   }
 
   @Test
@@ -392,6 +393,54 @@ public void testMultipleJoins() {
                 TEST_INDEX_STATE_COUNTRY));
   }
 
+  @Ignore // TODO seems a calcite bug
+  public void testMultipleJoinsWithRelationSubquery() {
+    JSONObject actual =
+        executeQuery(
+            String.format(
+                """
+                   source = %s
+                   | where country = 'Canada' OR country = 'England'
+                   | inner join left=a, right=b
+                       ON a.name = b.name AND a.year = 2023 AND a.month = 4 AND b.year = 2023 AND b.month = 4
+                       [
+                         source = %s
+                       ]
+                   | eval a_name = a.name
+                   | eval a_country = a.country
+                   | eval b_country = b.country
+                   | fields a_name, age, state, a_country, occupation, b_country, salary
+                   | left join left=a, right=b
+                       ON a.a_name = b.name
+                       [
+                         source = %s
+                       ]
+                   | eval aa_country = a.a_country
+                   | eval ab_country = a.b_country
+                   | eval bb_country = b.country
+                   | fields a_name, age, state, aa_country, occupation, ab_country, salary, bb_country, hobby, language
+                   | cross join left=a, right=b
+                       [
+                         source = %s
+                       ]
+                   | eval new_country = a.aa_country
+                   | eval new_salary = b.salary
+                   | stats avg(new_salary) as avg_salary by span(age, 5) as age_span, state
+                   | left semi join left=a, right=b
+                       ON a.state = b.state
+                       [
+                         source = %s
+                       ]
+                   | eval new_avg_salary = floor(avg_salary)
+                   | fields state, age_span, new_avg_salary
+                   """,
+                TEST_INDEX_STATE_COUNTRY,
+                TEST_INDEX_OCCUPATION,
+                TEST_INDEX_HOBBIES,
+                TEST_INDEX_OCCUPATION,
+                TEST_INDEX_STATE_COUNTRY));
+  }
+
   @Test
   public void testMultipleJoinsWithoutTableAliases() {
     JSONObject actual =
@@ -442,7 +491,7 @@ public void testMultipleJoinsWithPartTableAliases() {
   }
 
   @Test
-  public void testMultipleJoinsWithSelfJoin1() {
+  public void testMultipleJoinsWithSelfJoin() {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -469,8 +518,8 @@ public void testMultipleJoinsWithSelfJoin1() {
         rows("John", "John", "John", "John"));
   }
 
-  @Ignore // TODO table subquery not support
-  public void testMultipleJoinsWithSelfJoin2() {
+  @Test
+  public void testMultipleJoinsWithSubquerySelfJoin() {
     JSONObject actual =
         executeQuery(
             String.format(
@@ -481,10 +530,24 @@ public void testMultipleJoinsWithSelfJoin2() {
                 TEST_INDEX_OCCUPATION,
                 TEST_INDEX_HOBBIES,
                 TEST_INDEX_STATE_COUNTRY));
+    verifySchema(
+        actual,
+        schema("name", "string"),
+        schema("name0", "string"),
+        schema("name1", "string"),
+        schema("name2", "string"));
+    verifyDataRows(
+        actual,
+        rows("David", "David", "David", "David"),
+        rows("David", "David", "David", "David"),
+        rows("Hello", "Hello", "Hello", "Hello"),
+        rows("Jake", "Jake", "Jake", "Jake"),
+        rows("Jane", "Jane", "Jane", "Jane"),
+        rows("John", "John", "John", "John"));
   }
 
   @Test
-  public void testCheckAccessTheReferenceByAliases1() {
+  public void testCheckAccessTheReferenceByAliases() {
     String res1 =
         execute(
             String.format(
@@ -520,8 +583,8 @@ public void testCheckAccessTheReferenceByAliases1() {
     assertEquals(res4, res5);
   }
 
-  @Ignore // TODO table subquery not support
-  public void testCheckAccessTheReferenceByAliases2() {
+  @Test
+  public void testCheckAccessTheReferenceBySubqueryAliases() {
     String res1 =
         execute(
             String.format(
@@ -559,7 +622,7 @@ public void testCheckAccessTheReferenceByAliases2() {
   }
 
   @Test
-  public void testCheckAccessTheReferenceByAliases3() {
+  public void testCheckAccessTheReferenceByOverrideAliases() {
     String res1 =
         execute(
             String.format(
@@ -581,4 +644,109 @@ public void testCheckAccessTheReferenceByAliases3() {
     assertEquals(res1, res2);
     assertEquals(res1, res3);
   }
+
+  @Test
+  public void testCheckAccessTheReferenceByOverrideSubqueryAliases() {
+    String res1 =
+        execute(
+            String.format(
+                "source = %s | JOIN left = t1 right = t2 ON t1.name = t2.name [ source = %s as tt ]"
+                    + " | fields tt.name",
+                TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
+    String res2 =
+        execute(
+            String.format(
+                "source = %s | JOIN left = t1 right = t2 ON t1.name = t2.name [ source = %s as tt ]"
+                    + " as t2 | fields tt.name",
+                TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
+    String res3 =
+        execute(
+            String.format(
+                "source = %s | JOIN left = t1 right = t2 ON t1.name = t2.name [ source = %s ] as tt"
+                    + " | fields tt.name",
+                TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
+    assertEquals(res1, res2);
+    assertEquals(res1, res3);
+  }
+
+  @Test
+  public void testCheckAccessTheReferenceByOverrideSubqueryAliases2() {
+    String res1 =
+        execute(
+            String.format(
+                "source = %s | JOIN left = t1 right = t2 ON t1.name = t2.name [ source = %s as tt ]"
+                    + " | fields t2.name",
+                TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
+    String res2 =
+        execute(
+            String.format(
+                "source = %s | JOIN left = t1 right = t2 ON t1.name = t2.name [ source = %s as tt ]"
+                    + " as t2 | fields t2.name",
+                TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
+    String res3 =
+        execute(
+            String.format(
+                "source = %s | JOIN left = t1 right = t2 ON t1.name = t2.name [ source = %s ] as tt"
+                    + " | fields t2.name",
+                TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
+    assertEquals(res1, res2);
+    assertEquals(res1, res3);
+  }
+
+  @Test
+  public void testInnerJoinWithRelationSubquery() {
+    JSONObject actual =
+        executeQuery(
+            String.format(
+                """
+                   source = %s
+                   | where country = 'USA' OR country = 'England'
+                   | inner join left=a, right=b
+                       ON a.name = b.name
+                       [
+                         source = %s
+                         | where salary > 0
+                         | fields name, country, salary
+                         | sort salary
+                         | head 3
+                       ]
+                   | stats avg(salary) by span(age, 10) as age_span, b.country
+                   """,
+                TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
+    verifySchema(
+        actual,
+        schema("b.country", "string"),
+        schema("age_span", "double"),
+        schema("avg(salary)", "double"));
+    verifyDataRows(actual, rows("USA", 30, 70000.0), rows("England", 70, 100000));
+  }
+
+  @Test
+  public void testLeftJoinWithRelationSubquery() {
+    JSONObject actual =
+        executeQuery(
+            String.format(
+                """
+                   source = %s
+                   | where country = 'USA' OR country = 'England'
+                   | left join left=a, right=b
+                       ON a.name = b.name
+                       [
+                         source = %s
+                         | where salary > 0
+                         | fields name, country, salary
+                         | sort salary
+                         | head 3
+                       ]
+                   | stats avg(salary) by span(age, 10) as age_span, b.country
+                   """,
+                TEST_INDEX_STATE_COUNTRY, TEST_INDEX_OCCUPATION));
+    verifySchema(
+        actual,
+        schema("b.country", "string"),
+        schema("age_span", "double"),
+        schema("avg(salary)", "double"));
+    verifyDataRows(
+        actual, rows("USA", 30, 70000.0), rows("England", 70, 100000), rows(null, 40, 0));
+  }
 }

integ-test/src/test/java/org/opensearch/sql/calcite/standalone/CalcitePPLSortIT.java

@@ -6,9 +6,11 @@
 package org.opensearch.sql.calcite.standalone;
 
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK;
+import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_BANK_WITH_NULL_VALUES;
 import static org.opensearch.sql.util.MatcherUtils.rows;
 import static org.opensearch.sql.util.MatcherUtils.schema;
 import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
+import static org.opensearch.sql.util.MatcherUtils.verifyOrder;
 import static org.opensearch.sql.util.MatcherUtils.verifySchema;
 
 import java.io.IOException;
@@ -22,6 +24,7 @@ public void init() throws IOException {
     super.init();
 
     loadIndex(Index.BANK);
+    loadIndex(Index.BANK_WITH_NULL_VALUES);
   }
 
   @Test
@@ -191,4 +194,22 @@ public void testSortAgeNameAndFieldsNameAge() {
         rows("Amber JOHnny", 32),
         rows("Nanette", 28));
   }
+
+  @Test
+  public void testSortWithNullValue() throws IOException {
+    JSONObject result =
+        executeQuery(
+            String.format(
+                "source=%s | sort balance | fields firstname, balance",
+                TEST_INDEX_BANK_WITH_NULL_VALUES));
+    verifyOrder(
+        result,
+        rows("Dale", 4180),
+        rows("Nanette", 32838),
+        rows("Amber JOHnny", 39225),
+        rows("Dillard", 48086),
+        rows("Hattie", null),
+        rows("Elinor", null),
+        rows("Virginia", null));
+  }
 }

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

@@ -167,7 +167,7 @@ public CalciteOpenSearchIndexScan pushDownFilter(Filter filter) {
       // TODO: handle the case where condition contains a score function
       return newScan;
     } catch (Exception e) {
-      LOG.warn("Cannot analyze the filter condition {}", filter.getCondition(), e);
+      LOG.warn("Cannot pushdown the filter condition {}, ", filter.getCondition());
     }
     return null;
   }