From 11f7928f0c15d491ece35b58888fe0c57f32c812 Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Thu, 23 Jul 2026 10:33:35 -0700 Subject: [PATCH] ORC: fill initial defaults for missing id-bound fields For the generic and id-bound ORC read path, omit an absent field that declares an initial-default from the projection and materialize it as a per-file constant through the reader. Present values and explicit nulls are preserved; id-less/name-mapped files keep legacy null synthesis. Co-authored-by: Cursor --- .../iceberg/data/orc/GenericOrcReaders.java | 4 +- .../org/apache/iceberg/orc/ORCSchemaUtil.java | 33 +- .../org/apache/iceberg/orc/OrcIterable.java | 2 +- .../apache/iceberg/orc/OrcValueReaders.java | 15 + .../iceberg/orc/TestBuildOrcProjection.java | 135 +++++ .../iceberg/orc/TestOrcDefaultValues.java | 498 ++++++++++++++++++ 6 files changed, 681 insertions(+), 6 deletions(-) create mode 100644 orc/src/test/java/org/apache/iceberg/orc/TestOrcDefaultValues.java diff --git a/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReaders.java b/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReaders.java index 506c738015..ae9dc63199 100644 --- a/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReaders.java +++ b/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> readers, Types.StructType structType, Map idToConstant) { - super(orcType, readers, structType, idToConstant); + super( + orcType, readers, structType, idToConstant, IdentityPartitionConverters::convertConstant); this.template = GenericRecord.create(structType); } 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 fae1a76c37..1e66aea0c1 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java +++ b/orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java @@ -261,18 +261,37 @@ public static Schema convert(TypeDescription orcSchema) { */ public static TypeDescription buildOrcProjection( Schema schema, TypeDescription originalOrcSchema) { + return buildOrcProjection(schema, originalOrcSchema, false); + } + + static TypeDescription buildOrcProjection( + Schema schema, TypeDescription originalOrcSchema, boolean applyDefaults) { final Map icebergToOrc = icebergToOrcMapping("root", originalOrcSchema); - return buildOrcProjection(Integer.MIN_VALUE, schema.asStruct(), true, icebergToOrc); + return buildOrcProjection( + Integer.MIN_VALUE, schema.asStruct(), true, applyDefaults, icebergToOrc); + } + + private static boolean isOmittableDefault( + Types.NestedField field, boolean applyDefaults, Map mapping) { + return applyDefaults && field.initialDefault() != null && !mapping.containsKey(field.fieldId()); } private static TypeDescription buildOrcProjection( - Integer fieldId, Type type, boolean isRequired, Map mapping) { + Integer fieldId, + Type type, + boolean isRequired, + boolean applyDefaults, + Map mapping) { final TypeDescription orcType; switch (type.typeId()) { case STRUCT: orcType = TypeDescription.createStruct(); for (Types.NestedField nestedField : type.asStructType().fields()) { + if (isOmittableDefault(nestedField, applyDefaults, mapping)) { + 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 +304,7 @@ private static TypeDescription buildOrcProjection( nestedField.fieldId(), nestedField.type(), isRequired && nestedField.isRequired(), + applyDefaults, mapping); orcType.addField(name, childType); } @@ -296,16 +316,21 @@ private static TypeDescription buildOrcProjection( list.elementId(), list.elementType(), isRequired && list.isElementRequired(), + applyDefaults, 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, applyDefaults, mapping); TypeDescription valueType = buildOrcProjection( - map.valueId(), map.valueType(), isRequired && map.isValueRequired(), mapping); + map.valueId(), + map.valueType(), + isRequired && map.isValueRequired(), + applyDefaults, + mapping); orcType = TypeDescription.createMap(keyType, valueType); break; default: 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 58cf5d1f96..4a4315b6d0 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/OrcIterable.java +++ b/orc/src/main/java/org/apache/iceberg/orc/OrcIterable.java @@ -86,7 +86,7 @@ public CloseableIterator iterator() { TypeDescription fileSchema = orcFileReader.getSchema(); final TypeDescription readOrcSchema; if (ORCSchemaUtil.hasIds(fileSchema)) { - readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, fileSchema); + readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, fileSchema, true); } else { if (nameMapping == null) { nameMapping = MappingUtil.create(schema); diff --git a/orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java b/orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java index e13689eb20..5c8ee85abc 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java +++ b/orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java @@ -21,9 +21,11 @@ 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.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 +185,15 @@ protected StructReader( List> readers, Types.StructType struct, Map idToConstant) { + this(orcType, readers, struct, idToConstant, null); + } + + protected StructReader( + TypeDescription orcType, + List> readers, + Types.StructType struct, + Map idToConstant, + BiFunction convertConstant) { List fields = struct.fields(); this.readers = new OrcValueReader[fields.size()]; this.isConstantOrMetadataField = new boolean[fields.size()]; @@ -208,6 +219,10 @@ protected StructReader( this.isConstantOrMetadataField[pos] = false; this.orcFieldIndex[pos] = fieldIdToOrcIndex.getOrDefault(field.fieldId(), -1); this.readers[pos] = fileReader; + } else if (field.initialDefault() != null && convertConstant != 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); 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..734b6ccb96 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,136 @@ 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, 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 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); + 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, true); + 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 so the reader can fill it. + // 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, 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 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 (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, true); + TypeDescription nested = projection.findSubtype("s"); + assertEquals(0, nested.getChildren().size()); + } } diff --git a/orc/src/test/java/org/apache/iceberg/orc/TestOrcDefaultValues.java b/orc/src/test/java/org/apache/iceberg/orc/TestOrcDefaultValues.java new file mode 100644 index 0000000000..23a3a2b0ac --- /dev/null +++ b/orc/src/test/java/org/apache/iceberg/orc/TestOrcDefaultValues.java @@ -0,0 +1,498 @@ +/* + * 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.orc; + +import static org.apache.iceberg.types.Types.NestedField.optional; +import static org.apache.iceberg.types.Types.NestedField.required; + +import java.io.File; +import java.io.IOException; +import java.math.BigDecimal; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.iceberg.Files; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.data.orc.GenericOrcReader; +import org.apache.iceberg.data.orc.GenericOrcWriter; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.DataWriter; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.Types; +import org.apache.orc.OrcFile; +import org.apache.orc.TypeDescription; +import org.apache.orc.storage.ql.exec.vector.BytesColumnVector; +import org.apache.orc.storage.ql.exec.vector.LongColumnVector; +import org.apache.orc.storage.ql.exec.vector.VectorizedRowBatch; +import org.assertj.core.api.Assertions; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +/** Verifies that scalar {@code initial-default}s are filled on ORC read. */ +public class TestOrcDefaultValues { + + private static final Schema WRITE_SCHEMA = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "data", Types.StringType.get())); + + // Evolved: adds a top-level scalar with an initial-default that is absent from the written file. + private static final Schema READ_SCHEMA = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "data", Types.StringType.get()), + Types.NestedField.optional("country") + .withId(3) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .build()); + + private List records; + + @Rule public TemporaryFolder temp = new TemporaryFolder(); + + @Before + public void createRecords() { + GenericRecord record = GenericRecord.create(WRITE_SCHEMA); + records = + Lists.newArrayList( + record.copy(ImmutableMap.of("id", 1L, "data", "a")), + record.copy(ImmutableMap.of("id", 2L, "data", "b")), + record.copy(ImmutableMap.of("id", 3L, "data", "c"))); + } + + private OutputFile writeFile() throws IOException { + OutputFile file = Files.localOutput(temp.newFile()); + DataWriter writer = + ORC.writeData(file) + .schema(WRITE_SCHEMA) + .createWriterFunc(GenericOrcWriter::buildWriter) + .overwrite() + .withSpec(PartitionSpec.unpartitioned()) + .build(); + try { + for (Record record : records) { + writer.write(record); + } + } finally { + writer.close(); + } + return file; + } + + @Test + public void testReadFillsTopLevelScalarDefault() throws IOException { + OutputFile file = writeFile(); + + List read; + try (CloseableIterable reader = + ORC.read(file.toInputFile()) + .project(READ_SCHEMA) + .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(READ_SCHEMA, fileSchema)) + .build()) { + read = Lists.newArrayList(reader); + } + + Assertions.assertThat(read).hasSize(records.size()); + for (int i = 0; i < read.size(); i += 1) { + Assertions.assertThat(read.get(i).getField("id")).isEqualTo(records.get(i).getField("id")); + Assertions.assertThat(read.get(i).getField("data")) + .isEqualTo(records.get(i).getField("data")); + Assertions.assertThat(read.get(i).getField("country")).isEqualTo("US"); + } + } + + @Test + public void testReadSelectsOnlyDefaultColumn() throws IOException { + OutputFile file = writeFile(); + + Schema onlyDefault = + new Schema( + Types.NestedField.optional("country") + .withId(3) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .build()); + + List read; + try (CloseableIterable reader = + ORC.read(file.toInputFile()) + .project(onlyDefault) + .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(onlyDefault, fileSchema)) + .build()) { + read = Lists.newArrayList(reader); + } + + Assertions.assertThat(read).hasSize(records.size()); + for (Record record : read) { + Assertions.assertThat(record.getField("country")).isEqualTo("US"); + } + } + + @Test + public void testReadFillsScalarDefaultsAllTypes() throws IOException { + OutputFile file = writeFile(); + + Schema typed = + new Schema( + required(1, "id", Types.LongType.get()), + defaulted(10, "b", Types.BooleanType.get(), Expressions.lit(true)), + defaulted(11, "i", Types.IntegerType.get(), Expressions.lit(42)), + defaulted(12, "l", Types.LongType.get(), Expressions.lit(100L)), + defaulted(13, "f", Types.FloatType.get(), Expressions.lit(1.5f)), + defaulted(14, "d", Types.DoubleType.get(), Expressions.lit(2.5d)), + defaulted(15, "s", Types.StringType.get(), Expressions.lit("x")), + defaulted( + 16, "dec", Types.DecimalType.of(9, 2), Expressions.lit(new BigDecimal("1.50")))); + + List read; + try (CloseableIterable reader = + ORC.read(file.toInputFile()) + .project(typed) + .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(typed, fileSchema)) + .build()) { + read = Lists.newArrayList(reader); + } + + Assertions.assertThat(read).hasSize(records.size()); + for (Record record : read) { + Assertions.assertThat(record.getField("b")).isEqualTo(true); + Assertions.assertThat(record.getField("i")).isEqualTo(42); + Assertions.assertThat(record.getField("l")).isEqualTo(100L); + Assertions.assertThat(record.getField("f")).isEqualTo(1.5f); + Assertions.assertThat(record.getField("d")).isEqualTo(2.5d); + Assertions.assertThat(record.getField("s")).isEqualTo("x"); + Assertions.assertThat(record.getField("dec")).isEqualTo(new BigDecimal("1.50")); + } + } + + @Test + public void testReadFillsRequiredScalarDefault() throws IOException { + OutputFile file = writeFile(); + + // A required field absent from the file but declaring a default must be filled, not rejected. + Schema requiredDefault = + new Schema( + required(1, "id", Types.LongType.get()), + Types.NestedField.required("code") + .withId(20) + .ofType(Types.IntegerType.get()) + .withInitialDefault(Expressions.lit(7)) + .build()); + + List read; + try (CloseableIterable reader = + ORC.read(file.toInputFile()) + .project(requiredDefault) + .createReaderFunc( + fileSchema -> GenericOrcReader.buildReader(requiredDefault, fileSchema)) + .build()) { + read = Lists.newArrayList(reader); + } + + Assertions.assertThat(read).hasSize(records.size()); + for (Record record : read) { + Assertions.assertThat(record.getField("code")).isEqualTo(7); + } + } + + @Test + public void testReadDoesNotApplyDefaultToIdLessFile() throws IOException { + File file = writeIdLessFile(); + + List read; + try (CloseableIterable reader = + ORC.read(Files.localInput(file)) + .project(READ_SCHEMA) + .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(READ_SCHEMA, fileSchema)) + .build()) { + read = Lists.newArrayList(reader); + } + + Assertions.assertThat(read).hasSize(records.size()); + for (int i = 0; i < read.size(); i += 1) { + Assertions.assertThat(read.get(i).getField("id")).isEqualTo(records.get(i).getField("id")); + Assertions.assertThat(read.get(i).getField("data")) + .isEqualTo(records.get(i).getField("data")); + Assertions.assertThat(read.get(i).getField("country")).isNull(); + } + } + + @Test + public void testReadDoesNotOverridePresentColumn() throws IOException { + Schema writeSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "data", Types.StringType.get()), + optional(3, "country", Types.StringType.get())); + Record present = GenericRecord.create(writeSchema); + present.setField("id", 1L); + present.setField("data", "a"); + present.setField("country", "CA"); + Record presentNull = GenericRecord.create(writeSchema); + presentNull.setField("id", 2L); + presentNull.setField("data", "b"); + presentNull.setField("country", null); + + OutputFile file = writeRecords(writeSchema, Lists.newArrayList(present, presentNull)); + List read = read(file, READ_SCHEMA); + + Assertions.assertThat(read).hasSize(2); + Assertions.assertThat(read.get(0).getField("country")).isEqualTo("CA"); + Assertions.assertThat(read.get(1).getField("country")).isNull(); + } + + @Test + public void testNestedStructScalarDefault() throws IOException { + Schema writeSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 3, "nested", Types.StructType.of(required(4, "inner", Types.StringType.get())))); + Types.StructType writeNested = writeSchema.findField("nested").type().asStructType(); + + List recs = Lists.newArrayList(); + for (int i = 0; i < 3; i += 1) { + Record nested = GenericRecord.create(writeNested); + nested.setField("inner", "v" + i); + Record rec = GenericRecord.create(writeSchema); + rec.setField("id", (long) i); + rec.setField("nested", nested); + recs.add(rec); + } + // Row with a null nested struct: a default must not be fabricated when the parent struct is + // null (the struct stays null; the absent-only fill applies to present structs). + Record nullNested = GenericRecord.create(writeSchema); + nullNested.setField("id", 3L); + nullNested.setField("nested", null); + recs.add(nullNested); + + OutputFile file = writeRecords(writeSchema, recs); + + Schema readSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 3, + "nested", + Types.StructType.of( + required(4, "inner", Types.StringType.get()), + defaulted(5, "missing", Types.FloatType.get(), Expressions.lit(-0.0F))))); + + List read = read(file, readSchema); + Assertions.assertThat(read).hasSize(recs.size()); + for (int i = 0; i < 3; i += 1) { + Record nested = (Record) read.get(i).getField("nested"); + Assertions.assertThat(nested.getField("inner")).isEqualTo("v" + i); + Assertions.assertThat(nested.getField("missing")).isEqualTo(-0.0F); + } + Assertions.assertThat(read.get(3).getField("nested")).isNull(); + } + + @Test + public void testMapNestedScalarDefault() throws IOException { + Schema writeSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 3, + "m", + Types.MapType.ofOptional( + 4, + 5, + Types.StringType.get(), + Types.StructType.of(required(6, "v_str", Types.StringType.get()))))); + Types.StructType writeValue = + writeSchema.findField("m").type().asMapType().valueType().asStructType(); + + Record value = GenericRecord.create(writeValue); + value.setField("v_str", "s"); + Record rec = GenericRecord.create(writeSchema); + rec.setField("id", 1L); + rec.setField("m", Collections.singletonMap("k", value)); + OutputFile file = writeRecords(writeSchema, Lists.newArrayList(rec)); + + Schema readSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 3, + "m", + Types.MapType.ofOptional( + 4, + 5, + Types.StringType.get(), + Types.StructType.of( + required(6, "v_str", Types.StringType.get()), + defaulted(7, "v_int", Types.IntegerType.get(), Expressions.lit(34)))))); + + List read = read(file, readSchema); + Assertions.assertThat(read).hasSize(1); + Map map = (Map) read.get(0).getField("m"); + Assertions.assertThat(map).hasSize(1); + Record readValue = (Record) map.values().iterator().next(); + Assertions.assertThat(readValue.getField("v_str")).isEqualTo("s"); + Assertions.assertThat(readValue.getField("v_int")).isEqualTo(34); + } + + @Test + public void testListNestedScalarDefault() throws IOException { + Schema writeSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 3, + "l", + Types.ListType.ofOptional( + 4, Types.StructType.of(required(5, "e_str", Types.StringType.get()))))); + Types.StructType writeElement = + writeSchema.findField("l").type().asListType().elementType().asStructType(); + + Record element = GenericRecord.create(writeElement); + element.setField("e_str", "e"); + Record rec = GenericRecord.create(writeSchema); + rec.setField("id", 1L); + rec.setField("l", Collections.singletonList(element)); + OutputFile file = writeRecords(writeSchema, Lists.newArrayList(rec)); + + Schema readSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 3, + "l", + Types.ListType.ofOptional( + 4, + Types.StructType.of( + required(5, "e_str", Types.StringType.get()), + defaulted(7, "e_int", Types.IntegerType.get(), Expressions.lit(34)))))); + + List read = read(file, readSchema); + Assertions.assertThat(read).hasSize(1); + List list = (List) read.get(0).getField("l"); + Assertions.assertThat(list).hasSize(1); + Record readElement = (Record) list.get(0); + Assertions.assertThat(readElement.getField("e_str")).isEqualTo("e"); + Assertions.assertThat(readElement.getField("e_int")).isEqualTo(34); + } + + @Test + public void testNestedStructAllSubfieldsDefaulted() throws IOException { + // File: nested { a }. Read projects only a new defaulted subfield nested { b default 'x' } + // (a dropped), so the nested read struct is empty. The default must still fill. + Schema writeSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional(3, "nested", Types.StructType.of(required(4, "a", Types.LongType.get())))); + Types.StructType writeNested = writeSchema.findField("nested").type().asStructType(); + + Record nested = GenericRecord.create(writeNested); + nested.setField("a", 9L); + Record rec = GenericRecord.create(writeSchema); + rec.setField("id", 1L); + rec.setField("nested", nested); + OutputFile file = writeRecords(writeSchema, Lists.newArrayList(rec)); + + Schema readSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 3, + "nested", + Types.StructType.of( + defaulted(5, "b", Types.StringType.get(), Expressions.lit("x"))))); + + List read = read(file, readSchema); + Assertions.assertThat(read).hasSize(1); + Record readNested = (Record) read.get(0).getField("nested"); + Assertions.assertThat(readNested.getField("b")).isEqualTo("x"); + } + + private OutputFile writeRecords(Schema schema, List recs) throws IOException { + OutputFile file = Files.localOutput(temp.newFile()); + DataWriter writer = + ORC.writeData(file) + .schema(schema) + .createWriterFunc(GenericOrcWriter::buildWriter) + .overwrite() + .withSpec(PartitionSpec.unpartitioned()) + .build(); + try { + for (Record rec : recs) { + writer.write(rec); + } + } finally { + writer.close(); + } + return file; + } + + private File writeIdLessFile() throws IOException { + File file = temp.newFile(); + Assertions.assertThat(file.delete()).isTrue(); + TypeDescription writerSchema = TypeDescription.fromString("struct"); + try (org.apache.orc.Writer writer = + OrcFile.createWriter( + new Path(file.toString()), + OrcFile.writerOptions(new Configuration()).setSchema(writerSchema))) { + VectorizedRowBatch batch = writerSchema.createRowBatch(); + LongColumnVector ids = (LongColumnVector) batch.cols[0]; + BytesColumnVector data = (BytesColumnVector) batch.cols[1]; + for (Record record : records) { + int row = batch.size++; + ids.vector[row] = (Long) record.getField("id"); + data.setVal(row, record.getField("data").toString().getBytes(StandardCharsets.UTF_8)); + } + writer.addRowBatch(batch); + } + return file; + } + + private List read(OutputFile file, Schema readSchema) throws IOException { + try (CloseableIterable reader = + ORC.read(file.toInputFile()) + .project(readSchema) + .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(readSchema, fileSchema)) + .build()) { + return Lists.newArrayList(reader); + } + } + + private static Types.NestedField defaulted( + int id, + String name, + org.apache.iceberg.types.Type type, + org.apache.iceberg.expressions.Literal initial) { + return Types.NestedField.optional(name) + .withId(id) + .ofType(type) + .withInitialDefault(initial) + .build(); + } +}