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..75ea54579f55 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>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,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());
@@ -483,75 +452,65 @@ void observe(VariantValue value) {
       }
     }
 
-    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);
+    /**
+     * 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) {
+        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;
+      List<PhysicalType> family = familyOf(current);
+      if (family == null) {
+        return null;
+      }
+      return wider(current, candidate, family);
     }
 
-    private static boolean isIntegerType(PhysicalType type) {
-      return type == PhysicalType.INT8
-          || type == PhysicalType.INT16
-          || type == PhysicalType.INT32
-          || type == PhysicalType.INT64;
+    /** 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;
+      }
+      if (DECIMAL_TYPES.contains(type)) {
+        return DECIMAL_TYPES;
+      }
+      return null;
     }
 
-    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 fabcb9ae6cd1..bf3103570931 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,177 @@ 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 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");
+    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();
+  }
+
+  @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).
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);