Support basic nested queries when applying filter on a nested filed

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

Author: yuancu

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

@@ -8,6 +8,7 @@
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_ACCOUNT;
 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.legacy.TestsConstants.TEST_INDEX_DEEP_NESTED;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_LOGS;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_NESTED_SIMPLE;
 import static org.opensearch.sql.legacy.TestsConstants.TEST_INDEX_OTEL_LOGS;
@@ -24,6 +25,7 @@
 import org.junit.Ignore;
 import org.junit.Test;
 import org.opensearch.sql.common.setting.Settings;
+import org.opensearch.sql.common.utils.StringUtils;
 import org.opensearch.sql.ppl.ExplainIT;
 
 public class CalciteExplainIT extends ExplainIT {
@@ -42,6 +44,7 @@ public void init() throws Exception {
     loadIndex(Index.WORKER);
     loadIndex(Index.WORK_INFORMATION);
     loadIndex(Index.WEBLOG);
+    loadIndex(Index.DEEP_NESTED);
   }
 
   @Override
@@ -1778,4 +1781,15 @@ public void testInternalItemAccessOnStructs() throws IOException {
                     + " info.dummy_sub_field",
                 TEST_INDEX_WEBLOGS)));
   }
+
+  @Test
+  public void testFilterOnNestedFields() throws IOException {
+    assertYamlEqualsIgnoreId(
+        loadExpectedPlan("filter_on_nested.yaml"),
+        explainQueryYaml(
+            StringUtils.format(
+                "source=%s | eval proj_name_len=length(projects.name) | fields projects.name,"
+                    + " proj_name_len | where proj_name_len > 29",
+                TEST_INDEX_DEEP_NESTED)));
+  }
 }

integ-test/src/test/resources/expectedOutput/calcite/filter_on_nested.yaml

@@ -0,0 +1,9 @@
+calcite:
+  logical: |
+    LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
+      LogicalFilter(condition=[>($1, 29)])
+        LogicalProject(projects.name=[$3], proj_name_len=[CHAR_LENGTH($3)])
+          CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_deep_nested]])
+  physical: |
+    EnumerableCalc(expr#0=[{inputs}], expr#1=[CHAR_LENGTH($t0)], proj#0..1=[{exprs}])
+      CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_deep_nested]], PushDownContext=[[PROJECT->[projects.name], SCRIPT->>(CHAR_LENGTH($0), 29), LIMIT->10000], OpenSearchRequestBuilder(sourceBuilder={"from":0,"size":10000,"timeout":"1m","query":{"nested":{"query":{"script":{"script":{"source":"{\"langType\":\"calcite\",\"script\":\"rO0ABXQCIXsKICAib3AiOiB7CiAgICAibmFtZSI6ICI+IiwKICAgICJraW5kIjogIkdSRUFURVJfVEhBTiIsCiAgICAic3ludGF4IjogIkJJTkFSWSIKICB9LAogICJvcGVyYW5kcyI6IFsKICAgIHsKICAgICAgIm9wIjogewogICAgICAgICJuYW1lIjogIkNIQVJfTEVOR1RIIiwKICAgICAgICAia2luZCI6ICJDSEFSX0xFTkdUSCIsCiAgICAgICAgInN5bnRheCI6ICJGVU5DVElPTiIKICAgICAgfSwKICAgICAgIm9wZXJhbmRzIjogWwogICAgICAgIHsKICAgICAgICAgICJkeW5hbWljUGFyYW0iOiAwLAogICAgICAgICAgInR5cGUiOiB7CiAgICAgICAgICAgICJ0eXBlIjogIlZBUkNIQVIiLAogICAgICAgICAgICAibnVsbGFibGUiOiB0cnVlLAogICAgICAgICAgICAicHJlY2lzaW9uIjogLTEKICAgICAgICAgIH0KICAgICAgICB9CiAgICAgIF0KICAgIH0sCiAgICB7CiAgICAgICJkeW5hbWljUGFyYW0iOiAxLAogICAgICAidHlwZSI6IHsKICAgICAgICAidHlwZSI6ICJJTlRFR0VSIiwKICAgICAgICAibnVsbGFibGUiOiBmYWxzZQogICAgICB9CiAgICB9CiAgXQp9\"}","lang":"opensearch_compounded_script","params":{"utcTimestamp": 0,"SOURCES":[0,2],"DIGESTS":["projects.name",29]}},"boost":1.0}},"path":"projects","ignore_unmapped":false,"score_mode":"none","boost":1.0}},"_source":{"includes":["projects.name"],"excludes":[]}}, requestedTotalSize=10000, pageSize=null, startFrom=0)])

integ-test/src/test/resources/expectedOutput/calcite_no_pushdown/filter_on_nested.yaml

@@ -0,0 +1,10 @@
+calcite:
+  logical: |
+    LogicalSystemLimit(fetch=[10000], type=[QUERY_SIZE_LIMIT])
+      LogicalFilter(condition=[>($1, 29)])
+        LogicalProject(projects.name=[$3], proj_name_len=[CHAR_LENGTH($3)])
+          CalciteLogicalIndexScan(table=[[OpenSearch, opensearch-sql_test_index_deep_nested]])
+  physical: |
+    EnumerableLimit(fetch=[10000])
+      EnumerableCalc(expr#0..15=[{inputs}], expr#16=[CHAR_LENGTH($t3)], expr#17=[29], expr#18=[>($t16, $t17)], projects.name=[$t3], proj_name_len=[$t16], $condition=[$t18])
+        CalciteEnumerableIndexScan(table=[[OpenSearch, opensearch-sql_test_index_deep_nested]])

integ-test/src/yamlRestTest/resources/rest-api-spec/test/issues/4508.yml

@@ -0,0 +1,94 @@
+setup:
+  - do:
+      indices.create:
+        index: test_nested_eval_filter
+        body:
+          mappings:
+            properties:
+              "id":
+                type: keyword
+              "items":
+                type: nested
+                properties:
+                  "name":
+                    type: keyword
+  - do:
+      bulk:
+        index: test_nested_eval_filter
+        refresh: true
+        body:
+          - '{"index": {"_id": "1"}}'
+          - '{"id": "order1", "items": [{"name": "apple"}]}'
+          - '{"index": {"_id": "2"}}'
+          - '{"id": "order2", "items": [{"name": "banana"}]}'
+          - '{"index": {"_id": "3"}}'
+          - '{"id": "order3", "items": [{"name": "orange"}]}'
+
+---
+"eval on nested field without filter":
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=test_nested_eval_filter | eval NameLen=LENGTH(items.name) | fields id, items.name, NameLen
+
+  - match: { total: 3 }
+  - match: {"schema": [{"name": "id", "type": "string"}, {"name": "items.name", "type": "string"}, {"name": "NameLen", "type": "int"}]}
+  - match: {"datarows": [["order1", "apple", 5], ["order2", "banana", 6], ["order3", "orange", 6]]}
+
+---
+"eval on nested field with filter on computed field":
+  - skip:
+      features:
+        - headers
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=test_nested_eval_filter | eval NameLen=LENGTH(items.name) | fields id, items.name, NameLen | where NameLen > 5
+
+  - match: { total: 2 }
+  - match: {"schema": [{"name": "id", "type": "string"}, {"name": "items.name", "type": "string"}, {"name": "NameLen", "type": "int"}]}
+  - match: {"datarows": [["order2", "banana", 6], ["order3", "orange", 6]]}
+
+---
+"comparison with regular field - eval and filter works correctly":
+  - skip:
+      features:
+        - headers
+  - do:
+      indices.create:
+        index: test_regular_eval_filter
+        body:
+          mappings:
+            properties:
+              "id":
+                type: keyword
+              "name":
+                type: keyword
+  - do:
+      bulk:
+        index: test_regular_eval_filter
+        refresh: true
+        body:
+          - '{"index": {"_id": "1"}}'
+          - '{"id": "order1", "name": "apple"}'
+          - '{"index": {"_id": "2"}}'
+          - '{"id": "order2", "name": "banana"}'
+          - '{"index": {"_id": "3"}}'
+          - '{"id": "order3", "name": "orange"}'
+  - do:
+      headers:
+        Content-Type: 'application/json'
+      ppl:
+        body:
+          query: source=test_regular_eval_filter | eval NameLen=LENGTH(name) | fields id, name, NameLen | where NameLen > 5
+
+  - match: { total: 2 }
+  - match: {"schema": [{"name": "id", "type": "string"}, {"name": "name", "type": "string"}, {"name": "NameLen", "type": "int"}]}
+  - match: {"datarows": [["order2", "banana", 6], ["order3", "orange", 6]]}

opensearch/src/main/java/org/opensearch/sql/opensearch/request/PredicateAnalyzer.java

@@ -77,16 +77,19 @@
 import org.apache.calcite.util.NlsString;
 import org.apache.calcite.util.RangeSets;
 import org.apache.calcite.util.Sarg;
+import org.apache.lucene.search.join.ScoreMode;
 import org.opensearch.index.mapper.DateFieldMapper;
 import org.opensearch.index.query.BoolQueryBuilder;
 import org.opensearch.index.query.QueryBuilder;
+import org.opensearch.index.query.QueryBuilders;
 import org.opensearch.index.query.RangeQueryBuilder;
 import org.opensearch.index.query.ScriptQueryBuilder;
 import org.opensearch.script.Script;
 import org.opensearch.sql.calcite.plan.OpenSearchConstants;
 import org.opensearch.sql.calcite.type.ExprIPType;
 import org.opensearch.sql.calcite.type.ExprSqlType;
 import org.opensearch.sql.calcite.utils.OpenSearchTypeFactory.ExprUDT;
+import org.opensearch.sql.calcite.utils.PlanUtils;
 import org.opensearch.sql.calcite.utils.UserDefinedFunctionUtils;
 import org.opensearch.sql.data.model.ExprIpValue;
 import org.opensearch.sql.data.model.ExprTimestampValue;
@@ -1450,6 +1453,8 @@ public static class ScriptQueryExpression extends QueryExpression {
     private final Supplier<String> codeGenerator;
     private String generatedCode;
     private final ScriptParameterHelper parameterHelper;
+    private final Map<String, ExprType> fieldTypes;
+    private final List<String> referredFields;
 
     public ScriptQueryExpression(
         RexNode rexNode,
@@ -1471,6 +1476,12 @@ public ScriptQueryExpression(
           () ->
               SerializationWrapper.wrapWithLangType(
                   ScriptEngineType.CALCITE, serializer.serialize(rexNode, parameterHelper));
+      this.referredFields =
+          PlanUtils.getInputRefs(rexNode).stream()
+              .map(RexInputRef::getIndex)
+              .map(rowType.getFieldNames()::get)
+              .toList();
+      this.fieldTypes = fieldTypes;
     }
 
     // For filter script, this method will be called after planning phase;
@@ -1485,7 +1496,12 @@ private String getOrCreateGeneratedCode() {
 
     @Override
     public QueryBuilder builder() {
-      return new ScriptQueryBuilder(getScript());
+      ScriptQueryBuilder scriptQuery = QueryBuilders.scriptQuery(getScript());
+      String nestedPath = findNestedPath(fieldTypes);
+      if (nestedPath != null) {
+        return QueryBuilders.nestedQuery(nestedPath, scriptQuery, ScoreMode.None);
+      }
+      return scriptQuery;
     }
 
     public Script getScript() {
@@ -1511,6 +1527,45 @@ public void updateAnalyzedNodes(RexNode rexNode) {
     public List<RexNode> getUnAnalyzableNodes() {
       return List.of();
     }
+
+    /**
+     * Find the nested path for fields referenced in the expression. If multiple nested paths exist,
+     * returns the top one.
+     *
+     * @param fieldTypes Map of field names to their types
+     * @return The nested path, or null if no nested fields are found
+     */
+    private String findNestedPath(Map<String, ExprType> fieldTypes) {
+      if (fieldTypes == null || fieldTypes.isEmpty()) {
+        return null;
+      }
+
+      for (String fieldName : referredFields) {
+        // Check if the field is part of a nested structure
+        // For a field like "items.name", we need to check if "items" is nested
+        if (fieldName.contains(".")) {
+          String[] parts = fieldName.split("\\.");
+          StringBuilder pathBuilder = new StringBuilder();
+
+          // Build up the path progressively and check if any parent is nested
+          for (int i = 0; i < parts.length - 1; i++) {
+            if (i > 0) {
+              pathBuilder.append(".");
+            }
+            pathBuilder.append(parts[i]);
+            String currentPath = pathBuilder.toString();
+
+            // Check if this path exists in fieldTypes and is nested
+            ExprType pathType = fieldTypes.get(currentPath);
+            // OpenSearchDataType.Nested is mapped to ExprCoreType.ARRAY
+            if (pathType == ExprCoreType.ARRAY) {
+              return currentPath;
+            }
+          }
+        }
+      }
+      return null;
+    }
   }
 
   /**

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

@@ -153,10 +153,7 @@ public AbstractRelNode pushDownFilter(Filter filter) {
     try {
       RelDataType rowType = this.getRowType();
       List<String> schema = this.getRowType().getFieldNames();
-      Map<String, ExprType> fieldTypes =
-          this.osIndex.getAllFieldTypes().entrySet().stream()
-              .filter(entry -> schema.contains(entry.getKey()))
-              .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+      Map<String, ExprType> fieldTypes = this.osIndex.getAllFieldTypes();
       QueryExpression queryExpression =
           PredicateAnalyzer.analyzeExpression(
               filter.getCondition(), schema, fieldTypes, rowType, getCluster());