Spark, Arrow: Fix for vectorized builder variant handling

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) {

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");
+  }
+}

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<Row> 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')",

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<Row> 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')",