ORC: omit initial-defaults only for complete embedded field IDs

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,27 +287,60 @@ public static Schema convert(TypeDescription orcSchema) {
    */
   public static TypeDescription buildOrcProjection(
       Schema schema, TypeDescription originalOrcSchema) {
-    return buildOrcProjection(schema, originalOrcSchema, false);
+    // Callers that cannot establish ID provenance get the conservative behavior: never omit for
+    // defaults, matching this method's behavior before defaults were supported.
+    return buildOrcProjection(schema, originalOrcSchema, FieldIdSource.NAME_MAPPED, 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.
+   * <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, boolean supportsInitialDefaults) {
+      Schema schema,
+      TypeDescription originalOrcSchema,
+      FieldIdSource fieldIdSource,
+      boolean supportsInitialDefaults) {
     final Map<Integer, OrcField> icebergToOrc = icebergToOrcMapping("root", originalOrcSchema);
     return buildOrcProjection(
-        Integer.MIN_VALUE, schema.asStruct(), true, supportsInitialDefaults, icebergToOrc);
+        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,
+      FieldIdSource fieldIdSource,
       boolean supportsInitialDefaults,
       Map<Integer, OrcField> mapping) {
     final TypeDescription orcType;
@@ -290,10 +349,9 @@ private static TypeDescription buildOrcProjection(
       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) {
+          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
@@ -308,6 +366,7 @@ private static TypeDescription buildOrcProjection(
                   nestedField.fieldId(),
                   nestedField.type(),
                   isRequired && nestedField.isRequired(),
+                  fieldIdSource,
                   supportsInitialDefaults,
                   mapping);
           orcType.addField(name, childType);
@@ -320,6 +379,7 @@ private static TypeDescription buildOrcProjection(
                 list.elementId(),
                 list.elementType(),
                 isRequired && list.isElementRequired(),
+                fieldIdSource,
                 supportsInitialDefaults,
                 mapping);
         orcType = TypeDescription.createList(elementType);
@@ -328,12 +388,18 @@ private static TypeDescription buildOrcProjection(
         Types.MapType map = (Types.MapType) type;
         TypeDescription keyType =
             buildOrcProjection(
-                map.keyId(), map.keyType(), isRequired, supportsInitialDefaults, mapping);
+                map.keyId(),
+                map.keyType(),
+                isRequired,
+                fieldIdSource,
+                supportsInitialDefaults,
+                mapping);
         TypeDescription valueType =
             buildOrcProjection(
                 map.valueId(),
                 map.valueType(),
                 isRequired && map.isValueRequired(),
+                fieldIdSource,
                 supportsInitialDefaults,
                 mapping);
         orcType = TypeDescription.createMap(keyType, valueType);
@@ -462,6 +528,34 @@ static boolean hasIds(TypeDescription orcSchema) {
     return OrcSchemaVisitor.visit(orcSchema, new HasIds());
   }
 
+  /**
+   * Returns whether every column in the file carries its own Iceberg field ID.
+   *
+   * <p>{@link #hasIds(TypeDescription)} is satisfied by a single annotated column, which is the
+   * right test for choosing ID-based resolution over a name mapping. It is not sufficient for
+   * omitting a defaulted field: in a partially annotated file an unannotated physical column is
+   * indistinguishable from an absent one, so omitting it would replace real data with the default.
+   * Requiring complete annotation keeps such files on the prior null-synthesizing path.
+   *
+   * <p>The file's outermost struct is not itself annotated, so only its descendants are checked.
+   */
+  static boolean hasAllIds(TypeDescription orcSchema) {
+    List<TypeDescription> children = orcSchema.getChildren();
+    if (children == null) {
+      return true;
+    }
+
+    return children.stream().allMatch(ORCSchemaUtil::isFullyAnnotated);
+  }
+
+  private static boolean isFullyAnnotated(TypeDescription orcType) {
+    if (!icebergID(orcType).isPresent()) {
+      return false;
+    }
+
+    return hasAllIds(orcType);
+  }
+
   static TypeDescription applyNameMapping(TypeDescription orcSchema, NameMapping nameMapping) {
     return OrcSchemaVisitor.visit(orcSchema, new ApplyNameMapping(nameMapping));
   }

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

@@ -89,14 +89,21 @@ public CloseableIterator<T> iterator() {
     TypeDescription fileSchema = orcFileReader.getSchema();
     final TypeDescription readOrcSchema;
     if (ORCSchemaUtil.hasIds(fileSchema)) {
-      readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, fileSchema, supportsInitialDefaults);
+      // A partially annotated file cannot distinguish an absent field from an unannotated physical
+      // column, so defaults may only be omitted when every column carries its own field ID.
+      boolean omitDefaults = supportsInitialDefaults && ORCSchemaUtil.hasAllIds(fileSchema);
+      readOrcSchema =
+          ORCSchemaUtil.buildOrcProjection(
+              schema, fileSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED, omitDefaults);
     } else {
       if (nameMapping == null) {
         nameMapping = MappingUtil.create(schema);
       }
       TypeDescription typeWithIds = ORCSchemaUtil.applyNameMapping(fileSchema, nameMapping);
+      // Name-mapped IDs cannot prove a column was never written, so never omit for defaults.
       readOrcSchema =
-          ORCSchemaUtil.buildOrcProjection(schema, typeWithIds, supportsInitialDefaults);
+          ORCSchemaUtil.buildOrcProjection(
+              schema, typeWithIds, ORCSchemaUtil.FieldIdSource.NAME_MAPPED, false);
     }
 
     SearchArgument sarg = null;

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,163 @@ public void testRequiredNestedFieldMissingInFile() {
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessage("Field 4 of type long is required and was not found.");
   }
+
+  @Test
+  public void testOmitsTopLevelScalarDefaultWhenReaderSupportsDefaultsAndIdsAreEmbedded() {
+    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.FieldIdSource.EMBEDDED, true);
+    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 testSynthesizesNullForTopLevelScalarDefaultWhenReaderDoesNotSupportDefaults() {
+    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());
+
+    TypeDescription projection =
+        ORCSchemaUtil.buildOrcProjection(
+            evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED, false);
+
+    assertEquals(2, projection.getChildren().size());
+    assertNotNull(projection.findSubtype("country_r2"));
+  }
+
+  @Test
+  public void testSynthesizesNullForTopLevelScalarDefaultWhenIdsAreNameMapped() {
+    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());
+
+    // Name-mapped provenance (or the conservative public 2-arg API): an unmatched name does not
+    // prove the column is absent, so synthesize NULL rather than applying the default.
+    TypeDescription projection =
+        ORCSchemaUtil.buildOrcProjection(
+            evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.NAME_MAPPED, true);
+    assertEquals(2, projection.getChildren().size());
+    assertEquals(2, projection.findSubtype("country_r2").getId());
+    assertEquals(
+        TypeDescription.Category.STRING, projection.findSubtype("country_r2").getCategory());
+  }
+
+  @Test
+  public void testOmitsRequiredTopLevelScalarDefaultWhenReaderSupportsDefaults() {
+    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.FieldIdSource.EMBEDDED, true);
+    assertEquals(1, projection.getChildren().size());
+    assertFalse(
+        "required defaulted column must be omitted, not throw",
+        projection.getFieldNames().contains("code_r2"));
+  }
+
+  @Test
+  public void testOmitsNestedScalarDefaultWhenReaderSupportsDefaults() {
+    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.FieldIdSource.EMBEDDED, true);
+    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 testPreservesNestedStructWhenAllProjectedFieldsAreOmitted() {
+    // 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.
+    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.FieldIdSource.EMBEDDED, true);
+    TypeDescription nested = projection.findSubtype("s");
+    assertEquals(0, nested.getChildren().size());
+  }
 }