diff --git a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueReaders.java b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueReaders.java index fabb0868f8..06bc28ade9 100644 --- a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueReaders.java +++ b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueReaders.java @@ -25,6 +25,7 @@ import org.apache.iceberg.orc.OrcValueReaders; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.spark.source.BaseDataReader; import org.apache.iceberg.types.Types; import org.apache.orc.TypeDescription; import org.apache.orc.storage.ql.exec.vector.BytesColumnVector; @@ -168,7 +169,7 @@ protected StructReader( List> readers, Types.StructType struct, Map idToConstant) { - super(orcType, readers, struct, idToConstant); + super(orcType, readers, struct, idToConstant, BaseDataReader::convertConstant); this.numFields = struct.fields().size(); } 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/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")); + } + } +}