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.
• wider(): drop the unreachable firstIdx < 0 guard branch. The only caller resolves family from first via familyOf(current), so first is always in family; only second (the incoming candidate) can be out of family. Behavior is unchanged.
• Class-level javadoc updated, including a note that widening decides admission and the emitted typed_value type, not per-row routing: the writer shreds a row only on an exact physical-type match, so narrower-width rows fall to the residual value.

Follow up

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

Tests

• TestVariantShreddingAnalyzer: 13 new tests.
  • Rejection: mixed primitive types, int+decimal cross-family, mixed object+primitive, mixed array elements, float/double cross-family, timestamptz/timestamptz_nanos cross-family, and root-level mixed.
  • Widening admission: integer family (INT8+INT64 -> INT64) and decimal family.
  • Determinism: testShreddedSchemaIsOrderIndependent and testMixedTypesNotShreddedRegardlessOfOrder assert the shredded schema is identical across row order for both the admit and reject cases (the class javadoc's determinism contract).
  • Null handling: testAllNullFieldNotShredded (a field null in every observed row is not shredded) and testFieldWithNullsAndSingleTypeStillShreds (nulls mixed with one real type still shred that type).
  • 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]

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.

[nssalian]

Thanks for the reviews @RussellSpitzer. Addressed the comments. I dropped the redundant cached/widest-type state, and the decision now only depends on which families appear, rather than counts. Added a few tests for metrics and widening.

@pvary @huaxingao PTAL for the Spark and Flink related changes.

[Guosmilesmile]

LGTM overall. Can we make a small adjustment to preserve the newline after code blocks? https://iceberg.incubator.apache.org/contribute/?h=newline#block-spacing

[nssalian]

Made some changes since the last review. Added some tests TestVariantShreddingAnalyzer tests: schema order-independence (admit and reject) and null-field handling (all-null not shredded, nulls+one-type still shreds). Some branch cleanup in wider() and added some javadoc clarity.

[RussellSpitzer]

Thanks @nssalian for the pr, and thanks @Guosmilesmile for the review.