Skip to content

Arrow, Spark: Skip unprojected shredded variants in vectorized Parquet reads #18434

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
randreucetti wants to merge 2 commits into
base: main
Choose a base branch
from
+164 −0
Open

Arrow, Spark: Skip unprojected shredded variants in vectorized Parquet reads #18434

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
46c98d5
Arrow, Spark: Skip unprojected shredded variants in vectorized Parque…
randreucetti Oct 9, 2026
7485158
Address review: cover nested unprojected variants, rename new tests
randreucetti Oct 12, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedReaderBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,12 @@ public VectorizedReader<?> struct(

@Override
public ParquetVariantVisitor<VectorizedReader<?>> 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);
}
Expand Down
126 changes: 126 additions & 0 deletions arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestVectorizedReaderBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,20 +19,26 @@
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;
import org.apache.iceberg.types.Types.IntegerType;
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 {

Expand Down Expand Up @@ -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<Arguments> 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(
Expand All @@ -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);
}
}
32 changes: 32 additions & 0 deletions spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Row> 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)
Expand Down
Loading