Parquet: Fix NPE in ParquetAvroWriter when schema contains variant type

parquet/src/main/java/org/apache/iceberg/parquet/ParquetAvroWriter.java

@@ -103,6 +103,11 @@ public ParquetValueWriter<?> map(
           ParquetValueWriters.option(valueType, valueD, valueWriter));
     }
 
+    @Override
+    public ParquetValueWriter<?> variant(GroupType variant) {
+      throw new UnsupportedOperationException("Avro writer does not support variant types");
+    }
+
     @Override
     public ParquetValueWriter<?> primitive(PrimitiveType primitive) {
       ColumnDescriptor desc = type.getColumnDescription(currentPath());

parquet/src/test/java/org/apache/iceberg/parquet/TestParquet.java

@@ -30,6 +30,7 @@
 import static org.apache.iceberg.types.Types.NestedField.optional;
 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;
@@ -57,14 +58,17 @@
 import org.apache.iceberg.types.Types;
 import org.apache.iceberg.types.Types.IntegerType;
 import org.apache.iceberg.util.Pair;
+import org.apache.iceberg.variants.Variant;
 import org.apache.parquet.avro.AvroParquetWriter;
 import org.apache.parquet.column.statistics.Statistics;
 import org.apache.parquet.hadoop.ParquetFileReader;
 import org.apache.parquet.hadoop.ParquetWriter;
 import org.apache.parquet.hadoop.metadata.BlockMetaData;
 import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
 import org.apache.parquet.io.LocalOutputFile;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
 import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.io.TempDir;
 
@@ -314,6 +318,26 @@ public void testFooterMetricsWithNameMappingForFileWithoutIds() throws IOExcepti
     }
   }
 
+  @Test
+  public void testAvroWriterRejectsVariantType() {
+    MessageType schema =
+        org.apache.parquet.schema.Types.buildMessage()
+            .optional(PrimitiveTypeName.INT32)
+            .named("id")
+            .optionalGroup()
+            .as(LogicalTypeAnnotation.variantType(Variant.VARIANT_SPEC_VERSION))
+            .required(PrimitiveTypeName.BINARY)
+            .named("metadata")
+            .required(PrimitiveTypeName.BINARY)
+            .named("value")
+            .named("v")
+            .named("table");
+
+    assertThatThrownBy(() -> ParquetAvroWriter.buildWriter(schema))
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessage("Avro writer does not support variant types");
+  }
+
   private Pair<File, Long> generateFile(
       Function<MessageType, ParquetValueWriter<?>> createWriterFunc,
       int desiredRecordCount,