Repository navigation
Flink: Fix ClassCastException when filtering on ARRAY and MAP columns - #18508
yusufyildiz34 wants to merge 6 commits into
Conversation
0e02059 to
5b00924
Compare
RowDataWrapper had no getters for ARRAY and MAP columns and returned Flink's ArrayData/MapData where Iceberg expects List/Map, so filters re-evaluated by FlinkSourceFilter failed with a ClassCastException. Closes apache#18504
5b00924 to
6f9679e
Compare
mbutrovich
left a comment
There was a problem hiding this comment.
Thanks @yusufyildiz34 for tracking this down and for the regression tests.
The description says VARIANT has the same gap, but this PR closes #18504, which only covers ARRAY. Could you open a separate issue for VARIANT and link it here so the follow-up is tracked?
| } | ||
| } | ||
|
|
||
| private static Function<Object, Object> buildConverter(LogicalType logicalType, Type type) { |
There was a problem hiding this comment.
Should buildGetter be built on top of buildConverter? As I read it, the cases from TINYINT through ROW here repeat the conversions in buildGetter one for one. The only difference is that one reads from a RowData position and the other takes a value. That leaves two switches to keep in sync here and in the 1.20 and 2.2 backports.
One way is to read the raw value with FlinkRowData.createFieldGetter and convert it, keeping ROW separate so a top-level struct still reuses one wrapper:
private static PositionalGetter<?> buildGetter(LogicalType logicalType, Type type, int pos) {
if (logicalType instanceof RowType rowType) {
RowDataWrapper nestedWrapper = new RowDataWrapper(rowType, type.asStructType());
return (row, p) -> nestedWrapper.wrap(row.getRow(p, rowType.getFieldCount()));
}
RowData.FieldGetter fieldGetter = FlinkRowData.createFieldGetter(logicalType, pos);
Function<Object, Object> converter = buildConverter(logicalType, type);
return (row, p) -> converter.apply(fieldGetter.getFieldOrNull(row));
}The constructor would then call buildGetter(types[i], struct.fields().get(i).type(), i) and drop the null fallback. The randomized cases in RecordWrapperTestBase would then cover the same conversions that list and map elements use. The cost is one extra box per TIME value, and per SMALLINT value outside the Short cache, because the field getter boxes the raw int or short before the conversion.
There was a problem hiding this comment.
Done. buildGetter now reads the raw value with FlinkRowData.createFieldGetter and converts it with buildConverter, so there is only one switch.
I didn't keep ROW separate, though, because it conflicts with sharing getters across element wrappers (your ROW comment). The reused nested wrapper ends up inside a shared getter, so nested structs in different list elements overwrite each other. nestedStructsInListElementsAreIndependent fails with that version. Now each struct read gets its own small wrapper, which costs one allocation per struct read on top of the boxing you mentioned. If you'd rather keep the reuse for top-level structs, I can add a separate path for the element templates.
Since no getter uses the position anymore, I also replaced PositionalGetter with RowData.FieldGetter.
There was a problem hiding this comment.
@mbutrovich following pvary's suggestion, top-level structs reuse one wrapper again and only collection elements get new wrappers, so the extra allocation is gone.
| case ROW -> { | ||
| RowType rowType = (RowType) logicalType; | ||
| Types.StructType structType = type.asStructType(); | ||
| // Collection elements are read together, so each element needs its own wrapper. | ||
| yield value -> new RowDataWrapper(rowType, structType).wrap((RowData) value); | ||
| } |
There was a problem hiding this comment.
This calls the public constructor for every struct element of every row. The constructor allocates the types and getters arrays and calls buildGetter for each field, which also rebuilds any nested list or map converters. What do you think about building the getters once and sharing them across the element wrappers? For example, with a private constructor:
private RowDataWrapper(LogicalType[] types, PositionalGetter<?>[] getters) {
this.types = types;
this.getters = getters;
}| case ROW -> { | |
| RowType rowType = (RowType) logicalType; | |
| Types.StructType structType = type.asStructType(); | |
| // Collection elements are read together, so each element needs its own wrapper. | |
| yield value -> new RowDataWrapper(rowType, structType).wrap((RowData) value); | |
| } | |
| case ROW -> { | |
| RowDataWrapper template = new RowDataWrapper((RowType) logicalType, type.asStructType()); | |
| // Collection elements are read together, so each element needs its own wrapper. | |
| yield value -> new RowDataWrapper(template.types, template.getters).wrap((RowData) value); | |
| } |
There was a problem hiding this comment.
Done. The getters are now built once in a template and shared through a private constructor, via copyFor(RowData), the same pattern as InternalRecordWrapper.copyFor. Sharing them is also why ROW no longer reuses one nested wrapper; details in the buildGetter thread.
| private static ArrayData.ElementGetter elementGetter(LogicalType elementType) { | ||
| // Iceberg's unknown type is Flink's NULL type, which createElementGetter rejects. | ||
| if (elementType instanceof NullType) { | ||
| return (array, pos) -> null; | ||
| } | ||
|
|
||
| return ArrayData.createElementGetter(elementType); | ||
| } |
There was a problem hiding this comment.
Should this move to FlinkRowData as createElementGetter, next to createFieldGetter? That class already holds the same NullType workaround for row fields, so both Flink getter workarounds would live in one place for the backports.
There was a problem hiding this comment.
Done, moved it to FlinkRowData.createElementGetter, next to createFieldGetter. I made it package-private since it's only used in this package. For the 1.20 backport, it will also need the same null check as createFieldGetter, because Flink 1.20's element getter skips it for NOT NULL types.
| @Test | ||
| void mapKeysAndValuesAreConverted() { | ||
| Schema schema = | ||
| new Schema( | ||
| Types.NestedField.optional( | ||
| 1, | ||
| "prices", | ||
| Types.MapType.ofOptional( | ||
| 2, 3, Types.StringType.get(), Types.DecimalType.of(9, 2)))); | ||
| BigDecimal price = new BigDecimal("1.50"); | ||
| RowData row = | ||
| GenericRowData.of( | ||
| new GenericMapData( | ||
| ImmutableMap.of( | ||
| StringData.fromString("a"), DecimalData.fromBigDecimal(price, 9, 2)))); | ||
|
|
||
| assertThat(wrap(schema, row).get(0, Map.class)).isEqualTo(ImmutableMap.of("a", price)); | ||
| } |
There was a problem hiding this comment.
Could you add cases for an empty list, a map with a null value, and a list nested in a list? Those take paths the current tests don't reach: zero iterations in toList, convertOrNull on a map value, and the recursive ARRAY case in buildConverter.
There was a problem hiding this comment.
Added emptyListIsConverted, mapNullValueIsPreserved, and nestedListIsConverted for those three paths. I also added nestedStructsInListElementsAreIndependent, which caught a bug in my first version of the getter sharing; details in the buildGetter thread.
Use one type switch for top-level fields and collection elements, share getters across struct element wrappers so they are not rebuilt per element, keep the Flink NULL-type workarounds together in FlinkRowData, and give each struct read its own wrapper so values read together don't overwrite each other. Add tests for empty lists, nested lists, null map values, and nested structs in list elements. Generated-by: Claude (claude-opus-5-5)
| private static RowData.FieldGetter buildGetter(LogicalType logicalType, Type type, int pos) { | ||
| RowData.FieldGetter fieldGetter = FlinkRowData.createFieldGetter(logicalType, pos); | ||
| Function<Object, Object> converter = buildConverter(logicalType, type); | ||
| return row -> convertOrNull(fieldGetter.getFieldOrNull(row), converter); |
There was a problem hiding this comment.
Do we need this many orNull checks? Could we just short-circuit for nulls?
There was a problem hiding this comment.
@pvary Good point. get() already returns null fields before calling a getter, so I dropped the extra check there. For list and map elements, convertElement now checks isNullAt once and skips the getter and converter for nulls.
|
Is the issue only with collections? Shall we just create new wrappers when needed? |
Top-level struct fields are read one row at a time, so they reuse one wrapper again, and only struct values inside lists and maps get their own wrappers. Nulls are short-circuited once per field or element instead of being checked again after conversion. Generated-by: Claude (claude-opus-5-5)
|
Yes, values only overwrite each other inside collections. Top-level struct fields reuse one wrapper again, and only the element templates for lists and maps create a new wrapper per value. |
| buildConverter(mapType.getValueType(), icebergMapType.valueType()); | ||
| yield value -> toMap((MapData) value, keyGetter, keyConverter, valueGetter, valueConverter); | ||
| } | ||
| default -> Function.identity(); |
There was a problem hiding this comment.
This adds an unnecessary call. Shall we?
- Have buildConverter return null when no conversion is needed.
- In buildGetter, return fieldGetter directly in that case.
- In convertElement, skip the converter when it is null.
|
The more I see the copying, the more I hate it. What if we create a wrapper above Arrays and Maps, like: |
Lists and maps now convert their elements when they are read instead of being copied up front, and values that are already in Iceberg's representation skip the converter. Only primitive fields are kept beyond the current row, so views over reused Flink containers are safe.
|
Agreed, switched to views in fe53193. ArrayDataList and MapDataMap now convert elements when they are read. I copied before because readers reuse containers across rows, but the only values kept beyond the current row come from primitive fields (equality, partition and sort source fields), so views are safe here. |
| Types.NestedField.optional( | ||
| 8, "nested_tags", Types.ListType.ofOptional(9, Types.StringType.get()))))); | ||
|
|
||
| private static final RowData ROW = |
There was a problem hiding this comment.
Can we add a test which uses a row with null collections to check isNull returns true, notNull return false?
There was a problem hiding this comment.
Added nullCollectionIsNull in 2fa7125: on a row where the list, the map and the nested list are all null, isNull returns true and notNull returns false.
| 8, "nested_tags", Types.ListType.ofOptional(9, Types.StringType.get()))))); | ||
|
|
||
| private static final RowData ROW = | ||
| GenericRowData.of( |
There was a problem hiding this comment.
Can we test with BinaryRowData too? new RowDataSerializer(rowType).toBinaryRow(row))
There was a problem hiding this comment.
Done in 2fa7125. Both tests now also run on the row converted with RowDataSerializer.toBinaryRow, so the list and map views read BinaryArrayData and BinaryMapData.
|
@yusufyildiz34: Please update the PR description. |
The IS NULL / IS NOT NULL checks on list, map and nested list columns now also run on a row where those collections are null, and on rows converted to BinaryRowData, so the views over binary arrays and maps are exercised.
|
@pvary Updated the description, thanks! |
|
Thanks! One note: IS NULL / NOT NULL only check eval(struct) == null, so the binary variants never read list or map elements. BinaryArrayData/BinaryMapData element reads are still only covered for Generic data. Also, please add a row where nested itself is null, to cover the null-parent path for nested.nested_tags. |
The element conversion tests in TestRowDataWrapper now also run on rows converted to BinaryRowData, so elements are read through BinaryArrayData and BinaryMapData, and TestFlinkSourceFilter covers a list inside a null struct.
|
@pvary Good point, the null checks never read the elements. In d5ebfc3 the element tests in TestRowDataWrapper (strings, timestamps, decimals, structs, nested and empty lists, null map values) also run on BinaryRowData now, so they read through BinaryArrayData and BinaryMapData. The unknown-type case stays on GenericRowData because RowDataSerializer has no serializer for Flink's NULL type. TestFlinkSourceFilter also has a row where |
| @@ -37,28 +47,30 @@ | |||
| public class RowDataWrapper implements StructLike { | |||
There was a problem hiding this comment.
We could add some javadoc describing what can be used after a new wrap
| public class RowDataWrapper implements StructLike { | |
| /** | |
| * Wraps Flink {@link RowData} as an Iceberg {@link StructLike}. | |
| * | |
| * <p>{@link #wrap(RowData)} reuses this instance, and nested structs, lists and maps are views of | |
| * the wrapped data rather than copies. They are only valid until the next {@link #wrap(RowData)} | |
| * call or until the {@link RowData} is reused, so copy them to keep them. | |
| */ | |
| public class RowDataWrapper implements StructLike { |
FlinkSourceFilter re-evaluates pushed-down filters per row through RowDataWrapper, which had no getters for ARRAY and MAP, so IS NULL and IS NOT NULL on those columns threw a ClassCastException. Lists and maps are now exposed as views that convert each element when it is read. Readers may reuse containers across rows, but the values kept beyond the current row (equality, partition and sort source fields) are primitives, so views are safe here. Struct values inside lists and maps get their own wrappers so they don't overwrite each other, while top-level struct fields keep reusing one. Top-level fields and collection elements share one converter, so SMALLINT and TIME values pay one extra boxing. VARIANT has the same gap and is tracked in #18527. Backports to Flink 1.20 and 2.2 will follow; the 1.20 element getter also needs the null check from createFieldGetter.
Closes #18504
Test plan: TestFlinkSourceFilter checks IS NULL and IS NOT NULL on a list, a map, and a list nested in a struct, for null and non-null values and a null parent struct, with GenericRowData and BinaryRowData; the non-null cases fail on main. TestRowDataWrapper covers element conversion for strings, timestamps, decimals, structs and unknown; empty lists, nested lists, null map values, and nested structs in list elements, reading elements from both GenericRowData and BinaryRowData (except unknown, which RowDataSerializer can't write). Also ran the Flink 2.3 tests that use RowDataWrapper: sink partitioning, delta and position-delta writers, shuffle, and reader deletes.
AI Disclosure