Parquet, Spark, Flink: Type uniformity for variant shredding

[nssalian]

Rationale for the change

The current majority-based algorithm introduced in #14297 shreds mixed-type fields to the majority type. A field seen as 95% int + 5% string still emits an int-typed column that misses on the 5% string rows, which fall back to the residual value. That typed_value column pays its storage and I/O cost while covering only a fraction of the field's rows, and readers must always check both columns.

This change requires type uniformity: a field is shredded only when all its observations fall into a single type family after numeric widening. Mixed-type fields stay in the residual value; no typed_value column is emitted for them.

Benefits

• Reader I/O: no mostly-empty typed_value columns to open, decode, and skip. Wide-schema files get fewer column chunks and less Parquet metadata.
• Predicate pushdown and row-group skipping: when a typed_value column exists, its min/max stats cover the entire field, so filter pushdown and stats-based row-group skipping are clean. Under the majority algorithm, minority-type rows in value broke stat coverage and forced fallback scans.
• Predictable rule: uniform -> shredded, otherwise -> residual. No surprising tie-break outcomes (e.g., 50/50 int-vs-string silently picking STRING via priority table).

Clean single-type workloads are unchanged. Benchmark evidence (per-column scoring plus reader wall-clock across 11 workloads at 100k rows) shows uniformity ties or beats the majority algorithm on every workload and wins uniquely on mixed-type inputs (60/40, 95/5, 99/1 int/string, four-way polymorphic). Full numbers in the design document.

Design document with benchmarks: Google Doc

Changes

• VariantShreddingAnalyzer: add a uniformity check to admittedType(). Returns null when observations span more than one type family after widening; null propagates through existing handling in analyzeAndCreateSchema, buildFieldGroup, and createArrayTypedValue, so rejected fields fall out of the emitted schema and stay in residual value.
• Removed the now-unreachable TIE_BREAK_PRIORITY map; simplified type selection to families.iterator().next().
• Rename getMostCommonType -> admittedType (plus mostCommonCached/Computed and local renames). Change combinedCounts from Map<PhysicalType, Integer> to Set<PhysicalType> families since counts are no longer needed post-uniformity.
• Updated the class-level javadoc.

Follow up

• Will open follow-up issues for each implementation; the iceberg-go change I'll implement myself.

Tests

• TestVariantShreddingAnalyzer: 9 new tests covering mixed-primitive rejection, integer and decimal widening admission, int+decimal cross-family rejection, mixed object+primitive rejection, array-element mixing rejection, float/double and timestamptz/nanos cross-family rejection, and root-level mixed rejection. One existing test updated to use uniform-type observations.
• TestVariantShredding (Spark v4.0 and v4.1): 3 tests renamed and updated to assert non-shredding on mixed inputs (testInconsistentType, testPrimitiveDecimalType, testMixedTypeTieBreaking).
• TestFlinkVariantShreddingType (Flink v2.1): 4 tests renamed and updated similarly.

[RussellSpitzer]

I generally like to avoid having multiple local variables that only work if they are set in certain combinations and I think we can make this a bit tighter if instead of tracking all these things separately we do something like.

if admitted not set - set
if admitted set and this is not in the family - return null exit early on everything
if admtted set and this type is wider, set admitted to wider type

here is a quick draft i had the llm do

PhysicalType admittedType() {
  PhysicalType admitted = null;
  for (int i = 0; i < typeCounts.length; i++) {
    if (typeCounts[i] == 0) {
      continue;
    }
    PhysicalType merged = mergeFamily(admitted, PHYSICAL_TYPES[i]);
    if (merged == null) {
      // Mixed type families: do not shred this field.
      return null;
    }
    admitted = merged;
  }
  return admitted;
}
/**
 * Merges {@code candidate} into the currently admitted type.
 *
 * <p>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).
 */
private static PhysicalType mergeFamily(PhysicalType current, PhysicalType candidate) {
  if (current == null) {
    return candidate;
  }
  if (current == candidate) {
    return current;
  }
  if (isIntegerType(current) && isIntegerType(candidate)) {
    return integerRank(current) >= integerRank(candidate) ? current : candidate;
  }
  if (isDecimalType(current) && isDecimalType(candidate)) {
    return decimalRank(current) >= decimalRank(candidate) ? current : candidate;
  }
  return null;
}
private static int integerRank(PhysicalType type) {
  return switch (type) {
    case INT8 -> 0;
    case INT16 -> 1;
    case INT32 -> 2;
    case INT64 -> 3;
    default -> throw new IllegalArgumentException("Not an integer type: " + type);
  };
}
private static int decimalRank(PhysicalType type) {
  return switch (type) {
    case DECIMAL4 -> 0;
    case DECIMAL8 -> 1;
    case DECIMAL16 -> 2;
    default -> throw new IllegalArgumentException("Not a decimal type: " + type);
  };
}

[RussellSpitzer]

I'm not sure on the ranking code there, originally I had it with enum ordinal but thats a spotless issue. Maybe we just have static ordered type lists for each family?

[RussellSpitzer]

I would try iterating on that though ... see if we can remove all the state here we don't need

[nssalian]

Yeah let me see if this is complete. I'll noodle on this more and update.

[nssalian]

Dropped widestInteger, widestDecimal, admittedTypeCached, and the families Set. Remaining state (typeCounts, decimal scale/digits, observation count) is used for schema build and pruning.

[RussellSpitzer]

Overall I think this is good. I have some style/function comments on the code itself. I really think we should drop all the state we currently have since I think at the end of the day the only state we need is

"Have we decided to shred or not?"
"What is that shredded type?"

Currently we have a widest decimal, widest integer, cached , and set

I wrote this inline but in my head this could be

Are we trying to shred to a type and haven't picked one? Set it

Is the current type not in the family of the set type? We can't shred

Is the current type in the family of the set type?
is it narrower or equal? do nothing
is it wider? Set it

[Guosmilesmile]

@nssalian Thanks for the PR！I have small question about numeric widening.
A mixed integer family can be promoted to an INT64 typed_value, but the writer seems to match the original physical type exactly. Would that leave INT8/INT16/INT32 rows in residual value and make the typed_value min/max stats only cover the INT64 rows?

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

Lines 293 to 299 in f3d5b17

	if (typedWriter.types().contains(value.type()) && typedWriter.canWrite(value)) {
	typedWriter.write(repetitionLevel, value);
	writeNull(valueWriter, repetitionLevel, valueDefinitionLevel);
	} else {
	valueWriter.write(repetitionLevel, value);
	writeNull(typedWriter, repetitionLevel, typedDefinitionLevel);
	}

I also test testIntegerFamilyPromotion in Spark, get

String values =
        """
            (1, parse_json('{"value": 10}')),\
             (2, parse_json('{"value": 1000}')),\
             (3, parse_json('{"value": 100000}')),\
             (4, parse_json('{"value": 10000000000}'))\
            """;

rowGroup=0: valueCount=4, nullCount=3, min=10000000000, max=10000000000

Is there something I'm missing?

[nssalian]

@Guosmilesmile thanks for taking a look. The raw column min/max you see (10000000000 for both) is because integer widening (INT8+INT64 → INT64 schema, preserved in admittedType() from the old getMostCommonType) combined with ShreddedVariantWriter.write doing an exact-type check means INT8/INT16/INT32 rows fall to residual value. But Iceberg's variant bounds handle this correctly: ParquetMetrics.value() invalidates the typed bounds whenever the value residual has any non-null entry, so filter pushdown doesn't use those raw stats. I recently fixed iceberg-go in (apache/iceberg-go#1555) to match this behavior based on the spec. The spec permits either widening on write or falling to residual for out-of-schema types; Iceberg went with residual (see testMixedShredding in TestVariantWriters which relies on this to preserve exact width). This PR only adds the uniformity check above the widening logic.

[RussellSpitzer]

may be overkill if we never add another family .. but

private static PhysicalType mergeFamily(PhysicalType current, PhysicalType candidate) {
  if (current == null) {
    return candidate;
  }
  if (current == candidate) {
    return current;
  }
  List<PhysicalType> family = familyOf(current);
  if (family == null) {
    return null; // current has no widening family (STRING, FLOAT, …)
  }
  return wider(current, candidate, family); // null if candidate not in that family
}

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;
}

I dunno ... i'll let you decide :)

[nssalian]

I'll tell you I had this locally and chose not to do it because we had just two. Let me do this so it's better future looking.