ORC: Fill initial defaults for absent scalar columns on read

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

@@ -22,7 +22,9 @@
 import java.util.List;
 import java.util.Map;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.IdentityPartitionConverters;
 import org.apache.iceberg.data.Record;
+import org.apache.iceberg.orc.ORCSchemaUtil;
 import org.apache.iceberg.orc.OrcRowReader;
 import org.apache.iceberg.orc.OrcSchemaWithTypeVisitor;
 import org.apache.iceberg.orc.OrcValueReader;
@@ -76,7 +78,11 @@ public OrcValueReader<?> record(
         TypeDescription record,
         List<String> names,
         List<OrcValueReader<?>> fields) {
-      return GenericOrcReaders.struct(fields, expected, idToConstant);
+      return GenericOrcReaders.struct(
+          fields,
+          expected,
+          ORCSchemaUtil.idToConstantWithDefaults(
+              expected, record, idToConstant, IdentityPartitionConverters::convertConstant));
     }
 
     @Override

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

@@ -22,12 +22,15 @@
 import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.Set;
+import java.util.function.BiFunction;
 import java.util.stream.Collectors;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.mapping.NameMapping;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMultimap;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+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;
@@ -261,18 +264,47 @@ public static Schema convert(TypeDescription orcSchema) {
    */
   public static TypeDescription buildOrcProjection(
       Schema schema, TypeDescription originalOrcSchema) {
+    return buildOrcProjection(schema, originalOrcSchema, false);
+  }
+
+  /**
+   * Builds the ORC read schema for a file whose embedded field IDs are known to be trustworthy.
+   *
+   * <p>When {@code hasTrustedIds} is true, a scalar field at any nesting level that is absent from
+   * the data file and declares an {@code initial-default} is <em>omitted</em> from the read
+   * projection. A default-aware reader then fills it as a per-file constant through its existing
+   * {@code idToConstant} path (see {@link #idToConstantWithDefaults}). When false, the field is
+   * synthesized as a null column so a default is never name-matched onto an id-less file.
+   */
+  static TypeDescription buildOrcProjection(
+      Schema schema, TypeDescription originalOrcSchema, boolean hasTrustedIds) {
     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, hasTrustedIds, icebergToOrc);
+  }
+
+  private static boolean isOmittableDefault(
+      Types.NestedField field, boolean hasTrustedIds, Map<Integer, OrcField> mapping) {
+    return field.initialDefault() != null && !mapping.containsKey(field.fieldId()) && hasTrustedIds;
   }
 
   private static TypeDescription buildOrcProjection(
-      Integer fieldId, Type type, boolean isRequired, Map<Integer, OrcField> mapping) {
+      Integer fieldId,
+      Type type,
+      boolean isRequired,
+      boolean hasTrustedIds,
+      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, hasTrustedIds, mapping)) {
+            // The field declares a default, is absent, and the file carries trustworthy 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 +317,7 @@ private static TypeDescription buildOrcProjection(
                   nestedField.fieldId(),
                   nestedField.type(),
                   isRequired && nestedField.isRequired(),
+                  hasTrustedIds,
                   mapping);
           orcType.addField(name, childType);
         }
@@ -296,16 +329,21 @@ private static TypeDescription buildOrcProjection(
                 list.elementId(),
                 list.elementType(),
                 isRequired && list.isElementRequired(),
+                hasTrustedIds,
                 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, hasTrustedIds, mapping);
         TypeDescription valueType =
             buildOrcProjection(
-                map.valueId(), map.valueType(), isRequired && map.isValueRequired(), mapping);
+                map.valueId(),
+                map.valueType(),
+                isRequired && map.isValueRequired(),
+                hasTrustedIds,
+                mapping);
         orcType = TypeDescription.createMap(keyType, valueType);
         break;
       default:
@@ -432,6 +470,51 @@ static boolean hasIds(TypeDescription orcSchema) {
     return OrcSchemaVisitor.visit(orcSchema, new HasIds());
   }
 
+  /**
+   * Augments an {@code idToConstant} map with column initial-defaults for expected fields that are
+   * absent from the ORC read projection.
+   *
+   * <p>Used by the ORC readers' {@code record} visitor step: a field that {@link
+   * #buildOrcProjection(Schema, TypeDescription, boolean)} omitted (a scalar at any nesting level
+   * declaring an {@code initial-default} that is absent from a file with trustworthy IDs) is not
+   * present in {@code record}, so its converted default is injected as a constant. The reader then
+   * fills it for every row via its existing partition-constant path, consuming no column vector.
+   * Fields synthesized for id-less/name-mapped reads are present in {@code record} and therefore
+   * read NULL.
+   *
+   * @param expected the expected Iceberg struct for this record level
+   * @param record the ORC read projection for this record level
+   * @param idToConstant existing constants (e.g. partition values); not modified
+   * @param convertConstant converts an internal default value to the engine's in-memory form
+   * @return {@code idToConstant} unchanged when no defaults apply, otherwise a new merged map
+   */
+  public static Map<Integer, ?> idToConstantWithDefaults(
+      Types.StructType expected,
+      TypeDescription record,
+      Map<Integer, ?> idToConstant,
+      BiFunction<Type, Object, Object> convertConstant) {
+    Set<Integer> presentIds = Sets.newHashSet();
+    for (TypeDescription child : record.getChildren()) {
+      presentIds.add(fieldId(child));
+    }
+
+    Map<Integer, Object> withDefaults = null;
+    for (Types.NestedField field : expected.fields()) {
+      if (field.initialDefault() != null
+          && !presentIds.contains(field.fieldId())
+          && !idToConstant.containsKey(field.fieldId())) {
+        if (withDefaults == null) {
+          withDefaults = Maps.newHashMap();
+          withDefaults.putAll(idToConstant);
+        }
+        withDefaults.put(
+            field.fieldId(), convertConstant.apply(field.type(), field.initialDefault()));
+      }
+    }
+
+    return withDefaults == null ? idToConstant : withDefaults;
+  }
+
   static TypeDescription applyNameMapping(TypeDescription orcSchema, NameMapping nameMapping) {
     return OrcSchemaVisitor.visit(orcSchema, new ApplyNameMapping(nameMapping));
   }

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

@@ -84,19 +84,26 @@ public CloseableIterator<T> iterator() {
     addCloseable(orcFileReader);
 
     TypeDescription fileSchema = orcFileReader.getSchema();
+    boolean hasTrustedIds = ORCSchemaUtil.hasIds(fileSchema);
     final TypeDescription readOrcSchema;
-    if (ORCSchemaUtil.hasIds(fileSchema)) {
-      readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, fileSchema);
+    if (hasTrustedIds) {
+      // Embedded IDs make field identity trustworthy, so absent fields that declare an initial
+      // default can be omitted from the projection and filled by a default-aware reader.
+      readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, fileSchema, hasTrustedIds);
     } else {
       if (nameMapping == null) {
         nameMapping = MappingUtil.create(schema);
       }
       TypeDescription typeWithIds = ORCSchemaUtil.applyNameMapping(fileSchema, nameMapping);
-      readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, typeWithIds);
+      // Id-less (legacy/migrated) files are resolved by name mapping; never apply column defaults
+      // so an initial-default is never name-matched onto a file that lacks embedded field ids.
+      readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, typeWithIds, false);
     }
 
     SearchArgument sarg = null;
     if (filter != null) {
+      // Follow-up: predicates that reference absent defaulted fields need default-aware pushdown.
+      // Those fields are omitted when defaults are enabled and must be evaluated after read-fill.
       Expression boundFilter = Binder.bind(schema.asStruct(), filter, caseSensitive);
       sarg = ExpressionToSearchArgument.convert(boundFilter, readOrcSchema);
     }

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

@@ -21,8 +21,11 @@
 import static org.apache.iceberg.types.Types.NestedField.optional;
 import static org.apache.iceberg.types.Types.NestedField.required;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
 
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.types.Types;
 import org.apache.orc.TypeDescription;
 import org.assertj.core.api.Assertions;
@@ -159,4 +162,141 @@ public void testRequiredNestedFieldMissingInFile() {
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessage("Field 4 of type long is required and was not found.");
   }
+
+  @Test
+  public void testTopLevelScalarDefaultOmittedWhenFileIdentityIsTrustworthy() {
+    Schema baseSchema = new Schema(required(1, "id", Types.LongType.get()));
+    TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema);
+
+    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());
+
+    // The file carries embedded field IDs, so the absent field can be identified safely.
+    TypeDescription projection =
+        ORCSchemaUtil.buildOrcProjection(
+            evolvedSchema, baseOrcSchema, ORCSchemaUtil.hasIds(baseOrcSchema));
+    assertEquals(1, projection.getChildren().size());
+    assertNotNull(projection.findSubtype("id"));
+    assertFalse(
+        "defaulted column must be omitted from the read projection",
+        projection.getFieldNames().contains("country_r2"));
+  }
+
+  @Test
+  public void testTopLevelScalarDefaultSynthesizedWithoutTrustedFileIdentity() {
+    Schema baseSchema = new Schema(required(1, "id", Types.LongType.get()));
+    TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema);
+
+    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());
+
+    // Without trustworthy file identity, synthesize NULL rather than guessing that the field is
+    // absent and applying its default.
+    TypeDescription projection =
+        ORCSchemaUtil.buildOrcProjection(evolvedSchema, baseOrcSchema, false);
+    assertEquals(2, projection.getChildren().size());
+    assertEquals(2, projection.findSubtype("country_r2").getId());
+    assertEquals(
+        TypeDescription.Category.STRING, projection.findSubtype("country_r2").getCategory());
+  }
+
+  @Test
+  public void testTopLevelRequiredScalarDefaultOmitted() {
+    Schema baseSchema = new Schema(required(1, "id", Types.LongType.get()));
+    TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema);
+
+    // A required top-level field that is absent from the file but declares a default must be
+    // omitted (then filled), not rejected by the required-missing check.
+    Schema evolvedSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()),
+            Types.NestedField.required("code")
+                .withId(2)
+                .ofType(Types.IntegerType.get())
+                .withInitialDefault(Expressions.lit(7))
+                .build());
+
+    TypeDescription projection =
+        ORCSchemaUtil.buildOrcProjection(
+            evolvedSchema, baseOrcSchema, ORCSchemaUtil.hasIds(baseOrcSchema));
+    assertEquals(1, projection.getChildren().size());
+    assertFalse(
+        "required defaulted column must be omitted, not throw",
+        projection.getFieldNames().contains("code_r2"));
+  }
+
+  @Test
+  public void testNestedScalarDefaultOmitted() {
+    Schema baseSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()),
+            required(2, "s", Types.StructType.of(required(3, "a", Types.LongType.get()))));
+    TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema);
+
+    // A scalar default on a field nested inside a struct is omitted (then filled via idToConstant),
+    // at any nesting level. The present sibling "a" keeps the struct non-empty.
+    Schema evolvedSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()),
+            required(
+                2,
+                "s",
+                Types.StructType.of(
+                    required(3, "a", Types.LongType.get()),
+                    Types.NestedField.optional("b")
+                        .withId(4)
+                        .ofType(Types.StringType.get())
+                        .withInitialDefault(Expressions.lit("x"))
+                        .build())));
+
+    TypeDescription projection =
+        ORCSchemaUtil.buildOrcProjection(
+            evolvedSchema, baseOrcSchema, ORCSchemaUtil.hasIds(baseOrcSchema));
+    TypeDescription nested = projection.findSubtype("s");
+    assertEquals(1, nested.getChildren().size());
+    assertFalse("nested defaulted column must be omitted", nested.getFieldNames().contains("b_r4"));
+  }
+
+  @Test
+  public void testNestedStructEmptiedByOmit() {
+    // Base file: s { a }. Project only a new defaulted subfield s { b default 'x' } (drop a). Every
+    // projected subfield of s is absent + defaulted, so the nested read struct is omitted down to
+    // empty; the reader fills b via idToConstant (see TestOrcDefaultValues end-to-end coverage).
+    Schema baseSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()),
+            required(2, "s", Types.StructType.of(required(3, "a", Types.LongType.get()))));
+    TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema);
+
+    Schema evolvedSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()),
+            optional(
+                2,
+                "s",
+                Types.StructType.of(
+                    Types.NestedField.optional("b")
+                        .withId(4)
+                        .ofType(Types.StringType.get())
+                        .withInitialDefault(Expressions.lit("x"))
+                        .build())));
+
+    TypeDescription projection =
+        ORCSchemaUtil.buildOrcProjection(
+            evolvedSchema, baseOrcSchema, ORCSchemaUtil.hasIds(baseOrcSchema));
+    TypeDescription nested = projection.findSubtype("s");
+    assertEquals(0, nested.getChildren().size());
+  }
 }