Core: Add content stats evaluator

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

@@ -24,9 +24,11 @@
 import java.util.stream.Collectors;
 import org.apache.iceberg.transforms.Transform;
 import org.apache.iceberg.types.Comparators;
+import org.apache.iceberg.types.Type;
 import org.apache.iceberg.util.NaNUtil;
+import org.apache.iceberg.variants.VariantValue;
 
-abstract class InclusiveEvalVisitor extends ExpressionVisitors.BoundVisitor<Boolean> {
+public abstract class InclusiveEvalVisitor extends ExpressionVisitors.BoundVisitor<Boolean> {
   private static final int IN_PREDICATE_LIMIT = 200;
 
   protected static final boolean ROWS_MIGHT_MATCH = true;
@@ -64,6 +66,11 @@ protected <T> T extractUpperBound(BoundExtract<T> bound) {
     return null;
   }
 
+  /** Return a variant value as the given type, or null if it cannot be represented as that type. */
+  protected static <T> T castTo(VariantValue value, Type type) {
+    return VariantExpressionUtil.castTo(value, type);
+  }
+
   @Override
   public Boolean alwaysTrue() {
     return ROWS_MIGHT_MATCH; // all rows match

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

@@ -127,15 +127,15 @@ private Expression bindUnaryOperation(StructType struct, BoundTerm<T> boundTerm)
     switch (op()) {
       case IS_NULL:
         if (!boundTerm.producesNull()
-            && allAncestorFieldsAreRequired(struct, boundTerm.ref().fieldId())) {
+            && !TypeUtil.isNullable(struct.asSchema(), boundTerm.ref().fieldId())) {
           return Expressions.alwaysFalse();
         } else if (boundTerm.type().equals(Types.UnknownType.get())) {
           return Expressions.alwaysTrue();
         }
         return new BoundUnaryPredicate<>(Operation.IS_NULL, boundTerm);
       case NOT_NULL:
         if (!boundTerm.producesNull()
-            && allAncestorFieldsAreRequired(struct, boundTerm.ref().fieldId())) {
+            && !TypeUtil.isNullable(struct.asSchema(), boundTerm.ref().fieldId())) {
           return Expressions.alwaysTrue();
         } else if (boundTerm.type().equals(Types.UnknownType.get())) {
           return Expressions.alwaysFalse();
@@ -158,11 +158,6 @@ && allAncestorFieldsAreRequired(struct, boundTerm.ref().fieldId())) {
     }
   }
 
-  private boolean allAncestorFieldsAreRequired(StructType struct, int fieldId) {
-    return TypeUtil.ancestorFields(struct.asSchema(), fieldId).stream()
-        .allMatch(Types.NestedField::isRequired);
-  }
-
   private boolean floatingType(Type.TypeID typeID) {
     return Type.TypeID.DOUBLE.equals(typeID) || Type.TypeID.FLOAT.equals(typeID);
   }

api/src/main/java/org/apache/iceberg/types/TypeUtil.java

@@ -273,6 +273,25 @@ public static List<Types.NestedField> ancestorFields(Schema schema, int fieldId)
     return parents;
   }
 
+  /**
+   * Returns whether a field may contain null values.
+   *
+   * <p>A field may be null if it is optional or if any field that contains it is optional. A
+   * required field nested in an optional struct is null whenever that struct is null. A field that
+   * is not present in the schema may be null because its requirement is unknown.
+   *
+   * @param schema The schema that contains the field ID
+   * @param fieldId The field ID to check
+   * @return true if the field may be null, false if it cannot be null
+   */
+  public static boolean isNullable(Schema schema, int fieldId) {
+    Types.NestedField field = schema.findField(fieldId);
+
+    return field == null
+        || field.isOptional()
+        || ancestorFields(schema, fieldId).stream().anyMatch(Types.NestedField::isOptional);
+  }
+
   /**
    * Assigns fresh ids from the {@link NextID nextId function} for all fields in a type.
    *