ORC: fill initial defaults for missing id-bound fields

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

@@ -171,6 +171,24 @@ public class TestInclusiveMetricsEvaluator {
           // upper bounds
           ImmutableMap.of(3, toByteBuffer(StringType.get(), "イロハニホヘト")));
 
+  @Test
+  public void testRetainsFileWhenDefaultedFieldIsMissingFromMetrics() {
+    List<Types.NestedField> fields = Lists.newArrayList(SCHEMA.columns());
+    fields.add(
+        Types.NestedField.optional("country")
+            .withId(15)
+            .ofType(StringType.get())
+            .withInitialDefault(Expressions.lit("US"))
+            .build());
+    Schema evolvedSchema = new Schema(fields);
+
+    boolean mightMatch =
+        new InclusiveMetricsEvaluator(evolvedSchema, equal("country", "US")).eval(FILE);
+
+    Assert.assertTrue(
+        "Missing field stats must not prune rows that read as the default", mightMatch);
+  }
+
   @Test
   public void testAllNulls() {
     boolean shouldRead = new InclusiveMetricsEvaluator(SCHEMA, notNull("all_nulls")).eval(FILE);

data/src/main/java/org/apache/iceberg/data/GenericReader.java

@@ -137,6 +137,7 @@ private CloseableIterable<Record> openFile(FileScanTask task, Schema fileProject
                 .createReaderFunc(
                     fileSchema ->
                         GenericOrcReader.buildReader(fileProjection, fileSchema, partition))
+                .supportsInitialDefaults()
                 .split(task.start(), task.length())
                 .filter(task.residual());
 

orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReaders.java

@@ -31,6 +31,7 @@
 import java.util.Map;
 import java.util.UUID;
 import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.IdentityPartitionConverters;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.orc.OrcValueReader;
 import org.apache.iceberg.orc.OrcValueReaders;
@@ -238,7 +239,8 @@ protected StructReader(
         List<OrcValueReader<?>> readers,
         Types.StructType structType,
         Map<Integer, ?> idToConstant) {
-      super(orcType, readers, structType, idToConstant);
+      super(
+          orcType, readers, structType, idToConstant, IdentityPartitionConverters::convertConstant);
       this.template = GenericRecord.create(structType);
     }
 

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

@@ -270,7 +270,8 @@ 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

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

@@ -672,6 +672,7 @@ public static class ReadBuilder {
 
     private Function<TypeDescription, OrcRowReader<?>> readerFunc;
     private Function<TypeDescription, OrcBatchReader<?>> batchedReaderFunc;
+    private boolean supportsInitialDefaults = false;
     private int recordsPerBatch = VectorizedRowBatch.DEFAULT_SIZE;
 
     private ReadBuilder(InputFile file) {
@@ -725,6 +726,19 @@ public ReadBuilder createReaderFunc(Function<TypeDescription, OrcRowReader<?>> r
       return this;
     }
 
+    /**
+     * Signals that the configured row reader can fill initial defaults for fields omitted from an
+     * ORC file. Disabled by default so existing positional and vectorized readers retain their
+     * previous null-synthesizing projection behavior.
+     */
+    public ReadBuilder supportsInitialDefaults() {
+      Preconditions.checkState(
+          this.readerFunc != null,
+          "A row reader function must be configured before enabling initial defaults");
+      this.supportsInitialDefaults = true;
+      return this;
+    }
+
     public ReadBuilder filter(Expression newFilter) {
       this.filter = newFilter;
       return this;
@@ -733,7 +747,7 @@ public ReadBuilder filter(Expression newFilter) {
     public ReadBuilder createBatchedReaderFunc(
         Function<TypeDescription, OrcBatchReader<?>> batchReaderFunction) {
       Preconditions.checkArgument(
-          this.readerFunc == null,
+          this.readerFunc == null && !this.supportsInitialDefaults,
           "Batched reader function cannot be set since the non-batched version is already set");
       this.batchedReaderFunc = batchReaderFunction;
       return this;
@@ -759,6 +773,7 @@ public <D> CloseableIterable<D> build() {
           start,
           length,
           readerFunc,
+          supportsInitialDefaults,
           caseSensitive,
           filter,
           batchedReaderFunc,

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

@@ -47,6 +47,32 @@ public enum LongType {
     LONG
   }
 
+  /**
+   * Where the Iceberg field IDs in an ORC schema came from.
+   *
+   * <p>This is the provenance of the IDs, not a statement about the file's contents. It decides
+   * whether the absence of an ID may be read as "this column was never written", which in turn
+   * decides whether a declared {@code initial-default} may be filled on read.
+   */
+  enum FieldIdSource {
+    /**
+     * The IDs were read from {@code iceberg.id} column attributes written into the file by {@link
+     * ORCSchemaUtil#convert(Schema)}. A field with no ID was genuinely never written, so a declared
+     * default may be filled.
+     */
+    EMBEDDED,
+
+    /**
+     * The IDs were derived at read time by {@link ORCSchemaUtil#applyNameMapping} matching column
+     * names, because the file carried none of its own -- typically a legacy or Hive-migrated file.
+     * A field can look absent merely because its name did not match, for example after a rename the
+     * name mapping no longer covers, while its data is physically present in the file. Filling a
+     * default here would fabricate values over real data, so absent fields are synthesized as null
+     * columns instead.
+     */
+    NAME_MAPPED
+  }
+
   private static class OrcField {
     private final String name;
     private final TypeDescription type;
@@ -261,18 +287,73 @@ public static Schema convert(TypeDescription orcSchema) {
    */
   public static TypeDescription buildOrcProjection(
       Schema schema, TypeDescription originalOrcSchema) {
+    // Callers that cannot establish ID provenance get the conservative behavior: never fill
+    // defaults, matching this method's behavior before defaults were supported.
+    return buildOrcProjection(schema, originalOrcSchema, FieldIdSource.NAME_MAPPED, false);
+  }
+
+  /**
+   * Builds the ORC read schema, omitting absent fields that declare an {@code initial-default} so a
+   * default-aware reader can fill them.
+   *
+   * <p>A scalar field at any nesting level is <em>omitted</em> from the read projection when it
+   * declares an {@code initial-default}, is absent from the data file, {@code fieldIdSource} is
+   * {@link FieldIdSource#EMBEDDED}, and the configured reader supports initial defaults. An
+   * id-binding reader then sees no column for that field and fills the declared default as a
+   * per-file constant. Otherwise the field is synthesized as a null column, preserving the behavior
+   * of readers that have not opted in.
+   *
+   * @param fieldIdSource where the IDs in {@code originalOrcSchema} came from; see {@link
+   *     FieldIdSource}
+   * @param supportsInitialDefaults whether the configured reader can fill an omitted field's
+   *     initial default
+   */
+  static TypeDescription buildOrcProjection(
+      Schema schema,
+      TypeDescription originalOrcSchema,
+      FieldIdSource fieldIdSource,
+      boolean supportsInitialDefaults) {
     final Map<Integer, OrcField> icebergToOrc = icebergToOrcMapping("root", originalOrcSchema);
-    return buildOrcProjection(Integer.MIN_VALUE, schema.asStruct(), true, icebergToOrc);
+    return buildOrcProjection(
+        Integer.MIN_VALUE,
+        schema.asStruct(),
+        true,
+        fieldIdSource,
+        supportsInitialDefaults,
+        icebergToOrc);
+  }
+
+  private static boolean isOmittableDefault(
+      Types.NestedField field,
+      FieldIdSource fieldIdSource,
+      boolean supportsInitialDefaults,
+      Map<Integer, OrcField> mapping) {
+    // Only scalars reach here with a non-null default: Types.NestedField#castDefault rejects a
+    // default on any nested type at construction time.
+    return supportsInitialDefaults
+        && field.initialDefault() != null
+        && !mapping.containsKey(field.fieldId())
+        && fieldIdSource == FieldIdSource.EMBEDDED;
   }
 
   private static TypeDescription buildOrcProjection(
-      Integer fieldId, Type type, boolean isRequired, Map<Integer, OrcField> mapping) {
+      Integer fieldId,
+      Type type,
+      boolean isRequired,
+      FieldIdSource fieldIdSource,
+      boolean supportsInitialDefaults,
+      Map<Integer, OrcField> mapping) {
     final TypeDescription orcType;
 
     switch (type.typeId()) {
       case STRUCT:
         orcType = TypeDescription.createStruct();
         for (Types.NestedField nestedField : type.asStructType().fields()) {
+          if (isOmittableDefault(nestedField, fieldIdSource, supportsInitialDefaults, mapping)) {
+            // The field declares a default, is absent, and the file carries its own Iceberg field
+            // IDs. Omit it so a default-aware reader fills it through the existing constant path.
+            continue;
+          }
           // Using suffix _r to avoid potential underlying issues in ORC reader
           // with reused column names between ORC and Iceberg;
           // e.g. renaming column c -> d and adding new column d
@@ -285,6 +366,8 @@ private static TypeDescription buildOrcProjection(
                   nestedField.fieldId(),
                   nestedField.type(),
                   isRequired && nestedField.isRequired(),
+                  fieldIdSource,
+                  supportsInitialDefaults,
                   mapping);
           orcType.addField(name, childType);
         }
@@ -296,16 +379,29 @@ private static TypeDescription buildOrcProjection(
                 list.elementId(),
                 list.elementType(),
                 isRequired && list.isElementRequired(),
+                fieldIdSource,
+                supportsInitialDefaults,
                 mapping);
         orcType = TypeDescription.createList(elementType);
         break;
       case MAP:
         Types.MapType map = (Types.MapType) type;
         TypeDescription keyType =
-            buildOrcProjection(map.keyId(), map.keyType(), isRequired, mapping);
+            buildOrcProjection(
+                map.keyId(),
+                map.keyType(),
+                isRequired,
+                fieldIdSource,
+                supportsInitialDefaults,
+                mapping);
         TypeDescription valueType =
             buildOrcProjection(
-                map.valueId(), map.valueType(), isRequired && map.isValueRequired(), mapping);
+                map.valueId(),
+                map.valueType(),
+                isRequired && map.isValueRequired(),
+                fieldIdSource,
+                supportsInitialDefaults,
+                mapping);
         orcType = TypeDescription.createMap(keyType, valueType);
         break;
       default:

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

@@ -46,6 +46,7 @@ class OrcIterable<T> extends CloseableGroup implements CloseableIterable<T> {
   private final Long start;
   private final Long length;
   private final Function<TypeDescription, OrcRowReader<?>> readerFunction;
+  private final boolean supportsInitialDefaults;
   private final Expression filter;
   private final boolean caseSensitive;
   private final Function<TypeDescription, OrcBatchReader<?>> batchReaderFunction;
@@ -60,12 +61,14 @@ class OrcIterable<T> extends CloseableGroup implements CloseableIterable<T> {
       Long start,
       Long length,
       Function<TypeDescription, OrcRowReader<?>> readerFunction,
+      boolean supportsInitialDefaults,
       boolean caseSensitive,
       Expression filter,
       Function<TypeDescription, OrcBatchReader<?>> batchReaderFunction,
       int recordsPerBatch) {
     this.schema = schema;
     this.readerFunction = readerFunction;
+    this.supportsInitialDefaults = supportsInitialDefaults;
     this.file = file;
     this.nameMapping = nameMapping;
     this.start = start;
@@ -86,13 +89,20 @@ public CloseableIterator<T> iterator() {
     TypeDescription fileSchema = orcFileReader.getSchema();
     final TypeDescription readOrcSchema;
     if (ORCSchemaUtil.hasIds(fileSchema)) {
-      readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, fileSchema);
+      readOrcSchema =
+          ORCSchemaUtil.buildOrcProjection(
+              schema, fileSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED, supportsInitialDefaults);
     } else {
       if (nameMapping == null) {
         nameMapping = MappingUtil.create(schema);
       }
       TypeDescription typeWithIds = ORCSchemaUtil.applyNameMapping(fileSchema, nameMapping);
-      readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, typeWithIds);
+      readOrcSchema =
+          ORCSchemaUtil.buildOrcProjection(
+              schema,
+              typeWithIds,
+              ORCSchemaUtil.FieldIdSource.NAME_MAPPED,
+              supportsInitialDefaults);
     }
 
     SearchArgument sarg = null;

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

@@ -21,9 +21,12 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.function.BiFunction;
 import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.Schema;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 import org.apache.orc.TypeDescription;
 import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
@@ -183,6 +186,28 @@ protected StructReader(
         List<OrcValueReader<?>> readers,
         Types.StructType struct,
         Map<Integer, ?> idToConstant) {
+      this(orcType, readers, struct, idToConstant, null);
+    }
+
+    /**
+     * Binds struct fields by Iceberg field id, filling a field's {@code initial-default} when the
+     * file has no column for it.
+     *
+     * <p>A field reaches the default only when the read projection omitted it, which {@link
+     * ORCSchemaUtil#buildOrcProjection(Schema, TypeDescription, ORCSchemaUtil.FieldIdSource,
+     * boolean)} does only for an absent field that declares a default in a file carrying its own
+     * Iceberg field ids. The default is materialized as a per-file constant and consumes no column
+     * vector, so a present column -- including one holding an explicit null -- always wins.
+     *
+     * @param convertConstant converts a default to the engine's in-memory representation, or null
+     *     to disable default filling and keep the strict missing-reader failure
+     */
+    protected StructReader(
+        TypeDescription orcType,
+        List<OrcValueReader<?>> readers,
+        Types.StructType struct,
+        Map<Integer, ?> idToConstant,
+        BiFunction<Type, Object, Object> convertConstant) {
       List<Types.NestedField> fields = struct.fields();
       this.readers = new OrcValueReader[fields.size()];
       this.isConstantOrMetadataField = new boolean[fields.size()];
@@ -208,6 +233,10 @@ protected StructReader(
           this.isConstantOrMetadataField[pos] = false;
           this.orcFieldIndex[pos] = fieldIdToOrcIndex.getOrDefault(field.fieldId(), -1);
           this.readers[pos] = fileReader;
+        } else if (convertConstant != null && field.initialDefault() != null) {
+          this.isConstantOrMetadataField[pos] = true;
+          this.readers[pos] =
+              constants(convertConstant.apply(field.type(), field.initialDefault()));
         } else if (MetadataColumns.isMetadataColumn(field.name())) {
           this.isConstantOrMetadataField[pos] = true;
           this.readers[pos] = constants(null);