From 0e7b860572d454b8e57896fba506ff620522c01d Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Thu, 23 Jul 2026 13:55:00 -0700 Subject: [PATCH 1/5] ORC: fill initial defaults for missing id-bound fields Fill a field's initial-default when it is absent from an id-bearing ORC file, mirroring how the Parquet reader resolves defaults. buildOrcProjection omits an absent field that declares a default when the file's embedded ids are trustworthy, so the id-binding StructReader finds no column for it and materializes the declared default as a per-file constant. This reuses the reader's existing constant path, so a defaulted field consumes no column vector. A present column always wins, including one holding an explicit null. Id-less and name-mapped files keep legacy null synthesis, so a default is never name-matched onto a legacy or migrated file. Engines that have not opted into id binding are unaffected: default filling is enabled only by passing a constant converter, and the positional path keeps its strict missing-reader failure. Co-authored-by: Cursor --- .../iceberg/data/orc/GenericOrcReaders.java | 4 +- .../org/apache/iceberg/orc/ORCSchemaUtil.java | 43 +- .../org/apache/iceberg/orc/OrcIterable.java | 2 +- .../apache/iceberg/orc/OrcValueReaders.java | 29 + .../iceberg/orc/TestBuildOrcProjection.java | 135 +++++ .../iceberg/orc/TestOrcDefaultValues.java | 498 ++++++++++++++++++ 6 files changed, 705 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..8198131d85 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,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. + * + *

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 omitted from the read + * projection. An id-binding reader sees no column for that field and fills the declared default + * as a per-file constant. When false, the field is synthesized as a null column instead, so a + * default is never name-matched onto an id-less file. + */ + static TypeDescription buildOrcProjection( + Schema schema, TypeDescription originalOrcSchema, boolean hasTrustedIds) { final Map 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 mapping) { + return field.initialDefault() != null && !mapping.containsKey(field.fieldId()) && hasTrustedIds; } private static TypeDescription buildOrcProjection( - Integer fieldId, Type type, boolean isRequired, Map mapping) { + Integer fieldId, + Type type, + boolean isRequired, + boolean hasTrustedIds, + Map 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 +314,7 @@ private static TypeDescription buildOrcProjection( nestedField.fieldId(), nestedField.type(), isRequired && nestedField.isRequired(), + hasTrustedIds, mapping); orcType.addField(name, childType); } @@ -296,16 +326,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: 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..86151717a3 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,12 @@ 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.Schema; 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 +186,28 @@ protected StructReader( List> readers, Types.StructType struct, Map idToConstant) { + this(orcType, readers, struct, idToConstant, null); + } + + /** + * Binds struct fields by Iceberg field id, filling a field's {@code initial-default} when the + * file has no column for it. + * + *

A field reaches the default only when the read projection omitted it, which {@link + * ORCSchemaUtil#buildOrcProjection(Schema, TypeDescription, boolean)} does only for an absent + * field that declares a default in a file whose ids are trustworthy. The default is + * materialized as a per-file constant and consumes no column vector, so a present column -- + * including one holding an explicit null -- always wins. + * + * @param convertConstant converts a default to the engine's in-memory representation, or null + * to disable default filling and keep the strict missing-reader failure + */ + 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 +233,10 @@ protected StructReader( this.isConstantOrMetadataField[pos] = false; this.orcFieldIndex[pos] = fieldIdToOrcIndex.getOrDefault(field.fieldId(), -1); this.readers[pos] = fileReader; + } else if (convertConstant != null && field.initialDefault() != 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..35eda68647 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 (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, 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 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, 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(); + } +} From 6dfb91d0b358894e252bc42e7f3479fa7f3d5581 Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Sat, 1 Aug 2026 23:36:23 -0700 Subject: [PATCH 2/5] ORC: name field id provenance with a FieldIdSource enum Replace buildOrcProjection's boolean hasTrustedIds parameter with a FieldIdSource enum naming where a schema's Iceberg field ids came from: EMBEDDED for ids read from iceberg.id column attributes written into the file, NAME_MAPPED for ids derived at read time by matching column names. "Trusted" asserted a judgment without saying who trusts the ids or why, which a reviewer had to resolve by reading OrcIterable. The provenance is the checkable fact, and it is what decides whether an absent field may be read as "never written" and so have its declared default filled. Naming it at the call site also makes the two branches in OrcIterable read as the matched pair they are, rather than one passing a bare true and the other omitting the argument entirely. Each constant documents its own hazard: under NAME_MAPPED 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, so filling a default would fabricate values over real data. No behavior change. Also record why isOmittableDefault need not check that the field is scalar: Types.NestedField#castDefault already rejects a default on any nested type at construction time. --- .../org/apache/iceberg/orc/ORCSchemaUtil.java | 75 ++++++++++++++----- .../org/apache/iceberg/orc/OrcIterable.java | 8 +- .../apache/iceberg/orc/OrcValueReaders.java | 8 +- .../iceberg/orc/TestBuildOrcProjection.java | 20 +++-- 4 files changed, 78 insertions(+), 33 deletions(-) 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 8198131d85..7e2ce8c94d 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,35 +287,46 @@ 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 fill + // defaults, matching this method's behavior before defaults were supported. + return buildOrcProjection(schema, originalOrcSchema, FieldIdSource.NAME_MAPPED); } /** - * Builds the ORC read schema for a file whose embedded field IDs are known to be trustworthy. + * Builds the ORC read schema, omitting absent fields that declare an {@code initial-default} so a + * default-aware reader can fill them. + * + *

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, and {@code fieldIdSource} is + * {@link FieldIdSource#EMBEDDED}. An id-binding reader then sees no column for that field and + * fills the declared default as a per-file constant. Under {@link FieldIdSource#NAME_MAPPED} the + * field is synthesized as a null column instead, so a default is never name-matched onto a file + * that carries no IDs of its own. * - *

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 omitted from the read - * projection. An id-binding reader sees no column for that field and fills the declared default - * as a per-file constant. When false, the field is synthesized as a null column instead, so a - * default is never name-matched onto an id-less file. + * @param fieldIdSource where the IDs in {@code originalOrcSchema} came from; see {@link + * FieldIdSource} */ static TypeDescription buildOrcProjection( - Schema schema, TypeDescription originalOrcSchema, boolean hasTrustedIds) { + Schema schema, TypeDescription originalOrcSchema, FieldIdSource fieldIdSource) { final Map icebergToOrc = icebergToOrcMapping("root", originalOrcSchema); return buildOrcProjection( - Integer.MIN_VALUE, schema.asStruct(), true, hasTrustedIds, icebergToOrc); + Integer.MIN_VALUE, schema.asStruct(), true, fieldIdSource, icebergToOrc); } private static boolean isOmittableDefault( - Types.NestedField field, boolean hasTrustedIds, Map mapping) { - return field.initialDefault() != null && !mapping.containsKey(field.fieldId()) && hasTrustedIds; + Types.NestedField field, FieldIdSource fieldIdSource, 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 field.initialDefault() != null + && !mapping.containsKey(field.fieldId()) + && fieldIdSource == FieldIdSource.EMBEDDED; } private static TypeDescription buildOrcProjection( Integer fieldId, Type type, boolean isRequired, - boolean hasTrustedIds, + FieldIdSource fieldIdSource, Map mapping) { final TypeDescription orcType; @@ -297,9 +334,9 @@ private static TypeDescription buildOrcProjection( 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. + if (isOmittableDefault(nestedField, fieldIdSource, 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 @@ -314,7 +351,7 @@ private static TypeDescription buildOrcProjection( nestedField.fieldId(), nestedField.type(), isRequired && nestedField.isRequired(), - hasTrustedIds, + fieldIdSource, mapping); orcType.addField(name, childType); } @@ -326,20 +363,20 @@ private static TypeDescription buildOrcProjection( list.elementId(), list.elementType(), isRequired && list.isElementRequired(), - hasTrustedIds, + fieldIdSource, mapping); orcType = TypeDescription.createList(elementType); break; case MAP: Types.MapType map = (Types.MapType) type; TypeDescription keyType = - buildOrcProjection(map.keyId(), map.keyType(), isRequired, hasTrustedIds, mapping); + buildOrcProjection(map.keyId(), map.keyType(), isRequired, fieldIdSource, mapping); TypeDescription valueType = buildOrcProjection( map.valueId(), map.valueType(), isRequired && map.isValueRequired(), - hasTrustedIds, + fieldIdSource, mapping); orcType = TypeDescription.createMap(keyType, valueType); break; 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 4a4315b6d0..6c9f440e5d 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/OrcIterable.java +++ b/orc/src/main/java/org/apache/iceberg/orc/OrcIterable.java @@ -86,13 +86,17 @@ public CloseableIterator iterator() { TypeDescription fileSchema = orcFileReader.getSchema(); final TypeDescription readOrcSchema; if (ORCSchemaUtil.hasIds(fileSchema)) { - readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, fileSchema, true); + readOrcSchema = + ORCSchemaUtil.buildOrcProjection( + schema, fileSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED); } else { if (nameMapping == null) { nameMapping = MappingUtil.create(schema); } TypeDescription typeWithIds = ORCSchemaUtil.applyNameMapping(fileSchema, nameMapping); - readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, typeWithIds); + readOrcSchema = + ORCSchemaUtil.buildOrcProjection( + schema, typeWithIds, ORCSchemaUtil.FieldIdSource.NAME_MAPPED); } SearchArgument sarg = null; 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 86151717a3..3dca7f1dac 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java +++ b/orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java @@ -194,10 +194,10 @@ protected StructReader( * file has no column for it. * *

A field reaches the default only when the read projection omitted it, which {@link - * ORCSchemaUtil#buildOrcProjection(Schema, TypeDescription, boolean)} does only for an absent - * field that declares a default in a file whose ids are trustworthy. The default is - * materialized as a per-file constant and consumes no column vector, so a present column -- - * including one holding an explicit null -- always wins. + * ORCSchemaUtil#buildOrcProjection(Schema, TypeDescription, ORCSchemaUtil.FieldIdSource)} does + * only for an absent field that declares a default in a file carrying its own Iceberg field + * ids. The default is materialized as a per-file constant and consumes no column vector, so a + * present column -- including one holding an explicit null -- always wins. * * @param convertConstant converts a default to the engine's in-memory representation, or null * to disable default filling and keep the strict missing-reader failure 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 35eda68647..b8cf1c507a 100644 --- a/orc/src/test/java/org/apache/iceberg/orc/TestBuildOrcProjection.java +++ b/orc/src/test/java/org/apache/iceberg/orc/TestBuildOrcProjection.java @@ -164,7 +164,7 @@ public void testRequiredNestedFieldMissingInFile() { } @Test - public void testTopLevelScalarDefaultOmittedWhenFileIdentityIsTrustworthy() { + public void testTopLevelScalarDefaultOmittedWhenFileHasEmbeddedFieldIds() { Schema baseSchema = new Schema(required(1, "id", Types.LongType.get())); TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema); @@ -179,7 +179,8 @@ public void testTopLevelScalarDefaultOmittedWhenFileIdentityIsTrustworthy() { // The file carries embedded field IDs, so the absent field can be identified safely. TypeDescription projection = - ORCSchemaUtil.buildOrcProjection(evolvedSchema, baseOrcSchema, true); + ORCSchemaUtil.buildOrcProjection( + evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED); assertEquals(1, projection.getChildren().size()); assertNotNull(projection.findSubtype("id")); assertFalse( @@ -188,7 +189,7 @@ public void testTopLevelScalarDefaultOmittedWhenFileIdentityIsTrustworthy() { } @Test - public void testTopLevelScalarDefaultSynthesizedWithoutTrustedFileIdentity() { + public void testTopLevelScalarDefaultSynthesizedWhenFieldIdsAreNameMapped() { Schema baseSchema = new Schema(required(1, "id", Types.LongType.get())); TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema); @@ -201,8 +202,8 @@ public void testTopLevelScalarDefaultSynthesizedWithoutTrustedFileIdentity() { .withInitialDefault(Expressions.lit("US")) .build()); - // Without trustworthy file identity, synthesize NULL rather than guessing that the field is - // absent and applying its default. + // With name-mapped ids an unmatched name does not prove the column is absent, so synthesize + // NULL rather than guessing and applying the default. TypeDescription projection = ORCSchemaUtil.buildOrcProjection(evolvedSchema, baseOrcSchema); assertEquals(2, projection.getChildren().size()); assertEquals(2, projection.findSubtype("country_r2").getId()); @@ -227,7 +228,8 @@ public void testTopLevelRequiredScalarDefaultOmitted() { .build()); TypeDescription projection = - ORCSchemaUtil.buildOrcProjection(evolvedSchema, baseOrcSchema, true); + ORCSchemaUtil.buildOrcProjection( + evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED); assertEquals(1, projection.getChildren().size()); assertFalse( "required defaulted column must be omitted, not throw", @@ -259,7 +261,8 @@ public void testNestedScalarDefaultOmitted() { .build()))); TypeDescription projection = - ORCSchemaUtil.buildOrcProjection(evolvedSchema, baseOrcSchema, true); + ORCSchemaUtil.buildOrcProjection( + evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED); TypeDescription nested = projection.findSubtype("s"); assertEquals(1, nested.getChildren().size()); assertFalse("nested defaulted column must be omitted", nested.getFieldNames().contains("b_r4")); @@ -290,7 +293,8 @@ public void testNestedStructEmptiedByOmit() { .build()))); TypeDescription projection = - ORCSchemaUtil.buildOrcProjection(evolvedSchema, baseOrcSchema, true); + ORCSchemaUtil.buildOrcProjection( + evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED); TypeDescription nested = projection.findSubtype("s"); assertEquals(0, nested.getChildren().size()); } From 65e19962d8f11d28a224c96842c15a70fc89b08c Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Sun, 2 Aug 2026 00:59:51 -0700 Subject: [PATCH 3/5] ORC: gate initial-default projection by reader support --- .../apache/iceberg/data/GenericReader.java | 1 + .../orc/ExpressionToSearchArgument.java | 3 +- .../main/java/org/apache/iceberg/orc/ORC.java | 17 ++++++- .../org/apache/iceberg/orc/ORCSchemaUtil.java | 48 ++++++++++++++----- .../org/apache/iceberg/orc/OrcIterable.java | 10 +++- .../apache/iceberg/orc/OrcValueReaders.java | 8 ++-- .../iceberg/orc/TestBuildOrcProjection.java | 29 +++++++++-- .../iceberg/orc/TestOrcDefaultValues.java | 6 +++ 8 files changed, 98 insertions(+), 24 deletions(-) diff --git a/data/src/main/java/org/apache/iceberg/data/GenericReader.java b/data/src/main/java/org/apache/iceberg/data/GenericReader.java index 3637bf00bd..9c193fc291 100644 --- a/data/src/main/java/org/apache/iceberg/data/GenericReader.java +++ b/data/src/main/java/org/apache/iceberg/data/GenericReader.java @@ -137,6 +137,7 @@ private CloseableIterable openFile(FileScanTask task, Schema fileProject .createReaderFunc( fileSchema -> GenericOrcReader.buildReader(fileProjection, fileSchema, partition)) + .supportsInitialDefaults() .split(task.start(), task.length()) .filter(task.residual()); diff --git a/orc/src/main/java/org/apache/iceberg/orc/ExpressionToSearchArgument.java b/orc/src/main/java/org/apache/iceberg/orc/ExpressionToSearchArgument.java index ba05d966c8..ab6751657f 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/ExpressionToSearchArgument.java +++ b/orc/src/main/java/org/apache/iceberg/orc/ExpressionToSearchArgument.java @@ -270,7 +270,8 @@ public Action notStartsWith(Bound expr, Literal lit) { @Override public Action predicate(BoundPredicate pred) { - if (UNSUPPORTED_TYPES.contains(pred.ref().type().typeId())) { + if (!idToColumnName.containsKey(pred.ref().fieldId()) + || UNSUPPORTED_TYPES.contains(pred.ref().type().typeId())) { // Cannot push down predicates for types which cannot be represented in PredicateLeaf.Type, so // return // TruthValue.YES_NO_NULL which signifies that this predicate cannot help with filtering diff --git a/orc/src/main/java/org/apache/iceberg/orc/ORC.java b/orc/src/main/java/org/apache/iceberg/orc/ORC.java index 89cd1ad436..20cbef97f0 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/ORC.java +++ b/orc/src/main/java/org/apache/iceberg/orc/ORC.java @@ -672,6 +672,7 @@ public static class ReadBuilder { private Function> readerFunc; private Function> batchedReaderFunc; + private boolean supportsInitialDefaults = false; private int recordsPerBatch = VectorizedRowBatch.DEFAULT_SIZE; private ReadBuilder(InputFile file) { @@ -725,6 +726,19 @@ public ReadBuilder createReaderFunc(Function> r return this; } + /** + * Signals that the configured row reader can fill initial defaults for fields omitted from an + * ORC file. Disabled by default so existing positional and vectorized readers retain their + * previous null-synthesizing projection behavior. + */ + public ReadBuilder supportsInitialDefaults() { + Preconditions.checkState( + this.readerFunc != null, + "A row 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> 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 CloseableIterable build() { start, length, readerFunc, + supportsInitialDefaults, caseSensitive, filter, batchedReaderFunc, 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 7e2ce8c94d..3bac5599e5 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java +++ b/orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java @@ -289,7 +289,7 @@ public static TypeDescription buildOrcProjection( Schema schema, TypeDescription originalOrcSchema) { // Callers that cannot establish ID provenance get the conservative behavior: never fill // defaults, matching this method's behavior before defaults were supported. - return buildOrcProjection(schema, originalOrcSchema, FieldIdSource.NAME_MAPPED); + return buildOrcProjection(schema, originalOrcSchema, FieldIdSource.NAME_MAPPED, false); } /** @@ -297,27 +297,41 @@ public static TypeDescription buildOrcProjection( * default-aware reader can fill them. * *

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, and {@code fieldIdSource} is - * {@link FieldIdSource#EMBEDDED}. An id-binding reader then sees no column for that field and - * fills the declared default as a per-file constant. Under {@link FieldIdSource#NAME_MAPPED} the - * field is synthesized as a null column instead, so a default is never name-matched onto a file - * that carries no IDs of its own. + * 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, FieldIdSource fieldIdSource) { + Schema schema, + TypeDescription originalOrcSchema, + FieldIdSource fieldIdSource, + boolean supportsInitialDefaults) { final Map icebergToOrc = icebergToOrcMapping("root", originalOrcSchema); return buildOrcProjection( - Integer.MIN_VALUE, schema.asStruct(), true, fieldIdSource, icebergToOrc); + Integer.MIN_VALUE, + schema.asStruct(), + true, + fieldIdSource, + supportsInitialDefaults, + icebergToOrc); } private static boolean isOmittableDefault( - Types.NestedField field, FieldIdSource fieldIdSource, Map mapping) { + 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 field.initialDefault() != null + return supportsInitialDefaults + && field.initialDefault() != null && !mapping.containsKey(field.fieldId()) && fieldIdSource == FieldIdSource.EMBEDDED; } @@ -327,6 +341,7 @@ private static TypeDescription buildOrcProjection( Type type, boolean isRequired, FieldIdSource fieldIdSource, + boolean supportsInitialDefaults, Map mapping) { final TypeDescription orcType; @@ -334,7 +349,7 @@ private static TypeDescription buildOrcProjection( case STRUCT: orcType = TypeDescription.createStruct(); for (Types.NestedField nestedField : type.asStructType().fields()) { - if (isOmittableDefault(nestedField, fieldIdSource, mapping)) { + 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; @@ -352,6 +367,7 @@ private static TypeDescription buildOrcProjection( nestedField.type(), isRequired && nestedField.isRequired(), fieldIdSource, + supportsInitialDefaults, mapping); orcType.addField(name, childType); } @@ -364,19 +380,27 @@ private static TypeDescription buildOrcProjection( list.elementType(), isRequired && list.isElementRequired(), fieldIdSource, + supportsInitialDefaults, mapping); orcType = TypeDescription.createList(elementType); break; case MAP: Types.MapType map = (Types.MapType) type; TypeDescription keyType = - buildOrcProjection(map.keyId(), map.keyType(), isRequired, fieldIdSource, mapping); + buildOrcProjection( + 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); break; 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 6c9f440e5d..62def6f048 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/OrcIterable.java +++ b/orc/src/main/java/org/apache/iceberg/orc/OrcIterable.java @@ -46,6 +46,7 @@ class OrcIterable extends CloseableGroup implements CloseableIterable { private final Long start; private final Long length; private final Function> readerFunction; + private final boolean supportsInitialDefaults; private final Expression filter; private final boolean caseSensitive; private final Function> batchReaderFunction; @@ -60,12 +61,14 @@ class OrcIterable extends CloseableGroup implements CloseableIterable { Long start, Long length, Function> readerFunction, + boolean supportsInitialDefaults, boolean caseSensitive, Expression filter, Function> batchReaderFunction, int recordsPerBatch) { this.schema = schema; this.readerFunction = readerFunction; + this.supportsInitialDefaults = supportsInitialDefaults; this.file = file; this.nameMapping = nameMapping; this.start = start; @@ -88,7 +91,7 @@ public CloseableIterator iterator() { if (ORCSchemaUtil.hasIds(fileSchema)) { readOrcSchema = ORCSchemaUtil.buildOrcProjection( - schema, fileSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED); + schema, fileSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED, supportsInitialDefaults); } else { if (nameMapping == null) { nameMapping = MappingUtil.create(schema); @@ -96,7 +99,10 @@ public CloseableIterator iterator() { TypeDescription typeWithIds = ORCSchemaUtil.applyNameMapping(fileSchema, nameMapping); readOrcSchema = ORCSchemaUtil.buildOrcProjection( - schema, typeWithIds, ORCSchemaUtil.FieldIdSource.NAME_MAPPED); + schema, + typeWithIds, + ORCSchemaUtil.FieldIdSource.NAME_MAPPED, + supportsInitialDefaults); } SearchArgument sarg = null; 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 3dca7f1dac..b11c7dfc53 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java +++ b/orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java @@ -194,10 +194,10 @@ protected StructReader( * file has no column for it. * *

A field reaches the default only when the read projection omitted it, which {@link - * ORCSchemaUtil#buildOrcProjection(Schema, TypeDescription, ORCSchemaUtil.FieldIdSource)} does - * only for an absent field that declares a default in a file carrying its own Iceberg field - * ids. The default is materialized as a per-file constant and consumes no column vector, so a - * present column -- including one holding an explicit null -- always wins. + * ORCSchemaUtil#buildOrcProjection(Schema, TypeDescription, ORCSchemaUtil.FieldIdSource, + * boolean)} does only for an absent field that declares a default in a file carrying its own + * Iceberg field ids. The default is materialized as a per-file constant and consumes no column + * vector, so a present column -- including one holding an explicit null -- always wins. * * @param convertConstant converts a default to the engine's in-memory representation, or null * to disable default filling and keep the strict missing-reader failure 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 b8cf1c507a..68fef17684 100644 --- a/orc/src/test/java/org/apache/iceberg/orc/TestBuildOrcProjection.java +++ b/orc/src/test/java/org/apache/iceberg/orc/TestBuildOrcProjection.java @@ -180,7 +180,7 @@ public void testTopLevelScalarDefaultOmittedWhenFileHasEmbeddedFieldIds() { // The file carries embedded field IDs, so the absent field can be identified safely. TypeDescription projection = ORCSchemaUtil.buildOrcProjection( - evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED); + evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED, true); assertEquals(1, projection.getChildren().size()); assertNotNull(projection.findSubtype("id")); assertFalse( @@ -188,6 +188,27 @@ public void testTopLevelScalarDefaultOmittedWhenFileHasEmbeddedFieldIds() { projection.getFieldNames().contains("country_r2")); } + @Test + public void testTopLevelScalarDefaultSynthesizedWhenReaderDoesNotSupportDefaults() { + 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 testTopLevelScalarDefaultSynthesizedWhenFieldIdsAreNameMapped() { Schema baseSchema = new Schema(required(1, "id", Types.LongType.get())); @@ -229,7 +250,7 @@ public void testTopLevelRequiredScalarDefaultOmitted() { TypeDescription projection = ORCSchemaUtil.buildOrcProjection( - evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED); + evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED, true); assertEquals(1, projection.getChildren().size()); assertFalse( "required defaulted column must be omitted, not throw", @@ -262,7 +283,7 @@ public void testNestedScalarDefaultOmitted() { TypeDescription projection = ORCSchemaUtil.buildOrcProjection( - evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED); + 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")); @@ -294,7 +315,7 @@ public void testNestedStructEmptiedByOmit() { TypeDescription projection = ORCSchemaUtil.buildOrcProjection( - evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED); + 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/TestOrcDefaultValues.java b/orc/src/test/java/org/apache/iceberg/orc/TestOrcDefaultValues.java index 23a3a2b0ac..33e016e13c 100644 --- a/orc/src/test/java/org/apache/iceberg/orc/TestOrcDefaultValues.java +++ b/orc/src/test/java/org/apache/iceberg/orc/TestOrcDefaultValues.java @@ -115,6 +115,7 @@ public void testReadFillsTopLevelScalarDefault() throws IOException { ORC.read(file.toInputFile()) .project(READ_SCHEMA) .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(READ_SCHEMA, fileSchema)) + .supportsInitialDefaults() .build()) { read = Lists.newArrayList(reader); } @@ -145,6 +146,7 @@ public void testReadSelectsOnlyDefaultColumn() throws IOException { ORC.read(file.toInputFile()) .project(onlyDefault) .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(onlyDefault, fileSchema)) + .supportsInitialDefaults() .build()) { read = Lists.newArrayList(reader); } @@ -176,6 +178,7 @@ public void testReadFillsScalarDefaultsAllTypes() throws IOException { ORC.read(file.toInputFile()) .project(typed) .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(typed, fileSchema)) + .supportsInitialDefaults() .build()) { read = Lists.newArrayList(reader); } @@ -212,6 +215,7 @@ public void testReadFillsRequiredScalarDefault() throws IOException { .project(requiredDefault) .createReaderFunc( fileSchema -> GenericOrcReader.buildReader(requiredDefault, fileSchema)) + .supportsInitialDefaults() .build()) { read = Lists.newArrayList(reader); } @@ -231,6 +235,7 @@ public void testReadDoesNotApplyDefaultToIdLessFile() throws IOException { ORC.read(Files.localInput(file)) .project(READ_SCHEMA) .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(READ_SCHEMA, fileSchema)) + .supportsInitialDefaults() .build()) { read = Lists.newArrayList(reader); } @@ -479,6 +484,7 @@ private List read(OutputFile file, Schema readSchema) throws IOException ORC.read(file.toInputFile()) .project(readSchema) .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(readSchema, fileSchema)) + .supportsInitialDefaults() .build()) { return Lists.newArrayList(reader); } From ca18b890545f584622b225ad70bb72100d100c9a Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Sun, 2 Aug 2026 19:02:41 -0700 Subject: [PATCH 4/5] ORC: add initial-default compatibility tests --- .../TestInclusiveMetricsEvaluator.java | 18 +++ .../iceberg/orc/TestBuildOrcProjection.java | 12 +- .../orc/TestExpressionToSearchArgument.java | 27 ++++ .../iceberg/orc/TestOrcDefaultValues.java | 151 ++++++++++++++++-- 4 files changed, 192 insertions(+), 16 deletions(-) diff --git a/api/src/test/java/org/apache/iceberg/expressions/TestInclusiveMetricsEvaluator.java b/api/src/test/java/org/apache/iceberg/expressions/TestInclusiveMetricsEvaluator.java index d1d0f8d8aa..91e925a88c 100644 --- a/api/src/test/java/org/apache/iceberg/expressions/TestInclusiveMetricsEvaluator.java +++ b/api/src/test/java/org/apache/iceberg/expressions/TestInclusiveMetricsEvaluator.java @@ -171,6 +171,24 @@ public class TestInclusiveMetricsEvaluator { // upper bounds ImmutableMap.of(3, toByteBuffer(StringType.get(), "イロハニホヘト"))); + @Test + public void testRetainsFileWhenDefaultedFieldIsMissingFromMetrics() { + List fields = Lists.newArrayList(SCHEMA.columns()); + fields.add( + Types.NestedField.optional("country") + .withId(15) + .ofType(StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .build()); + Schema evolvedSchema = new Schema(fields); + + boolean mightMatch = + new InclusiveMetricsEvaluator(evolvedSchema, equal("country", "US")).eval(FILE); + + Assert.assertTrue( + "Missing field stats must not prune rows that read as the default", mightMatch); + } + @Test public void testAllNulls() { boolean shouldRead = new InclusiveMetricsEvaluator(SCHEMA, notNull("all_nulls")).eval(FILE); 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 68fef17684..5acf17140b 100644 --- a/orc/src/test/java/org/apache/iceberg/orc/TestBuildOrcProjection.java +++ b/orc/src/test/java/org/apache/iceberg/orc/TestBuildOrcProjection.java @@ -164,7 +164,7 @@ public void testRequiredNestedFieldMissingInFile() { } @Test - public void testTopLevelScalarDefaultOmittedWhenFileHasEmbeddedFieldIds() { + public void testOmitsTopLevelScalarDefaultWhenReaderSupportsDefaultsAndIdsAreEmbedded() { Schema baseSchema = new Schema(required(1, "id", Types.LongType.get())); TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema); @@ -189,7 +189,7 @@ public void testTopLevelScalarDefaultOmittedWhenFileHasEmbeddedFieldIds() { } @Test - public void testTopLevelScalarDefaultSynthesizedWhenReaderDoesNotSupportDefaults() { + public void testSynthesizesNullForTopLevelScalarDefaultWhenReaderDoesNotSupportDefaults() { Schema baseSchema = new Schema(required(1, "id", Types.LongType.get())); TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema); Schema evolvedSchema = @@ -210,7 +210,7 @@ public void testTopLevelScalarDefaultSynthesizedWhenReaderDoesNotSupportDefaults } @Test - public void testTopLevelScalarDefaultSynthesizedWhenFieldIdsAreNameMapped() { + public void testSynthesizesNullForTopLevelScalarDefaultWhenIdsAreNameMapped() { Schema baseSchema = new Schema(required(1, "id", Types.LongType.get())); TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema); @@ -233,7 +233,7 @@ public void testTopLevelScalarDefaultSynthesizedWhenFieldIdsAreNameMapped() { } @Test - public void testTopLevelRequiredScalarDefaultOmitted() { + public void testOmitsRequiredTopLevelScalarDefaultWhenReaderSupportsDefaults() { Schema baseSchema = new Schema(required(1, "id", Types.LongType.get())); TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema); @@ -258,7 +258,7 @@ public void testTopLevelRequiredScalarDefaultOmitted() { } @Test - public void testNestedScalarDefaultOmitted() { + public void testOmitsNestedScalarDefaultWhenReaderSupportsDefaults() { Schema baseSchema = new Schema( required(1, "id", Types.LongType.get()), @@ -290,7 +290,7 @@ public void testNestedScalarDefaultOmitted() { } @Test - public void testNestedStructEmptiedByOmit() { + 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 (see TestOrcDefaultValues end-to-end coverage). diff --git a/orc/src/test/java/org/apache/iceberg/orc/TestExpressionToSearchArgument.java b/orc/src/test/java/org/apache/iceberg/orc/TestExpressionToSearchArgument.java index 4af8768786..afc0dfc78a 100644 --- a/orc/src/test/java/org/apache/iceberg/orc/TestExpressionToSearchArgument.java +++ b/orc/src/test/java/org/apache/iceberg/orc/TestExpressionToSearchArgument.java @@ -49,6 +49,7 @@ import org.apache.iceberg.Schema; import org.apache.iceberg.expressions.Binder; import org.apache.iceberg.expressions.Expression; +import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.mapping.MappingUtil; import org.apache.iceberg.mapping.NameMapping; import org.apache.iceberg.types.Types; @@ -329,6 +330,32 @@ public void testEvolvedSchema() { Assert.assertEquals(expected.toString(), actual.toString()); } + @Test + public void testDisablesPushdownWhenDefaultedFieldIsOmitted() { + Schema fileSchema = new Schema(required(1, "id", Types.LongType.get())); + 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 readSchema = + ORCSchemaUtil.buildOrcProjection( + evolvedSchema, + ORCSchemaUtil.convert(fileSchema), + ORCSchemaUtil.FieldIdSource.EMBEDDED, + true); + Expression boundFilter = Binder.bind(evolvedSchema.asStruct(), equal("country", "US"), true); + SearchArgument expected = + SearchArgumentFactory.newBuilder().literal(TruthValue.YES_NO_NULL).build(); + + SearchArgument actual = ExpressionToSearchArgument.convert(boundFilter, readSchema); + + Assert.assertEquals(expected.toString(), actual.toString()); + } + @Test public void testOriginalSchemaNameMapping() { Schema originalSchema = diff --git a/orc/src/test/java/org/apache/iceberg/orc/TestOrcDefaultValues.java b/orc/src/test/java/org/apache/iceberg/orc/TestOrcDefaultValues.java index 33e016e13c..aef6104057 100644 --- a/orc/src/test/java/org/apache/iceberg/orc/TestOrcDefaultValues.java +++ b/orc/src/test/java/org/apache/iceberg/orc/TestOrcDefaultValues.java @@ -36,6 +36,7 @@ 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.GenericOrcReaders; import org.apache.iceberg.data.orc.GenericOrcWriter; import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.io.CloseableIterable; @@ -43,11 +44,13 @@ 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.Type; 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.StructColumnVector; import org.apache.orc.storage.ql.exec.vector.VectorizedRowBatch; import org.assertj.core.api.Assertions; import org.junit.Before; @@ -107,7 +110,7 @@ private OutputFile writeFile() throws IOException { } @Test - public void testReadFillsTopLevelScalarDefault() throws IOException { + public void testFillsTopLevelScalarDefaultWhenFieldIsAbsent() throws IOException { OutputFile file = writeFile(); List read; @@ -130,7 +133,22 @@ public void testReadFillsTopLevelScalarDefault() throws IOException { } @Test - public void testReadSelectsOnlyDefaultColumn() throws IOException { + public void testSynthesizesNullWhenReaderDoesNotSupportDefaults() throws IOException { + OutputFile file = writeFile(); + + try (CloseableIterable reader = + ORC.read(file.toInputFile()) + .project(READ_SCHEMA) + .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(READ_SCHEMA, fileSchema)) + .build()) { + List read = Lists.newArrayList(reader); + Assertions.assertThat(read).hasSize(records.size()); + Assertions.assertThat(read).allMatch(record -> record.getField("country") == null); + } + } + + @Test + public void testFillsDefaultWhenOnlyDefaultedFieldIsProjected() throws IOException { OutputFile file = writeFile(); Schema onlyDefault = @@ -158,7 +176,55 @@ public void testReadSelectsOnlyDefaultColumn() throws IOException { } @Test - public void testReadFillsScalarDefaultsAllTypes() throws IOException { + public void testFillsMiddleDefaultWithoutShiftingTrailingPhysicalField() throws IOException { + Schema writeSchema = + new Schema( + required(1, "id", Types.LongType.get()), optional(3, "data", Types.StringType.get())); + Schema readSchema = + new Schema( + required(1, "id", Types.LongType.get()), + defaulted(2, "country", Types.StringType.get(), Expressions.lit("US")), + optional(3, "data", Types.StringType.get())); + Record record = GenericRecord.create(writeSchema); + record.setField("id", 1L); + record.setField("data", "trailing"); + OutputFile file = writeRecords(writeSchema, Lists.newArrayList(record)); + + Record readRecord = read(file, readSchema).get(0); + + Assertions.assertThat(readRecord.getField("country")).isEqualTo("US"); + Assertions.assertThat(readRecord.getField("data")).isEqualTo("trailing"); + } + + @Test + public void testPositionalReaderRetainsNullForMiddleDefaultWithoutShiftingTrailingField() + throws IOException { + Schema writeSchema = + new Schema( + required(1, "id", Types.LongType.get()), optional(3, "data", Types.StringType.get())); + Schema readSchema = + new Schema( + required(1, "id", Types.LongType.get()), + defaulted(2, "country", Types.StringType.get(), Expressions.lit("US")), + optional(3, "data", Types.StringType.get())); + Record record = GenericRecord.create(writeSchema); + record.setField("id", 1L); + record.setField("data", "trailing"); + OutputFile file = writeRecords(writeSchema, Lists.newArrayList(record)); + + try (CloseableIterable reader = + ORC.read(file.toInputFile()) + .project(readSchema) + .createReaderFunc(readOrcSchema -> new PositionalOrcReader(readSchema, readOrcSchema)) + .build()) { + Object[] readRecord = Lists.newArrayList(reader).get(0); + Assertions.assertThat(readRecord[1]).isNull(); + Assertions.assertThat(readRecord[2]).isEqualTo("trailing"); + } + } + + @Test + public void testFillsDefaultsForBooleanNumericStringAndDecimalTypes() throws IOException { OutputFile file = writeFile(); Schema typed = @@ -196,7 +262,7 @@ public void testReadFillsScalarDefaultsAllTypes() throws IOException { } @Test - public void testReadFillsRequiredScalarDefault() throws IOException { + public void testFillsRequiredDefaultWhenFieldIsAbsent() throws IOException { OutputFile file = writeFile(); // A required field absent from the file but declaring a default must be filled, not rejected. @@ -227,7 +293,7 @@ public void testReadFillsRequiredScalarDefault() throws IOException { } @Test - public void testReadDoesNotApplyDefaultToIdLessFile() throws IOException { + public void testSynthesizesNullWhenFileHasNoEmbeddedIds() throws IOException { File file = writeIdLessFile(); List read; @@ -250,7 +316,7 @@ public void testReadDoesNotApplyDefaultToIdLessFile() throws IOException { } @Test - public void testReadDoesNotOverridePresentColumn() throws IOException { + public void testPreservesPhysicalValueAndNullWhenFieldIsPresent() throws IOException { Schema writeSchema = new Schema( required(1, "id", Types.LongType.get()), @@ -274,7 +340,7 @@ public void testReadDoesNotOverridePresentColumn() throws IOException { } @Test - public void testNestedStructScalarDefault() throws IOException { + public void testFillsNestedStructDefaultWhenFieldIsAbsent() throws IOException { Schema writeSchema = new Schema( required(1, "id", Types.LongType.get()), @@ -321,7 +387,7 @@ public void testNestedStructScalarDefault() throws IOException { } @Test - public void testMapNestedScalarDefault() throws IOException { + public void testFillsMapValueStructDefaultWhenFieldIsAbsent() throws IOException { Schema writeSchema = new Schema( required(1, "id", Types.LongType.get()), @@ -367,7 +433,7 @@ public void testMapNestedScalarDefault() throws IOException { } @Test - public void testListNestedScalarDefault() throws IOException { + public void testFillsListElementStructDefaultWhenFieldIsAbsent() throws IOException { Schema writeSchema = new Schema( required(1, "id", Types.LongType.get()), @@ -408,7 +474,7 @@ public void testListNestedScalarDefault() throws IOException { } @Test - public void testNestedStructAllSubfieldsDefaulted() throws IOException { + public void testFillsDefaultsWhenAllProjectedStructFieldsAreAbsent() 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 = @@ -490,6 +556,71 @@ private List read(OutputFile file, Schema readSchema) throws IOException } } + private static class PositionalOrcReader implements OrcRowReader { + private final OrcValueReader reader; + + private PositionalOrcReader(Schema expectedSchema, TypeDescription readSchema) { + this.reader = + OrcSchemaWithTypeVisitor.visit( + expectedSchema, + readSchema, + new OrcSchemaWithTypeVisitor>() { + @Override + public OrcValueReader record( + Types.StructType expected, + TypeDescription record, + List names, + List> fields) { + return new PositionalStructReader(fields, expected); + } + + @Override + public OrcValueReader primitive( + Type.PrimitiveType expected, TypeDescription primitive) { + switch (expected.typeId()) { + case LONG: + return OrcValueReaders.longs(); + case STRING: + return GenericOrcReaders.strings(); + default: + throw new IllegalArgumentException("Unsupported test type: " + expected); + } + } + }); + } + + @Override + public Object[] read(VectorizedRowBatch batch, int row) { + return (Object[]) reader.read(new StructColumnVector(batch.size, batch.cols), row); + } + + @Override + public void setBatchContext(long batchOffsetInFile) { + reader.setBatchContext(batchOffsetInFile); + } + } + + @SuppressWarnings("deprecation") + private static class PositionalStructReader extends OrcValueReaders.StructReader { + private final int size; + + private PositionalStructReader( + List> readers, Types.StructType expectedStruct) { + super(readers, expectedStruct, Collections.emptyMap()); + this.size = expectedStruct.fields().size(); + } + + @Override + protected Object[] create() { + return new Object[size]; + } + + @Override + protected void set(Object[] struct, int pos, Object value) { + struct[pos] = value; + } + } + private static Types.NestedField defaulted( int id, String name, From 37c3c78cc46ef152c7464eb093dedd10a488ed3b Mon Sep 17 00:00:00 2001 From: Christian Bush Date: Mon, 3 Aug 2026 00:24:29 -0700 Subject: [PATCH 5/5] Spark 3.1: route projected defaults to iterative ORC reads --- .../iceberg/spark/source/SparkBatchScan.java | 11 ++- .../TestSparkBatchScanInitialDefaults.java | 69 +++++++++++++++++++ 2 files changed, 79 insertions(+), 1 deletion(-) create mode 100644 spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkBatchScanInitialDefaults.java diff --git a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchScan.java b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchScan.java index 63489d5056..0fc1bba17d 100644 --- a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchScan.java +++ b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkBatchScan.java @@ -38,6 +38,7 @@ import org.apache.iceberg.spark.SparkReadConf; import org.apache.iceberg.spark.SparkSchemaUtil; import org.apache.iceberg.spark.SparkUtil; +import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.util.PropertyUtil; import org.apache.iceberg.util.TableScanUtil; import org.apache.iceberg.util.Tasks; @@ -193,7 +194,10 @@ public PartitionReaderFactory createReaderFactory() { boolean batchReadsEnabled = batchReadsEnabled(allParquetFileScanTasks, allOrcFileScanTasks); - boolean batchReadOrc = hasNoDeleteFiles && allOrcFileScanTasks; + boolean hasNoInitialDefaults = hasNoInitialDefaults(expectedSchema); + + // Initial defaults are currently supported by the iterative ORC reader only. + boolean batchReadOrc = hasNoDeleteFiles && allOrcFileScanTasks && hasNoInitialDefaults; boolean batchReadParquet = hasNoEqDeleteFiles && allParquetFileScanTasks && atLeastOneColumn && onlyPrimitives; @@ -205,6 +209,11 @@ public PartitionReaderFactory createReaderFactory() { return new ReaderFactory(batchSize); } + static boolean hasNoInitialDefaults(Schema schema) { + return TypeUtil.indexById(schema.asStruct()).values().stream() + .noneMatch(field -> field.initialDefault() != null); + } + private boolean batchReadsEnabled(boolean isParquetOnly, boolean isOrcOnly) { if (isParquetOnly) { return readConf.parquetVectorizationEnabled(); diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkBatchScanInitialDefaults.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkBatchScanInitialDefaults.java new file mode 100644 index 0000000000..7d0fc98074 --- /dev/null +++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkBatchScanInitialDefaults.java @@ -0,0 +1,69 @@ +/* + * 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.source; + +import org.apache.iceberg.Schema; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.types.Types; +import org.junit.Assert; +import org.junit.Test; + +public class TestSparkBatchScanInitialDefaults { + + @Test + public void testDisablesVectorizationWhenTopLevelDefaultIsProjected() { + Schema schema = + new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional("country") + .withId(2) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .build()); + + Assert.assertFalse(SparkBatchScan.hasNoInitialDefaults(schema)); + } + + @Test + public void testDisablesVectorizationWhenNestedDefaultIsProjected() { + Schema schema = + new Schema( + Types.NestedField.required( + 1, + "location", + Types.StructType.of( + Types.NestedField.optional("country") + .withId(2) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .build()))); + + Assert.assertFalse(SparkBatchScan.hasNoInitialDefaults(schema)); + } + + @Test + public void testAllowsVectorizationWhenNoDefaultIsProjected() { + Schema schema = + new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional(2, "country", Types.StringType.get())); + + Assert.assertTrue(SparkBatchScan.hasNoInitialDefaults(schema)); + } +}