Spark: Add vectorized Parquet reads for variant columns

arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorHolder.java

@@ -185,4 +185,34 @@ public int numValues() {
       return numRows;
     }
   }
+
+  public static class VariantVectorHolder extends VectorHolder {
+    private final VectorHolder metadataHolder;
+    private final VectorHolder valueHolder;
+    private final int numRows;
+
+    public VariantVectorHolder(
+        Types.NestedField icebergField,
+        int numRows,
+        VectorHolder metadataHolder,
+        VectorHolder valueHolder) {
+      super(icebergField);
+      this.numRows = numRows;
+      this.metadataHolder = metadataHolder;
+      this.valueHolder = valueHolder;
+    }
+
+    @Override
+    public int numValues() {
+      return numRows;
+    }
+
+    public VectorHolder metadataHolder() {
+      return metadataHolder;
+    }
+
+    public VectorHolder valueHolder() {
+      return valueHolder;
+    }
+  }
 }

arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedArrowReader.java

@@ -965,4 +965,57 @@ public String toString() {
     @Override
     public void setBatchSize(int batchSize) {}
   }
+
+  public static class VectorizedVariantReader extends VectorizedArrowReader {
+    private final VectorizedArrowReader metadataReader;
+    private final VectorizedArrowReader valueReader;
+
+    public VectorizedVariantReader(
+        Types.NestedField icebergField,
+        VectorizedArrowReader metadataReader,
+        VectorizedArrowReader valueReader) {
+      super(icebergField);
+      this.metadataReader = metadataReader;
+      this.valueReader = valueReader;
+    }
+
+    @Override
+    public VectorHolder read(VectorHolder reuse, int numValsToRead) {
+      VectorHolder reuseMetadata = null;
+      VectorHolder reuseValue = null;
+      if (reuse instanceof VectorHolder.VariantVectorHolder) {
+        VectorHolder.VariantVectorHolder variantReuse = (VectorHolder.VariantVectorHolder) reuse;
+        reuseMetadata = variantReuse.metadataHolder();
+        reuseValue = variantReuse.valueHolder();
+      }
+      VectorHolder metadataHolder = metadataReader.read(reuseMetadata, numValsToRead);
+      VectorHolder valueHolder = valueReader.read(reuseValue, numValsToRead);
+      return new VectorHolder.VariantVectorHolder(
+          icebergField(), numValsToRead, metadataHolder, valueHolder);
+    }
+
+    @Override
+    public void setRowGroupInfo(
+        PageReadStore source, Map<ColumnPath, ColumnChunkMetaData> metadata) {
+      metadataReader.setRowGroupInfo(source, metadata);
+      valueReader.setRowGroupInfo(source, metadata);
+    }
+
+    @Override
+    public void setBatchSize(int batchSize) {
+      metadataReader.setBatchSize(batchSize);
+      valueReader.setBatchSize(batchSize);
+    }
+
+    @Override
+    public void close() {
+      metadataReader.close();
+      valueReader.close();
+    }
+
+    @Override
+    public String toString() {
+      return "VectorizedVariantReader";
+    }
+  }
 }

arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedReaderBuilder.java

@@ -27,6 +27,7 @@
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.arrow.ArrowAllocation;
 import org.apache.iceberg.arrow.vectorized.VectorizedArrowReader.ConstantVectorReader;
+import org.apache.iceberg.parquet.ParquetVariantVisitor;
 import org.apache.iceberg.parquet.TypeWithSchemaVisitor;
 import org.apache.iceberg.parquet.VectorizedReader;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -154,14 +155,16 @@ public VectorizedReader<?> struct(
     return null;
   }
 
+  @Override
+  public ParquetVariantVisitor<VectorizedReader<?>> variantVisitor() {
+    return new VectorizedVariantVisitor(
+        currentPath(), parquetSchema, icebergSchema, rootAllocator, setArrowValidityVector);
+  }
+
   @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;
+    return result;
   }
 
   @Override

arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedVariantVisitor.java

@@ -0,0 +1,137 @@
+/*
+ * 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 java.util.List;
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.parquet.ParquetVariantVisitor;
+import org.apache.iceberg.parquet.VectorizedReader;
+import org.apache.iceberg.types.Types;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.io.InvalidRecordException;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
+
+/** Creates Arrow readers for a variant column's metadata and value leaves. */
+class VectorizedVariantVisitor extends ParquetVariantVisitor<VectorizedReader<?>> {
+  private final String[] variantGroupPath;
+  private final MessageType parquetSchema;
+  private final Schema icebergSchema;
+  private final BufferAllocator allocator;
+  private final boolean setArrowValidityVector;
+
+  VectorizedVariantVisitor(
+      String[] variantGroupPath,
+      MessageType parquetSchema,
+      Schema icebergSchema,
+      BufferAllocator allocator,
+      boolean setArrowValidityVector) {
+    this.variantGroupPath = variantGroupPath;
+    this.parquetSchema = parquetSchema;
+    this.icebergSchema = icebergSchema;
+    this.allocator = allocator;
+    this.setArrowValidityVector = setArrowValidityVector;
+  }
+
+  @Override
+  public VectorizedReader<?> metadata(PrimitiveType metadata) {
+    ColumnDescriptor desc = resolveDescriptor(metadata);
+    if (desc == null) {
+      return null;
+    }
+
+    Types.NestedField field =
+        Types.NestedField.required(-1, metadata.getName(), Types.BinaryType.get());
+    return new VectorizedArrowReader(desc, field, allocator, setArrowValidityVector);
+  }
+
+  @Override
+  public VectorizedReader<?> serialized(PrimitiveType value) {
+    ColumnDescriptor desc = resolveDescriptor(value);
+    if (desc == null) {
+      return null;
+    }
+
+    Types.NestedField field =
+        Types.NestedField.optional(-1, value.getName(), Types.BinaryType.get());
+    return new VectorizedArrowReader(desc, field, allocator, setArrowValidityVector);
+  }
+
+  @Override
+  public VectorizedReader<?> variant(
+      GroupType variant, VectorizedReader<?> metadataResult, VectorizedReader<?> valueResult) {
+    if (metadataResult instanceof VectorizedArrowReader
+        && valueResult instanceof VectorizedArrowReader) {
+      Types.NestedField field = findVariantField(variant);
+      if (field != null) {
+        return new VectorizedArrowReader.VectorizedVariantReader(
+            field, (VectorizedArrowReader) metadataResult, (VectorizedArrowReader) valueResult);
+      }
+    }
+
+    return null;
+  }
+
+  @Override
+  public VectorizedReader<?> primitive(PrimitiveType primitive) {
+    return null;
+  }
+
+  @Override
+  public VectorizedReader<?> value(
+      GroupType value, VectorizedReader<?> valueResult, VectorizedReader<?> typedResult) {
+    return valueResult;
+  }
+
+  @Override
+  public VectorizedReader<?> object(
+      GroupType object, VectorizedReader<?> valueResult, List<VectorizedReader<?>> fieldResults) {
+    return valueResult;
+  }
+
+  @Override
+  public VectorizedReader<?> array(
+      GroupType array, VectorizedReader<?> valueResult, VectorizedReader<?> elementResult) {
+    return valueResult;
+  }
+
+  private ColumnDescriptor resolveDescriptor(PrimitiveType primitive) {
+    // Build full column path: variant group path + primitive name
+    // e.g., ["v1", "metadata"] or ["v2", "value"]
+    String[] path = new String[variantGroupPath.length + 1];
+    System.arraycopy(variantGroupPath, 0, path, 0, variantGroupPath.length);
+    path[variantGroupPath.length] = primitive.getName();
+
+    try {
+      return parquetSchema.getColumnDescription(path);
+    } catch (InvalidRecordException e) {
+      return null;
+    }
+  }
+
+  private Types.NestedField findVariantField(GroupType variant) {
+    if (variant.getId() != null) {
+      return icebergSchema.findField(variant.getId().intValue());
+    }
+
+    return null;
+  }
+}

arrow/src/test/java/org/apache/iceberg/arrow/vectorized/TestVectorizedReaderBuilder.java

@@ -19,7 +19,6 @@
 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;
@@ -38,7 +37,7 @@
 public class TestVectorizedReaderBuilder {
 
   @Test
-  public void testVariantNotSupportedInVectorizedReads() {
+  public void testVariantSupportedInVectorizedReads() {
     Schema icebergSchema =
         new Schema(
             NestedField.required(1, "id", IntegerType.get()),
@@ -50,10 +49,9 @@ public void testVariantNotSupportedInVectorizedReads() {
         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");
+    assertThatNoException()
+        .isThrownBy(
+            () -> TypeWithSchemaVisitor.visit(icebergSchema.asStruct(), parquetSchema, builder));
   }
 
   @Test

spark/v4.0/spark-extensions/src/test/java/org/apache/iceberg/spark/extensions/TestSnapshotTableProcedure.java

@@ -394,6 +394,11 @@ public void testSnapshotWithVariant() throws IOException {
             catalogName, SOURCE_NAME, tableName);
     assertThat(result).as("Should have added one file").isEqualTo(1L);
 
+    // Disable vectorized reads for variant imported from non-Iceberg files (not yet supported)
+    sql(
+        "ALTER TABLE %s SET TBLPROPERTIES ('read.parquet.vectorization.enabled'='false')",
+        tableName);
+
     assertEquals(
         "Should have expected rows",
         ImmutableList.of(row(1L, 123)),
@@ -420,6 +425,11 @@ public void testSnapshotPartitionedWithVariant() throws IOException {
             catalogName, SOURCE_NAME, tableName);
     assertThat(result).as("Should have added two files").isEqualTo(2L);
 
+    // Disable vectorized reads for variant imported from non-Iceberg files (not yet supported)
+    sql(
+        "ALTER TABLE %s SET TBLPROPERTIES ('read.parquet.vectorization.enabled'='false')",
+        tableName);
+
     assertEquals(
         "Should have expected rows",
         ImmutableList.of(row(123, 1L), row(456, 2L)),

spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnVectorBuilder.java

@@ -27,7 +27,9 @@
 class ColumnVectorBuilder {
 
   public ColumnVector build(VectorHolder holder, int numRows) {
-    if (holder.isDummy()) {
+    if (holder instanceof VectorHolder.VariantVectorHolder) {
+      return new VariantColumnVector((VectorHolder.VariantVectorHolder) holder);
+    } else if (holder.isDummy()) {
       if (holder instanceof VectorHolder.DeletedVectorHolder) {
         return new DeletedColumnVector(Types.BooleanType.get());
       } else if (holder instanceof ConstantVectorHolder) {

spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnVectorWithFilter.java

@@ -20,6 +20,7 @@
 
 import org.apache.spark.sql.types.Decimal;
 import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.types.VariantType;
 import org.apache.spark.sql.vectorized.ColumnVector;
 import org.apache.spark.sql.vectorized.ColumnarArray;
 import org.apache.spark.sql.vectorized.ColumnarMap;
@@ -143,6 +144,12 @@ public ColumnVector getChild(int ordinal) {
             for (int index = 0; index < structType.length(); index++) {
               children[index] = new ColumnVectorWithFilter(delegate.getChild(index), rowIdMapping);
             }
+          } else if (dataType() instanceof VariantType) {
+            this.children =
+                new ColumnVectorWithFilter[] {
+                  new ColumnVectorWithFilter(delegate.getChild(0), rowIdMapping),
+                  new ColumnVectorWithFilter(delegate.getChild(1), rowIdMapping)
+                };
           } else {
             throw new UnsupportedOperationException("Unsupported nested type: " + dataType());
           }