diff --git a/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java b/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java index e58f76a8de..06d2428b12 100644 --- a/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java +++ b/api/src/main/java/org/apache/iceberg/types/AssignFreshIds.java @@ -88,11 +88,7 @@ public Type struct(Types.StructType struct, Iterable futures) { for (int i = 0; i < length; i += 1) { Types.NestedField field = fields.get(i); Type type = types.next(); - if (field.isOptional()) { - newFields.add(Types.NestedField.optional(newIds.get(i), field.name(), type, field.doc())); - } else { - newFields.add(Types.NestedField.required(newIds.get(i), field.name(), type, field.doc())); - } + newFields.add(Types.NestedField.from(field).withId(newIds.get(i)).ofType(type).build()); } return Types.StructType.of(newFields); diff --git a/api/src/main/java/org/apache/iceberg/types/PruneColumns.java b/api/src/main/java/org/apache/iceberg/types/PruneColumns.java index daf2e6bbc0..5554f550d6 100644 --- a/api/src/main/java/org/apache/iceberg/types/PruneColumns.java +++ b/api/src/main/java/org/apache/iceberg/types/PruneColumns.java @@ -65,15 +65,7 @@ public Type struct(Types.StructType struct, List fieldResults) { selectedFields.add(field); } else if (projectedType != null) { sameTypes = false; // signal that some types were altered - if (field.isOptional()) { - selectedFields.add( - Types.NestedField.optional( - field.fieldId(), field.name(), projectedType, field.doc())); - } else { - selectedFields.add( - Types.NestedField.required( - field.fieldId(), field.name(), projectedType, field.doc())); - } + selectedFields.add(Types.NestedField.from(field).ofType(projectedType).build()); } } diff --git a/api/src/main/java/org/apache/iceberg/types/ReassignIds.java b/api/src/main/java/org/apache/iceberg/types/ReassignIds.java index d31fea98e5..565ceee2a9 100644 --- a/api/src/main/java/org/apache/iceberg/types/ReassignIds.java +++ b/api/src/main/java/org/apache/iceberg/types/ReassignIds.java @@ -79,11 +79,7 @@ public Type struct(Types.StructType struct, Iterable fieldTypes) { for (int i = 0; i < length; i += 1) { Types.NestedField field = fields.get(i); int fieldId = id(sourceStruct, field.name()); - if (field.isRequired()) { - newFields.add(Types.NestedField.required(fieldId, field.name(), types.get(i), field.doc())); - } else { - newFields.add(Types.NestedField.optional(fieldId, field.name(), types.get(i), field.doc())); - } + newFields.add(Types.NestedField.from(field).withId(fieldId).ofType(types.get(i)).build()); } return Types.StructType.of(newFields); diff --git a/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java b/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java index 81f32f8ee9..6d3db868b9 100644 --- a/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java +++ b/api/src/test/java/org/apache/iceberg/types/TestTypeUtil.java @@ -25,6 +25,7 @@ import java.util.Set; import org.apache.iceberg.Schema; +import org.apache.iceberg.expressions.Expressions; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Sets; import org.apache.iceberg.types.Types.IntegerType; @@ -65,6 +66,24 @@ public void testReassignIdsWithIdentifier() { .isEqualTo(sourceSchema.identifierFieldIds()); } + @Test + public void testAssignFreshIdsPreservesDefaults() { + Types.NestedField field = + Types.NestedField.optional("country") + .withId(10) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .withWriteDefault(Expressions.lit("CA")) + .build(); + + Schema reassigned = TypeUtil.assignIncreasingFreshIds(new Schema(field)); + Types.NestedField reassignedField = reassigned.findField("country"); + + assertThat(reassignedField.fieldId()).isEqualTo(1); + assertThat(reassignedField.initialDefault()).isEqualTo("US"); + assertThat(reassignedField.writeDefault()).isEqualTo("CA"); + } + @Test public void testAssignIncreasingFreshIdWithIdentifier() { Schema schema = 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 609c8aedb5..5b7ed6cbbc 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/ExpressionToSearchArgument.java +++ b/orc/src/main/java/org/apache/iceberg/orc/ExpressionToSearchArgument.java @@ -31,6 +31,7 @@ import org.apache.iceberg.expressions.Expression; import org.apache.iceberg.expressions.ExpressionVisitors; import org.apache.iceberg.expressions.Literal; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Type.TypeID; @@ -46,8 +47,7 @@ class ExpressionToSearchArgument extends ExpressionVisitors.BoundVisitor { static SearchArgument convert(Expression expr, TypeDescription readSchema) { - Map idToColumnName = - ORCSchemaUtil.idToOrcName(ORCSchemaUtil.convert(readSchema)); + Map idToColumnName = columnNamesForPushdown(readSchema); SearchArgument.Builder builder = SearchArgumentFactory.newBuilder(); ExpressionVisitors.visit(expr, new ExpressionToSearchArgument(builder, idToColumnName)) .invoke(); @@ -69,6 +69,17 @@ private ExpressionToSearchArgument( this.idToColumnName = idToColumnName; } + // convert() requires at least one Iceberg field. Omitting defaulted children can leave nested + // empty structs (struct>) that fail convert even when the root is non-empty. + // Treat that as no bindable columns so predicates become YES_NO_NULL and defaults can be filled. + private static Map columnNamesForPushdown(TypeDescription readSchema) { + try { + return ORCSchemaUtil.idToOrcName(ORCSchemaUtil.convert(readSchema)); + } catch (IllegalArgumentException e) { + return ImmutableMap.of(); + } + } + @Override public Action alwaysTrue() { return () -> this.builder.literal(TruthValue.YES); @@ -271,11 +282,13 @@ 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()) || !(pred.term() instanceof BoundReference)) { // 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 + // (including fields omitted from the read projection so defaults can be filled). return () -> this.builder.literal(TruthValue.YES_NO_NULL); } else { return super.predicate(pred); 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 c7c7a8aeb0..26ed92a576 100644 --- a/orc/src/test/java/org/apache/iceberg/orc/TestExpressionToSearchArgument.java +++ b/orc/src/test/java/org/apache/iceberg/orc/TestExpressionToSearchArgument.java @@ -50,6 +50,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; @@ -492,4 +493,127 @@ public void testExpressionContainsNonReferenceTerm() { ExpressionToSearchArgument.convert(boundFilter, ORCSchemaUtil.convert(schema)); Assertions.assertThat(actual.toString()).isEqualTo(expected.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); + + Assertions.assertThat(actual.toString()).isEqualTo(expected.toString()); + } + + @Test + public void testDisablesPushdownWhenReadProjectionIsEmpty() { + Schema fileSchema = new Schema(required(1, "id", Types.LongType.get())); + Schema evolvedSchema = + new Schema( + 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); + Assertions.assertThat(readSchema.getChildren()).hasSize(0); + + Expression boundFilter = Binder.bind(evolvedSchema.asStruct(), equal("country", "bar"), true); + SearchArgument expected = + SearchArgumentFactory.newBuilder().literal(TruthValue.YES_NO_NULL).build(); + + SearchArgument actual = ExpressionToSearchArgument.convert(boundFilter, readSchema); + + Assertions.assertThat(actual.toString()).isEqualTo(expected.toString()); + } + + @Test + public void testDisablesPushdownWhenNestedReadProjectionIsEmpty() { + // File has loc as an empty struct. Projecting only loc.country (a new initial-default) omits + // that child and leaves struct>, which is non-empty at the root. + Schema fileSchema = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "loc", Types.StructType.of())); + Schema evolvedSchema = + new Schema( + optional( + 2, + "loc", + Types.StructType.of( + Types.NestedField.optional("country") + .withId(3) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .build()))); + TypeDescription readSchema = + ORCSchemaUtil.buildOrcProjection( + evolvedSchema, + ORCSchemaUtil.convert(fileSchema), + ORCSchemaUtil.FieldIdSource.EMBEDDED, + true); + Assertions.assertThat(readSchema.getChildren()).hasSize(1); + Assertions.assertThat(readSchema.findSubtype("loc").getChildren()).hasSize(0); + + Expression boundFilter = + Binder.bind(evolvedSchema.asStruct(), equal("loc.country", "bar"), true); + SearchArgument expected = + SearchArgumentFactory.newBuilder().literal(TruthValue.YES_NO_NULL).build(); + + SearchArgument actual = ExpressionToSearchArgument.convert(boundFilter, readSchema); + + Assertions.assertThat(actual.toString()).isEqualTo(expected.toString()); + } + + @Test + public void testMixedPhysicalAndOmittedDefaultPredicates() { + 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(), and(equal("id", 1L), equal("country", "US")), true); + SearchArgument expected = + SearchArgumentFactory.newBuilder() + .startAnd() + .equals("`id`", Type.LONG, 1L) + .literal(TruthValue.YES_NO_NULL) + .end() + .build(); + + SearchArgument actual = ExpressionToSearchArgument.convert(boundFilter, readSchema); + + Assertions.assertThat(actual.toString()).isEqualTo(expected.toString()); + } } diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderForFieldsWithDefaultValue.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderForFieldsWithDefaultValue.java index fa86eadc83..277cb6a27f 100644 --- a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderForFieldsWithDefaultValue.java +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderForFieldsWithDefaultValue.java @@ -30,6 +30,7 @@ import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.orc.ORC; import org.apache.iceberg.orc.ORCSchemaUtil; +import org.apache.iceberg.relocated.com.google.common.collect.Iterators; import org.apache.iceberg.types.Types; import org.apache.orc.OrcFile; import org.apache.orc.TypeDescription; @@ -165,6 +166,66 @@ public void testPartialEmbeddedIdsDoNotFillDefault() throws IOException { } } + @Test + public void testFilterOnOnlyOmittedDefaultDoesNotThrow() throws IOException { + // Projecting and filtering only a defaulted column yields an empty ORC schema. Convert must + // not throw; SARG is disabled (YES_NO_NULL). Row-level filtering is Spark's job. + Schema writeSchema = new Schema(Types.NestedField.required(1, "col1", Types.IntegerType.get())); + TypeDescription orcSchema = ORCSchemaUtil.convert(writeSchema); + Schema readSchema = + new Schema( + Types.NestedField.optional("col2") + .withId(2) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("foo")) + .build()); + File orcFile = writeOrcWithIntColumn(orcSchema, 10); + + try (CloseableIterable reader = + ORC.read(Files.localInput(orcFile)) + .project(readSchema) + .filter(Expressions.equal("col2", "bar")) + .createReaderFunc(readOrcSchema -> new SparkOrcReader(readSchema, readOrcSchema)) + .supportsInitialDefaults() + .build()) { + Assert.assertEquals(10, Iterators.size(reader.iterator())); + } + } + + @Test + public void testFilterOnOnlyOmittedNestedDefaultDoesNotThrow() throws IOException { + // Projecting and filtering only loc.country leaves struct>. Convert must not + // throw; SARG is disabled (YES_NO_NULL). Row-level filtering is Spark's job. + Schema writeSchema = + new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional("loc").withId(2).ofType(Types.StructType.of()).build()); + TypeDescription orcSchema = ORCSchemaUtil.convert(writeSchema); + Schema readSchema = + new Schema( + Types.NestedField.optional("loc") + .withId(2) + .ofType( + Types.StructType.of( + Types.NestedField.optional("country") + .withId(3) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .build())) + .build()); + File orcFile = writeOrcWithIdAndEmptyLoc(orcSchema, 10); + + try (CloseableIterable reader = + ORC.read(Files.localInput(orcFile)) + .project(readSchema) + .filter(Expressions.equal("loc.country", "bar")) + .createReaderFunc(readOrcSchema -> new SparkOrcReader(readSchema, readOrcSchema)) + .supportsInitialDefaults() + .build()) { + Assert.assertEquals(10, Iterators.size(reader.iterator())); + } + } + private File writeOrcWithIntColumn(TypeDescription orcSchema, int numRows) throws IOException { Configuration conf = new Configuration(); File orcFile = temp.newFile(); diff --git a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkOrcInitialDefaultScan.java b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkOrcInitialDefaultScan.java new file mode 100644 index 0000000000..22e871f24b --- /dev/null +++ b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkOrcInitialDefaultScan.java @@ -0,0 +1,360 @@ +/* + * 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 static org.apache.iceberg.Files.localInput; +import static org.apache.iceberg.Files.localOutput; +import static org.assertj.core.api.Assertions.tuple; + +import java.io.File; +import java.io.IOException; +import java.util.List; +import java.util.stream.Collectors; +import org.apache.hadoop.conf.Configuration; +import org.apache.iceberg.DataFile; +import org.apache.iceberg.DataFiles; +import org.apache.iceberg.FileFormat; +import org.apache.iceberg.PartitionSpec; +import org.apache.iceberg.Schema; +import org.apache.iceberg.Table; +import org.apache.iceberg.TableProperties; +import org.apache.iceberg.data.GenericAppenderFactory; +import org.apache.iceberg.data.GenericRecord; +import org.apache.iceberg.data.Record; +import org.apache.iceberg.expressions.Expressions; +import org.apache.iceberg.hadoop.HadoopTables; +import org.apache.iceberg.io.FileAppender; +import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; +import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.iceberg.spark.SparkReadOptions; +import org.apache.iceberg.spark.SparkSchemaUtil; +import org.apache.iceberg.types.Types; +import org.apache.spark.sql.Dataset; +import org.apache.spark.sql.Row; +import org.apache.spark.sql.SparkSession; +import org.apache.spark.sql.connector.read.Batch; +import org.apache.spark.sql.connector.read.InputPartition; +import org.apache.spark.sql.connector.read.PartitionReaderFactory; +import org.apache.spark.sql.util.CaseInsensitiveStringMap; +import org.assertj.core.api.Assertions; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +/** + * Spark SQL coverage for ORC initial-defaults: empty-projection filters, mixed physical+defaulted + * predicates, and nested defaults when the parent struct is null. Defaulted projections are forced + * onto the row reader even when ORC vectorization is enabled. + */ +public class TestSparkOrcInitialDefaultScan { + private static final Schema TABLE_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(), + Types.NestedField.optional(3, "data", Types.StringType.get())); + private static final Schema OLD_FILE_SCHEMA = + new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional(3, "data", Types.StringType.get())); + private static final Schema NESTED_TABLE_SCHEMA = + new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional("loc") + .withId(2) + .ofType( + Types.StructType.of( + Types.NestedField.optional(3, "city", Types.StringType.get()), + Types.NestedField.optional("country") + .withId(4) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .build())) + .build()); + private static final Schema NESTED_OLD_FILE_SCHEMA = + new Schema( + Types.NestedField.required(1, "id", Types.LongType.get()), + Types.NestedField.optional("loc") + .withId(2) + .ofType( + Types.StructType.of( + Types.NestedField.optional(3, "city", Types.StringType.get()))) + .build()); + private static final Configuration CONF = new Configuration(); + private static SparkSession spark; + + @Rule public TemporaryFolder temp = new TemporaryFolder(); + private File tableLocation; + private Table table; + + @BeforeClass + public static void startSpark() { + spark = SparkSession.builder().master("local[2]").getOrCreate(); + } + + @AfterClass + public static void stopSpark() { + spark.stop(); + spark = null; + } + + @Before + public void createMixedFileTable() throws IOException { + tableLocation = temp.newFolder("mixed-default-files"); + table = + new HadoopTables(CONF) + .create(TABLE_SCHEMA, PartitionSpec.unpartitioned(), tableLocation.toString()); + table + .updateProperties() + .set(TableProperties.FORMAT_VERSION, "2") + .set(TableProperties.DEFAULT_FILE_FORMAT, FileFormat.ORC.name()) + .set(TableProperties.ORC_VECTORIZATION_ENABLED, "true") + .commit(); + Assertions.assertThat(table.schema().findField("country").initialDefault()).isEqualTo("US"); + + Record old = GenericRecord.create(OLD_FILE_SCHEMA); + old.setField("id", 1L); + old.setField("data", "old"); + + Record present = GenericRecord.create(TABLE_SCHEMA); + present.setField("id", 2L); + present.setField("country", "CA"); + present.setField("data", "new-value"); + + Record presentNull = GenericRecord.create(TABLE_SCHEMA); + presentNull.setField("id", 3L); + presentNull.setField("country", null); + presentNull.setField("data", "new-null"); + + table + .newAppend() + .appendFile(writeFile(OLD_FILE_SCHEMA, Lists.newArrayList(old), "old.orc")) + .appendFile(writeFile(TABLE_SCHEMA, Lists.newArrayList(present, presentNull), "new.orc")) + .commit(); + } + + @Test + public void testDefaultedScansAreRoutedToTheRowReader() { + Assertions.assertThat(supportsColumnarReads(TABLE_SCHEMA)).isFalse(); + Schema countryOnly = + new Schema( + Types.NestedField.optional("country") + .withId(2) + .ofType(Types.StringType.get()) + .withInitialDefault(Expressions.lit("US")) + .build()); + Assertions.assertThat(supportsColumnarReads(countryOnly)).isFalse(); + } + + @Test + public void testMixedFilesApplyDefaultPerFile() { + List rows = read().select("id", "country", "data").orderBy("id").collectAsList(); + + Assertions.assertThat(rows) + .extracting(row -> row.getLong(0), row -> row.getString(1), row -> row.getString(2)) + .containsExactly( + tuple(1L, "US", "old"), tuple(2L, "CA", "new-value"), tuple(3L, null, "new-null")); + } + + @Test + public void testFilterWhenOnlyOmittedDefaultColumnIsProjected() { + // Projection+filter on only the defaulted column yields an empty ORC schema for the old file. + // SARG must disable pushdown so Spark can evaluate the filled default. + Assertions.assertThat( + read().select("country").filter("country = 'US'").collectAsList().stream() + .map(row -> row.getString(0)) + .collect(Collectors.toList())) + .containsExactly("US"); + + Assertions.assertThat(read().select("country").filter("country = 'bar'").collectAsList()) + .isEmpty(); + + Assertions.assertThat( + read().select("country").filter("country IS NULL").collectAsList().stream() + .map(row -> row.getString(0)) + .collect(Collectors.toList())) + .containsExactly((String) null); + } + + @Test + public void testFilterMixedPhysicalAndDefaultedColumn() { + List ids = + read().filter("id = 1 AND country = 'US'").select("id").collectAsList().stream() + .map(row -> row.getLong(0)) + .collect(Collectors.toList()); + Assertions.assertThat(ids).containsExactly(1L); + + Assertions.assertThat(read().filter("id = 1 AND country = 'CA'").collectAsList()).isEmpty(); + } + + @Test + public void testFilterDefaultedColumnAcrossMixedFiles() { + // Old file reads country as 'US'; new file has 'CA' and an explicit null. SARG is disabled for + // the omitted column so Spark filters the filled values. + Assertions.assertThat( + read().filter("country IS NOT NULL").select("id").orderBy("id").collectAsList().stream() + .map(row -> row.getLong(0)) + .collect(Collectors.toList())) + .containsExactly(1L, 2L); + + Assertions.assertThat( + read().filter("upper(country) = 'US'").select("id").orderBy("id").collectAsList() + .stream() + .map(row -> row.getLong(0)) + .collect(Collectors.toList())) + .containsExactly(1L); + + Assertions.assertThat( + read().filter("country = 'CA'").select("id").collectAsList().stream() + .map(row -> row.getLong(0)) + .collect(Collectors.toList())) + .containsExactly(2L); + } + + @Test + public void testFilterNestedDefaultWhenParentStructIsNull() throws IOException { + File nestedLocation = temp.newFolder("nested-default-files"); + Table nestedTable = + new HadoopTables(CONF) + .create(NESTED_TABLE_SCHEMA, PartitionSpec.unpartitioned(), nestedLocation.toString()); + nestedTable + .updateProperties() + .set(TableProperties.FORMAT_VERSION, "2") + .set(TableProperties.DEFAULT_FILE_FORMAT, FileFormat.ORC.name()) + .set(TableProperties.ORC_VECTORIZATION_ENABLED, "true") + .commit(); + + Types.StructType oldLocType = NESTED_OLD_FILE_SCHEMA.findType("loc").asStructType(); + Record locSf = GenericRecord.create(oldLocType); + locSf.setField("city", "San Francisco"); + Record presentLoc = GenericRecord.create(NESTED_OLD_FILE_SCHEMA); + presentLoc.setField("id", 1L); + presentLoc.setField("loc", locSf); + + Record nullLoc = GenericRecord.create(NESTED_OLD_FILE_SCHEMA); + nullLoc.setField("id", 2L); + nullLoc.setField("loc", null); + + Types.StructType newLocType = NESTED_TABLE_SCHEMA.findType("loc").asStructType(); + Record locCa = GenericRecord.create(newLocType); + locCa.setField("city", "Toronto"); + locCa.setField("country", "CA"); + Record presentCa = GenericRecord.create(NESTED_TABLE_SCHEMA); + presentCa.setField("id", 3L); + presentCa.setField("loc", locCa); + + nestedTable + .newAppend() + .appendFile( + writeFile( + nestedLocation, + NESTED_OLD_FILE_SCHEMA, + Lists.newArrayList(presentLoc, nullLoc), + "old-nested.orc")) + .appendFile( + writeFile( + nestedLocation, + NESTED_TABLE_SCHEMA, + Lists.newArrayList(presentCa), + "new-nested.orc")) + .commit(); + + Dataset nestedRead = + spark + .read() + .format("iceberg") + .option(SparkReadOptions.VECTORIZATION_ENABLED, "true") + .load(nestedLocation.toString()); + + // Present loc fills country from initial-default; a null parent stays null. + Assertions.assertThat( + nestedRead.selectExpr("id", "loc.country as country").orderBy("id").collectAsList() + .stream() + .map(row -> tuple(row.getLong(0), row.isNullAt(1) ? null : row.getString(1))) + .collect(Collectors.toList())) + .containsExactly(tuple(1L, "US"), tuple(2L, null), tuple(3L, "CA")); + + Assertions.assertThat( + nestedRead.filter("loc.country = 'US'").select("id").collectAsList().stream() + .map(row -> row.getLong(0)) + .collect(Collectors.toList())) + .containsExactly(1L); + + Assertions.assertThat( + nestedRead.filter("loc.country IS NULL").select("id").collectAsList().stream() + .map(row -> row.getLong(0)) + .collect(Collectors.toList())) + .containsExactly(2L); + + Assertions.assertThat( + nestedRead.filter("loc.country = 'CA'").select("id").collectAsList().stream() + .map(row -> row.getLong(0)) + .collect(Collectors.toList())) + .containsExactly(3L); + } + + private Dataset read() { + return spark + .read() + .format("iceberg") + .option(SparkReadOptions.VECTORIZATION_ENABLED, "true") + .load(tableLocation.toString()); + } + + private boolean supportsColumnarReads(Schema projection) { + CaseInsensitiveStringMap options = + new CaseInsensitiveStringMap( + ImmutableMap.of( + "path", tableLocation.toString(), SparkReadOptions.VECTORIZATION_ENABLED, "true")); + SparkScanBuilder builder = new SparkScanBuilder(spark, table, options); + builder.pruneColumns(SparkSchemaUtil.convert(projection)); + Batch batch = builder.build().toBatch(); + InputPartition partition = batch.planInputPartitions()[0]; + PartitionReaderFactory factory = batch.createReaderFactory(); + return factory.supportColumnarReads(partition); + } + + private DataFile writeFile(Schema schema, List records, String name) throws IOException { + return writeFile(tableLocation, schema, records, name); + } + + private DataFile writeFile(File location, Schema schema, List records, String name) + throws IOException { + File file = new File(new File(location, "data"), name); + Assertions.assertThat(file.getParentFile().mkdirs() || file.getParentFile().isDirectory()) + .isTrue(); + FileAppender appender = + new GenericAppenderFactory(schema).newAppender(localOutput(file), FileFormat.ORC); + try { + appender.addAll(records); + } finally { + appender.close(); + } + return DataFiles.builder(PartitionSpec.unpartitioned()) + .withInputFile(localInput(file)) + .withMetrics(appender.metrics()) + .build(); + } +}