Spark: Fix NPE when reading/writing variant type with ORC format

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

@@ -123,7 +123,7 @@ public OrcValueReader<?> primitive(Type.PrimitiveType iPrimitive, TypeDescriptio
         case STRING:
           return SparkOrcValueReaders.utf8String();
         case BINARY:
-          if (Type.TypeID.UUID == iPrimitive.typeId()) {
+          if (iPrimitive != null && Type.TypeID.UUID == iPrimitive.typeId()) {
             return SparkOrcValueReaders.uuids();
           }
           return OrcValueReaders.bytes();

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

@@ -113,7 +113,7 @@ public OrcValueWriter<?> primitive(Type.PrimitiveType iPrimitive, TypeDescriptio
         case DOUBLE:
           return GenericOrcWriters.doubles(ORCSchemaUtil.fieldId(primitive));
         case BINARY:
-          if (Type.TypeID.UUID == iPrimitive.typeId()) {
+          if (iPrimitive != null && Type.TypeID.UUID == iPrimitive.typeId()) {
             return SparkOrcValueWriters.uuids();
           }
           return GenericOrcWriters.byteArrays();

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

@@ -156,7 +156,7 @@ public Converter primitive(Type.PrimitiveType iPrimitive, TypeDescription primit
           break;
         case BINARY:
           primitiveValueReader =
-              Type.TypeID.UUID == iPrimitive.typeId()
+              iPrimitive != null && Type.TypeID.UUID == iPrimitive.typeId()
                   ? SparkOrcValueReaders.uuids()
                   : OrcValueReaders.bytes();
           break;

spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java

@@ -20,6 +20,7 @@
 
 import static org.apache.iceberg.types.Types.NestedField.optional;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.io.File;
 import java.io.IOException;
@@ -56,4 +57,24 @@ public void splitOffsets() throws IOException {
     writer.close();
     assertThat(writer.splitOffsets()).as("Split offsets not present").isNotEmpty();
   }
+
+  @Test
+  public void variantIsNotSupported() throws IOException {
+    Schema variantSchema =
+        new Schema(
+            optional(1, "id", Types.IntegerType.get()),
+            optional(2, "payload", Types.VariantType.get()));
+
+    File testFile = File.createTempFile("junit", null, temp.toFile());
+    assertThat(testFile.delete()).as("Delete should succeed").isTrue();
+
+    assertThatThrownBy(
+            () ->
+                ORC.write(Files.localOutput(testFile))
+                    .createWriterFunc(SparkOrcWriter::new)
+                    .schema(variantSchema)
+                    .build())
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessage("Variant is not supported");
+  }
 }

spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java

@@ -30,18 +30,30 @@
 import java.util.List;
 import org.apache.iceberg.Files;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.orc.GenericOrcWriter;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.orc.ORC;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
 import org.apache.iceberg.spark.data.vectorized.VectorizedSparkOrcReaders;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.Variants;
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.vectorized.ColumnarBatch;
 import org.junit.jupiter.api.Test;
 
 public class TestSparkOrcReader extends AvroDataTestBase {
+  private static final Schema VARIANT_SCHEMA =
+      new Schema(
+          required(100, "id", Types.LongType.get()),
+          required(101, "payload", Types.VariantType.get()));
+
   @Override
   protected void writeAndValidate(Schema schema) throws IOException {
     final Iterable<InternalRow> expected = RandomData.generateSpark(schema, 100, 0L);
@@ -124,4 +136,60 @@ public void testUnknownMapType() {
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessageStartingWith("Cannot create MapType with unknown value type");
   }
+
+  @Test
+  public void variantIsNotSupported() throws IOException {
+    File testFile = writeVariantOrcFile();
+
+    try (CloseableIterable<InternalRow> reader =
+        ORC.read(Files.localInput(testFile))
+            .project(VARIANT_SCHEMA)
+            .createReaderFunc(readOrcSchema -> new SparkOrcReader(VARIANT_SCHEMA, readOrcSchema))
+            .build()) {
+      assertThatThrownBy(reader::iterator)
+          .isInstanceOf(UnsupportedOperationException.class)
+          .hasMessage("Variant is not supported");
+    }
+  }
+
+  @Test
+  public void variantIsNotSupportedWhenVectorized() throws IOException {
+    File testFile = writeVariantOrcFile();
+
+    try (CloseableIterable<ColumnarBatch> reader =
+        ORC.read(Files.localInput(testFile))
+            .project(VARIANT_SCHEMA)
+            .createBatchedReaderFunc(
+                readOrcSchema ->
+                    VectorizedSparkOrcReaders.buildReader(
+                        VARIANT_SCHEMA, readOrcSchema, ImmutableMap.of()))
+            .build()) {
+      assertThatThrownBy(reader::iterator)
+          .isInstanceOf(UnsupportedOperationException.class)
+          .hasMessage("Variant is not supported");
+    }
+  }
+
+  private File writeVariantOrcFile() throws IOException {
+    File testFile = File.createTempFile("junit", null, temp.toFile());
+    assertThat(testFile.delete()).as("Delete should succeed").isTrue();
+
+    VariantMetadata metadata = Variants.metadata("name");
+    ShreddedObject payload = Variants.object(metadata);
+    payload.put("name", Variants.of("alice"));
+
+    GenericRecord record = GenericRecord.create(VARIANT_SCHEMA);
+    record.setField("id", 1L);
+    record.setField("payload", Variant.of(metadata, payload));
+
+    try (FileAppender<Record> writer =
+        ORC.write(Files.localOutput(testFile))
+            .createWriterFunc(GenericOrcWriter::buildWriter)
+            .schema(VARIANT_SCHEMA)
+            .build()) {
+      writer.add(record);
+    }
+
+    return testFile;
+  }
 }

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcReader.java

@@ -123,7 +123,7 @@ public OrcValueReader<?> primitive(Type.PrimitiveType iPrimitive, TypeDescriptio
         case STRING:
           return SparkOrcValueReaders.utf8String();
         case BINARY:
-          if (Type.TypeID.UUID == iPrimitive.typeId()) {
+          if (iPrimitive != null && Type.TypeID.UUID == iPrimitive.typeId()) {
             return SparkOrcValueReaders.uuids();
           }
           return OrcValueReaders.bytes();

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcWriter.java

@@ -113,7 +113,7 @@ public OrcValueWriter<?> primitive(Type.PrimitiveType iPrimitive, TypeDescriptio
         case DOUBLE:
           return GenericOrcWriters.doubles(ORCSchemaUtil.fieldId(primitive));
         case BINARY:
-          if (Type.TypeID.UUID == iPrimitive.typeId()) {
+          if (iPrimitive != null && Type.TypeID.UUID == iPrimitive.typeId()) {
             return SparkOrcValueWriters.uuids();
           }
           return GenericOrcWriters.byteArrays();

spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/VectorizedSparkOrcReaders.java

@@ -156,7 +156,7 @@ public Converter primitive(Type.PrimitiveType iPrimitive, TypeDescription primit
           break;
         case BINARY:
           primitiveValueReader =
-              Type.TypeID.UUID == iPrimitive.typeId()
+              iPrimitive != null && Type.TypeID.UUID == iPrimitive.typeId()
                   ? SparkOrcValueReaders.uuids()
                   : OrcValueReaders.bytes();
           break;

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java

@@ -20,6 +20,7 @@
 
 import static org.apache.iceberg.types.Types.NestedField.optional;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.io.File;
 import java.io.IOException;
@@ -56,4 +57,24 @@ public void splitOffsets() throws IOException {
     writer.close();
     assertThat(writer.splitOffsets()).as("Split offsets not present").isNotEmpty();
   }
+
+  @Test
+  public void variantIsNotSupported() throws IOException {
+    Schema variantSchema =
+        new Schema(
+            optional(1, "id", Types.IntegerType.get()),
+            optional(2, "payload", Types.VariantType.get()));
+
+    File testFile = File.createTempFile("junit", null, temp.toFile());
+    assertThat(testFile.delete()).as("Delete should succeed").isTrue();
+
+    assertThatThrownBy(
+            () ->
+                ORC.write(Files.localOutput(testFile))
+                    .createWriterFunc(SparkOrcWriter::new)
+                    .schema(variantSchema)
+                    .build())
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessage("Variant is not supported");
+  }
 }

spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java

@@ -30,18 +30,30 @@
 import java.util.List;
 import org.apache.iceberg.Files;
 import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.data.orc.GenericOrcWriter;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.orc.ORC;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
 import org.apache.iceberg.spark.data.vectorized.VectorizedSparkOrcReaders;
 import org.apache.iceberg.types.Types;
+import org.apache.iceberg.variants.ShreddedObject;
+import org.apache.iceberg.variants.Variant;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.Variants;
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.vectorized.ColumnarBatch;
 import org.junit.jupiter.api.Test;
 
 public class TestSparkOrcReader extends AvroDataTestBase {
+  private static final Schema VARIANT_SCHEMA =
+      new Schema(
+          required(100, "id", Types.LongType.get()),
+          required(101, "payload", Types.VariantType.get()));
+
   @Override
   protected void writeAndValidate(Schema schema) throws IOException {
     final Iterable<InternalRow> expected = RandomData.generateSpark(schema, 100, 0L);
@@ -124,4 +136,60 @@ public void testUnknownMapType() {
         .isInstanceOf(IllegalArgumentException.class)
         .hasMessageStartingWith("Cannot create MapType with unknown value type");
   }
+
+  @Test
+  public void variantIsNotSupported() throws IOException {
+    File testFile = writeVariantOrcFile();
+
+    try (CloseableIterable<InternalRow> reader =
+        ORC.read(Files.localInput(testFile))
+            .project(VARIANT_SCHEMA)
+            .createReaderFunc(readOrcSchema -> new SparkOrcReader(VARIANT_SCHEMA, readOrcSchema))
+            .build()) {
+      assertThatThrownBy(reader::iterator)
+          .isInstanceOf(UnsupportedOperationException.class)
+          .hasMessage("Variant is not supported");
+    }
+  }
+
+  @Test
+  public void variantIsNotSupportedWhenVectorized() throws IOException {
+    File testFile = writeVariantOrcFile();
+
+    try (CloseableIterable<ColumnarBatch> reader =
+        ORC.read(Files.localInput(testFile))
+            .project(VARIANT_SCHEMA)
+            .createBatchedReaderFunc(
+                readOrcSchema ->
+                    VectorizedSparkOrcReaders.buildReader(
+                        VARIANT_SCHEMA, readOrcSchema, ImmutableMap.of()))
+            .build()) {
+      assertThatThrownBy(reader::iterator)
+          .isInstanceOf(UnsupportedOperationException.class)
+          .hasMessage("Variant is not supported");
+    }
+  }
+
+  private File writeVariantOrcFile() throws IOException {
+    File testFile = File.createTempFile("junit", null, temp.toFile());
+    assertThat(testFile.delete()).as("Delete should succeed").isTrue();
+
+    VariantMetadata metadata = Variants.metadata("name");
+    ShreddedObject payload = Variants.object(metadata);
+    payload.put("name", Variants.of("alice"));
+
+    GenericRecord record = GenericRecord.create(VARIANT_SCHEMA);
+    record.setField("id", 1L);
+    record.setField("payload", Variant.of(metadata, payload));
+
+    try (FileAppender<Record> writer =
+        ORC.write(Files.localOutput(testFile))
+            .createWriterFunc(GenericOrcWriter::buildWriter)
+            .schema(VARIANT_SCHEMA)
+            .build()) {
+      writer.add(record);
+    }
+
+    return testFile;
+  }
 }

spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcReader.java

@@ -123,7 +123,7 @@ public OrcValueReader<?> primitive(Type.PrimitiveType iPrimitive, TypeDescriptio
         case STRING:
           return SparkOrcValueReaders.utf8String();
         case BINARY:
-          if (Type.TypeID.UUID == iPrimitive.typeId()) {
+          if (iPrimitive != null && Type.TypeID.UUID == iPrimitive.typeId()) {
             return SparkOrcValueReaders.uuids();
           }
           return OrcValueReaders.bytes();

spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcWriter.java

@@ -113,7 +113,7 @@ public OrcValueWriter<?> primitive(Type.PrimitiveType iPrimitive, TypeDescriptio
         case DOUBLE:
           return GenericOrcWriters.doubles(ORCSchemaUtil.fieldId(primitive));
         case BINARY:
-          if (Type.TypeID.UUID == iPrimitive.typeId()) {
+          if (iPrimitive != null && Type.TypeID.UUID == iPrimitive.typeId()) {
             return SparkOrcValueWriters.uuids();
           }
           return GenericOrcWriters.byteArrays();

spark/v4.2/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/VectorizedSparkOrcReaders.java

@@ -156,7 +156,7 @@ public Converter primitive(Type.PrimitiveType iPrimitive, TypeDescription primit
           break;
         case BINARY:
           primitiveValueReader =
-              Type.TypeID.UUID == iPrimitive.typeId()
+              iPrimitive != null && Type.TypeID.UUID == iPrimitive.typeId()
                   ? SparkOrcValueReaders.uuids()
                   : OrcValueReaders.bytes();
           break;

spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java

@@ -20,6 +20,7 @@
 
 import static org.apache.iceberg.types.Types.NestedField.optional;
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.io.File;
 import java.io.IOException;
@@ -56,4 +57,24 @@ public void splitOffsets() throws IOException {
     writer.close();
     assertThat(writer.splitOffsets()).as("Split offsets not present").isNotEmpty();
   }
+
+  @Test
+  public void variantIsNotSupported() throws IOException {
+    Schema variantSchema =
+        new Schema(
+            optional(1, "id", Types.IntegerType.get()),
+            optional(2, "payload", Types.VariantType.get()));
+
+    File testFile = File.createTempFile("junit", null, temp.toFile());
+    assertThat(testFile.delete()).as("Delete should succeed").isTrue();
+
+    assertThatThrownBy(
+            () ->
+                ORC.write(Files.localOutput(testFile))
+                    .createWriterFunc(SparkOrcWriter::new)
+                    .schema(variantSchema)
+                    .build())
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessage("Variant is not supported");
+  }
 }