Skip to content

Parquet, Spark, Flink: Type uniformity for variant shredding #17424

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
nssalian wants to merge 6 commits into
base: main
Choose a base branch
from
Open

Parquet, Spark, Flink: Type uniformity for variant shredding #17424

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
c8454fd
Parquet, Spark, Flink: Type uniformity for variant shredding
nssalian Jul 29, 2026
4f717b6
Reduce cyclomatic complexity in variant admittedType
nssalian Jul 29, 2026
63a1d4c
PR comments fixes for ranking and cleanup
nssalian Jul 30, 2026
69d6e88
mergefamily cleanup and docstring cleanup
nssalian Jul 30, 2026
c566dcc
Add variant shredding metrics bounds tests and dedupe
nssalian Aug 3, 2026
88731f6
Merge remote-tracking branch 'apache/main' into variant-uniform-shred…
nssalian Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 8 additions & 31 deletions flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"}')),
Expand All @@ -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')),
Expand All @@ -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')),
Expand All @@ -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);
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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);
Expand Down
116 changes: 43 additions & 73 deletions parquet/src/main/java/org/apache/iceberg/parquet/VariantShreddingAnalyzer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -51,10 +51,10 @@
*
* <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>A field is admitted only if all observations fall into a single type family after numeric
Comment thread
nssalian marked this conversation as resolved.
Outdated
* widening; mixed-type fields remain in the residual {@code value}.
* <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.
* widest observed within their family.
* <li>Fields below {@code MIN_FIELD_FREQUENCY} are pruned. Above {@code MAX_SHREDDED_FIELDS}, the
Comment thread
nssalian marked this conversation as resolved.
* most frequent are kept with alphabetical tie-breaking.
* <li>Recursion into nested objects/arrays stops at {@code MAX_SHREDDING_DEPTH} (default 50).
Expand Down Expand Up @@ -96,7 +96,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;
}
Expand Down Expand Up @@ -251,12 +251,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;
}
Expand Down Expand Up @@ -303,7 +303,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;
}
Expand Down Expand Up @@ -425,8 +425,8 @@ private static class FieldInfo {
private int maxDecimalScale = 0;
private int maxDecimalIntegerDigits = 0;
private int observationCount = 0;
private boolean mostCommonComputed = false;
private PhysicalType mostCommonCached = null;
private boolean admittedTypeComputed = false;
private PhysicalType admittedTypeCached = null;

private static final Map<PhysicalType, Integer> INTEGER_PRIORITY =
ImmutableMap.of(
Expand All @@ -441,32 +441,8 @@ private static class FieldInfo {
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;
admittedTypeComputed = false;
observationCount++;
// Use BOOLEAN_TRUE for both TRUE/FALSE values
PhysicalType type =
Expand All @@ -483,62 +459,56 @@ void observe(VariantValue value) {
}
}

PhysicalType getMostCommonType() {
if (mostCommonComputed) {
return mostCommonCached;
PhysicalType admittedType() {
if (admittedTypeComputed) {
return admittedTypeCached;
}

Map<PhysicalType, Integer> combinedCounts = Maps.newHashMap();

int integerTotalCount = 0;
PhysicalType mostCapableInteger = null;

int decimalTotalCount = 0;
PhysicalType mostCapableDecimal = null;
Set<PhysicalType> families = Sets.newHashSet();

@RussellSpitzer RussellSpitzer Jul 30, 2026 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 RussellSpitzer Jul 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 RussellSpitzer Jul 30, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@nssalian nssalian Jul 30, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@nssalian nssalian Jul 30, 2026

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

PhysicalType widestInteger = null;
PhysicalType widestDecimal = 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;
}
widestInteger = widerType(widestInteger, type, INTEGER_PRIORITY);
} else if (isDecimalType(type)) {
decimalTotalCount += count;
if (mostCapableDecimal == null
|| DECIMAL_PRIORITY.get(type) > DECIMAL_PRIORITY.get(mostCapableDecimal)) {
mostCapableDecimal = type;
}
widestDecimal = widerType(widestDecimal, type, DECIMAL_PRIORITY);
} else {
combinedCounts.put(type, count);
families.add(type);
}
}

if (mostCapableInteger != null) {
combinedCounts.put(mostCapableInteger, integerTotalCount);
if (widestInteger != null) {
families.add(widestInteger);
}

if (mostCapableDecimal != null) {
combinedCounts.put(mostCapableDecimal, decimalTotalCount);
if (widestDecimal != null) {
families.add(widestDecimal);
}

// 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;
// Type-uniformity: admit only if exactly one family remains after widening.
if (families.size() != 1) {
admittedTypeCached = null;
admittedTypeComputed = true;
return null;
}

admittedTypeCached = families.iterator().next();
admittedTypeComputed = true;
return admittedTypeCached;
}

private static PhysicalType widerType(
PhysicalType current, PhysicalType candidate, Map<PhysicalType, Integer> priority) {
if (current == null) {
return candidate;
}
return priority.get(candidate) > priority.get(current) ? candidate : current;
}

private static boolean isIntegerType(PhysicalType type) {
Expand Down
Loading
Loading