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/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/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/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 fae1a76c37..3bac5599e5 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,18 +287,73 @@ public static Schema convert(TypeDescription orcSchema) { */ 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, false); + } + + /** + * 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, {@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, + boolean supportsInitialDefaults) { final Map icebergToOrc = icebergToOrcMapping("root", originalOrcSchema); - return buildOrcProjection(Integer.MIN_VALUE, schema.asStruct(), true, icebergToOrc); + return buildOrcProjection( + Integer.MIN_VALUE, + schema.asStruct(), + true, + fieldIdSource, + supportsInitialDefaults, + icebergToOrc); + } + + private static boolean isOmittableDefault( + Types.NestedField field, + FieldIdSource fieldIdSource, + boolean supportsInitialDefaults, + Map mapping) { + // Only scalars reach here with a non-null default: Types.NestedField#castDefault rejects a + // default on any nested type at construction time. + return supportsInitialDefaults + && field.initialDefault() != null + && !mapping.containsKey(field.fieldId()) + && fieldIdSource == FieldIdSource.EMBEDDED; } private static TypeDescription buildOrcProjection( - Integer fieldId, Type type, boolean isRequired, Map mapping) { + Integer fieldId, + Type type, + boolean isRequired, + FieldIdSource fieldIdSource, + boolean supportsInitialDefaults, + Map mapping) { final TypeDescription orcType; switch (type.typeId()) { case STRUCT: orcType = TypeDescription.createStruct(); for (Types.NestedField nestedField : type.asStructType().fields()) { + if (isOmittableDefault(nestedField, fieldIdSource, supportsInitialDefaults, mapping)) { + // The field declares a default, is absent, and the file carries its own Iceberg field + // IDs. Omit it so a default-aware reader fills it through the existing constant path. + continue; + } // Using suffix _r to avoid potential underlying issues in ORC reader // with reused column names between ORC and Iceberg; // e.g. renaming column c -> d and adding new column d @@ -285,6 +366,8 @@ private static TypeDescription buildOrcProjection( nestedField.fieldId(), nestedField.type(), isRequired && nestedField.isRequired(), + fieldIdSource, + supportsInitialDefaults, mapping); orcType.addField(name, childType); } @@ -296,16 +379,29 @@ private static TypeDescription buildOrcProjection( list.elementId(), 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, mapping); + buildOrcProjection( + map.keyId(), + map.keyType(), + isRequired, + fieldIdSource, + supportsInitialDefaults, + mapping); TypeDescription valueType = buildOrcProjection( - map.valueId(), map.valueType(), isRequired && map.isValueRequired(), mapping); + map.valueId(), + map.valueType(), + isRequired && map.isValueRequired(), + fieldIdSource, + supportsInitialDefaults, + 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..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; @@ -86,13 +89,20 @@ public CloseableIterator iterator() { TypeDescription fileSchema = orcFileReader.getSchema(); final TypeDescription readOrcSchema; if (ORCSchemaUtil.hasIds(fileSchema)) { - readOrcSchema = ORCSchemaUtil.buildOrcProjection(schema, fileSchema); + readOrcSchema = + ORCSchemaUtil.buildOrcProjection( + schema, fileSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED, supportsInitialDefaults); } 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, + 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 e13689eb20..b11c7dfc53 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, 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 + */ + 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..5acf17140b 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,161 @@ public void testRequiredNestedFieldMissingInFile() { .isInstanceOf(IllegalArgumentException.class) .hasMessage("Field 4 of type long is required and was not found."); } + + @Test + public void testOmitsTopLevelScalarDefaultWhenReaderSupportsDefaultsAndIdsAreEmbedded() { + Schema baseSchema = new Schema(required(1, "id", Types.LongType.get())); + TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema); + + Schema evolvedSchema = + new Schema( + required(1, "id", Types.LongType.get()), + Types.NestedField.optional("country") + .withId(2) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .build()); + + // The file carries embedded field IDs, so the absent field can be identified safely. + TypeDescription projection = + ORCSchemaUtil.buildOrcProjection( + evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED, true); + assertEquals(1, projection.getChildren().size()); + assertNotNull(projection.findSubtype("id")); + assertFalse( + "defaulted column must be omitted from the read projection", + projection.getFieldNames().contains("country_r2")); + } + + @Test + public void testSynthesizesNullForTopLevelScalarDefaultWhenReaderDoesNotSupportDefaults() { + Schema baseSchema = new Schema(required(1, "id", Types.LongType.get())); + TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema); + Schema evolvedSchema = + new Schema( + required(1, "id", Types.LongType.get()), + Types.NestedField.optional("country") + .withId(2) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .build()); + + TypeDescription projection = + ORCSchemaUtil.buildOrcProjection( + evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED, false); + + assertEquals(2, projection.getChildren().size()); + assertNotNull(projection.findSubtype("country_r2")); + } + + @Test + public void testSynthesizesNullForTopLevelScalarDefaultWhenIdsAreNameMapped() { + Schema baseSchema = new Schema(required(1, "id", Types.LongType.get())); + TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema); + + Schema evolvedSchema = + new Schema( + required(1, "id", Types.LongType.get()), + Types.NestedField.optional("country") + .withId(2) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .build()); + + // 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()); + assertEquals( + TypeDescription.Category.STRING, projection.findSubtype("country_r2").getCategory()); + } + + @Test + public void testOmitsRequiredTopLevelScalarDefaultWhenReaderSupportsDefaults() { + Schema baseSchema = new Schema(required(1, "id", Types.LongType.get())); + TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema); + + // A required top-level field that is absent from the file but declares a default must be + // omitted (then filled), not rejected by the required-missing check. + Schema evolvedSchema = + new Schema( + required(1, "id", Types.LongType.get()), + Types.NestedField.required("code") + .withId(2) + .ofType(Types.IntegerType.get()) + .withInitialDefault(Expressions.lit(7)) + .build()); + + TypeDescription projection = + ORCSchemaUtil.buildOrcProjection( + evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED, true); + assertEquals(1, projection.getChildren().size()); + assertFalse( + "required defaulted column must be omitted, not throw", + projection.getFieldNames().contains("code_r2")); + } + + @Test + public void testOmitsNestedScalarDefaultWhenReaderSupportsDefaults() { + Schema baseSchema = + new Schema( + required(1, "id", Types.LongType.get()), + required(2, "s", Types.StructType.of(required(3, "a", Types.LongType.get())))); + TypeDescription baseOrcSchema = ORCSchemaUtil.convert(baseSchema); + + // A scalar default on a field nested inside a struct is omitted (then filled via idToConstant), + // at any nesting level. The present sibling "a" keeps the struct non-empty. + Schema evolvedSchema = + new Schema( + required(1, "id", Types.LongType.get()), + required( + 2, + "s", + Types.StructType.of( + required(3, "a", Types.LongType.get()), + Types.NestedField.optional("b") + .withId(4) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("x")) + .build()))); + + TypeDescription projection = + ORCSchemaUtil.buildOrcProjection( + evolvedSchema, baseOrcSchema, ORCSchemaUtil.FieldIdSource.EMBEDDED, true); + TypeDescription nested = projection.findSubtype("s"); + assertEquals(1, nested.getChildren().size()); + assertFalse("nested defaulted column must be omitted", nested.getFieldNames().contains("b_r4")); + } + + @Test + public void testPreservesNestedStructWhenAllProjectedFieldsAreOmitted() { + // Base file: s { a }. Project only a new defaulted subfield s { b default 'x' } (drop a). Every + // projected subfield of s is absent + defaulted, so the nested read struct is omitted down to + // empty; the reader fills b via idToConstant (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, 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/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 new file mode 100644 index 0000000000..aef6104057 --- /dev/null +++ b/orc/src/test/java/org/apache/iceberg/orc/TestOrcDefaultValues.java @@ -0,0 +1,635 @@ +/* + * 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.GenericOrcReaders; +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.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; +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 testFillsTopLevelScalarDefaultWhenFieldIsAbsent() throws IOException { + OutputFile file = writeFile(); + + List read; + try (CloseableIterable reader = + ORC.read(file.toInputFile()) + .project(READ_SCHEMA) + .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(READ_SCHEMA, fileSchema)) + .supportsInitialDefaults() + .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 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 = + 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)) + .supportsInitialDefaults() + .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 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 = + 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)) + .supportsInitialDefaults() + .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 testFillsRequiredDefaultWhenFieldIsAbsent() 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)) + .supportsInitialDefaults() + .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 testSynthesizesNullWhenFileHasNoEmbeddedIds() 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)) + .supportsInitialDefaults() + .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 testPreservesPhysicalValueAndNullWhenFieldIsPresent() 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 testFillsNestedStructDefaultWhenFieldIsAbsent() 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 testFillsMapValueStructDefaultWhenFieldIsAbsent() 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 testFillsListElementStructDefaultWhenFieldIsAbsent() 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 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 = + 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)) + .supportsInitialDefaults() + .build()) { + return Lists.newArrayList(reader); + } + } + + 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, + 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/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)); + } +}