API: Detect whether required fields nested within optionals can produce nulls

api/src/main/java/org/apache/iceberg/Accessor.java

@@ -25,9 +25,4 @@ public interface Accessor<T> extends Serializable {
   Object get(T container);
 
   Type type();
-
-  /** Returns true if the current field or any ancestor in the access path is optional. */
-  default boolean hasOptionalFieldInPath() {
-    return false;
-  }
 }

api/src/main/java/org/apache/iceberg/Accessors.java

@@ -59,13 +59,11 @@ private static class PositionAccessor implements Accessor<StructLike> {
     private final int position;
     private final Type type;
     private final Class<?> javaClass;
-    private final boolean hasOptionalFieldInPath;
 
-    PositionAccessor(int pos, Type type, boolean isOptional) {
+    PositionAccessor(int pos, Type type) {
       this.position = pos;
       this.type = type;
       this.javaClass = type.typeId().javaClass();
-      this.hasOptionalFieldInPath = isOptional;
     }
 
     @Override
@@ -86,11 +84,6 @@ public Class<?> javaClass() {
       return javaClass;
     }
 
-    @Override
-    public boolean hasOptionalFieldInPath() {
-      return hasOptionalFieldInPath;
-    }
-
     @Override
     public String toString() {
       return "Accessor(positions=[" + position + "], type=" + type + ")";
@@ -102,14 +95,12 @@ private static class Position2Accessor implements Accessor<StructLike> {
     private final int p1;
     private final Type type;
     private final Class<?> javaClass;
-    private final boolean hasOptionalFieldInPath;
 
-    Position2Accessor(int pos, PositionAccessor wrapped, boolean isOptional) {
+    Position2Accessor(int pos, PositionAccessor wrapped) {
       this.p0 = pos;
       this.p1 = wrapped.position();
       this.type = wrapped.type();
       this.javaClass = wrapped.javaClass();
-      this.hasOptionalFieldInPath = isOptional || wrapped.hasOptionalFieldInPath();
     }
 
     @Override
@@ -126,11 +117,6 @@ public Class<?> javaClass() {
       return javaClass;
     }
 
-    @Override
-    public boolean hasOptionalFieldInPath() {
-      return hasOptionalFieldInPath;
-    }
-
     @Override
     public String toString() {
       return "Accessor(positions=[" + p0 + ", " + p1 + "], type=" + type + ")";
@@ -143,15 +129,13 @@ private static class Position3Accessor implements Accessor<StructLike> {
     private final int p2;
     private final Type type;
     private final Class<?> javaClass;
-    private final boolean hasOptionalFieldInPath;
 
-    Position3Accessor(int pos, Position2Accessor wrapped, boolean isOptional) {
+    Position3Accessor(int pos, Position2Accessor wrapped) {
       this.p0 = pos;
       this.p1 = wrapped.p0;
       this.p2 = wrapped.p1;
       this.type = wrapped.type();
       this.javaClass = wrapped.javaClass();
-      this.hasOptionalFieldInPath = isOptional || wrapped.hasOptionalFieldInPath();
     }
 
     @Override
@@ -164,11 +148,6 @@ public Type type() {
       return type;
     }
 
-    @Override
-    public boolean hasOptionalFieldInPath() {
-      return hasOptionalFieldInPath;
-    }
-
     @Override
     public String toString() {
       return "Accessor(positions=[" + p0 + ", " + p1 + ", " + p2 + "], type=" + type + ")";
@@ -178,12 +157,10 @@ public String toString() {
   private static class WrappedPositionAccessor implements Accessor<StructLike> {
     private final int position;
     private final Accessor<StructLike> accessor;
-    private final boolean hasOptionalFieldInPath;
 
-    WrappedPositionAccessor(int pos, Accessor<StructLike> accessor, boolean isOptional) {
+    WrappedPositionAccessor(int pos, Accessor<StructLike> accessor) {
       this.position = pos;
       this.accessor = accessor;
-      this.hasOptionalFieldInPath = isOptional || accessor.hasOptionalFieldInPath();
     }
 
     @Override
@@ -200,32 +177,27 @@ public Type type() {
       return accessor.type();
     }
 
-    @Override
-    public boolean hasOptionalFieldInPath() {
-      return hasOptionalFieldInPath;
-    }
-
     @Override
     public String toString() {
       return "WrappedAccessor(position=" + position + ", wrapped=" + accessor + ")";
     }
   }
 
-  private static Accessor<StructLike> newAccessor(int pos, boolean isOptional, Type type) {
-    return new PositionAccessor(pos, type, isOptional);
+  private static Accessor<StructLike> newAccessor(int pos, Type type) {
+    return new PositionAccessor(pos, type);
   }
 
   private static Accessor<StructLike> newAccessor(
       int pos, boolean isOptional, Accessor<StructLike> accessor) {
     if (isOptional) {
       // the wrapped position handles null layers
-      return new WrappedPositionAccessor(pos, accessor, isOptional);
+      return new WrappedPositionAccessor(pos, accessor);
     } else if (accessor.getClass() == PositionAccessor.class) {
-      return new Position2Accessor(pos, (PositionAccessor) accessor, isOptional);
+      return new Position2Accessor(pos, (PositionAccessor) accessor);
     } else if (accessor instanceof Position2Accessor) {
-      return new Position3Accessor(pos, (Position2Accessor) accessor, isOptional);
+      return new Position3Accessor(pos, (Position2Accessor) accessor);
     } else {
-      return new WrappedPositionAccessor(pos, accessor, isOptional);
+      return new WrappedPositionAccessor(pos, accessor);
     }
   }
 
@@ -254,7 +226,7 @@ public Map<Integer, Accessor<StructLike>> struct(
         }
 
         // Add an accessor for this field as an Object (may or may not be primitive).
-        accessors.put(field.fieldId(), newAccessor(i, field.isOptional(), field.type()));
+        accessors.put(field.fieldId(), newAccessor(i, field.type()));
       }
 
       return accessors;

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

@@ -57,9 +57,7 @@ public Type type() {
 
   @Override
   public boolean producesNull() {
-    // A leaf required field can evaluate to null if it is optional itself or any
-    // ancestor on the path is optional.
-    return accessor.hasOptionalFieldInPath();
+    return field.isOptional();
   }
 
   @Override

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

@@ -27,6 +27,7 @@
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
 import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.TypeUtil;
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.StructType;
 import org.apache.iceberg.util.CharSequenceSet;
@@ -112,7 +113,7 @@ public Expression bind(StructType struct, boolean caseSensitive) {
     BoundTerm<T> bound = term().bind(struct, caseSensitive);
 
     if (literals == null) {
-      return bindUnaryOperation(bound);
+      return bindUnaryOperation(struct, bound);
     }
 
     if (op() == Operation.IN || op() == Operation.NOT_IN) {
@@ -122,17 +123,19 @@ public Expression bind(StructType struct, boolean caseSensitive) {
     return bindLiteralOperation(bound);
   }
 
-  private Expression bindUnaryOperation(BoundTerm<T> boundTerm) {
+  private Expression bindUnaryOperation(StructType struct, BoundTerm<T> boundTerm) {
     switch (op()) {
       case IS_NULL:
-        if (!boundTerm.producesNull()) {
+        if (!boundTerm.producesNull()
+            && allAncestorFieldsAreRequired(struct, 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()) {
+        if (!boundTerm.producesNull()
+            && allAncestorFieldsAreRequired(struct, boundTerm.ref().fieldId())) {
           return Expressions.alwaysTrue();
         } else if (boundTerm.type().equals(Types.UnknownType.get())) {
           return Expressions.alwaysFalse();
@@ -155,6 +158,11 @@ private Expression bindUnaryOperation(BoundTerm<T> boundTerm) {
     }
   }
 
+  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

@@ -222,6 +222,30 @@ public static Map<Integer, Integer> indexParents(Types.StructType struct) {
     return ImmutableMap.copyOf(visit(struct, new IndexParents()));
   }
 
+  /**
+   * Searches in the given schema for all ancestor fields of the given field ID. If the field ID is
+   * defined in a nested type, then all of its ancestor fields are returned. If the field ID is not
+   * nested, an empty list is returned.
+   *
+   * @param schema The schema to search for the field ID
+   * @param fieldId The field ID to find the parents of
+   * @return A list of all ancestor fields of the given field ID if the field ID points to a nested
+   *     field. If the field ID is not a nested field, then an empty list is returned.
+   */
+  public static List<Types.NestedField> ancestorFields(Schema schema, int fieldId) {
+    Map<Integer, Integer> idToParent = TypeUtil.indexParents(schema.asStruct());
+    List<Types.NestedField> parents = Lists.newArrayList();
+    if (idToParent.containsKey(fieldId)) {
+      Integer parentId = idToParent.get(fieldId);
+      while (parentId != null) {
+        parents.add(schema.findField(parentId));
+        parentId = idToParent.get(parentId);
+      }
+    }
+
+    return parents;
+  }
+
   /**
    * Assigns fresh ids from the {@link NextID nextId function} for all fields in a type.
    *