From c8454fd8d0a2dfab2d22c0cc4f37974f732c95fa Mon Sep 17 00:00:00 2001
From: Neelesh Salian <n_salian@apple.com>
Date: Wed, 29 Jul 2026 15:09:20 -0700
Subject: [PATCH 1/4] Parquet, Spark, Flink: Type uniformity for variant
 shredding

---
 .../flink/TestFlinkVariantShreddingType.java  |  39 ++-----
 .../parquet/VariantShreddingAnalyzer.java     | 110 +++++++-----------
 .../parquet/TestVariantShreddingAnalyzer.java | 110 +++++++++++++++++-
 .../spark/variant/TestVariantShredding.java   |  31 ++---
 .../spark/variant/TestVariantShredding.java   |  31 ++---
 5 files changed, 168 insertions(+), 153 deletions(-)

diff --git a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java
index c4bd2efb4960..6cb035ddff9c 100644
--- a/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java
+++ b/flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java
@@ -124,7 +124,7 @@ public void testExcludingNullValue() throws IOException {
   }
 
   @TestTemplate
-  public void testConsistentType() throws IOException {
+  public void testInconsistentTypeNotShredded() throws IOException {
     String values =
         """
             (1, parse_json('{"age": "25"}')),
@@ -133,18 +133,13 @@ public void testConsistentType() throws IOException {
             """;
     sql("INSERT INTO %s VALUES %s", TABLE_NAME, values);
 
-    GroupType age =
-        field(
-            "age",
-            shreddedPrimitive(
-                PrimitiveType.PrimitiveTypeName.BINARY, LogicalTypeAnnotation.stringType()));
-    GroupType address = variant("address", 2, Type.Repetition.REQUIRED, objectFields(age));
+    GroupType address = variant("address", 2, Type.Repetition.REQUIRED);
     MessageType expectedSchema = parquetSchema(address);
     verifyParquetSchema(icebergTable, expectedSchema);
   }
 
   @TestTemplate
-  public void testPrimitiveType() throws IOException {
+  public void testMixedPrimitiveTypeAtRootNotShredded() throws IOException {
     String values =
         """
             (1, parse_json('123')),
@@ -153,13 +148,7 @@ public void testPrimitiveType() throws IOException {
             """;
     sql("INSERT INTO %s VALUES %s", TABLE_NAME, values);
 
-    GroupType address =
-        variant(
-            "address",
-            2,
-            Type.Repetition.REQUIRED,
-            shreddedPrimitive(
-                PrimitiveType.PrimitiveTypeName.INT32, LogicalTypeAnnotation.intType(8, true)));
+    GroupType address = variant("address", 2, Type.Repetition.REQUIRED);
     MessageType expectedSchema = parquetSchema(address);
 
     assertThat(SimpleDataUtil.tableRecords(icebergTable)).hasSize(3);
@@ -167,7 +156,7 @@ public void testPrimitiveType() throws IOException {
   }
 
   @TestTemplate
-  public void testPrimitiveDecimalType() throws IOException {
+  public void testMixedDecimalAndStringAtRootNotShredded() throws IOException {
     String values =
         """
             (1, parse_json('123.56')),
@@ -176,13 +165,7 @@ public void testPrimitiveDecimalType() throws IOException {
             """;
     sql("INSERT INTO %s VALUES %s", TABLE_NAME, values);
 
-    GroupType address =
-        variant(
-            "address",
-            2,
-            Type.Repetition.REQUIRED,
-            shreddedPrimitive(
-                PrimitiveType.PrimitiveTypeName.INT32, LogicalTypeAnnotation.decimalType(2, 5)));
+    GroupType address = variant("address", 2, Type.Repetition.REQUIRED);
     MessageType expectedSchema = parquetSchema(address);
     assertThat(SimpleDataUtil.tableRecords(icebergTable)).hasSize(3);
     verifyParquetSchema(icebergTable, expectedSchema);
@@ -587,7 +570,7 @@ public void testInfrequentFieldPruning() throws IOException {
   }
 
   @TestTemplate
-  public void testMixedTypeTieBreaking() throws IOException {
+  public void testMixedTypeFieldNotShredded() throws IOException {
     StringBuilder valuesBuilder = new StringBuilder();
     for (int i = 1; i <= 10; i++) {
       if (i > 1) {
@@ -617,13 +600,7 @@ public void testMixedTypeTieBreaking() throws IOException {
 
     sql("INSERT INTO %s VALUES %s", TABLE_NAME, valuesBuilder.toString());
 
-    // 5 ints + 5 strings is a tie so STRING wins (higher TIE_BREAK_PRIORITY)
-    GroupType val =
-        field(
-            "val",
-            shreddedPrimitive(
-                PrimitiveType.PrimitiveTypeName.BINARY, LogicalTypeAnnotation.stringType()));
-    GroupType address = variant("address", 2, Type.Repetition.REQUIRED, objectFields(val));
+    GroupType address = variant("address", 2, Type.Repetition.REQUIRED);
     MessageType expectedSchema = parquetSchema(address);
 
     verifyParquetSchema(icebergTable, expectedSchema);
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java b/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
index 3e6694c38c05..c8e6ef26655a 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
@@ -51,10 +51,10 @@
  *
  * <ul>
  *   <li>Object fields are emitted in alphabetical order in the shredded schema.
- *   <li>Type selection picks the most common type with explicit tie-break priority (see {@link
- *       FieldInfo#TIE_BREAK_PRIORITY}), not enum ordinal.
+ *   <li>A field is admitted only if all observations fall into a single type family after numeric
+ *       widening; mixed-type fields remain in the residual {@code value}.
  *   <li>Integer types (INT8/16/32/64) and decimal types (DECIMAL4/8/16) are each promoted to the
- *       widest observed before competing with other types.
+ *       widest observed within their family.
  *   <li>Fields below {@code MIN_FIELD_FREQUENCY} are pruned. Above {@code MAX_SHREDDED_FIELDS}, the
  *       most frequent are kept with alphabetical tie-breaking.
  *   <li>Recursion into nested objects/arrays stops at {@code MAX_SHREDDING_DEPTH} (default 50).
@@ -96,7 +96,7 @@ public Type analyzeAndCreateSchema(List<T> bufferedRows, int variantFieldIndex)
     }
 
     PathNode root = buildPathTree(variantValues);
-    PhysicalType rootType = root.info.getMostCommonType();
+    PhysicalType rootType = root.info.admittedType();
     if (rootType == null) {
       return null;
     }
@@ -251,12 +251,12 @@ private static void traverseArray(PathNode node, VariantArray array, int depth)
   }
 
   private static Type buildFieldGroup(PathNode node) {
-    PhysicalType commonType = node.info.getMostCommonType();
-    if (commonType == null) {
+    PhysicalType admittedType = node.info.admittedType();
+    if (admittedType == null) {
       return null;
     }
 
-    Type typedValue = buildTypedValue(node, commonType);
+    Type typedValue = buildTypedValue(node, admittedType);
     if (typedValue == null) {
       return null;
     }
@@ -303,7 +303,7 @@ private static Type createArrayTypedValue(PathNode node) {
     if (elementNode == null) {
       return null;
     }
-    PhysicalType elementType = elementNode.info.getMostCommonType();
+    PhysicalType elementType = elementNode.info.admittedType();
     if (elementType == null) {
       return null;
     }
@@ -425,8 +425,8 @@ private static class FieldInfo {
     private int maxDecimalScale = 0;
     private int maxDecimalIntegerDigits = 0;
     private int observationCount = 0;
-    private boolean mostCommonComputed = false;
-    private PhysicalType mostCommonCached = null;
+    private boolean admittedTypeComputed = false;
+    private PhysicalType admittedTypeCached = null;
 
     private static final Map<PhysicalType, Integer> INTEGER_PRIORITY =
         ImmutableMap.of(
@@ -441,32 +441,8 @@ private static class FieldInfo {
             PhysicalType.DECIMAL8, 1,
             PhysicalType.DECIMAL16, 2);
 
-    /** Tie-break ordering when two physical types have equal counts. Higher value wins. */
-    private static final Map<PhysicalType, Integer> TIE_BREAK_PRIORITY =
-        ImmutableMap.<PhysicalType, Integer>builder()
-            .put(PhysicalType.BOOLEAN_TRUE, 0)
-            .put(PhysicalType.INT8, 1)
-            .put(PhysicalType.INT16, 2)
-            .put(PhysicalType.INT32, 3)
-            .put(PhysicalType.INT64, 4)
-            .put(PhysicalType.FLOAT, 5)
-            .put(PhysicalType.DOUBLE, 6)
-            .put(PhysicalType.DECIMAL4, 7)
-            .put(PhysicalType.DECIMAL8, 8)
-            .put(PhysicalType.DECIMAL16, 9)
-            .put(PhysicalType.DATE, 10)
-            .put(PhysicalType.TIME, 11)
-            .put(PhysicalType.TIMESTAMPTZ, 12)
-            .put(PhysicalType.TIMESTAMPNTZ, 13)
-            .put(PhysicalType.BINARY, 14)
-            .put(PhysicalType.STRING, 15)
-            .put(PhysicalType.TIMESTAMPTZ_NANOS, 16)
-            .put(PhysicalType.TIMESTAMPNTZ_NANOS, 17)
-            .put(PhysicalType.UUID, 18)
-            .buildOrThrow();
-
     void observe(VariantValue value) {
-      mostCommonComputed = false;
+      admittedTypeComputed = false;
       observationCount++;
       // Use BOOLEAN_TRUE for both TRUE/FALSE values
       PhysicalType type =
@@ -483,62 +459,54 @@ void observe(VariantValue value) {
       }
     }
 
-    PhysicalType getMostCommonType() {
-      if (mostCommonComputed) {
-        return mostCommonCached;
+    PhysicalType admittedType() {
+      if (admittedTypeComputed) {
+        return admittedTypeCached;
       }
 
-      Map<PhysicalType, Integer> combinedCounts = Maps.newHashMap();
-
-      int integerTotalCount = 0;
-      PhysicalType mostCapableInteger = null;
-
-      int decimalTotalCount = 0;
-      PhysicalType mostCapableDecimal = null;
+      Set<PhysicalType> families = Sets.newHashSet();
+      PhysicalType widestInteger = null;
+      PhysicalType widestDecimal = null;
 
       for (int i = 0; i < typeCounts.length; i++) {
-        int count = typeCounts[i];
-        if (count == 0) {
+        if (typeCounts[i] == 0) {
           continue;
         }
         PhysicalType type = PHYSICAL_TYPES[i];
 
         if (isIntegerType(type)) {
-          integerTotalCount += count;
-          if (mostCapableInteger == null
-              || INTEGER_PRIORITY.get(type) > INTEGER_PRIORITY.get(mostCapableInteger)) {
-            mostCapableInteger = type;
+          if (widestInteger == null
+              || INTEGER_PRIORITY.get(type) > INTEGER_PRIORITY.get(widestInteger)) {
+            widestInteger = type;
           }
         } else if (isDecimalType(type)) {
-          decimalTotalCount += count;
-          if (mostCapableDecimal == null
-              || DECIMAL_PRIORITY.get(type) > DECIMAL_PRIORITY.get(mostCapableDecimal)) {
-            mostCapableDecimal = type;
+          if (widestDecimal == null
+              || DECIMAL_PRIORITY.get(type) > DECIMAL_PRIORITY.get(widestDecimal)) {
+            widestDecimal = type;
           }
         } else {
-          combinedCounts.put(type, count);
+          families.add(type);
         }
       }
 
-      if (mostCapableInteger != null) {
-        combinedCounts.put(mostCapableInteger, integerTotalCount);
+      if (widestInteger != null) {
+        families.add(widestInteger);
+      }
+
+      if (widestDecimal != null) {
+        families.add(widestDecimal);
       }
 
-      if (mostCapableDecimal != null) {
-        combinedCounts.put(mostCapableDecimal, decimalTotalCount);
+      // Type-uniformity: admit only if exactly one family remains after widening.
+      if (families.size() != 1) {
+        admittedTypeCached = null;
+        admittedTypeComputed = true;
+        return null;
       }
 
-      // Pick the most common type with tie-breaking
-      mostCommonCached =
-          combinedCounts.entrySet().stream()
-              .max(
-                  Map.Entry.<PhysicalType, Integer>comparingByValue()
-                      .thenComparingInt(
-                          entry -> TIE_BREAK_PRIORITY.getOrDefault(entry.getKey(), -1)))
-              .map(Map.Entry::getKey)
-              .orElse(null);
-      mostCommonComputed = true;
-      return mostCommonCached;
+      admittedTypeCached = families.iterator().next();
+      admittedTypeComputed = true;
+      return admittedTypeCached;
     }
 
     private static boolean isIntegerType(PhysicalType type) {
diff --git a/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantShreddingAnalyzer.java b/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantShreddingAnalyzer.java
index fabcb9ae6cd1..3114c2febdf6 100644
--- a/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantShreddingAnalyzer.java
+++ b/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantShreddingAnalyzer.java
@@ -160,7 +160,7 @@ public void testFieldCapAllowsExistingFieldUpdates() {
 
     ShreddedObject row2 = Variants.object(meta);
     for (int i = 0; i < 10; i++) {
-      row2.put(fieldNames[i], Variants.of("text"));
+      row2.put(fieldNames[i], Variants.of(7));
     }
 
     ShreddedObject row3 = Variants.object(meta);
@@ -432,8 +432,6 @@ public void testLongArrayInFewRowsSurvivesPruning() {
           item.put("key", Variants.of(j));
           arr.add(item);
         }
-      } else {
-        arr.add(Variants.of("no_key"));
       }
       rows.add(arr);
     }
@@ -471,6 +469,112 @@ public void testUuidFieldIsTrackedAndShredded() {
         .isInstanceOf(LogicalTypeAnnotation.UUIDLogicalTypeAnnotation.class);
   }
 
+  @Test
+  public void testMixedPrimitiveTypesFieldNotShredded() {
+    VariantMetadata meta = Variants.metadata("mixed", "keep");
+    ShreddedObject row1 = Variants.object(meta);
+    row1.put("mixed", Variants.of(42));
+    row1.put("keep", Variants.of(1));
+    ShreddedObject row2 = Variants.object(meta);
+    row2.put("mixed", Variants.of("text"));
+    row2.put("keep", Variants.of(2));
+
+    DirectAnalyzer analyzer = new DirectAnalyzer();
+    Type schema = analyzer.analyzeAndCreateSchema(List.of(row1, row2), 0);
+
+    assertThat(schema).isNotNull().isInstanceOf(GroupType.class);
+    GroupType typedValue = (GroupType) schema;
+    assertThat(typedValue.containsField("mixed")).isFalse();
+    assertThat(typedValue.containsField("keep")).isTrue();
+  }
+
+  @Test
+  public void testIntegerWideningAdmitsField() {
+    VariantMetadata meta = Variants.metadata("n");
+    ShreddedObject row1 = Variants.object(meta);
+    row1.put("n", Variants.of(42));
+    ShreddedObject row2 = Variants.object(meta);
+    row2.put("n", Variants.of(5_000_000_000L));
+
+    DirectAnalyzer analyzer = new DirectAnalyzer();
+    Type schema = analyzer.analyzeAndCreateSchema(List.of(row1, row2), 0);
+
+    assertThat(schema).isNotNull().isInstanceOf(GroupType.class);
+    GroupType typedValue = (GroupType) schema;
+    assertThat(typedValue.containsField("n")).isTrue();
+    GroupType nGroup = typedValue.getType("n").asGroupType();
+    assertThat(nGroup.getType("typed_value").asPrimitiveType().getPrimitiveTypeName())
+        .isEqualTo(PrimitiveType.PrimitiveTypeName.INT64);
+  }
+
+  @Test
+  public void testIntAndDecimalAtSameFieldNotShredded() {
+    VariantMetadata meta = Variants.metadata("mixed", "keep");
+    ShreddedObject row1 = Variants.object(meta);
+    row1.put("mixed", Variants.of(42));
+    row1.put("keep", Variants.of(1));
+    ShreddedObject row2 = Variants.object(meta);
+    row2.put("mixed", Variants.of(new BigDecimal("3.14")));
+    row2.put("keep", Variants.of(2));
+
+    DirectAnalyzer analyzer = new DirectAnalyzer();
+    Type schema = analyzer.analyzeAndCreateSchema(List.of(row1, row2), 0);
+
+    assertThat(schema).isNotNull().isInstanceOf(GroupType.class);
+    GroupType typedValue = (GroupType) schema;
+    assertThat(typedValue.containsField("mixed")).isFalse();
+    assertThat(typedValue.containsField("keep")).isTrue();
+  }
+
+  @Test
+  public void testMixedObjectAndPrimitiveAtFieldNotShredded() {
+    VariantMetadata outerMeta = Variants.metadata("mixed", "keep");
+    VariantMetadata innerMeta = Variants.metadata("x");
+    ShreddedObject inner = Variants.object(innerMeta);
+    inner.put("x", Variants.of(1));
+
+    ShreddedObject row1 = Variants.object(outerMeta);
+    row1.put("mixed", inner);
+    row1.put("keep", Variants.of(1));
+    ShreddedObject row2 = Variants.object(outerMeta);
+    row2.put("mixed", Variants.of("hello"));
+    row2.put("keep", Variants.of(2));
+
+    DirectAnalyzer analyzer = new DirectAnalyzer();
+    Type schema = analyzer.analyzeAndCreateSchema(List.of(row1, row2), 0);
+
+    assertThat(schema).isNotNull().isInstanceOf(GroupType.class);
+    GroupType typedValue = (GroupType) schema;
+    assertThat(typedValue.containsField("mixed")).isFalse();
+    assertThat(typedValue.containsField("keep")).isTrue();
+  }
+
+  @Test
+  public void testArrayWithMixedElementTypesNotShredded() {
+    VariantMetadata meta = Variants.metadata("arr", "keep");
+
+    ValueArray arr1 = Variants.array();
+    arr1.add(Variants.of(1));
+    arr1.add(Variants.of(2));
+    ShreddedObject row1 = Variants.object(meta);
+    row1.put("arr", arr1);
+    row1.put("keep", Variants.of(1));
+
+    ValueArray arr2 = Variants.array();
+    arr2.add(Variants.of("text"));
+    ShreddedObject row2 = Variants.object(meta);
+    row2.put("arr", arr2);
+    row2.put("keep", Variants.of(2));
+
+    DirectAnalyzer analyzer = new DirectAnalyzer();
+    Type schema = analyzer.analyzeAndCreateSchema(List.of(row1, row2), 0);
+
+    assertThat(schema).isNotNull().isInstanceOf(GroupType.class);
+    GroupType typedValue = (GroupType) schema;
+    assertThat(typedValue.containsField("arr")).isFalse();
+    assertThat(typedValue.containsField("keep")).isTrue();
+  }
+
   /**
    * Builds 100 variant rows where "common" appears in every row and "rare" appears in only {@code
    * rareCount} rows (below MIN_FIELD_FREQUENCY = 0.10 when rareCount < 10).
diff --git a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/variant/TestVariantShredding.java b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/variant/TestVariantShredding.java
index 4b14f32fb13c..9b07e1372428 100644
--- a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/variant/TestVariantShredding.java
+++ b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/variant/TestVariantShredding.java
@@ -169,7 +169,7 @@ public void testExcludingNullValue() throws IOException {
   }
 
   @TestTemplate
-  public void testInconsistentType() throws IOException {
+  public void testInconsistentTypeNotShredded() throws IOException {
     spark.conf().set(SparkSQLProperties.SHRED_VARIANTS, "true");
 
     String values =
@@ -180,12 +180,7 @@ public void testInconsistentType() throws IOException {
             """;
     sql("INSERT INTO %s VALUES %s", tableName, values);
 
-    GroupType age =
-        field(
-            "age",
-            shreddedPrimitive(
-                PrimitiveType.PrimitiveTypeName.BINARY, LogicalTypeAnnotation.stringType()));
-    GroupType address = variant("address", 2, Type.Repetition.REQUIRED, objectFields(age));
+    GroupType address = variant("address", 2, Type.Repetition.REQUIRED);
     MessageType expectedSchema = parquetSchema(address);
 
     Table table = validationCatalog.loadTable(tableIdent);
@@ -218,20 +213,14 @@ public void testPrimitiveType() throws IOException {
   }
 
   @TestTemplate
-  public void testPrimitiveDecimalType() throws IOException {
+  public void testMixedPrimitiveTypesAtRootNotShredded() throws IOException {
     spark.conf().set(SparkSQLProperties.SHRED_VARIANTS, "true");
 
     String values =
         "(1, parse_json('123.56')), (2, parse_json('\"abc\"')), (3, parse_json('12.56'))";
     sql("INSERT INTO %s VALUES %s", tableName, values);
 
-    GroupType address =
-        variant(
-            "address",
-            2,
-            Type.Repetition.REQUIRED,
-            shreddedPrimitive(
-                PrimitiveType.PrimitiveTypeName.INT32, LogicalTypeAnnotation.decimalType(2, 5)));
+    GroupType address = variant("address", 2, Type.Repetition.REQUIRED);
     MessageType expectedSchema = parquetSchema(address);
 
     Table table = validationCatalog.loadTable(tableIdent);
@@ -785,7 +774,7 @@ public void testInfrequentFieldPruning() throws IOException {
   }
 
   @TestTemplate
-  public void testMixedTypeTieBreaking() throws IOException {
+  public void testMixedTypeFieldNotShredded() throws IOException {
     spark.conf().set(SparkSQLProperties.SHRED_VARIANTS, "true");
     spark.conf().set(SparkSQLProperties.VARIANT_INFERENCE_BUFFER_SIZE, "10");
 
@@ -802,19 +791,13 @@ public void testMixedTypeTieBreaking() throws IOException {
     }
     sql("INSERT INTO %s VALUES %s", tableName, valuesBuilder.toString());
 
-    // 5 ints + 5 strings is a tie so STRING wins (higher TIE_BREAK_PRIORITY)
-    GroupType val =
-        field(
-            "val",
-            shreddedPrimitive(
-                PrimitiveType.PrimitiveTypeName.BINARY, LogicalTypeAnnotation.stringType()));
-    GroupType address = variant("address", 2, Type.Repetition.REQUIRED, objectFields(val));
+    GroupType address = variant("address", 2, Type.Repetition.REQUIRED);
     MessageType expectedSchema = parquetSchema(address);
 
     Table table = validationCatalog.loadTable(tableIdent);
     verifyParquetSchema(table, expectedSchema);
 
-    // Verify data round-trips correctly
+    // Verify data round-trips correctly through the residual value column
     List<Object[]> rows =
         sql("SELECT id, variant_get(address, '$.val', 'string') FROM %s ORDER BY id", tableName);
     assertThat(rows).hasSize(10);
diff --git a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/variant/TestVariantShredding.java b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/variant/TestVariantShredding.java
index 4b14f32fb13c..9b07e1372428 100644
--- a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/variant/TestVariantShredding.java
+++ b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/variant/TestVariantShredding.java
@@ -169,7 +169,7 @@ public void testExcludingNullValue() throws IOException {
   }
 
   @TestTemplate
-  public void testInconsistentType() throws IOException {
+  public void testInconsistentTypeNotShredded() throws IOException {
     spark.conf().set(SparkSQLProperties.SHRED_VARIANTS, "true");
 
     String values =
@@ -180,12 +180,7 @@ public void testInconsistentType() throws IOException {
             """;
     sql("INSERT INTO %s VALUES %s", tableName, values);
 
-    GroupType age =
-        field(
-            "age",
-            shreddedPrimitive(
-                PrimitiveType.PrimitiveTypeName.BINARY, LogicalTypeAnnotation.stringType()));
-    GroupType address = variant("address", 2, Type.Repetition.REQUIRED, objectFields(age));
+    GroupType address = variant("address", 2, Type.Repetition.REQUIRED);
     MessageType expectedSchema = parquetSchema(address);
 
     Table table = validationCatalog.loadTable(tableIdent);
@@ -218,20 +213,14 @@ public void testPrimitiveType() throws IOException {
   }
 
   @TestTemplate
-  public void testPrimitiveDecimalType() throws IOException {
+  public void testMixedPrimitiveTypesAtRootNotShredded() throws IOException {
     spark.conf().set(SparkSQLProperties.SHRED_VARIANTS, "true");
 
     String values =
         "(1, parse_json('123.56')), (2, parse_json('\"abc\"')), (3, parse_json('12.56'))";
     sql("INSERT INTO %s VALUES %s", tableName, values);
 
-    GroupType address =
-        variant(
-            "address",
-            2,
-            Type.Repetition.REQUIRED,
-            shreddedPrimitive(
-                PrimitiveType.PrimitiveTypeName.INT32, LogicalTypeAnnotation.decimalType(2, 5)));
+    GroupType address = variant("address", 2, Type.Repetition.REQUIRED);
     MessageType expectedSchema = parquetSchema(address);
 
     Table table = validationCatalog.loadTable(tableIdent);
@@ -785,7 +774,7 @@ public void testInfrequentFieldPruning() throws IOException {
   }
 
   @TestTemplate
-  public void testMixedTypeTieBreaking() throws IOException {
+  public void testMixedTypeFieldNotShredded() throws IOException {
     spark.conf().set(SparkSQLProperties.SHRED_VARIANTS, "true");
     spark.conf().set(SparkSQLProperties.VARIANT_INFERENCE_BUFFER_SIZE, "10");
 
@@ -802,19 +791,13 @@ public void testMixedTypeTieBreaking() throws IOException {
     }
     sql("INSERT INTO %s VALUES %s", tableName, valuesBuilder.toString());
 
-    // 5 ints + 5 strings is a tie so STRING wins (higher TIE_BREAK_PRIORITY)
-    GroupType val =
-        field(
-            "val",
-            shreddedPrimitive(
-                PrimitiveType.PrimitiveTypeName.BINARY, LogicalTypeAnnotation.stringType()));
-    GroupType address = variant("address", 2, Type.Repetition.REQUIRED, objectFields(val));
+    GroupType address = variant("address", 2, Type.Repetition.REQUIRED);
     MessageType expectedSchema = parquetSchema(address);
 
     Table table = validationCatalog.loadTable(tableIdent);
     verifyParquetSchema(table, expectedSchema);
 
-    // Verify data round-trips correctly
+    // Verify data round-trips correctly through the residual value column
     List<Object[]> rows =
         sql("SELECT id, variant_get(address, '$.val', 'string') FROM %s ORDER BY id", tableName);
     assertThat(rows).hasSize(10);

From 4f717b6c6ac9b8a11e3ed11a45d705bf2198ae6f Mon Sep 17 00:00:00 2001
From: Neelesh Salian <n_salian@apple.com>
Date: Wed, 29 Jul 2026 15:53:12 -0700
Subject: [PATCH 2/4] Reduce cyclomatic complexity in variant admittedType

---
 .../parquet/VariantShreddingAnalyzer.java      | 18 ++++++++++--------
 1 file changed, 10 insertions(+), 8 deletions(-)

diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java b/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
index c8e6ef26655a..d7dca695aa77 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
@@ -475,15 +475,9 @@ PhysicalType admittedType() {
         PhysicalType type = PHYSICAL_TYPES[i];
 
         if (isIntegerType(type)) {
-          if (widestInteger == null
-              || INTEGER_PRIORITY.get(type) > INTEGER_PRIORITY.get(widestInteger)) {
-            widestInteger = type;
-          }
+          widestInteger = widerType(widestInteger, type, INTEGER_PRIORITY);
         } else if (isDecimalType(type)) {
-          if (widestDecimal == null
-              || DECIMAL_PRIORITY.get(type) > DECIMAL_PRIORITY.get(widestDecimal)) {
-            widestDecimal = type;
-          }
+          widestDecimal = widerType(widestDecimal, type, DECIMAL_PRIORITY);
         } else {
           families.add(type);
         }
@@ -509,6 +503,14 @@ PhysicalType admittedType() {
       return admittedTypeCached;
     }
 
+    private static PhysicalType widerType(
+        PhysicalType current, PhysicalType candidate, Map<PhysicalType, Integer> priority) {
+      if (current == null) {
+        return candidate;
+      }
+      return priority.get(candidate) > priority.get(current) ? candidate : current;
+    }
+
     private static boolean isIntegerType(PhysicalType type) {
       return type == PhysicalType.INT8
           || type == PhysicalType.INT16

From 63a1d4cacc3d7aafff5aaa76c4c03ec4d5514e67 Mon Sep 17 00:00:00 2001
From: Neelesh Salian <n_salian@apple.com>
Date: Thu, 30 Jul 2026 11:59:56 -0700
Subject: [PATCH 3/4] PR comments fixes for ranking and cleanup

---
 .../parquet/VariantShreddingAnalyzer.java     | 123 ++++++++----------
 .../parquet/TestVariantShreddingAnalyzer.java |  65 +++++++++
 2 files changed, 117 insertions(+), 71 deletions(-)

diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java b/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
index d7dca695aa77..a774ce06635e 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
@@ -25,7 +25,6 @@
 import java.util.PriorityQueue;
 import java.util.Set;
 import org.apache.iceberg.Schema;
-import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.relocated.com.google.common.collect.Maps;
 import org.apache.iceberg.relocated.com.google.common.collect.Sets;
@@ -51,10 +50,14 @@
  *
  * <ul>
  *   <li>Object fields are emitted in alphabetical order in the shredded schema.
- *   <li>A field is admitted only if all observations fall into a single type family after numeric
- *       widening; mixed-type fields remain in the residual {@code value}.
- *   <li>Integer types (INT8/16/32/64) and decimal types (DECIMAL4/8/16) are each promoted to the
- *       widest observed within their family.
+ *   <li>A field is admitted only if all its observations fall into a single type family after
+ *       numeric widening. Integer types (INT8/16/32/64) widen within their family; decimal types
+ *       (DECIMAL4/8/16) widen within theirs. All other physical types - including {@code FLOAT} vs
+ *       {@code DOUBLE} and {@code TIMESTAMPTZ} vs {@code TIMESTAMPTZ_NANOS} - are treated as
+ *       separate families.
+ *   <li>When the top-level variant's own observations span multiple families, the whole variant is
+ *       written without any typed_value. When a nested field's observations are mixed, only that
+ *       field stays in the residual value; sibling fields still shred.
  *   <li>Fields below {@code MIN_FIELD_FREQUENCY} are pruned. Above {@code MAX_SHREDDED_FIELDS}, the
  *       most frequent are kept with alphabetical tie-breaking.
  *   <li>Recursion into nested objects/arrays stops at {@code MAX_SHREDDING_DEPTH} (default 50).
@@ -421,28 +424,18 @@ private static Type createPrimitiveTypedValue(FieldInfo info, PhysicalType primi
   private static class FieldInfo {
     private static final PhysicalType[] PHYSICAL_TYPES = PhysicalType.values();
 
+    private static final List<PhysicalType> INTEGER_TYPES =
+        List.of(PhysicalType.INT8, PhysicalType.INT16, PhysicalType.INT32, PhysicalType.INT64);
+
+    private static final List<PhysicalType> DECIMAL_TYPES =
+        List.of(PhysicalType.DECIMAL4, PhysicalType.DECIMAL8, PhysicalType.DECIMAL16);
+
     private final int[] typeCounts = new int[PHYSICAL_TYPES.length];
     private int maxDecimalScale = 0;
     private int maxDecimalIntegerDigits = 0;
     private int observationCount = 0;
-    private boolean admittedTypeComputed = false;
-    private PhysicalType admittedTypeCached = null;
-
-    private static final Map<PhysicalType, Integer> INTEGER_PRIORITY =
-        ImmutableMap.of(
-            PhysicalType.INT8, 0,
-            PhysicalType.INT16, 1,
-            PhysicalType.INT32, 2,
-            PhysicalType.INT64, 3);
-
-    private static final Map<PhysicalType, Integer> DECIMAL_PRIORITY =
-        ImmutableMap.of(
-            PhysicalType.DECIMAL4, 0,
-            PhysicalType.DECIMAL8, 1,
-            PhysicalType.DECIMAL16, 2);
 
     void observe(VariantValue value) {
-      admittedTypeComputed = false;
       observationCount++;
       // Use BOOLEAN_TRUE for both TRUE/FALSE values
       PhysicalType type =
@@ -451,7 +444,7 @@ void observe(VariantValue value) {
       typeCounts[type.ordinal()]++;
 
       // Track max precision and scale for decimal types
-      if (isDecimalType(type)) {
+      if (DECIMAL_TYPES.contains(type)) {
         if (value.asPrimitive().get() instanceof BigDecimal bd) {
           maxDecimalIntegerDigits = Math.max(maxDecimalIntegerDigits, bd.precision() - bd.scale());
           maxDecimalScale = Math.max(maxDecimalScale, bd.scale());
@@ -459,69 +452,57 @@ void observe(VariantValue value) {
       }
     }
 
+    /**
+     * Returns the single type family that all observations fall into after numeric widening, or
+     * null if observations span multiple families.
+     */
     PhysicalType admittedType() {
-      if (admittedTypeComputed) {
-        return admittedTypeCached;
-      }
-
-      Set<PhysicalType> families = Sets.newHashSet();
-      PhysicalType widestInteger = null;
-      PhysicalType widestDecimal = null;
-
+      PhysicalType admitted = null;
       for (int i = 0; i < typeCounts.length; i++) {
         if (typeCounts[i] == 0) {
           continue;
         }
-        PhysicalType type = PHYSICAL_TYPES[i];
-
-        if (isIntegerType(type)) {
-          widestInteger = widerType(widestInteger, type, INTEGER_PRIORITY);
-        } else if (isDecimalType(type)) {
-          widestDecimal = widerType(widestDecimal, type, DECIMAL_PRIORITY);
-        } else {
-          families.add(type);
+        PhysicalType merged = mergeFamily(admitted, PHYSICAL_TYPES[i]);
+        if (merged == null) {
+          return null;
         }
+        admitted = merged;
       }
-
-      if (widestInteger != null) {
-        families.add(widestInteger);
-      }
-
-      if (widestDecimal != null) {
-        families.add(widestDecimal);
-      }
-
-      // Type-uniformity: admit only if exactly one family remains after widening.
-      if (families.size() != 1) {
-        admittedTypeCached = null;
-        admittedTypeComputed = true;
-        return null;
-      }
-
-      admittedTypeCached = families.iterator().next();
-      admittedTypeComputed = true;
-      return admittedTypeCached;
+      return admitted;
     }
 
-    private static PhysicalType widerType(
-        PhysicalType current, PhysicalType candidate, Map<PhysicalType, Integer> priority) {
+    /**
+     * Merges {@code candidate} into the currently admitted type. Returns the wider type when both
+     * are in the same integer or decimal family, {@code candidate} when nothing is admitted yet,
+     * {@code current} when the types are identical, and null when the types are from incompatible
+     * families (including FLOAT vs DOUBLE, TIMESTAMPTZ vs TIMESTAMPTZ_NANOS, etc.).
+     */
+    private static PhysicalType mergeFamily(PhysicalType current, PhysicalType candidate) {
       if (current == null) {
         return candidate;
       }
-      return priority.get(candidate) > priority.get(current) ? candidate : current;
-    }
-
-    private static boolean isIntegerType(PhysicalType type) {
-      return type == PhysicalType.INT8
-          || type == PhysicalType.INT16
-          || type == PhysicalType.INT32
-          || type == PhysicalType.INT64;
+      if (current == candidate) {
+        return current;
+      }
+      PhysicalType widened = wider(current, candidate, INTEGER_TYPES);
+      if (widened != null) {
+        return widened;
+      }
+      return wider(current, candidate, DECIMAL_TYPES);
     }
 
-    private static boolean isDecimalType(PhysicalType type) {
-      return type == PhysicalType.DECIMAL4
-          || type == PhysicalType.DECIMAL8
-          || type == PhysicalType.DECIMAL16;
+    /**
+     * Returns the wider of {@code first} and {@code second} when both are in the given family
+     * (positions later in {@code family} are wider), or null when either is not in the family.
+     */
+    private static PhysicalType wider(
+        PhysicalType first, PhysicalType second, List<PhysicalType> family) {
+      int firstIdx = family.indexOf(first);
+      int secondIdx = family.indexOf(second);
+      if (firstIdx < 0 || secondIdx < 0) {
+        return null;
+      }
+      return firstIdx >= secondIdx ? first : second;
     }
   }
 }
diff --git a/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantShreddingAnalyzer.java b/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantShreddingAnalyzer.java
index 3114c2febdf6..bf3103570931 100644
--- a/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantShreddingAnalyzer.java
+++ b/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantShreddingAnalyzer.java
@@ -507,6 +507,25 @@ public void testIntegerWideningAdmitsField() {
         .isEqualTo(PrimitiveType.PrimitiveTypeName.INT64);
   }
 
+  @Test
+  public void testDecimalWideningAdmitsField() {
+    VariantMetadata meta = Variants.metadata("n");
+    ShreddedObject row1 = Variants.object(meta);
+    row1.put("n", Variants.of(new BigDecimal("1.5")));
+    ShreddedObject row2 = Variants.object(meta);
+    row2.put("n", Variants.of(new BigDecimal("9876543210.123")));
+
+    DirectAnalyzer analyzer = new DirectAnalyzer();
+    Type schema = analyzer.analyzeAndCreateSchema(List.of(row1, row2), 0);
+
+    assertThat(schema).isNotNull().isInstanceOf(GroupType.class);
+    GroupType typedValue = (GroupType) schema;
+    assertThat(typedValue.containsField("n")).isTrue();
+    GroupType nGroup = typedValue.getType("n").asGroupType();
+    assertThat(nGroup.getType("typed_value").asPrimitiveType().getPrimitiveTypeName())
+        .isEqualTo(PrimitiveType.PrimitiveTypeName.INT64);
+  }
+
   @Test
   public void testIntAndDecimalAtSameFieldNotShredded() {
     VariantMetadata meta = Variants.metadata("mixed", "keep");
@@ -575,6 +594,52 @@ public void testArrayWithMixedElementTypesNotShredded() {
     assertThat(typedValue.containsField("keep")).isTrue();
   }
 
+  @Test
+  public void testMixedFloatAndDoubleNotShredded() {
+    VariantMetadata meta = Variants.metadata("mixed", "keep");
+    ShreddedObject row1 = Variants.object(meta);
+    row1.put("mixed", Variants.of(1.5F));
+    row1.put("keep", Variants.of(1));
+    ShreddedObject row2 = Variants.object(meta);
+    row2.put("mixed", Variants.of(2.5D));
+    row2.put("keep", Variants.of(2));
+
+    DirectAnalyzer analyzer = new DirectAnalyzer();
+    Type schema = analyzer.analyzeAndCreateSchema(List.of(row1, row2), 0);
+
+    assertThat(schema).isNotNull().isInstanceOf(GroupType.class);
+    GroupType typedValue = (GroupType) schema;
+    assertThat(typedValue.containsField("mixed")).isFalse();
+    assertThat(typedValue.containsField("keep")).isTrue();
+  }
+
+  @Test
+  public void testMixedTimestampTzAndNanosNotShredded() {
+    VariantMetadata meta = Variants.metadata("mixed", "keep");
+    ShreddedObject row1 = Variants.object(meta);
+    row1.put("mixed", Variants.ofIsoTimestamptz("2024-11-07T12:33:54.123456+00:00"));
+    row1.put("keep", Variants.of(1));
+    ShreddedObject row2 = Variants.object(meta);
+    row2.put("mixed", Variants.ofIsoTimestamptzNanos("2024-11-07T12:33:54.123456789+00:00"));
+    row2.put("keep", Variants.of(2));
+
+    DirectAnalyzer analyzer = new DirectAnalyzer();
+    Type schema = analyzer.analyzeAndCreateSchema(List.of(row1, row2), 0);
+
+    assertThat(schema).isNotNull().isInstanceOf(GroupType.class);
+    GroupType typedValue = (GroupType) schema;
+    assertThat(typedValue.containsField("mixed")).isFalse();
+    assertThat(typedValue.containsField("keep")).isTrue();
+  }
+
+  @Test
+  public void testRootLevelMixedTypesReturnsNull() {
+    DirectAnalyzer analyzer = new DirectAnalyzer();
+    Type schema = analyzer.analyzeAndCreateSchema(List.of(Variants.of(42), Variants.of("text")), 0);
+
+    assertThat(schema).isNull();
+  }
+
   /**
    * Builds 100 variant rows where "common" appears in every row and "rare" appears in only {@code
    * rareCount} rows (below MIN_FIELD_FREQUENCY = 0.10 when rareCount < 10).

From 69d6e882de3ba3d7c9a36c44082bae7d61cd60d3 Mon Sep 17 00:00:00 2001
From: Neelesh Salian <n_salian@apple.com>
Date: Thu, 30 Jul 2026 15:10:01 -0700
Subject: [PATCH 4/4] mergefamily cleanup and docstring cleanup

---
 .../parquet/VariantShreddingAnalyzer.java     | 24 ++++++++++++-------
 1 file changed, 16 insertions(+), 8 deletions(-)

diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java b/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
index a774ce06635e..75ea54579f55 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
@@ -472,10 +472,7 @@ PhysicalType admittedType() {
     }
 
     /**
-     * Merges {@code candidate} into the currently admitted type. Returns the wider type when both
-     * are in the same integer or decimal family, {@code candidate} when nothing is admitted yet,
-     * {@code current} when the types are identical, and null when the types are from incompatible
-     * families (including FLOAT vs DOUBLE, TIMESTAMPTZ vs TIMESTAMPTZ_NANOS, etc.).
+     * Widens {@code current} with {@code candidate}, or null if they belong to different families.
      */
     private static PhysicalType mergeFamily(PhysicalType current, PhysicalType candidate) {
       if (current == null) {
@@ -484,11 +481,22 @@ private static PhysicalType mergeFamily(PhysicalType current, PhysicalType candi
       if (current == candidate) {
         return current;
       }
-      PhysicalType widened = wider(current, candidate, INTEGER_TYPES);
-      if (widened != null) {
-        return widened;
+      List<PhysicalType> family = familyOf(current);
+      if (family == null) {
+        return null;
+      }
+      return wider(current, candidate, family);
+    }
+
+    /** Returns the widening family for {@code type}, or null if none applies. */
+    private static List<PhysicalType> familyOf(PhysicalType type) {
+      if (INTEGER_TYPES.contains(type)) {
+        return INTEGER_TYPES;
       }
-      return wider(current, candidate, DECIMAL_TYPES);
+      if (DECIMAL_TYPES.contains(type)) {
+        return DECIMAL_TYPES;
+      }
+      return null;
     }
 
     /**