Support JOIN query on object field with unexpanded name (#1229)

* Resolve sub object field in search hit source Signed-off-by: Chen Dai <[email protected]> * Rename to unexpanded object Signed-off-by: Chen Dai <[email protected]> * Update IT with where condition Signed-off-by: Chen Dai <[email protected]> * Fix test index mapping Signed-off-by: Chen Dai <[email protected]> Signed-off-by: Chen Dai <[email protected]> (cherry picked from commit 151f4cc)
opensearch-project · Jan 9, 2023 · a1ec0bb · a1ec0bb
1 parent 628be97
commit a1ec0bb
Show file tree

Hide file tree

Showing 8 changed files with 139 additions and 0 deletions.
diff --git a/integ-test/src/test/java/org/opensearch/sql/legacy/HashJoinIT.java b/integ-test/src/test/java/org/opensearch/sql/legacy/HashJoinIT.java
@@ -9,6 +9,11 @@
 import static org.hamcrest.Matchers.equalTo;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_ACCOUNT;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_GAME_OF_THRONES;
+import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_UNEXPANDED_OBJECT;
+import static org.opensearch.sql.util.MatcherUtils.columnName;
+import static org.opensearch.sql.util.MatcherUtils.rows;
+import static org.opensearch.sql.util.MatcherUtils.verifyColumn;
+import static org.opensearch.sql.util.MatcherUtils.verifyDataRows;
 
 import java.io.IOException;
 import java.util.HashSet;
@@ -55,6 +60,7 @@ public class HashJoinIT extends SQLIntegTestCase {
   protected void init() throws Exception {
     loadIndex(Index.ACCOUNT);
     loadIndex(Index.GAME_OF_THRONES);
+    loadIndex(Index.UNEXPANDED_OBJECT);
   }
 
   @Test
@@ -69,6 +75,21 @@ public void leftJoin() throws IOException {
     testJoin("LEFT JOIN");
   }
 
+  @Test
+  public void innerJoinUnexpandedObjectField() {
+    String query = String.format(Locale.ROOT,
+        "SELECT " +
+            "a.id.serial, b.id.serial " +
+            "FROM %1$s AS a " +
+            "JOIN %1$s AS b " +
+            "ON a.id.serial = b.attributes.hardware.correlate_id " +
+            "WHERE b.attributes.hardware.platform = 'Linux' ",
+        TEST_INDEX_UNEXPANDED_OBJECT);
+
+    JSONObject response = executeJdbcRequest(query);
+    verifyDataRows(response, rows(3, 1), rows(3, 3));
+  }
+
   @Test
   public void innerJoinWithObjectField() throws IOException {
     testJoinWithObjectField("INNER JOIN", "");

diff --git a/integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java b/integ-test/src/test/java/org/opensearch/sql/legacy/SQLIntegTestCase.java
@@ -57,6 +57,7 @@
 import static org.opensearch.sql.legacy.TestUtils.getPhraseIndexMapping;
 import static org.opensearch.sql.legacy.TestUtils.getResponseBody;
 import static org.opensearch.sql.legacy.TestUtils.getStringIndexMapping;
+import static org.opensearch.sql.legacy.TestUtils.getUnexpandedObjectIndexMapping;
 import static org.opensearch.sql.legacy.TestUtils.getWeblogsIndexMapping;
 import static org.opensearch.sql.legacy.TestUtils.isIndexExist;
 import static org.opensearch.sql.legacy.TestUtils.loadDataByRestClient;
@@ -517,6 +518,10 @@ public enum Index {
         "joinType",
         getJoinTypeIndexMapping(),
         "src/test/resources/join_objects.json"),
+    UNEXPANDED_OBJECT(TestsConstants.TEST_INDEX_UNEXPANDED_OBJECT,
+        "unexpandedObject",
+        getUnexpandedObjectIndexMapping(),
+        "src/test/resources/unexpanded_objects.json"),
     BANK(TestsConstants.TEST_INDEX_BANK,
         "account",
         getBankIndexMapping(),

diff --git a/integ-test/src/test/java/org/opensearch/sql/legacy/TestUtils.java b/integ-test/src/test/java/org/opensearch/sql/legacy/TestUtils.java
@@ -188,6 +188,11 @@ public static String getJoinTypeIndexMapping() {
     return getMappingFile(mappingFile);
   }
 
+  public static String getUnexpandedObjectIndexMapping() {
+    String mappingFile = "unexpanded_object_index_mapping.json";
+    return getMappingFile(mappingFile);
+  }
+
   public static String getBankIndexMapping() {
     String mappingFile = "bank_index_mapping.json";
     return getMappingFile(mappingFile);

diff --git a/integ-test/src/test/java/org/opensearch/sql/legacy/TestsConstants.java b/integ-test/src/test/java/org/opensearch/sql/legacy/TestsConstants.java
@@ -36,6 +36,7 @@ public class TestsConstants {
       TEST_INDEX + "_nested_type_with_quotes";
   public final static String TEST_INDEX_EMPLOYEE_NESTED = TEST_INDEX + "_employee_nested";
   public final static String TEST_INDEX_JOIN_TYPE = TEST_INDEX + "_join_type";
+  public final static String TEST_INDEX_UNEXPANDED_OBJECT = TEST_INDEX + "_unexpanded_object";
   public final static String TEST_INDEX_BANK = TEST_INDEX + "_bank";
   public final static String TEST_INDEX_BANK_TWO = TEST_INDEX_BANK + "_two";
   public final static String TEST_INDEX_BANK_WITH_NULL_VALUES =

diff --git a/integ-test/src/test/resources/indexDefinitions/unexpanded_object_index_mapping.json b/integ-test/src/test/resources/indexDefinitions/unexpanded_object_index_mapping.json
@@ -0,0 +1,27 @@
+{
+  "mappings": {
+    "properties": {
+      "id": {
+        "properties": {
+          "serial": {
+            "type": "integer"
+          }
+        }
+      },
+      "attributes": {
+        "properties": {
+          "hardware": {
+            "properties": {
+              "correlate_id": {
+                "type": "integer"
+              },
+              "platform": {
+                "type": "keyword"
+              }
+            }
+          }
+        }
+      }
+    }
+  }
+}
diff --git a/integ-test/src/test/resources/unexpanded_objects.json b/integ-test/src/test/resources/unexpanded_objects.json
@@ -0,0 +1,8 @@
+{"index":{"_id":"1"}}
+{"id.serial" : 1 , "attributes.hardware.correlate_id": 3, "attributes.hardware.platform": "Linux"}
+{"index":{"_id":"2"}}
+{"id.serial" : 2 , "attributes.hardware.correlate_id": 3, "attributes.hardware.platform": "Windows"}
+{"index":{"_id":"3"}}
+{"id.serial" : 3 , "attributes.hardware.correlate_id": 3, "attributes.hardware.platform": "Linux"}
+{"index":{"_id":"4"}}
+{"id.serial" : 4 , "attributes.hardware.correlate_id": 100, "attributes.hardware.platform": "Linux"}
diff --git a/.../main/java/org/opensearch/sql/legacy/query/planner/physical/node/scroll/SearchHitRow.java b/.../main/java/org/opensearch/sql/legacy/query/planner/physical/node/scroll/SearchHitRow.java
@@ -134,6 +134,13 @@ private Object getValueOfPath(Object source, String path, boolean isIgnoreFirstD
         if (dot == -1) {
             return ((Map) source).get(path);
         }
+
+        // Object field name maybe unexpanded without recursive object structure
+        // ex. {"a.b.c": value} instead of {"a": {"b": {"c": value}}}}
+        if (((Map) source).containsKey(path)) {
+            return ((Map) source).get(path);
+        }
+
         return getValueOfPath(
                 ((Map) source).get(path.substring(0, dot)),
                 path.substring(dot + 1),

diff --git a/...t/java/org/opensearch/sql/legacy/query/planner/physical/node/scroll/SearchHitRowTest.java b/...t/java/org/opensearch/sql/legacy/query/planner/physical/node/scroll/SearchHitRowTest.java
@@ -0,0 +1,65 @@
+/*
+ * Copyright OpenSearch Contributors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package org.opensearch.sql.legacy.query.planner.physical.node.scroll;
+
+import static org.junit.Assert.assertEquals;
+import static org.opensearch.sql.legacy.query.planner.physical.Row.RowKey;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Test;
+import org.opensearch.common.bytes.BytesArray;
+import org.opensearch.search.SearchHit;
+
+public class SearchHitRowTest {
+
+  @Test
+  public void testKeyWithObjectField() {
+    SearchHit hit = new SearchHit(1);
+    hit.sourceRef(new BytesArray("{\"id\": {\"serial\": 3}}"));
+    SearchHitRow row = new SearchHitRow(hit, "a");
+    RowKey key = row.key(new String[]{"id.serial"});
+
+    Object[] data = key.keys();
+    assertEquals(1, data.length);
+    assertEquals(3, data[0]);
+  }
+
+  @Test
+  public void testKeyWithUnexpandedObjectField() {
+    SearchHit hit = new SearchHit(1);
+    hit.sourceRef(new BytesArray("{\"attributes.hardware.correlate_id\": 10}"));
+    SearchHitRow row = new SearchHitRow(hit, "a");
+    RowKey key = row.key(new String[]{"attributes.hardware.correlate_id"});
+
+    Object[] data = key.keys();
+    assertEquals(1, data.length);
+    assertEquals(10, data[0]);
+  }
+
+  @Test
+  public void testRetainWithObjectField() {
+    SearchHit hit = new SearchHit(1);
+    hit.sourceRef(new BytesArray("{\"a.id\": {\"serial\": 3}}"));
+    SearchHitRow row = new SearchHitRow(hit, "");
+    row.retain(ImmutableMap.of("a.id.serial", ""));
+
+    SearchHit expected = new SearchHit(1);
+    expected.sourceRef(new BytesArray("{\"a.id\": {\"serial\": 3}}"));
+    assertEquals(expected, row.data());
+  }
+
+  @Test
+  public void testRetainWithUnexpandedObjectField() {
+    SearchHit hit = new SearchHit(1);
+    hit.sourceRef(new BytesArray("{\"a.attributes.hardware.correlate_id\": 10}"));
+    SearchHitRow row = new SearchHitRow(hit, "");
+    row.retain(ImmutableMap.of("a.attributes.hardware.correlate_id", ""));
+
+    SearchHit expected = new SearchHit(1);
+    expected.sourceRef(new BytesArray("{\"a.attributes.hardware.correlate_id\": 10}"));
+    assertEquals(expected, row.data());
+  }
+}