Prevents filtering with isnull/isnotnull conditions on nested fields

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

Author: yuancu

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

@@ -10,6 +10,7 @@
 import static org.opensearch.sql.util.MatcherUtils.rows;
 
 import java.io.IOException;
+import java.util.Locale;
 import org.json.JSONObject;
 import org.junit.jupiter.api.Test;
 import org.opensearch.client.Request;
@@ -24,7 +25,7 @@ public void init() throws Exception {
     loadIndex(Index.STATE_COUNTRY);
     loadIndex(Index.STATE_COUNTRY_WITH_NULL);
     loadIndex(Index.CALCS);
-    loadIndex(Index.NESTED_WITHOUT_ARRAYS);
+    loadIndex(Index.NESTED_SIMPLE);
     loadIndex(Index.BIG5);
     Request request1 =
         new Request("PUT", "/" + TEST_INDEX_STATE_COUNTRY_WITH_NULL + "/_doc/7?refresh=true");
@@ -49,13 +50,27 @@ public void testIsNull() throws IOException {
     verifySchema(actual, schema("age", "int"));
 
     verifyDataRows(actual, rows(10));
+  }
 
-    // Test isNull on struct objects
-    actual = executeQuery("source=big5 | where isnull(aws) | fields aws");
+  @Test
+  public void testIsNullWithStruct() throws IOException {
+    JSONObject actual = executeQuery("source=big5 | where isnull(aws) | fields aws");
     verifySchema(actual, schema("aws", "struct"));
     verifyNumOfRows(actual, 0);
   }
 
+  @Test
+  public void testIsNullWithNested() throws IOException {
+    JSONObject actual =
+        executeQuery(
+            String.format(
+                Locale.ROOT,
+                "source=%s | where isnull(address) | fields address",
+                TEST_INDEX_NESTED_SIMPLE));
+    verifySchema(actual, schema("address", "array"));
+    verifyNumOfRows(actual, 0);
+  }
+
   @Test
   public void testIsNotNull() throws IOException {
     JSONObject actual =
@@ -75,13 +90,27 @@ public void testIsNotNull() throws IOException {
         rows("Kevin"),
         rows("    "),
         rows(""));
+  }
 
-    // Test isNotNull on struct objects
-    actual = executeQuery("source=big5 | where isnotnull(aws) | fields aws");
+  @Test
+  public void testIsNotNullWithStruct() throws IOException {
+    JSONObject actual = executeQuery("source=big5 | where isnotnull(aws) | fields aws");
     verifySchema(actual, schema("aws", "struct"));
     verifyNumOfRows(actual, 1);
   }
 
+  @Test
+  public void testIsNotNullWithNested() throws IOException {
+    JSONObject actual =
+        executeQuery(
+            String.format(
+                Locale.ROOT,
+                "source=%s | where isnotnull(address) | fields address",
+                TEST_INDEX_NESTED_SIMPLE));
+    verifySchema(actual, schema("address", "array"));
+    verifyNumOfRows(actual, 5);
+  }
+
   @Test
   public void testNullIf() throws IOException {
     JSONObject actual =

opensearch/src/main/java/org/opensearch/sql/opensearch/planner/physical/OpenSearchFilterIndexScanRule.java

@@ -54,6 +54,7 @@ public interface Config extends RelRule.Config {
             .withOperandSupplier(
                 b0 ->
                     b0.operand(LogicalFilter.class)
+                        .predicate(Predicate.not(OpenSearchIndexScanRule::isConditionExcluded))
                         .oneInput(
                             b1 ->
                                 b1.operand(CalciteLogicalIndexScan.class)

opensearch/src/main/java/org/opensearch/sql/opensearch/planner/physical/OpenSearchIndexScanRule.java

@@ -6,13 +6,20 @@
 package org.opensearch.sql.opensearch.planner.physical;
 
 import java.util.HashSet;
+import java.util.List;
 import java.util.Set;
 import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.rel.core.Filter;
 import org.apache.calcite.rel.core.Sort;
 import org.apache.calcite.rel.logical.LogicalProject;
 import org.apache.calcite.rel.logical.LogicalSort;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.rex.RexOver;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.ArraySqlType;
 import org.opensearch.sql.opensearch.storage.OpenSearchIndex;
 import org.opensearch.sql.opensearch.storage.scan.AbstractCalciteIndexScan;
 
@@ -62,4 +69,39 @@ static boolean isLogicalSortLimit(LogicalSort sort) {
   static boolean sortByFieldsOnly(Sort sort) {
     return !sort.getCollation().getFieldCollations().isEmpty() && sort.fetch == null;
   }
+
+  /**
+   * Filters with conditions of these kinds should not be pushed down since DSL does not handle such
+   * cases correctly
+   */
+  Set<SqlKind> FILTER_PUSHDOWN_EXCLUDED_FUNCTION = Set.of(SqlKind.IS_NULL, SqlKind.IS_NOT_NULL);
+
+  /**
+   * Determines whether a filter condition should be excluded from pushdown to OpenSearch.
+   * Currently, this method only excludes IS NULL and IS NOT NULL operations on nested fields, as
+   * OpenSearch DSL does not handle these conditions correctly.
+   *
+   * @param filter The filter operation to check
+   * @return true if the condition should be excluded from pushdown, false otherwise
+   */
+  static boolean isConditionExcluded(Filter filter) {
+    RexNode condition = filter.getCondition();
+    if (condition instanceof RexCall conditionCall) {
+      return FILTER_PUSHDOWN_EXCLUDED_FUNCTION.contains(conditionCall.getKind())
+          && conditionContainsNestedField(condition, filter.getRowType().getFieldList());
+    }
+    return false;
+  }
+
+  private static boolean conditionContainsNestedField(
+      RexNode condition, List<RelDataTypeField> fields) {
+    if (condition instanceof RexCall conditionCall) {
+      return conditionCall.getOperands().stream()
+          .filter(operand -> operand instanceof RexInputRef)
+          .map(operand -> (RexInputRef) operand)
+          // Nested fields are of type ArraySqlType
+          .anyMatch(inputRef -> fields.get(inputRef.getIndex()).getType() instanceof ArraySqlType);
+    }
+    return false;
+  }
 }