From 06db44a94a8345d69be3d9e5381ead3a7defb1db Mon Sep 17 00:00:00 2001 From: Neelesh Salian Date: Wed, 22 Apr 2026 14:57:25 -0700 Subject: [PATCH 1/5] Fix for vectorized builder variant handling --- .../vectorized/VectorizedReaderBuilder.java | 10 ++++ .../spark/sql/TestSparkVariantRead.java | 50 +++++++++++++++++++ 2 files changed, 60 insertions(+) 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 15b55fb48d4a..3fbd797c26fb 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 @@ -154,6 +154,16 @@ public VectorizedReader struct( return null; } + @Override + public VectorizedReader variant( + Types.VariantType iVariant, GroupType variant, VectorizedReader result) { + if (iVariant != null) { + throw new UnsupportedOperationException( + "Vectorized reads are not supported yet for variant fields"); + } + return null; + } + @Override public VectorizedReader primitive( org.apache.iceberg.types.Type.PrimitiveType expected, PrimitiveType primitive) { diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java index 599bf591e9a4..a4998f269cd9 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java @@ -302,6 +302,56 @@ public void testNestedMapVariant(boolean vectorized) { sql("DROP TABLE IF EXISTS %s", mapTable); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testMergeIntoWithVariant(boolean vectorized) { + // Variant columns are not vectorized yet, but MERGE INTO should not crash regardless of the + // vectorization setting. The reader falls back to non-vectorized for variant columns. + String mergeTable = CATALOG + ".default.var_merge"; + sql("DROP TABLE IF EXISTS %s", mergeTable); + sql( + "CREATE TABLE %s (id BIGINT, data VARIANT) USING iceberg " + + "TBLPROPERTIES ('format-version'='3')", + mergeTable); + setVectorization(mergeTable, vectorized); + + sql( + "INSERT INTO %s VALUES " + + "(1, parse_json('{\"name\":\"alice\",\"age\":30}')), " + + "(2, parse_json('{\"name\":\"bob\",\"age\":25}'))", + mergeTable); + + sql( + "MERGE INTO %s AS target " + + "USING (SELECT 1 AS id, parse_json('{\"name\":\"alice\",\"age\":31}') AS data) AS source " + + "ON target.id = source.id " + + " AND variant_get(target.data, '$.name', 'string') = variant_get(source.data, '$.name', 'string') " + + "WHEN MATCHED THEN UPDATE SET target.data = source.data " + + "WHEN NOT MATCHED THEN INSERT *", + mergeTable); + + List rows = spark.table(mergeTable).select("id", "data").orderBy("id").collectAsList(); + + assertThat(rows).hasSize(2); + assertThat(rows.get(0).getLong(0)).isEqualTo(1L); + Variant v1 = + new Variant( + ((VariantVal) rows.get(0).get(1)).getValue(), + ((VariantVal) rows.get(0).get(1)).getMetadata()); + assertThat(v1.getFieldByKey("name").getString()).isEqualTo("alice"); + assertThat(v1.getFieldByKey("age").getLong()).isEqualTo(31L); + + assertThat(rows.get(1).getLong(0)).isEqualTo(2L); + Variant v2 = + new Variant( + ((VariantVal) rows.get(1).get(1)).getValue(), + ((VariantVal) rows.get(1).get(1)).getMetadata()); + assertThat(v2.getFieldByKey("name").getString()).isEqualTo("bob"); + assertThat(v2.getFieldByKey("age").getLong()).isEqualTo(25L); + + sql("DROP TABLE IF EXISTS %s", mergeTable); + } + private void setVectorization(boolean on) { sql( "ALTER TABLE %s SET TBLPROPERTIES ('read.parquet.vectorization.enabled'='%s')", From be22175aa896247c568ffd52dbb42efcc63c5778 Mon Sep 17 00:00:00 2001 From: Neelesh Salian Date: Fri, 24 Apr 2026 09:01:44 -0700 Subject: [PATCH 2/5] Simplify test query and add reg test --- .../TestVectorizedReaderBuilder.java | 74 +++++++++++++++++++ .../spark/sql/TestSparkVariantRead.java | 1 - 2 files changed, 74 insertions(+), 1 deletion(-) create mode 100644 arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestVectorizedReaderBuilder.java 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 new file mode 100644 index 000000000000..3d6b7713a20d --- /dev/null +++ b/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestVectorizedReaderBuilder.java @@ -0,0 +1,74 @@ +/* + * 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.arrow.vectorized; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +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.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; + +public class TestVectorizedReaderBuilder { + + @Test + public void testVariantNotSupportedInVectorizedReads() { + 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( + Types.buildGroup(Type.Repetition.OPTIONAL) + .as(LogicalTypeAnnotation.variantType(Variant.VARIANT_SPEC_VERSION)) + .addField( + Types.primitive(PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED) + .named("metadata")) + .addField( + Types.primitive(PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED) + .named("value")) + .id(2) + .named("data")) + .named("table"); + + VectorizedReaderBuilder builder = + new VectorizedReaderBuilder( + icebergSchema, parquetSchema, false, ImmutableMap.of(), readers -> null); + + assertThatThrownBy( + () -> TypeWithSchemaVisitor.visit(icebergSchema.asStruct(), parquetSchema, builder)) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("Vectorized reads are not supported yet for variant fields"); + } +} diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java index a4998f269cd9..1b1e89bd3825 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java @@ -325,7 +325,6 @@ public void testMergeIntoWithVariant(boolean vectorized) { "MERGE INTO %s AS target " + "USING (SELECT 1 AS id, parse_json('{\"name\":\"alice\",\"age\":31}') AS data) AS source " + "ON target.id = source.id " - + " AND variant_get(target.data, '$.name', 'string') = variant_get(source.data, '$.name', 'string') " + "WHEN MATCHED THEN UPDATE SET target.data = source.data " + "WHEN NOT MATCHED THEN INSERT *", mergeTable); From b9082f16b3508a288b3683ea583d5ebe55b88d62 Mon Sep 17 00:00:00 2001 From: Neelesh Salian Date: Tue, 28 Apr 2026 12:49:20 -0700 Subject: [PATCH 3/5] PR comment: add describedAs for keys --- .../apache/iceberg/spark/sql/TestSparkVariantRead.java | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java index 1b1e89bd3825..2d6e919a91ee 100644 --- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java +++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java @@ -337,16 +337,16 @@ public void testMergeIntoWithVariant(boolean vectorized) { new Variant( ((VariantVal) rows.get(0).get(1)).getValue(), ((VariantVal) rows.get(0).get(1)).getMetadata()); - assertThat(v1.getFieldByKey("name").getString()).isEqualTo("alice"); - assertThat(v1.getFieldByKey("age").getLong()).isEqualTo(31L); + assertThat(v1.getFieldByKey("name").getString()).describedAs("v1.name").isEqualTo("alice"); + assertThat(v1.getFieldByKey("age").getLong()).describedAs("v1.age").isEqualTo(31L); assertThat(rows.get(1).getLong(0)).isEqualTo(2L); Variant v2 = new Variant( ((VariantVal) rows.get(1).get(1)).getValue(), ((VariantVal) rows.get(1).get(1)).getMetadata()); - assertThat(v2.getFieldByKey("name").getString()).isEqualTo("bob"); - assertThat(v2.getFieldByKey("age").getLong()).isEqualTo(25L); + assertThat(v2.getFieldByKey("name").getString()).describedAs("v2.name").isEqualTo("bob"); + assertThat(v2.getFieldByKey("age").getLong()).describedAs("v2.age").isEqualTo(25L); sql("DROP TABLE IF EXISTS %s", mergeTable); } From b31ee26bf582239c45df78d4abe30b15931460af Mon Sep 17 00:00:00 2001 From: Neelesh Salian Date: Fri, 1 May 2026 09:26:34 -0700 Subject: [PATCH 4/5] Add merge into test for spark 4.0 --- .../spark/sql/TestSparkVariantRead.java | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java index 599bf591e9a4..2d6e919a91ee 100644 --- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java +++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java @@ -302,6 +302,55 @@ public void testNestedMapVariant(boolean vectorized) { sql("DROP TABLE IF EXISTS %s", mapTable); } + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testMergeIntoWithVariant(boolean vectorized) { + // Variant columns are not vectorized yet, but MERGE INTO should not crash regardless of the + // vectorization setting. The reader falls back to non-vectorized for variant columns. + String mergeTable = CATALOG + ".default.var_merge"; + sql("DROP TABLE IF EXISTS %s", mergeTable); + sql( + "CREATE TABLE %s (id BIGINT, data VARIANT) USING iceberg " + + "TBLPROPERTIES ('format-version'='3')", + mergeTable); + setVectorization(mergeTable, vectorized); + + sql( + "INSERT INTO %s VALUES " + + "(1, parse_json('{\"name\":\"alice\",\"age\":30}')), " + + "(2, parse_json('{\"name\":\"bob\",\"age\":25}'))", + mergeTable); + + sql( + "MERGE INTO %s AS target " + + "USING (SELECT 1 AS id, parse_json('{\"name\":\"alice\",\"age\":31}') AS data) AS source " + + "ON target.id = source.id " + + "WHEN MATCHED THEN UPDATE SET target.data = source.data " + + "WHEN NOT MATCHED THEN INSERT *", + mergeTable); + + List rows = spark.table(mergeTable).select("id", "data").orderBy("id").collectAsList(); + + assertThat(rows).hasSize(2); + assertThat(rows.get(0).getLong(0)).isEqualTo(1L); + Variant v1 = + new Variant( + ((VariantVal) rows.get(0).get(1)).getValue(), + ((VariantVal) rows.get(0).get(1)).getMetadata()); + assertThat(v1.getFieldByKey("name").getString()).describedAs("v1.name").isEqualTo("alice"); + assertThat(v1.getFieldByKey("age").getLong()).describedAs("v1.age").isEqualTo(31L); + + assertThat(rows.get(1).getLong(0)).isEqualTo(2L); + Variant v2 = + new Variant( + ((VariantVal) rows.get(1).get(1)).getValue(), + ((VariantVal) rows.get(1).get(1)).getMetadata()); + assertThat(v2.getFieldByKey("name").getString()).describedAs("v2.name").isEqualTo("bob"); + assertThat(v2.getFieldByKey("age").getLong()).describedAs("v2.age").isEqualTo(25L); + + sql("DROP TABLE IF EXISTS %s", mergeTable); + } + private void setVectorization(boolean on) { sql( "ALTER TABLE %s SET TBLPROPERTIES ('read.parquet.vectorization.enabled'='%s')", From 6c85a36b7231a591a7cdf9049b4e62b0edf1b178 Mon Sep 17 00:00:00 2001 From: Neelesh Salian Date: Fri, 1 May 2026 16:14:19 -0700 Subject: [PATCH 5/5] PR comment: Add test for variant not in projection --- .../TestVectorizedReaderBuilder.java | 54 ++++++++++++------- 1 file changed, 36 insertions(+), 18 deletions(-) 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 3d6b7713a20d..e3d76515bcc7 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 @@ -18,6 +18,7 @@ */ package org.apache.iceberg.arrow.vectorized; +import static org.assertj.core.api.Assertions.assertThatNoException; import static org.assertj.core.api.Assertions.assertThatThrownBy; import org.apache.iceberg.Schema; @@ -43,24 +44,7 @@ public void testVariantNotSupportedInVectorizedReads() { 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( - Types.buildGroup(Type.Repetition.OPTIONAL) - .as(LogicalTypeAnnotation.variantType(Variant.VARIANT_SPEC_VERSION)) - .addField( - Types.primitive(PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED) - .named("metadata")) - .addField( - Types.primitive(PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED) - .named("value")) - .id(2) - .named("data")) - .named("table"); + MessageType parquetSchema = parquetSchemaWithVariant(); VectorizedReaderBuilder builder = new VectorizedReaderBuilder( @@ -71,4 +55,38 @@ public void testVariantNotSupportedInVectorizedReads() { .isInstanceOf(UnsupportedOperationException.class) .hasMessageContaining("Vectorized reads are not supported yet for variant fields"); } + + @Test + public void testVariantSkippedWhenNotInProjection() { + Schema icebergSchema = new Schema(NestedField.required(1, "id", IntegerType.get())); + + MessageType parquetSchema = parquetSchemaWithVariant(); + + VectorizedReaderBuilder builder = + new VectorizedReaderBuilder( + icebergSchema, parquetSchema, false, ImmutableMap.of(), readers -> null); + + assertThatNoException() + .describedAs("Variant not in projection should not throw") + .isThrownBy( + () -> TypeWithSchemaVisitor.visit(icebergSchema.asStruct(), parquetSchema, builder)); + } + + private static MessageType parquetSchemaWithVariant() { + return Types.buildMessage() + .addField( + Types.primitive(PrimitiveTypeName.INT32, Type.Repetition.REQUIRED).id(1).named("id")) + .addField( + Types.buildGroup(Type.Repetition.OPTIONAL) + .as(LogicalTypeAnnotation.variantType(Variant.VARIANT_SPEC_VERSION)) + .addField( + Types.primitive(PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED) + .named("metadata")) + .addField( + Types.primitive(PrimitiveTypeName.BINARY, Type.Repetition.REQUIRED) + .named("value")) + .id(2) + .named("data")) + .named("table"); + } }