Parquet, Spark, Flink: Type uniformity for variant shredding - #17424
Parquet, Spark, Flink: Type uniformity for variant shredding#17424nssalian wants to merge 8 commits into
Conversation
|
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?" 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? |
There was a problem hiding this comment.
@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?
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?
|
@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. |
|
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
left a comment
There was a problem hiding this comment.
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
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. Thattyped_valuecolumn 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; notyped_valuecolumn is emitted for them.Benefits
valuebroke stat coverage and forced fallback scans.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 toadmittedType(). Returns null when observations span more than one type family after widening; null propagates through existing handling inanalyzeAndCreateSchema,buildFieldGroup, andcreateArrayTypedValue, so rejected fields fall out of the emitted schema and stay in residualvalue.TIE_BREAK_PRIORITYmap; simplified type selection tofamilies.iterator().next().getMostCommonType->admittedType(plusmostCommonCached/Computedand local renames). ChangecombinedCountsfromMap<PhysicalType, Integer>toSet<PhysicalType> familiessince counts are no longer needed post-uniformity.Follow up
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.