ORC/Spark: forward-port default-value reads via idToConstant (LI #76)

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 reader can fill {@code initial-default} values for fields omitted
+     * from an ORC file. Disabled by default so existing readers retain null-synthesizing projection
+     * behavior.
+     */
+    public ReadBuilder supportsInitialDefaults() {
+      Preconditions.checkState(
+          this.readerFunc != null || this.batchedReaderFunc != null,
+          "A 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

@@ -261,18 +261,41 @@ public static Schema convert(TypeDescription orcSchema) {
    */
   public static TypeDescription buildOrcProjection(
       Schema schema, TypeDescription originalOrcSchema) {
+    return buildOrcProjection(schema, originalOrcSchema, false);
+  }
+
+  /**
+   * Builds the ORC read projection, optionally omitting absent fields that declare an {@code
+   * initial-default} so a default-aware reader can fill them via {@code idToConstant}.
+   *
+   * <p>When {@code supportsInitialDefaults} is true and a field is absent from the file but
+   * declares {@code initialDefault()}, it is omitted instead of being synthesized as a null column.
+   */
+  static TypeDescription buildOrcProjection(
+      Schema schema, TypeDescription originalOrcSchema, 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, supportsInitialDefaults, icebergToOrc);
   }
 
   private static TypeDescription buildOrcProjection(
-      Integer fieldId, Type type, boolean isRequired, Map<Integer, OrcField> mapping) {
+      Integer fieldId,
+      Type type,
+      boolean isRequired,
+      boolean supportsInitialDefaults,
+      Map<Integer, OrcField> mapping) {
     final TypeDescription orcType;
 
     switch (type.typeId()) {
       case STRUCT:
         orcType = TypeDescription.createStruct();
         for (Types.NestedField nestedField : type.asStructType().fields()) {
+          // Omit so the reader fills via idToConstant instead of a synthetic null column.
+          if (supportsInitialDefaults
+              && mapping.get(nestedField.fieldId()) == null
+              && nestedField.initialDefault() != null) {
+            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 +308,7 @@ private static TypeDescription buildOrcProjection(
                   nestedField.fieldId(),
                   nestedField.type(),
                   isRequired && nestedField.isRequired(),
+                  supportsInitialDefaults,
                   mapping);
           orcType.addField(name, childType);
         }
@@ -296,16 +320,22 @@ private static TypeDescription buildOrcProjection(
                 list.elementId(),
                 list.elementType(),
                 isRequired && list.isElementRequired(),
+                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, supportsInitialDefaults, mapping);
         TypeDescription valueType =
             buildOrcProjection(
-                map.valueId(), map.valueType(), isRequired && map.isValueRequired(), mapping);
+                map.valueId(),
+                map.valueType(),
+                isRequired && map.isValueRequired(),
+                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,14 @@ 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, 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, supportsInitialDefaults);
     }
 
     SearchArgument sarg = null;

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

@@ -36,7 +36,7 @@ public static <T> T visit(
       Type iType, TypeDescription schema, OrcSchemaWithTypeVisitor<T> visitor) {
     switch (schema.getCategory()) {
       case STRUCT:
-        return visitRecord(iType != null ? iType.asStructType() : null, schema, visitor);
+        return visitor.visitRecord(iType != null ? iType.asStructType() : null, schema, visitor);
 
       case UNION:
         throw new UnsupportedOperationException("Cannot handle " + schema);
@@ -61,7 +61,11 @@ public static <T> T visit(
     }
   }
 
-  private static <T> T visitRecord(
+  /**
+   * Visits a struct. Overridden by Spark to inject {@code initial-default} values into {@code
+   * idToConstant} for fields omitted from the ORC projection.
+   */
+  protected T visitRecord(
       Types.StructType struct, TypeDescription record, OrcSchemaWithTypeVisitor<T> visitor) {
     List<TypeDescription> fields = record.getChildren();
     List<String> names = record.getFieldNames();

spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/OrcSchemaWithTypeVisitorSpark.java

@@ -0,0 +1,122 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.spark;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.orc.ORCSchemaUtil;
+import org.apache.iceberg.orc.OrcSchemaWithTypeVisitor;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.spark.source.BaseDataReader;
+import org.apache.iceberg.types.Types;
+import org.apache.orc.TypeDescription;
+
+/**
+ * Spark ORC schema visitor that injects {@code initial-default} values into {@code idToConstant}
+ * for fields omitted from the per-file ORC projection.
+ *
+ * <p>{@link org.apache.iceberg.spark.data.SparkOrcReader} uses this inject. Vectorized ORC does
+ * not; {@code SparkBatchScan} keeps defaulted projections on the row reader.
+ */
+public abstract class OrcSchemaWithTypeVisitorSpark<T> extends OrcSchemaWithTypeVisitor<T> {
+
+  private final Map<Integer, Object> idToConstant;
+
+  public Map<Integer, Object> getIdToConstant() {
+    return idToConstant;
+  }
+
+  protected OrcSchemaWithTypeVisitorSpark(Map<Integer, ?> idToConstant) {
+    this.idToConstant = Maps.newHashMap();
+    this.idToConstant.putAll(idToConstant);
+  }
+
+  @Override
+  protected T visitRecord(
+      Types.StructType struct, TypeDescription record, OrcSchemaWithTypeVisitor<T> visitor) {
+    Preconditions.checkState(
+        icebergFieldIdsContainOrcFieldIdsInOrder(struct, record),
+        "Iceberg schema and ORC schema doesn't align, please call ORCSchemaUtil.buildOrcProjection"
+            + " to get an aligned ORC schema first!");
+    List<Types.NestedField> iFields = struct.fields();
+    List<TypeDescription> fields = record.getChildren();
+    List<String> names = record.getFieldNames();
+    List<T> results = Lists.newArrayListWithExpectedSize(fields.size());
+
+    for (int i = 0, j = 0; i < iFields.size(); i++) {
+      Types.NestedField iField = iFields.get(i);
+      TypeDescription field = j < fields.size() ? fields.get(j) : null;
+      if (field == null || (iField.fieldId() != ORCSchemaUtil.fieldId(field))) {
+        // Cases that use idToConstant for an iField:
+        // 1. MetadataColumns.ROW_POSITION → RowPositionReader
+        // 2. Partition column → ConstantReader (already in idToConstant from PartitionUtil)
+        // 3. Field omitted because it declares initial-default → ConstantReader (inject here)
+        if (MetadataColumns.nonMetadataColumn(iField.name())
+            && !idToConstant.containsKey(iField.fieldId())
+            && iField.initialDefault() != null) {
+          idToConstant.put(
+              iField.fieldId(),
+              BaseDataReader.convertConstant(iField.type(), iField.initialDefault()));
+        }
+      } else {
+        results.add(visit(iField.type(), field, visitor));
+        j++;
+      }
+    }
+    return visitor.record(struct, record, names, results);
+  }
+
+  private static boolean icebergFieldIdsContainOrcFieldIdsInOrder(
+      Types.StructType struct, TypeDescription record) {
+    List<Integer> icebergIDList =
+        struct.fields().stream().map(Types.NestedField::fieldId).collect(Collectors.toList());
+    List<Integer> orcIDList =
+        record.getChildren().stream().map(ORCSchemaUtil::fieldId).collect(Collectors.toList());
+
+    return containsInOrder(icebergIDList, orcIDList);
+  }
+
+  /**
+   * Checks whether {@code list1} contains all integers from {@code list2} in the same relative
+   * order. {@code list1} may contain extra integers that {@code list2} does not.
+   */
+  private static boolean containsInOrder(List<Integer> list1, List<Integer> list2) {
+    if (list1.size() < list2.size()) {
+      return false;
+    }
+
+    for (int i = 0, j = 0; j < list2.size(); j++) {
+      if (i >= list1.size()) {
+        return false;
+      }
+      while (!list1.get(i).equals(list2.get(j))) {
+        i++;
+        if (i >= list1.size()) {
+          return false;
+        }
+      }
+      i++;
+    }
+    return true;
+  }
+}

spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcReader.java

@@ -25,6 +25,7 @@
 import org.apache.iceberg.orc.OrcValueReader;
 import org.apache.iceberg.orc.OrcValueReaders;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.OrcSchemaWithTypeVisitorSpark;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 import org.apache.orc.TypeDescription;
@@ -64,11 +65,10 @@ public void setBatchContext(long batchOffsetInFile) {
     reader.setBatchContext(batchOffsetInFile);
   }
 
-  private static class ReadBuilder extends OrcSchemaWithTypeVisitor<OrcValueReader<?>> {
-    private final Map<Integer, ?> idToConstant;
+  private static class ReadBuilder extends OrcSchemaWithTypeVisitorSpark<OrcValueReader<?>> {
 
     private ReadBuilder(Map<Integer, ?> idToConstant) {
-      this.idToConstant = idToConstant;
+      super(idToConstant);
     }
 
     @Override
@@ -77,7 +77,7 @@ public OrcValueReader<?> record(
         TypeDescription record,
         List<String> names,
         List<OrcValueReader<?>> fields) {
-      return SparkOrcValueReaders.struct(record, fields, expected, idToConstant);
+      return SparkOrcValueReaders.struct(record, fields, expected, getIdToConstant());
     }
 
     @Override