API: Guard against null in IN/NOT_IN predicates (#17014) (#17183)

(cherry picked from commit 187da3e)

Author: ebyhr

api/src/main/java/org/apache/iceberg/expressions/BoundSetPredicate.java

@@ -58,9 +58,9 @@ public Set<T> literalSet() {
   public boolean test(T value) {
     switch (op()) {
       case IN:
-        return literalSet.contains(value);
+        return value != null && literalSet.contains(value);
       case NOT_IN:
-        return !literalSet.contains(value);
+        return value == null || !literalSet.contains(value);
       default:
         throw new IllegalStateException("Invalid operation for BoundSetPredicate: " + op());
     }

api/src/main/java/org/apache/iceberg/expressions/Evaluator.java

@@ -138,7 +138,8 @@ public <T> Boolean notEq(Bound<T> valueExpr, Literal<T> lit) {
 
     @Override
     public <T> Boolean in(Bound<T> valueExpr, Set<T> literalSet) {
-      return literalSet.contains(valueExpr.eval(struct));
+      T value = valueExpr.eval(struct);
+      return value != null && literalSet.contains(value);
     }
 
     @Override

api/src/main/java/org/apache/iceberg/expressions/ResidualEvaluator.java

@@ -200,12 +200,14 @@ public <T> Expression notEq(BoundReference<T> ref, Literal<T> lit) {
 
     @Override
     public <T> Expression in(BoundReference<T> ref, Set<T> literalSet) {
-      return literalSet.contains(ref.eval(struct)) ? alwaysTrue() : alwaysFalse();
+      T value = ref.eval(struct);
+      return value != null && literalSet.contains(value) ? alwaysTrue() : alwaysFalse();
     }
 
     @Override
     public <T> Expression notIn(BoundReference<T> ref, Set<T> literalSet) {
-      return literalSet.contains(ref.eval(struct)) ? alwaysFalse() : alwaysTrue();
+      T value = ref.eval(struct);
+      return value == null || !literalSet.contains(value) ? alwaysTrue() : alwaysFalse();
     }
 
     @Override

api/src/test/java/org/apache/iceberg/expressions/TestEvaluator.java

@@ -674,6 +674,12 @@ public void testIn() {
     assertThat(charSeqEvaluator.eval(TestHelpers.Row.of(new Utf8("abcd"))))
         .as("utf8(abcd) in [string(abc), string(abd)] => false")
         .isFalse();
+
+    StructType nullableStringStruct = StructType.of(optional(35, "s", Types.StringType.get()));
+    Evaluator nullStringEvaluator = new Evaluator(nullableStringStruct, in("s", "abc", "abd"));
+    assertThat(nullStringEvaluator.eval(TestHelpers.Row.of((String) null)))
+        .as("null in [abc, abd] => false")
+        .isFalse();
   }
 
   @Test
@@ -776,6 +782,12 @@ public void testNotIn() {
     assertThat(charSeqEvaluator.eval(TestHelpers.Row.of(new Utf8("abcd"))))
         .as("utf8(abcd) not in [string(abc), string(abd)] => true")
         .isTrue();
+
+    StructType nullableStringStruct = StructType.of(optional(35, "s", Types.StringType.get()));
+    Evaluator nullStringEvaluator = new Evaluator(nullableStringStruct, notIn("s", "abc", "abd"));
+    assertThat(nullStringEvaluator.eval(TestHelpers.Row.of((String) null)))
+        .as("null not in [abc, abd] => true")
+        .isTrue();
   }
 
   @Test

api/src/test/java/org/apache/iceberg/expressions/TestPredicateBinding.java

@@ -528,6 +528,30 @@ public void testInPredicateBindingConversionToExpression() {
         .isEqualTo(Expressions.alwaysFalse());
   }
 
+  @Test
+  public void testInPredicateTestWithNull() {
+    StructType struct = StructType.of(optional(36, "s", Types.StringType.get()));
+
+    Expression expr = Expressions.in("s", "abc", "abd").bind(struct);
+    BoundSetPredicate<CharSequence> bound = assertAndUnwrapBoundSet(expr);
+
+    assertThat(bound.test((CharSequence) null)).as("null in [abc, abd] => false").isFalse();
+    assertThat(bound.test("abc")).as("abc in [abc, abd] => true").isTrue();
+    assertThat(bound.test("xyz")).as("xyz in [abc, abd] => false").isFalse();
+  }
+
+  @Test
+  public void testNotInPredicateTestWithNull() {
+    StructType struct = StructType.of(optional(37, "s", Types.StringType.get()));
+
+    Expression expr = Expressions.notIn("s", "abc", "abd").bind(struct);
+    BoundSetPredicate<CharSequence> bound = assertAndUnwrapBoundSet(expr);
+
+    assertThat(bound.test((CharSequence) null)).as("null not in [abc, abd] => true").isTrue();
+    assertThat(bound.test("abc")).as("abc not in [abc, abd] => false").isFalse();
+    assertThat(bound.test("xyz")).as("xyz not in [abc, abd] => true").isTrue();
+  }
+
   @Test
   public void testNotInPredicateBinding() {
     StructType struct =

api/src/test/java/org/apache/iceberg/transforms/TestResiduals.java

@@ -193,6 +193,29 @@ public void testIn() {
 
     residual = resEval.residualFor(Row.of(20180815));
     assertThat(residual).isEqualTo(alwaysFalse());
+
+    residual = resEval.residualFor(Row.of((Object) null));
+    assertThat(residual)
+        .as("null in [20170815, 20170816, 20170817] => false")
+        .isEqualTo(alwaysFalse());
+  }
+
+  @Test
+  public void testInWithNullStringPartition() {
+    Schema schema = new Schema(Types.NestedField.optional(50, "category", Types.StringType.get()));
+
+    PartitionSpec spec = PartitionSpec.builderFor(schema).identity("category").build();
+
+    ResidualEvaluator resEval = ResidualEvaluator.of(spec, in("category", "a", "b", "c"), true);
+
+    Expression residual = resEval.residualFor(Row.of("a"));
+    assertThat(residual).isEqualTo(alwaysTrue());
+
+    residual = resEval.residualFor(Row.of("d"));
+    assertThat(residual).isEqualTo(alwaysFalse());
+
+    residual = resEval.residualFor(Row.of((Object) null));
+    assertThat(residual).as("null in [a, b, c] => false").isEqualTo(alwaysFalse());
   }
 
   @Test
@@ -241,6 +264,29 @@ public void testNotIn() {
 
     residual = resEval.residualFor(Row.of(20170815));
     assertThat(residual).isEqualTo(alwaysFalse());
+
+    residual = resEval.residualFor(Row.of((Object) null));
+    assertThat(residual)
+        .as("null not in [20170815, 20170816, 20170817] => true")
+        .isEqualTo(alwaysTrue());
+  }
+
+  @Test
+  public void testNotInWithNullStringPartition() {
+    Schema schema = new Schema(Types.NestedField.optional(50, "category", Types.StringType.get()));
+
+    PartitionSpec spec = PartitionSpec.builderFor(schema).identity("category").build();
+
+    ResidualEvaluator resEval = ResidualEvaluator.of(spec, notIn("category", "a", "b", "c"), true);
+
+    Expression residual = resEval.residualFor(Row.of("d"));
+    assertThat(residual).isEqualTo(alwaysTrue());
+
+    residual = resEval.residualFor(Row.of("a"));
+    assertThat(residual).isEqualTo(alwaysFalse());
+
+    residual = resEval.residualFor(Row.of((Object) null));
+    assertThat(residual).as("null not in [a, b, c] => true").isEqualTo(alwaysTrue());
   }
 
   @Test