diff --git a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedReaderBuilder.java b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedReaderBuilder.java index 59023fe7c877..12b741135c45 100644 --- a/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedReaderBuilder.java +++ b/arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedReaderBuilder.java @@ -157,6 +157,12 @@ public VectorizedReader struct( @Override public ParquetVariantVisitor> variantVisitor() { + // skip variants that are not projected, like primitives, so their shredded layout is not read + Type variant = parquetSchema.getType(currentPath()); + if (variant.getId() == null || icebergSchema.findField(variant.getId().intValue()) == null) { + return null; + } + return new VectorizedVariantVisitor( currentPath(), parquetSchema, icebergSchema, rootAllocator, setArrowValidityVector); } diff --git a/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestVectorizedReaderBuilder.java b/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestVectorizedReaderBuilder.java index 148d0c0ecb82..7c416c3751b9 100644 --- a/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestVectorizedReaderBuilder.java +++ b/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestVectorizedReaderBuilder.java @@ -19,7 +19,9 @@ package org.apache.iceberg.arrow.vectorized; import static org.assertj.core.api.Assertions.assertThatNoException; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import java.util.stream.Stream; import org.apache.iceberg.Schema; import org.apache.iceberg.parquet.TypeWithSchemaVisitor; import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; @@ -27,12 +29,16 @@ import org.apache.iceberg.types.Types.NestedField; import org.apache.iceberg.types.Types.VariantType; import org.apache.iceberg.variants.Variant; +import org.apache.parquet.schema.GroupType; import org.apache.parquet.schema.LogicalTypeAnnotation; import org.apache.parquet.schema.MessageType; import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; import org.apache.parquet.schema.Type; import org.apache.parquet.schema.Types; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; public class TestVectorizedReaderBuilder { @@ -70,6 +76,105 @@ public void testVariantSkippedWhenNotInProjection() { () -> TypeWithSchemaVisitor.visit(icebergSchema.asStruct(), parquetSchema, builder)); } + @ParameterizedTest(name = "{0}") + @MethodSource("unprojectedShreddedVariants") + void shreddedVariantSkippedWhenNotInProjection(String location, Type field) { + Schema icebergSchema = new Schema(NestedField.required(1, "id", IntegerType.get())); + + MessageType parquetSchema = + Types.buildMessage() + .addField( + Types.primitive(PrimitiveTypeName.INT32, Type.Repetition.REQUIRED) + .id(1) + .named("id")) + .addField(field) + .named("table"); + + VectorizedReaderBuilder builder = + new VectorizedReaderBuilder( + icebergSchema, parquetSchema, false, ImmutableMap.of(), readers -> null); + + assertThatNoException() + .describedAs("Shredded variant (%s) not in projection should not throw", location) + .isThrownBy( + () -> TypeWithSchemaVisitor.visit(icebergSchema.asStruct(), parquetSchema, builder)); + } + + private static Stream unprojectedShreddedVariants() { + return Stream.of( + Arguments.of("top-level", shreddedVariant(Type.Repetition.OPTIONAL, 2, "data")), + Arguments.of( + "struct field", + Types.buildGroup(Type.Repetition.OPTIONAL) + .addField(shreddedVariant(Type.Repetition.OPTIONAL, 3, "data")) + .id(2) + .named("struct")), + Arguments.of( + "list element", + Types.buildGroup(Type.Repetition.OPTIONAL) + .as(LogicalTypeAnnotation.listType()) + .addField( + Types.buildGroup(Type.Repetition.REPEATED) + .addField(shreddedVariant(Type.Repetition.OPTIONAL, 3, "element")) + .named("list")) + .id(2) + .named("list")), + Arguments.of( + "map key", + Types.buildGroup(Type.Repetition.OPTIONAL) + .as(LogicalTypeAnnotation.mapType()) + .addField( + Types.buildGroup(Type.Repetition.REPEATED) + .addField(shreddedVariant(Type.Repetition.REQUIRED, 3, "key")) + .addField( + Types.primitive(PrimitiveTypeName.INT32, Type.Repetition.OPTIONAL) + .id(4) + .named("value")) + .named("key_value")) + .id(2) + .named("map")), + Arguments.of( + "map value", + Types.buildGroup(Type.Repetition.OPTIONAL) + .as(LogicalTypeAnnotation.mapType()) + .addField( + Types.buildGroup(Type.Repetition.REPEATED) + .addField( + Types.primitive(PrimitiveTypeName.INT32, Type.Repetition.REQUIRED) + .id(3) + .named("key")) + .addField(shreddedVariant(Type.Repetition.OPTIONAL, 4, "value")) + .named("key_value")) + .id(2) + .named("map"))); + } + + @Test + void projectingShreddedVariantUnsupported() { + Schema icebergSchema = + new Schema( + NestedField.required(1, "id", IntegerType.get()), + NestedField.optional(2, "data", VariantType.get())); + + MessageType parquetSchema = + Types.buildMessage() + .addField( + Types.primitive(PrimitiveTypeName.INT32, Type.Repetition.REQUIRED) + .id(1) + .named("id")) + .addField(shreddedVariant(Type.Repetition.OPTIONAL, 2, "data")) + .named("table"); + + VectorizedReaderBuilder builder = + new VectorizedReaderBuilder( + icebergSchema, parquetSchema, false, ImmutableMap.of(), readers -> null); + + assertThatThrownBy( + () -> TypeWithSchemaVisitor.visit(icebergSchema.asStruct(), parquetSchema, builder)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessage("Unsupported variant: shredded typed_value primitive"); + } + private static MessageType parquetSchemaWithVariant() { return Types.buildMessage() .addField( @@ -87,4 +192,25 @@ private static MessageType parquetSchemaWithVariant() { .named("data")) .named("table"); } + + private static GroupType shreddedVariant(Type.Repetition repetition, int id, String name) { + GroupType shreddedField = + Types.buildGroup(Type.Repetition.REQUIRED) + .addField( + Types.primitive(PrimitiveTypeName.BINARY, Type.Repetition.OPTIONAL).named("value")) + .addField( + Types.primitive(PrimitiveTypeName.INT32, Type.Repetition.OPTIONAL) + .named("typed_value")) + .named("a"); + return Types.buildGroup(repetition) + .as(LogicalTypeAnnotation.variantType(Variant.VARIANT_SPEC_VERSION)) + .addField( + Types.primitive(PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED).named("metadata")) + .addField( + Types.primitive(PrimitiveTypeName.BINARY, Type.Repetition.OPTIONAL).named("value")) + .addField( + Types.buildGroup(Type.Repetition.OPTIONAL).addField(shreddedField).named("typed_value")) + .id(id) + .named(name); + } } diff --git a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java index 79e52d6e339c..4e8d303c4e91 100644 --- a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java +++ b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java @@ -464,6 +464,38 @@ public void testReadShreddedAfterPropertyToggled(boolean vectorized) sql("DROP TABLE IF EXISTS %s", toggleTable); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void readShreddedWithoutProjectingVariant(boolean vectorized) + throws IOException, NoSuchTableException, ParseException { + String shreddedTable = CATALOG + ".default.var_shredded_unprojected"; + sql("DROP TABLE IF EXISTS %s", shreddedTable); + sql( + "CREATE TABLE %s (id BIGINT, v VARIANT) USING iceberg " + + "TBLPROPERTIES ('format-version'='3', 'write.parquet.shred-variants'='true')", + shreddedTable); + + spark.conf().set("spark.sql.iceberg.shred-variants", "true"); + try { + sql( + "INSERT INTO %s VALUES " + + "(1, parse_json('{\"name\":\"alice\",\"age\":30}')), " + + "(2, parse_json('{\"name\":\"bob\",\"age\":25}'))", + shreddedTable); + } finally { + spark.conf().unset("spark.sql.iceberg.shred-variants"); + } + + Table table = Spark3Util.loadIcebergTable(spark, shreddedTable); + assertHasTypedValueSubtree(table); + setVectorization(shreddedTable, vectorized); + + List rows = spark.table(shreddedTable).select("id").orderBy("id").collectAsList(); + assertThat(rows).extracting(r -> r.getLong(0)).containsExactly(1L, 2L); + + sql("DROP TABLE IF EXISTS %s", shreddedTable); + } + @ParameterizedTest @ValueSource(strings = {"none", "counts"}) public void testReadShreddedWithMetricsDisabled(String metricsMode)