This is an automated email from the ASF dual-hosted git repository.
pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 4e2a9bda37 ORC: Fix NPE when reading/writing variant type with ORC
format (#18387)
4e2a9bda37 is described below
commit 4e2a9bda3705c55bf37473c7d728d4ffecb698f6
Author: Akash R Nilugal <[email protected]>
AuthorDate: Fri Nov 6 18:05:46 2026 +0530
ORC: Fix NPE when reading/writing variant type with ORC format (#18387)
---
.../flink/data/TestFlinkOrcReaderWriter.java | 46 +++++++++++++++
.../flink/data/TestFlinkOrcReaderWriter.java | 46 +++++++++++++++
.../flink/data/TestFlinkOrcReaderWriter.java | 46 +++++++++++++++
.../iceberg/orc/OrcSchemaWithTypeVisitor.java | 6 +-
.../apache/iceberg/spark/data/TestOrcWrite.java | 21 +++++++
.../iceberg/spark/data/TestSparkOrcReader.java | 68 ++++++++++++++++++++++
.../apache/iceberg/spark/data/TestOrcWrite.java | 21 +++++++
.../iceberg/spark/data/TestSparkOrcReader.java | 68 ++++++++++++++++++++++
.../apache/iceberg/spark/data/TestOrcWrite.java | 21 +++++++
.../iceberg/spark/data/TestSparkOrcReader.java | 68 ++++++++++++++++++++++
10 files changed, 406 insertions(+), 5 deletions(-)
diff --git
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java
index b7b0a54156..e1df627d87 100644
---
a/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java
+++
b/flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java
@@ -18,7 +18,9 @@
*/
package org.apache.iceberg.flink.data;
+import static org.apache.iceberg.types.Types.NestedField.required;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.File;
import java.io.IOException;
@@ -29,6 +31,7 @@ import org.apache.flink.table.types.logical.RowType;
import org.apache.iceberg.Files;
import org.apache.iceberg.Schema;
import org.apache.iceberg.data.DataTestBase;
+import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.RandomGenericData;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.data.orc.GenericOrcReader;
@@ -39,6 +42,12 @@ 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.Lists;
+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.junit.jupiter.api.Test;
public class TestFlinkOrcReaderWriter extends DataTestBase {
private static final int NUM_RECORDS = 100;
@@ -120,4 +129,41 @@ public class TestFlinkOrcReaderWriter extends DataTestBase
{
assertThat(records).isExhausted();
}
}
+
+ @Test
+ public void variantIsNotSupported() throws IOException {
+ Schema variantSchema =
+ new Schema(
+ required(100, "id", Types.LongType.get()),
+ required(101, "payload", Types.VariantType.get()));
+
+ File testFile = File.createTempFile("junit", null, temp.toFile());
+ assertThat(testFile.delete()).isTrue();
+
+ VariantMetadata metadata = Variants.metadata("name");
+ ShreddedObject payload = Variants.object(metadata);
+ payload.put("name", Variants.of("alice"));
+
+ GenericRecord record = GenericRecord.create(variantSchema);
+ record.setField("id", 1L);
+ record.setField("payload", Variant.of(metadata, payload));
+
+ try (FileAppender<Record> writer =
+ ORC.write(Files.localOutput(testFile))
+ .schema(variantSchema)
+ .createWriterFunc(GenericOrcWriter::buildWriter)
+ .build()) {
+ writer.add(record);
+ }
+
+ try (CloseableIterable<RowData> reader =
+ ORC.read(Files.localInput(testFile))
+ .project(variantSchema)
+ .createReaderFunc(type -> new FlinkOrcReader(variantSchema, type))
+ .build()) {
+ assertThatThrownBy(reader::iterator)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Variant is not supported");
+ }
+ }
}
diff --git
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java
index b7b0a54156..e1df627d87 100644
---
a/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java
+++
b/flink/v2.2/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java
@@ -18,7 +18,9 @@
*/
package org.apache.iceberg.flink.data;
+import static org.apache.iceberg.types.Types.NestedField.required;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.File;
import java.io.IOException;
@@ -29,6 +31,7 @@ import org.apache.flink.table.types.logical.RowType;
import org.apache.iceberg.Files;
import org.apache.iceberg.Schema;
import org.apache.iceberg.data.DataTestBase;
+import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.RandomGenericData;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.data.orc.GenericOrcReader;
@@ -39,6 +42,12 @@ 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.Lists;
+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.junit.jupiter.api.Test;
public class TestFlinkOrcReaderWriter extends DataTestBase {
private static final int NUM_RECORDS = 100;
@@ -120,4 +129,41 @@ public class TestFlinkOrcReaderWriter extends DataTestBase
{
assertThat(records).isExhausted();
}
}
+
+ @Test
+ public void variantIsNotSupported() throws IOException {
+ Schema variantSchema =
+ new Schema(
+ required(100, "id", Types.LongType.get()),
+ required(101, "payload", Types.VariantType.get()));
+
+ File testFile = File.createTempFile("junit", null, temp.toFile());
+ assertThat(testFile.delete()).isTrue();
+
+ VariantMetadata metadata = Variants.metadata("name");
+ ShreddedObject payload = Variants.object(metadata);
+ payload.put("name", Variants.of("alice"));
+
+ GenericRecord record = GenericRecord.create(variantSchema);
+ record.setField("id", 1L);
+ record.setField("payload", Variant.of(metadata, payload));
+
+ try (FileAppender<Record> writer =
+ ORC.write(Files.localOutput(testFile))
+ .schema(variantSchema)
+ .createWriterFunc(GenericOrcWriter::buildWriter)
+ .build()) {
+ writer.add(record);
+ }
+
+ try (CloseableIterable<RowData> reader =
+ ORC.read(Files.localInput(testFile))
+ .project(variantSchema)
+ .createReaderFunc(type -> new FlinkOrcReader(variantSchema, type))
+ .build()) {
+ assertThatThrownBy(reader::iterator)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Variant is not supported");
+ }
+ }
}
diff --git
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java
index b7b0a54156..e1df627d87 100644
---
a/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java
+++
b/flink/v2.3/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkOrcReaderWriter.java
@@ -18,7 +18,9 @@
*/
package org.apache.iceberg.flink.data;
+import static org.apache.iceberg.types.Types.NestedField.required;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import java.io.File;
import java.io.IOException;
@@ -29,6 +31,7 @@ import org.apache.flink.table.types.logical.RowType;
import org.apache.iceberg.Files;
import org.apache.iceberg.Schema;
import org.apache.iceberg.data.DataTestBase;
+import org.apache.iceberg.data.GenericRecord;
import org.apache.iceberg.data.RandomGenericData;
import org.apache.iceberg.data.Record;
import org.apache.iceberg.data.orc.GenericOrcReader;
@@ -39,6 +42,12 @@ 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.Lists;
+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.junit.jupiter.api.Test;
public class TestFlinkOrcReaderWriter extends DataTestBase {
private static final int NUM_RECORDS = 100;
@@ -120,4 +129,41 @@ public class TestFlinkOrcReaderWriter extends DataTestBase
{
assertThat(records).isExhausted();
}
}
+
+ @Test
+ public void variantIsNotSupported() throws IOException {
+ Schema variantSchema =
+ new Schema(
+ required(100, "id", Types.LongType.get()),
+ required(101, "payload", Types.VariantType.get()));
+
+ File testFile = File.createTempFile("junit", null, temp.toFile());
+ assertThat(testFile.delete()).isTrue();
+
+ VariantMetadata metadata = Variants.metadata("name");
+ ShreddedObject payload = Variants.object(metadata);
+ payload.put("name", Variants.of("alice"));
+
+ GenericRecord record = GenericRecord.create(variantSchema);
+ record.setField("id", 1L);
+ record.setField("payload", Variant.of(metadata, payload));
+
+ try (FileAppender<Record> writer =
+ ORC.write(Files.localOutput(testFile))
+ .schema(variantSchema)
+ .createWriterFunc(GenericOrcWriter::buildWriter)
+ .build()) {
+ writer.add(record);
+ }
+
+ try (CloseableIterable<RowData> reader =
+ ORC.read(Files.localInput(testFile))
+ .project(variantSchema)
+ .createReaderFunc(type -> new FlinkOrcReader(variantSchema, type))
+ .build()) {
+ assertThatThrownBy(reader::iterator)
+ .isInstanceOf(UnsupportedOperationException.class)
+ .hasMessage("Variant is not supported");
+ }
+ }
}
diff --git
a/orc/src/main/java/org/apache/iceberg/orc/OrcSchemaWithTypeVisitor.java
b/orc/src/main/java/org/apache/iceberg/orc/OrcSchemaWithTypeVisitor.java
index 222a00c3e1..c701ab7099 100644
--- a/orc/src/main/java/org/apache/iceberg/orc/OrcSchemaWithTypeVisitor.java
+++ b/orc/src/main/java/org/apache/iceberg/orc/OrcSchemaWithTypeVisitor.java
@@ -90,11 +90,7 @@ public abstract class OrcSchemaWithTypeVisitor<T> {
"Invalid variant metadata fields: %s",
String.join(", ", names));
- List<TypeDescription> children = variant.getChildren();
- T metadataResult = visit((Type) null, children.get(0), visitor);
- T valueResult = visit((Type) null, children.get(1), visitor);
-
- return visitor.variant(iVariant, variant, metadataResult, valueResult);
+ return visitor.variant(iVariant, variant, null, null);
}
public T record(
diff --git
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java
index e149e57e81..55656be6c0 100644
---
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java
+++
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java
@@ -20,6 +20,7 @@ package org.apache.iceberg.spark.data;
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 class TestOrcWrite {
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");
+ }
}
diff --git
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java
index 3fcfe6845c..3b4493638d 100644
---
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java
+++
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java
@@ -30,6 +30,9 @@ import java.util.Iterator;
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;
@@ -37,11 +40,20 @@ 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 class TestSparkOrcReader extends AvroDataTestBase {
.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;
+ }
}
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java
index e149e57e81..55656be6c0 100644
---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java
+++
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java
@@ -20,6 +20,7 @@ package org.apache.iceberg.spark.data;
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 class TestOrcWrite {
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");
+ }
}
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java
index 3fcfe6845c..3b4493638d 100644
---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java
+++
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java
@@ -30,6 +30,9 @@ import java.util.Iterator;
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;
@@ -37,11 +40,20 @@ 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 class TestSparkOrcReader extends AvroDataTestBase {
.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;
+ }
}
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java
index e149e57e81..55656be6c0 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestOrcWrite.java
@@ -20,6 +20,7 @@ package org.apache.iceberg.spark.data;
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 class TestOrcWrite {
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");
+ }
}
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java
index 3fcfe6845c..3b4493638d 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReader.java
@@ -30,6 +30,9 @@ import java.util.Iterator;
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;
@@ -37,11 +40,20 @@ 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 class TestSparkOrcReader extends AvroDataTestBase {
.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;
+ }
}
