Parquet, Spark, Flink: Type uniformity for variant shredding

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,21 +148,15 @@ 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);
     verifyParquetSchema(icebergTable, expectedSchema);
   }
 
   @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);

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>Type selection picks the most common type with explicit tie-break priority (see {@link
- *       FieldInfo#TIE_BREAK_PRIORITY}), not enum ordinal.
- *   <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.
+ *   <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).
@@ -96,7 +99,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 +254,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 +306,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;
     }
@@ -421,52 +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 mostCommonComputed = false;
-    private PhysicalType mostCommonCached = 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);
-
-    /** 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;
       observationCount++;
       // Use BOOLEAN_TRUE for both TRUE/FALSE values
       PhysicalType type =
@@ -475,83 +444,65 @@ 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());
         }
       }
     }
 
-    PhysicalType getMostCommonType() {
-      if (mostCommonComputed) {
-        return mostCommonCached;
-      }
-
-      Map<PhysicalType, Integer> combinedCounts = Maps.newHashMap();
-
-      int integerTotalCount = 0;
-      PhysicalType mostCapableInteger = null;
-
-      int decimalTotalCount = 0;
-      PhysicalType mostCapableDecimal = null;
-
+    /**
+     * Returns the single type family that all observations fall into after numeric widening, or
+     * null if observations span multiple families.
+     */
+    PhysicalType admittedType() {
+      PhysicalType admitted = 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;
-          }
-        } else if (isDecimalType(type)) {
-          decimalTotalCount += count;
-          if (mostCapableDecimal == null
-              || DECIMAL_PRIORITY.get(type) > DECIMAL_PRIORITY.get(mostCapableDecimal)) {
-            mostCapableDecimal = type;
-          }
-        } else {
-          combinedCounts.put(type, count);
+        PhysicalType merged = mergeFamily(admitted, PHYSICAL_TYPES[i]);
+        if (merged == null) {
+          return null;
         }
+        admitted = merged;
       }
+      return admitted;
+    }
 
-      if (mostCapableInteger != null) {
-        combinedCounts.put(mostCapableInteger, integerTotalCount);
+    /**
+     * 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;
       }
-
-      if (mostCapableDecimal != null) {
-        combinedCounts.put(mostCapableDecimal, decimalTotalCount);
+      if (current == candidate) {
+        return current;
       }
-
-      // 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;
-    }
-
-    private static boolean isIntegerType(PhysicalType type) {
-      return type == PhysicalType.INT8
-          || type == PhysicalType.INT16
-          || type == PhysicalType.INT32
-          || type == PhysicalType.INT64;
+      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;
     }
   }
 }