ORC: don't push file filters for omitted initial-default fields

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

@@ -88,11 +88,7 @@ public Type struct(Types.StructType struct, Iterable<Type> futures) {
     for (int i = 0; i < length; i += 1) {
       Types.NestedField field = fields.get(i);
       Type type = types.next();
-      if (field.isOptional()) {
-        newFields.add(Types.NestedField.optional(newIds.get(i), field.name(), type, field.doc()));
-      } else {
-        newFields.add(Types.NestedField.required(newIds.get(i), field.name(), type, field.doc()));
-      }
+      newFields.add(Types.NestedField.from(field).withId(newIds.get(i)).ofType(type).build());
     }
 
     return Types.StructType.of(newFields);

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

@@ -65,15 +65,7 @@ public Type struct(Types.StructType struct, List<Type> fieldResults) {
         selectedFields.add(field);
       } else if (projectedType != null) {
         sameTypes = false; // signal that some types were altered
-        if (field.isOptional()) {
-          selectedFields.add(
-              Types.NestedField.optional(
-                  field.fieldId(), field.name(), projectedType, field.doc()));
-        } else {
-          selectedFields.add(
-              Types.NestedField.required(
-                  field.fieldId(), field.name(), projectedType, field.doc()));
-        }
+        selectedFields.add(Types.NestedField.from(field).ofType(projectedType).build());
       }
     }
 

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

@@ -79,11 +79,7 @@ public Type struct(Types.StructType struct, Iterable<Type> fieldTypes) {
     for (int i = 0; i < length; i += 1) {
       Types.NestedField field = fields.get(i);
       int fieldId = id(sourceStruct, field.name());
-      if (field.isRequired()) {
-        newFields.add(Types.NestedField.required(fieldId, field.name(), types.get(i), field.doc()));
-      } else {
-        newFields.add(Types.NestedField.optional(fieldId, field.name(), types.get(i), field.doc()));
-      }
+      newFields.add(Types.NestedField.from(field).withId(fieldId).ofType(types.get(i)).build());
     }
 
     return Types.StructType.of(newFields);

api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java

@@ -24,6 +24,7 @@
 import java.util.Set;
 import org.apache.iceberg.AssertHelpers;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.expressions.Expressions;
 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.Types.IntegerType;
@@ -66,6 +67,24 @@ public void testReassignIdsWithIdentifier() {
         actualSchema.identifierFieldIds());
   }
 
+  @Test
+  public void testAssignFreshIdsPreservesDefaults() {
+    Types.NestedField field =
+        Types.NestedField.optional("country")
+            .withId(10)
+            .ofType(Types.StringType.get())
+            .withInitialDefault(Expressions.lit("US"))
+            .withWriteDefault(Expressions.lit("CA"))
+            .build();
+
+    Schema reassigned = TypeUtil.assignIncreasingFreshIds(new Schema(field));
+    Types.NestedField reassignedField = reassigned.findField("country");
+
+    Assert.assertEquals(1, reassignedField.fieldId());
+    Assert.assertEquals("US", reassignedField.initialDefault());
+    Assert.assertEquals("CA", reassignedField.writeDefault());
+  }
+
   @Test
   public void testAssignIncreasingFreshIdWithIdentifier() {
     Schema schema =

orc/src/main/java/org/apache/iceberg/orc/ExpressionToSearchArgument.java

@@ -30,6 +30,7 @@
 import org.apache.iceberg.expressions.Expression;
 import org.apache.iceberg.expressions.ExpressionVisitors;
 import org.apache.iceberg.expressions.Literal;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Type.TypeID;
@@ -45,8 +46,7 @@ class ExpressionToSearchArgument
     extends ExpressionVisitors.BoundVisitor<ExpressionToSearchArgument.Action> {
 
   static SearchArgument convert(Expression expr, TypeDescription readSchema) {
-    Map<Integer, String> idToColumnName =
-        ORCSchemaUtil.idToOrcName(ORCSchemaUtil.convert(readSchema));
+    Map<Integer, String> idToColumnName = columnNamesForPushdown(readSchema);
     SearchArgument.Builder builder = SearchArgumentFactory.newBuilder();
     ExpressionVisitors.visit(expr, new ExpressionToSearchArgument(builder, idToColumnName))
         .invoke();
@@ -68,6 +68,17 @@ private ExpressionToSearchArgument(
     this.idToColumnName = idToColumnName;
   }
 
+  // convert() requires at least one Iceberg field. Omitting defaulted children can leave nested
+  // empty structs (struct<loc:struct<>>) that fail convert even when the root is non-empty.
+  // Treat that as no bindable columns so predicates become YES_NO_NULL and defaults can be filled.
+  private static Map<Integer, String> columnNamesForPushdown(TypeDescription readSchema) {
+    try {
+      return ORCSchemaUtil.idToOrcName(ORCSchemaUtil.convert(readSchema));
+    } catch (IllegalArgumentException e) {
+      return ImmutableMap.of();
+    }
+  }
+
   @Override
   public Action alwaysTrue() {
     return () -> this.builder.literal(TruthValue.YES);
@@ -270,10 +281,12 @@ public <T> Action notStartsWith(Bound<T> expr, Literal<T> lit) {
 
   @Override
   public <T> Action predicate(BoundPredicate<T> pred) {
-    if (UNSUPPORTED_TYPES.contains(pred.ref().type().typeId())) {
+    if (!idToColumnName.containsKey(pred.ref().fieldId())
+        || UNSUPPORTED_TYPES.contains(pred.ref().type().typeId())) {
       // Cannot push down predicates for types which cannot be represented in PredicateLeaf.Type, so
       // return
       // TruthValue.YES_NO_NULL which signifies that this predicate cannot help with filtering
+      // (including fields omitted from the read projection so defaults can be filled).
       return () -> this.builder.literal(TruthValue.YES_NO_NULL);
     } else {
       return super.predicate(pred);

orc/src/test/java/org/apache/iceberg/orc/TestExpressionToSearchArgument.java

@@ -49,6 +49,7 @@
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.expressions.Binder;
 import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.mapping.MappingUtil;
 import org.apache.iceberg.mapping.NameMapping;
 import org.apache.iceberg.types.Types;
@@ -476,4 +477,127 @@ public void testModifiedComplexSchemaNameMapping() {
     SearchArgument actual = ExpressionToSearchArgument.convert(boundFilter, readSchema);
     Assert.assertEquals(expected.toString(), actual.toString());
   }
+
+  @Test
+  public void testDisablesPushdownWhenDefaultedFieldIsOmitted() {
+    Schema fileSchema = new Schema(required(1, "id", Types.LongType.get()));
+    Schema evolvedSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()),
+            Types.NestedField.optional("country")
+                .withId(2)
+                .ofType(Types.StringType.get())
+                .withInitialDefault(Expressions.lit("US"))
+                .build());
+    TypeDescription readSchema =
+        ORCSchemaUtil.buildOrcProjection(
+            evolvedSchema,
+            ORCSchemaUtil.convert(fileSchema),
+            ORCSchemaUtil.FieldIdSource.EMBEDDED,
+            true);
+    Expression boundFilter = Binder.bind(evolvedSchema.asStruct(), equal("country", "US"), true);
+    SearchArgument expected =
+        SearchArgumentFactory.newBuilder().literal(TruthValue.YES_NO_NULL).build();
+
+    SearchArgument actual = ExpressionToSearchArgument.convert(boundFilter, readSchema);
+
+    Assert.assertEquals(expected.toString(), actual.toString());
+  }
+
+  @Test
+  public void testDisablesPushdownWhenReadProjectionIsEmpty() {
+    Schema fileSchema = new Schema(required(1, "id", Types.LongType.get()));
+    Schema evolvedSchema =
+        new Schema(
+            Types.NestedField.optional("country")
+                .withId(2)
+                .ofType(Types.StringType.get())
+                .withInitialDefault(Expressions.lit("US"))
+                .build());
+    TypeDescription readSchema =
+        ORCSchemaUtil.buildOrcProjection(
+            evolvedSchema,
+            ORCSchemaUtil.convert(fileSchema),
+            ORCSchemaUtil.FieldIdSource.EMBEDDED,
+            true);
+    Assert.assertEquals(0, readSchema.getChildren().size());
+
+    Expression boundFilter = Binder.bind(evolvedSchema.asStruct(), equal("country", "bar"), true);
+    SearchArgument expected =
+        SearchArgumentFactory.newBuilder().literal(TruthValue.YES_NO_NULL).build();
+
+    SearchArgument actual = ExpressionToSearchArgument.convert(boundFilter, readSchema);
+
+    Assert.assertEquals(expected.toString(), actual.toString());
+  }
+
+  @Test
+  public void testDisablesPushdownWhenNestedReadProjectionIsEmpty() {
+    // File has loc as an empty struct. Projecting only loc.country (a new initial-default) omits
+    // that child and leaves struct<loc:struct<>>, which is non-empty at the root.
+    Schema fileSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()), optional(2, "loc", Types.StructType.of()));
+    Schema evolvedSchema =
+        new Schema(
+            optional(
+                2,
+                "loc",
+                Types.StructType.of(
+                    Types.NestedField.optional("country")
+                        .withId(3)
+                        .ofType(Types.StringType.get())
+                        .withInitialDefault(Expressions.lit("US"))
+                        .build())));
+    TypeDescription readSchema =
+        ORCSchemaUtil.buildOrcProjection(
+            evolvedSchema,
+            ORCSchemaUtil.convert(fileSchema),
+            ORCSchemaUtil.FieldIdSource.EMBEDDED,
+            true);
+    Assert.assertEquals(1, readSchema.getChildren().size());
+    Assert.assertEquals(0, readSchema.findSubtype("loc").getChildren().size());
+
+    Expression boundFilter =
+        Binder.bind(evolvedSchema.asStruct(), equal("loc.country", "bar"), true);
+    SearchArgument expected =
+        SearchArgumentFactory.newBuilder().literal(TruthValue.YES_NO_NULL).build();
+
+    SearchArgument actual = ExpressionToSearchArgument.convert(boundFilter, readSchema);
+
+    Assert.assertEquals(expected.toString(), actual.toString());
+  }
+
+  @Test
+  public void testMixedPhysicalAndOmittedDefaultPredicates() {
+    Schema fileSchema = new Schema(required(1, "id", Types.LongType.get()));
+    Schema evolvedSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()),
+            Types.NestedField.optional("country")
+                .withId(2)
+                .ofType(Types.StringType.get())
+                .withInitialDefault(Expressions.lit("US"))
+                .build());
+    TypeDescription readSchema =
+        ORCSchemaUtil.buildOrcProjection(
+            evolvedSchema,
+            ORCSchemaUtil.convert(fileSchema),
+            ORCSchemaUtil.FieldIdSource.EMBEDDED,
+            true);
+
+    Expression boundFilter =
+        Binder.bind(evolvedSchema.asStruct(), and(equal("id", 1L), equal("country", "US")), true);
+    SearchArgument expected =
+        SearchArgumentFactory.newBuilder()
+            .startAnd()
+            .equals("`id`", Type.LONG, 1L)
+            .literal(TruthValue.YES_NO_NULL)
+            .end()
+            .build();
+
+    SearchArgument actual = ExpressionToSearchArgument.convert(boundFilter, readSchema);
+
+    Assert.assertEquals(expected.toString(), actual.toString());
+  }
 }

spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderForFieldsWithDefaultValue.java

@@ -30,6 +30,7 @@
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.orc.ORC;
 import org.apache.iceberg.orc.ORCSchemaUtil;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
 import org.apache.iceberg.types.Types;
 import org.apache.orc.OrcFile;
 import org.apache.orc.TypeDescription;
@@ -165,6 +166,66 @@ public void testPartialEmbeddedIdsDoNotFillDefault() throws IOException {
     }
   }
 
+  @Test
+  public void testFilterOnOnlyOmittedDefaultDoesNotThrow() throws IOException {
+    // Projecting and filtering only a defaulted column yields an empty ORC schema. Convert must
+    // not throw; SARG is disabled (YES_NO_NULL). Row-level filtering is Spark's job.
+    Schema writeSchema = new Schema(Types.NestedField.required(1, "col1", Types.IntegerType.get()));
+    TypeDescription orcSchema = ORCSchemaUtil.convert(writeSchema);
+    Schema readSchema =
+        new Schema(
+            Types.NestedField.optional("col2")
+                .withId(2)
+                .ofType(Types.StringType.get())
+                .withInitialDefault(Expressions.lit("foo"))
+                .build());
+    File orcFile = writeOrcWithIntColumn(orcSchema, 10);
+
+    try (CloseableIterable<InternalRow> reader =
+        ORC.read(Files.localInput(orcFile))
+            .project(readSchema)
+            .filter(Expressions.equal("col2", "bar"))
+            .createReaderFunc(readOrcSchema -> new SparkOrcReader(readSchema, readOrcSchema))
+            .supportsInitialDefaults()
+            .build()) {
+      Assert.assertEquals(10, Iterators.size(reader.iterator()));
+    }
+  }
+
+  @Test
+  public void testFilterOnOnlyOmittedNestedDefaultDoesNotThrow() throws IOException {
+    // Projecting and filtering only loc.country leaves struct<loc:struct<>>. Convert must not
+    // throw; SARG is disabled (YES_NO_NULL). Row-level filtering is Spark's job.
+    Schema writeSchema =
+        new Schema(
+            Types.NestedField.required(1, "id", Types.LongType.get()),
+            Types.NestedField.optional("loc").withId(2).ofType(Types.StructType.of()).build());
+    TypeDescription orcSchema = ORCSchemaUtil.convert(writeSchema);
+    Schema readSchema =
+        new Schema(
+            Types.NestedField.optional("loc")
+                .withId(2)
+                .ofType(
+                    Types.StructType.of(
+                        Types.NestedField.optional("country")
+                            .withId(3)
+                            .ofType(Types.StringType.get())
+                            .withInitialDefault(Expressions.lit("US"))
+                            .build()))
+                .build());
+    File orcFile = writeOrcWithIdAndEmptyLoc(orcSchema, 10);
+
+    try (CloseableIterable<InternalRow> reader =
+        ORC.read(Files.localInput(orcFile))
+            .project(readSchema)
+            .filter(Expressions.equal("loc.country", "bar"))
+            .createReaderFunc(readOrcSchema -> new SparkOrcReader(readSchema, readOrcSchema))
+            .supportsInitialDefaults()
+            .build()) {
+      Assert.assertEquals(10, Iterators.size(reader.iterator()));
+    }
+  }
+
   private File writeOrcWithIntColumn(TypeDescription orcSchema, int numRows) throws IOException {
     Configuration conf = new Configuration();
     File orcFile = temp.newFile();