diff --git a/orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java b/orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java index e641c53dd4..7a60f75168 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java +++ b/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. + * + *

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}. * - *

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. + *

A scalar field at any nesting level is omitted 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 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 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 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. + * + *

{@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. + * + *

The file's outermost struct is not itself annotated, so only its descendants are checked. + */ + static boolean hasAllIds(TypeDescription orcSchema) { + List 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)); } diff --git a/orc/src/main/java/org/apache/iceberg/orc/OrcIterable.java b/orc/src/main/java/org/apache/iceberg/orc/OrcIterable.java index cd8148fdb5..40422ccd4c 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/OrcIterable.java +++ b/orc/src/main/java/org/apache/iceberg/orc/OrcIterable.java @@ -89,14 +89,21 @@ public CloseableIterator 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; diff --git a/orc/src/test/java/org/apache/iceberg/orc/TestBuildOrcProjection.java b/orc/src/test/java/org/apache/iceberg/orc/TestBuildOrcProjection.java index ec42b26f1b..dbd06a75b6 100644 --- a/orc/src/test/java/org/apache/iceberg/orc/TestBuildOrcProjection.java +++ b/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()); + } } diff --git a/orc/src/test/java/org/apache/iceberg/orc/TestORCSchemaUtil.java b/orc/src/test/java/org/apache/iceberg/orc/TestORCSchemaUtil.java index d64e2dd610..46d66cd8e5 100644 --- a/orc/src/test/java/org/apache/iceberg/orc/TestORCSchemaUtil.java +++ b/orc/src/test/java/org/apache/iceberg/orc/TestORCSchemaUtil.java @@ -335,6 +335,28 @@ public void testHasIds() { assertTrue("Should have Ids after adding one type with Id", ORCSchemaUtil.hasIds(orcSchema)); } + @Test + public void testHasAllIds() { + Schema schema = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "data", Types.StringType.get())); + + TypeDescription fullyAnnotated = ORCSchemaUtil.convert(schema); + assertTrue( + "Iceberg-written schemas annotate every column", ORCSchemaUtil.hasAllIds(fullyAnnotated)); + + TypeDescription noneAnnotated = ORCSchemaUtil.removeIds(fullyAnnotated); + assertFalse( + "A file with no ids is not fully annotated", ORCSchemaUtil.hasAllIds(noneAnnotated)); + + // hasIds is true if any column is annotated; hasAllIds requires every column. + TypeDescription partial = ORCSchemaUtil.convert(schema); + TypeDescription extra = TypeDescription.createString(); + partial.addField("unannotated", extra); + assertTrue("One annotated column is enough for hasIds", ORCSchemaUtil.hasIds(partial)); + assertFalse("An unannotated sibling must fail hasAllIds", ORCSchemaUtil.hasAllIds(partial)); + } + @Test public void testAssignIdsByNameMapping() { Types.StructType structType = diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderForFieldsWithDefaultValue.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderForFieldsWithDefaultValue.java index 0f6de50518..fa86eadc83 100644 --- a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderForFieldsWithDefaultValue.java +++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderForFieldsWithDefaultValue.java @@ -29,6 +29,7 @@ import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.orc.ORC; +import org.apache.iceberg.orc.ORCSchemaUtil; import org.apache.iceberg.types.Types; import org.apache.orc.OrcFile; import org.apache.orc.TypeDescription; @@ -38,6 +39,7 @@ import org.apache.spark.sql.catalyst.InternalRow; import org.apache.spark.sql.catalyst.expressions.GenericInternalRow; import org.apache.spark.unsafe.types.UTF8String; +import org.junit.Assert; import org.junit.Rule; import org.junit.Test; import org.junit.rules.TemporaryFolder; @@ -58,7 +60,10 @@ public void testOrcScalarDefaultValues() throws IOException { expectedFirstRow.update(0, 0); expectedFirstRow.update(1, UTF8String.fromString("foo")); - TypeDescription orcSchema = TypeDescription.fromString("struct"); + // Write with Iceberg-embedded field ids (production ORC path). Bare ORC schemas without ids + // take the name-mapped path, which does not omit for defaults. + Schema writeSchema = new Schema(Types.NestedField.required(1, "col1", Types.IntegerType.get())); + TypeDescription orcSchema = ORCSchemaUtil.convert(writeSchema); Schema readSchema = new Schema( @@ -95,7 +100,12 @@ public void testOrcNestedScalarDefaultValues() throws IOException { expectedFirstRow.update(1, expectedLoc); // Empty loc struct in the file: country is absent and will be filled from initial-default. - TypeDescription orcSchema = TypeDescription.fromString("struct>"); + // Use convert() so the file carries embedded field ids (required to omit for defaults). + 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( @@ -124,6 +134,37 @@ public void testOrcNestedScalarDefaultValues() throws IOException { } } + @Test + public void testPartialEmbeddedIdsDoNotFillDefault() throws IOException { + // A file with some iceberg.id attributes takes the EMBEDDED path, but hasAllIds is false, so + // an unannotated physical column must not be treated as absent and filled with the default. + TypeDescription orcSchema = TypeDescription.fromString("struct"); + orcSchema.getChildren().get(0).setAttribute("iceberg.id", "1"); + + Schema readSchema = + new Schema( + Types.NestedField.required(1, "col1", Types.IntegerType.get()), + Types.NestedField.optional("col2") + .withId(2) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("foo")) + .build()); + + File orcFile = writeOrcWithIntAndString(orcSchema, 1, "CA"); + + try (CloseableIterable reader = + ORC.read(Files.localInput(orcFile)) + .project(readSchema) + .createReaderFunc(readOrcSchema -> new SparkOrcReader(readSchema, readOrcSchema)) + .supportsInitialDefaults() + .build()) { + InternalRow row = reader.iterator().next(); + Assert.assertEquals(1, row.getInt(0)); + Assert.assertTrue( + "unannotated physical column must not be replaced by the default", row.isNullAt(1)); + } + } + private File writeOrcWithIntColumn(TypeDescription orcSchema, int numRows) throws IOException { Configuration conf = new Configuration(); File orcFile = temp.newFile(); @@ -151,6 +192,28 @@ private File writeOrcWithIntColumn(TypeDescription orcSchema, int numRows) throw return orcFile; } + private File writeOrcWithIntAndString(TypeDescription orcSchema, int intValue, String stringValue) + throws IOException { + Configuration conf = new Configuration(); + File orcFile = temp.newFile(); + Path orcFilePath = new Path(orcFile.getPath()); + + Writer writer = + OrcFile.createWriter( + orcFilePath, OrcFile.writerOptions(conf).setSchema(orcSchema).overwrite(true)); + + VectorizedRowBatch batch = orcSchema.createRowBatch(); + LongColumnVector intCol = (LongColumnVector) batch.cols[0]; + org.apache.orc.storage.ql.exec.vector.BytesColumnVector strCol = + (org.apache.orc.storage.ql.exec.vector.BytesColumnVector) batch.cols[1]; + int row = batch.size++; + intCol.vector[row] = intValue; + strCol.setVal(row, stringValue.getBytes(java.nio.charset.StandardCharsets.UTF_8)); + writer.addRowBatch(batch); + writer.close(); + return orcFile; + } + private File writeOrcWithIdAndEmptyLoc(TypeDescription orcSchema, int numRows) throws IOException { Configuration conf = new Configuration();