diff --git a/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReader.java b/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReader.java index f4d816baf2..cba6617b83 100644 --- a/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReader.java +++ b/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReader.java @@ -22,7 +22,9 @@ import java.util.List; import java.util.Map; import org.apache.iceberg.Schema; +import org.apache.iceberg.data.IdentityPartitionConverters; import org.apache.iceberg.data.Record; +import org.apache.iceberg.orc.ORCSchemaUtil; import org.apache.iceberg.orc.OrcRowReader; import org.apache.iceberg.orc.OrcSchemaWithTypeVisitor; import org.apache.iceberg.orc.OrcValueReader; @@ -76,7 +78,11 @@ public OrcValueReader record( TypeDescription record, List names, List> fields) { - return GenericOrcReaders.struct(fields, expected, idToConstant); + return GenericOrcReaders.struct( + fields, + expected, + ORCSchemaUtil.idToConstantWithDefaults( + expected, record, idToConstant, IdentityPartitionConverters::convertConstant)); } @Override 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..05c1c02205 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java +++ b/orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java @@ -22,12 +22,15 @@ import java.util.Map; import java.util.Objects; import java.util.Optional; +import java.util.Set; +import java.util.function.BiFunction; import java.util.stream.Collectors; import org.apache.iceberg.Schema; import org.apache.iceberg.mapping.NameMapping; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMultimap; import org.apache.iceberg.relocated.com.google.common.collect.Maps; +import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.TypeUtil; import org.apache.iceberg.types.Types; @@ -261,18 +264,47 @@ public static Schema convert(TypeDescription orcSchema) { */ public static TypeDescription buildOrcProjection( Schema schema, TypeDescription originalOrcSchema) { + return buildOrcProjection(schema, originalOrcSchema, false); + } + + /** + * Builds the ORC read schema for a file whose embedded field IDs are known to be trustworthy. + * + *

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. A default-aware reader then fills it as a per-file constant through its existing + * {@code idToConstant} path (see {@link #idToConstantWithDefaults}). When false, the field is + * synthesized as a null column so a default is never name-matched onto an id-less file. + */ + static TypeDescription buildOrcProjection( + Schema schema, TypeDescription originalOrcSchema, boolean hasTrustedIds) { final Map 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 +317,7 @@ private static TypeDescription buildOrcProjection( nestedField.fieldId(), nestedField.type(), isRequired && nestedField.isRequired(), + hasTrustedIds, mapping); orcType.addField(name, childType); } @@ -296,16 +329,21 @@ private static TypeDescription buildOrcProjection( list.elementId(), list.elementType(), isRequired && list.isElementRequired(), + hasTrustedIds, mapping); orcType = TypeDescription.createList(elementType); break; case MAP: Types.MapType map = (Types.MapType) type; TypeDescription keyType = - buildOrcProjection(map.keyId(), map.keyType(), isRequired, mapping); + buildOrcProjection(map.keyId(), map.keyType(), isRequired, hasTrustedIds, mapping); TypeDescription valueType = buildOrcProjection( - map.valueId(), map.valueType(), isRequired && map.isValueRequired(), mapping); + map.valueId(), + map.valueType(), + isRequired && map.isValueRequired(), + hasTrustedIds, + mapping); orcType = TypeDescription.createMap(keyType, valueType); break; default: @@ -432,6 +470,51 @@ static boolean hasIds(TypeDescription orcSchema) { return OrcSchemaVisitor.visit(orcSchema, new HasIds()); } + /** + * Augments an {@code idToConstant} map with column initial-defaults for expected fields that are + * absent from the ORC read projection. + * + *

Used by the ORC readers' {@code record} visitor step: a field that {@link + * #buildOrcProjection(Schema, TypeDescription, boolean)} omitted (a scalar at any nesting level + * declaring an {@code initial-default} that is absent from a file with trustworthy IDs) is not + * present in {@code record}, so its converted default is injected as a constant. The reader then + * fills it for every row via its existing partition-constant path, consuming no column vector. + * Fields synthesized for id-less/name-mapped reads are present in {@code record} and therefore + * read NULL. + * + * @param expected the expected Iceberg struct for this record level + * @param record the ORC read projection for this record level + * @param idToConstant existing constants (e.g. partition values); not modified + * @param convertConstant converts an internal default value to the engine's in-memory form + * @return {@code idToConstant} unchanged when no defaults apply, otherwise a new merged map + */ + public static Map idToConstantWithDefaults( + Types.StructType expected, + TypeDescription record, + Map idToConstant, + BiFunction convertConstant) { + Set presentIds = Sets.newHashSet(); + for (TypeDescription child : record.getChildren()) { + presentIds.add(fieldId(child)); + } + + Map withDefaults = null; + for (Types.NestedField field : expected.fields()) { + if (field.initialDefault() != null + && !presentIds.contains(field.fieldId()) + && !idToConstant.containsKey(field.fieldId())) { + if (withDefaults == null) { + withDefaults = Maps.newHashMap(); + withDefaults.putAll(idToConstant); + } + withDefaults.put( + field.fieldId(), convertConstant.apply(field.type(), field.initialDefault())); + } + } + + return withDefaults == null ? idToConstant : withDefaults; + } + static TypeDescription applyNameMapping(TypeDescription orcSchema, NameMapping nameMapping) { return OrcSchemaVisitor.visit(orcSchema, new ApplyNameMapping(nameMapping)); } 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/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..b1cd6fb3b0 --- /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 m = (Map) read.get(0).getField("m"); + Assertions.assertThat(m).hasSize(1); + Record readValue = (Record) m.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 l = (List) read.get(0).getField("l"); + Assertions.assertThat(l).hasSize(1); + Record readElement = (Record) l.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(); + } +} diff --git a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcReader.java b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcReader.java index 78db137054..cf6bd2025b 100644 --- a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcReader.java +++ b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcReader.java @@ -20,11 +20,13 @@ import java.util.List; import java.util.Map; +import org.apache.iceberg.orc.ORCSchemaUtil; import org.apache.iceberg.orc.OrcRowReader; import org.apache.iceberg.orc.OrcSchemaWithTypeVisitor; import org.apache.iceberg.orc.OrcValueReader; import org.apache.iceberg.orc.OrcValueReaders; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.spark.source.BaseDataReader; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; import org.apache.orc.TypeDescription; @@ -77,7 +79,11 @@ public OrcValueReader record( TypeDescription record, List names, List> fields) { - return SparkOrcValueReaders.struct(fields, expected, idToConstant); + return SparkOrcValueReaders.struct( + fields, + expected, + ORCSchemaUtil.idToConstantWithDefaults( + expected, record, idToConstant, BaseDataReader::convertConstant)); } @Override diff --git a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/BaseDataReader.java b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/BaseDataReader.java index f3ddd50eef..8d35fbca71 100644 --- a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/BaseDataReader.java +++ b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/BaseDataReader.java @@ -59,7 +59,7 @@ * * @param is the Java class returned by this reader whose objects contain one or more rows. */ -abstract class BaseDataReader implements Closeable { +public abstract class BaseDataReader implements Closeable { private static final Logger LOG = LoggerFactory.getLogger(BaseDataReader.class); private final Table table; @@ -160,7 +160,7 @@ protected InputFile getInputFile(String location) { } } - protected static Object convertConstant(Type type, Object value) { + public static Object convertConstant(Type type, Object value) { if (value == null) { return null; } diff --git a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java index 68e98ba913..a598e81c5a 100644 --- a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java +++ b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/source/BatchDataReader.java @@ -114,6 +114,8 @@ CloseableIterator open(FileScanTask task) { Sets.union(constantFieldIds, metadataFieldIds); Schema schemaWithoutConstantAndMetadataFields = TypeUtil.selectNot(expectedSchema, constantAndMetadataFieldIds); + // Follow-up: wire initial-default constants into VectorizedSparkOrcReaders. Tables whose + // in-memory schema declares defaults must remain on the row reader until this path is wired. ORC.ReadBuilder builder = ORC.read(location) .project(schemaWithoutConstantAndMetadataFields) diff --git a/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderDefaults.java b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderDefaults.java new file mode 100644 index 0000000000..69187493fe --- /dev/null +++ b/spark/v3.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderDefaults.java @@ -0,0 +1,170 @@ +/* + * 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.data; + +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.util.List; +import org.apache.iceberg.Files; +import org.apache.iceberg.Schema; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.io.CloseableIterable; +import org.apache.iceberg.io.FileAppender; +import org.apache.iceberg.orc.ORC; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.types.Types; +import org.apache.spark.sql.catalyst.InternalRow; +import org.apache.spark.sql.catalyst.expressions.GenericInternalRow; +import org.apache.spark.unsafe.types.UTF8String; +import org.assertj.core.api.Assertions; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +/** + * Verifies that the row {@code SparkOrcReader} fills a scalar {@code initial-default} only when the + * field declares one and is absent from a file with embedded field IDs. Vectorized ORC defaults are + * a follow-up. + */ +public class TestSparkOrcReaderDefaults { + + private static final Schema WRITE_SCHEMA = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "data", Types.StringType.get())); + + 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 static final UTF8String EXPECTED_DEFAULT = UTF8String.fromString("US"); + + @Rule public TemporaryFolder temp = new TemporaryFolder(); + + private File writeFile() throws IOException { + List rows = + Lists.newArrayList( + new GenericInternalRow(new Object[] {1L, UTF8String.fromString("a")}), + new GenericInternalRow(new Object[] {2L, UTF8String.fromString("b")}), + new GenericInternalRow(new Object[] {3L, UTF8String.fromString("c")})); + return writeFile(WRITE_SCHEMA, rows); + } + + private File writeFile(Schema schema, List rows) throws IOException { + File testFile = temp.newFile(); + Assertions.assertThat(testFile.delete()).isTrue(); + try (FileAppender writer = + ORC.write(Files.localOutput(testFile)) + .createWriterFunc(SparkOrcWriter::new) + .schema(schema) + .build()) { + writer.addAll(rows); + } + return testFile; + } + + @Test + public void testRowReadFillsDeclaredDefault() throws IOException { + File testFile = writeFile(); + + try (CloseableIterable reader = + ORC.read(Files.localInput(testFile)) + .project(READ_SCHEMA) + .createReaderFunc(readOrcSchema -> new SparkOrcReader(READ_SCHEMA, readOrcSchema)) + .build()) { + int count = 0; + for (InternalRow row : reader) { + Assertions.assertThat(row.getUTF8String(2)).isEqualTo(EXPECTED_DEFAULT); + count += 1; + } + Assertions.assertThat(count).isEqualTo(3); + } + } + + @Test + public void testRowReadReadsNullWhenFieldHasNoDefault() throws IOException { + File testFile = writeFile(); + Schema schemaWithoutDefault = + new Schema( + required(1, "id", Types.LongType.get()), + optional(2, "data", Types.StringType.get()), + optional(3, "country", Types.StringType.get())); + + try (CloseableIterable reader = + ORC.read(Files.localInput(testFile)) + .project(schemaWithoutDefault) + .createReaderFunc( + readOrcSchema -> new SparkOrcReader(schemaWithoutDefault, readOrcSchema)) + .build()) { + int count = 0; + for (InternalRow row : reader) { + Assertions.assertThat(row.isNullAt(2)).isTrue(); + count += 1; + } + Assertions.assertThat(count).isEqualTo(3); + } + } + + @Test + public void testRowReadFillsNestedDeclaredDefault() throws IOException { + Schema writeSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, "nested", Types.StructType.of(required(3, "value", Types.StringType.get())))); + Schema readSchema = + new Schema( + required(1, "id", Types.LongType.get()), + optional( + 2, + "nested", + Types.StructType.of( + required(3, "value", Types.StringType.get()), + Types.NestedField.optional("missing") + .withId(4) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("filled")) + .build()))); + InternalRow nested = new GenericInternalRow(new Object[] {UTF8String.fromString("present")}); + File testFile = + writeFile( + writeSchema, Lists.newArrayList(new GenericInternalRow(new Object[] {1L, nested}))); + + try (CloseableIterable reader = + ORC.read(Files.localInput(testFile)) + .project(readSchema) + .createReaderFunc(readOrcSchema -> new SparkOrcReader(readSchema, readOrcSchema)) + .build()) { + List rows = Lists.newArrayList(reader); + Assertions.assertThat(rows).hasSize(1); + InternalRow readNested = rows.get(0).getStruct(1, 2); + Assertions.assertThat(readNested.getUTF8String(0)) + .isEqualTo(UTF8String.fromString("present")); + Assertions.assertThat(readNested.getUTF8String(1)).isEqualTo(UTF8String.fromString("filled")); + } + } +}