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/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..e3d76515bcc7 --- /dev/null +++ b/arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestVectorizedReaderBuilder.java @@ -0,0 +1,92 @@ +/* + * 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.assertThatNoException; +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 = parquetSchemaWithVariant(); + + 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"); + } + + @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"); + } +} 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')", 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..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 @@ -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')",