Implement ppl scalar subquery command with Calcite

core/src/main/java/org/opensearch/sql/calcite/CalciteRelNodeVisitor.java

@@ -32,12 +32,11 @@
 import org.apache.calcite.util.Holder;
 import org.checkerframework.checker.nullness.qual.Nullable;
 import org.opensearch.sql.ast.AbstractNodeVisitor;
+import org.opensearch.sql.ast.Node;
 import org.opensearch.sql.ast.expression.AllFields;
 import org.opensearch.sql.ast.expression.Argument;
-import org.opensearch.sql.ast.expression.Compare;
 import org.opensearch.sql.ast.expression.Field;
 import org.opensearch.sql.ast.expression.Let;
-import org.opensearch.sql.ast.expression.Not;
 import org.opensearch.sql.ast.expression.QualifiedName;
 import org.opensearch.sql.ast.expression.UnresolvedExpression;
 import org.opensearch.sql.ast.expression.subquery.ExistsSubquery;
@@ -111,31 +110,38 @@ public RelNode visitFilter(Filter node, CalcitePlanContext context) {
     return context.relBuilder.peek();
   }
 
-  private boolean containsExistsSubquery(Object condition) {
-    if (condition instanceof ExistsSubquery) {
+  private boolean containsExistsSubquery(Node expr) {
+    if (expr == null) {
+      return false;
+    }
+    if (expr instanceof ExistsSubquery) {
       return true;
     }
-    if (condition instanceof Not n) {
-      return containsExistsSubquery(n.getExpression());
+    if (expr instanceof Let l) {
+      return containsExistsSubquery(l.getExpression());
     }
-    if (condition instanceof Compare c) {
-      return containsExistsSubquery(c.getLeft()) || containsExistsSubquery(c.getRight());
+    for (Node child : expr.getChild()) {
+      if (containsExistsSubquery(child)) {
+        return true;
+      }
     }
     return false;
   }
 
-  private boolean containsScalarSubquery(Object expr) {
+  private boolean containsScalarSubquery(Node expr) {
+    if (expr == null) {
+      return false;
+    }
     if (expr instanceof ScalarSubquery) {
       return true;
     }
-    if (expr instanceof Not n) {
-      return containsScalarSubquery(n.getExpression());
-    }
     if (expr instanceof Let l) {
       return containsScalarSubquery(l.getExpression());
     }
-    if (expr instanceof Compare c) {
-      return containsScalarSubquery(c.getLeft()) || containsScalarSubquery(c.getRight());
+    for (Node child : expr.getChild()) {
+      if (containsScalarSubquery(child)) {
+        return true;
+      }
     }
     return false;
   }

integ-test/src/test/java/org/opensearch/sql/calcite/standalone/CalcitePPLScalaSubqueryIT.java → integ-test/src/test/java/org/opensearch/sql/calcite/standalone/CalcitePPLScalarSubqueryIT.java

@@ -18,7 +18,7 @@
 import org.junit.Test;
 import org.opensearch.client.Request;
 
-public class CalcitePPLScalaSubqueryIT extends CalcitePPLIntegTestCase {
+public class CalcitePPLScalarSubqueryIT extends CalcitePPLIntegTestCase {
 
   @Override
   public void init() throws IOException {
@@ -244,7 +244,7 @@ public void testDisjunctiveCorrelatedScalarSubquery() {
   }
 
   @Test
-  public void testTwoCorrelatedScalarSubqueriesInOr() {
+  public void testTwoUncorrelatedScalarSubqueriesInOr() {
     JSONObject result =
         executeQuery(
             String.format(
@@ -262,6 +262,31 @@ public void testTwoCorrelatedScalarSubqueriesInOr() {
     verifyDataRows(result, rows(1002, "John"), rows(1006, "Tommy"));
   }
 
+  @Test
+  public void testTwoCorrelatedScalarSubqueriesInOr() {
+    JSONObject result =
+        executeQuery(
+            String.format(
+                """
+                   source = %s
+                   | where id = [
+                       source = %s | where id = uid | stats max(uid)
+                     ] OR id = [
+                       source = %s | sort uid | where department = 'DATA' | stats min(uid)
+                     ]
+                   | fields id, name
+                   """,
+                TEST_INDEX_WORKER, TEST_INDEX_WORK_INFORMATION, TEST_INDEX_WORK_INFORMATION));
+    verifySchema(result, schema("id", "integer"), schema("name", "string"));
+    verifyDataRows(
+        result,
+        rows(1000, "Jake"),
+        rows(1002, "John"),
+        rows(1003, "David"),
+        rows(1005, "Jane"),
+        rows(1006, "Tommy"));
+  }
+
   @Test
   public void testNestedScalarSubquery() {
     JSONObject result =

ppl/src/test/java/org/opensearch/sql/ppl/calcite/CalcitePPLScalarSubqueryTest.java

@@ -258,7 +258,7 @@ public void testTwoScalarSubqueriesInOr() {
             + "  LogicalSort(sort0=[$2], dir0=[DESC])\n"
             + "    LogicalFilter(condition=[>($1, 1000.0E0:DOUBLE)])\n"
             + "      LogicalTableScan(table=[[scott, SALGRADE]])\n"
-            + "})))])\n"
+            + "})))], variablesSet=[[$cor0]])\n"
             + "  LogicalTableScan(table=[[scott, EMP]])\n";
     verifyLogical(root, expectedLogical);