Parquet: Fix variant metrics crash when value column has no stats (apache#16585)

* Parquet: Fix variant metrics crash when value column has no stats

* Fixup after 16327

Author: nssalian

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

@@ -477,6 +477,11 @@ public Iterable<ParquetVariantUtil.VariantMetrics> value(
           return typedResult;
         }
 
+        if (Iterables.isEmpty(valueResult)) {
+          // missing value stats invalidate typed bounds
+          return ImmutableList.of();
+        }
+
         ParquetVariantUtil.VariantMetrics valueMetrics = Iterables.getOnlyElement(valueResult);
         if (typedResult != null && valueMetrics.valueCount() == valueMetrics.nullCount()) {
           // all the variant-encoded values are null, so the typed stats can be used

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

@@ -26,14 +26,17 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.Metrics;
+import org.apache.iceberg.MetricsConfig;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.data.parquet.InternalWriter;
 import org.apache.iceberg.inmemory.InMemoryOutputFile;
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.types.Conversions;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
@@ -48,6 +51,10 @@
 import org.apache.iceberg.variants.VariantTestUtil;
 import org.apache.iceberg.variants.VariantValue;
 import org.apache.iceberg.variants.Variants;
+import org.apache.parquet.column.ParquetProperties;
+import org.apache.parquet.hadoop.ParquetFileWriter;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.apache.parquet.schema.MessageType;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.FieldSource;
@@ -479,6 +486,46 @@ public void testShreddedObjectFieldTypeMismatch() throws IOException {
         .isEqualTo(Map.of(1, Types.LongType.get(), 2, Types.VariantType.get()));
   }
 
+  @Test
+  public void testShreddedValueColumnWithEmptyStats() throws IOException {
+    // typed bounds are dropped when value-column stats are missing on a shredded variant
+    OutputFile out = new InMemoryOutputFile();
+    GenericRecord record = GenericRecord.create(SCHEMA);
+
+    VariantShreddingFunction shredding =
+        (id, name) -> ParquetVariantUtil.toParquetSchema(Variants.of((byte) 0));
+    MessageType parquetSchema = ParquetSchemaUtil.convert(SCHEMA, "table", shredding);
+    ParquetProperties props = ParquetProperties.builder().withStatisticsEnabled(false).build();
+
+    // Parquet.write() cannot disable stats on variant sub-columns (no field IDs)
+    ParquetWriter<Record> writer =
+        new ParquetWriter<>(
+            new Configuration(),
+            out,
+            SCHEMA,
+            parquetSchema,
+            1024,
+            ImmutableMap.of(),
+            (s, m) -> InternalWriter.create(s.asStruct(), m),
+            CompressionCodecName.SNAPPY,
+            props,
+            MetricsConfig.getDefault(),
+            ParquetFileWriter.Mode.CREATE,
+            null,
+            false);
+
+    try (writer) {
+      record.setField("id", 1L);
+      record.setField("var", Variant.of(EMPTY, Variants.of((byte) 5)));
+      writer.add(record);
+    }
+
+    Metrics metrics = writer.metrics();
+    assertThat(metrics.recordCount()).isEqualTo(1L);
+    assertThat(metrics.lowerBounds()).doesNotContainKey(2);
+    assertThat(metrics.upperBounds()).doesNotContainKey(2);
+  }
+
   private Metrics writeParquet(VariantShreddingFunction shredding, Variant... variants)
       throws IOException {
     OutputFile out = new InMemoryOutputFile();