Use left join in expand to keep documents where their expanded array field is empty

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

Author: yuancu

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

@@ -1030,7 +1030,9 @@ public RelNode visitExpand(Expand expand, CalcitePlanContext context) {
     // array field.
     // The last parameter has to refer to the array to be expanded on the left side. It will
     // be used by the right side to correlate with the left side.
-    relBuilder.correlate(JoinRelType.INNER, correlVariable.get().id, List.of(arrayFieldRex));
+    // Using left join to keep the records where the array field is empty. The corresponding
+    // field in the result will be null.
+    relBuilder.correlate(JoinRelType.LEFT, correlVariable.get().id, List.of(arrayFieldRex));
 
     // 9. Remove the original array field from the output.
     // TODO: RFC: should we keep the original array field when alias is present?

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

@@ -19,6 +19,7 @@
 import org.json.JSONObject;
 import org.junit.Ignore;
 import org.junit.jupiter.api.Test;
+import org.opensearch.client.Request;
 import org.opensearch.sql.ppl.PPLIntegTestCase;
 
 public class CalciteExpandCommandIT extends PPLIntegTestCase {
@@ -41,7 +42,6 @@ public void testExpandOnNested() throws Exception {
         schema("age", "bigint"),
         schema("id", "bigint"),
         schema("address", "struct"));
-    verifyNumOfRows(response, 11);
     verifyDataRows(
         response,
         rows(
@@ -276,4 +276,126 @@ public void testExpandWithEval() throws Exception {
         schema("addr", "struct"));
     verifyNumOfRows(response, 11);
   }
+
+  @Test
+  public void testExpandEmptyArray() throws Exception {
+    final int docId = 6;
+    Request insertRequest =
+        new Request(
+            "PUT", String.format("/%s/_doc/%d?refresh=true", TEST_INDEX_NESTED_SIMPLE, docId));
+    insertRequest.setJsonEntity("{\"name\":\"ben\",\"age\":47, \"id\": 437821, \"address\":[]}\n");
+    client().performRequest(insertRequest);
+
+    JSONObject response =
+        executeQuery(String.format("source=%s | expand address", TEST_INDEX_NESTED_SIMPLE));
+    verifySchema(
+        response,
+        schema("name", "string"),
+        schema("age", "bigint"),
+        schema("id", "bigint"),
+        schema("address", "struct"));
+    verifyDataRows(
+        response,
+        rows(
+            "abbas",
+            null,
+            24,
+            new JSONObject()
+                .put("city", "New york city")
+                .put("state", "NY")
+                .put("moveInDate", new JSONObject().put("dateAndTime", "1984-04-12 09:07:42"))),
+        rows(
+            "abbas",
+            null,
+            24,
+            new JSONObject()
+                .put("city", "bellevue")
+                .put("state", "WA")
+                .put(
+                    "moveInDate",
+                    new JSONArray()
+                        .put(new JSONObject().put("dateAndTime", "2023-05-03 08:07:42"))
+                        .put(new JSONObject().put("dateAndTime", "2001-11-11 04:07:44")))),
+        rows(
+            "abbas",
+            null,
+            24,
+            new JSONObject()
+                .put("city", "seattle")
+                .put("state", "WA")
+                .put("moveInDate", new JSONObject().put("dateAndTime", "1966-03-19 03:04:55"))),
+        rows(
+            "abbas",
+            null,
+            24,
+            new JSONObject()
+                .put("city", "chicago")
+                .put("state", "IL")
+                .put("moveInDate", new JSONObject().put("dateAndTime", "2011-06-01 01:01:42"))),
+        rows(
+            "chen",
+            null,
+            32,
+            new JSONObject()
+                .put("city", "Miami")
+                .put("state", "Florida")
+                .put("moveInDate", new JSONObject().put("dateAndTime", "1901-08-11 04:03:33"))),
+        rows(
+            "chen",
+            null,
+            32,
+            new JSONObject()
+                .put("city", "los angeles")
+                .put("state", "CA")
+                .put("moveInDate", new JSONObject().put("dateAndTime", "2023-05-03 08:07:42"))),
+        rows(
+            "peng",
+            null,
+            26,
+            new JSONObject()
+                .put("city", "san diego")
+                .put("state", "CA")
+                .put("moveInDate", new JSONObject().put("dateAndTime", "2001-11-11 04:07:44"))),
+        rows(
+            "peng",
+            null,
+            26,
+            new JSONObject()
+                .put("city", "austin")
+                .put("state", "TX")
+                .put("moveInDate", new JSONObject().put("dateAndTime", "1977-07-13 09:04:41"))),
+        rows(
+            "andy",
+            4,
+            19,
+            new JSONObject()
+                .put("city", "houston")
+                .put("state", "TX")
+                .put("moveInDate", new JSONObject().put("dateAndTime", "1933-12-12 05:05:45"))),
+        rows(
+            "david",
+            null,
+            25,
+            new JSONObject()
+                .put("city", "raleigh")
+                .put("state", "NC")
+                .put("moveInDate", new JSONObject().put("dateAndTime", "1909-06-17 01:04:21"))),
+        rows(
+            "david",
+            null,
+            25,
+            new JSONObject()
+                .put("city", "charlotte")
+                .put("state", "SC")
+                .put(
+                    "moveInDate",
+                    new JSONArray()
+                        .put(new JSONObject().put("dateAndTime", "2001-11-11 04:07:44")))),
+        rows("ben", 437821, 47, null));
+
+    verifyNumOfRows(response, 12);
+    Request deleteRequest =
+        new Request("DELETE", String.format("/%s/_doc/%d", TEST_INDEX_NESTED_SIMPLE, docId));
+    client().performRequest(deleteRequest);
+  }
 }