Repository navigation
fix(flink): unwrap the list element type when creating vectors for nested arrays - #20337
skywalker0618 wants to merge 3 commits into
Conversation
fb2c4fa to
bb17ac2
Compare
github-actions
Bot
added
size:L
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This makes the Flink 1.18 COW reader build the vector tree for arrays by unwrapping the LIST element with the same rules as the ParquetField tree, which fixes the NPE for arrays of rows that contain arrays of rows; there's one non-blocking note inline about an existing schema-evolution edge case. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of small readability suggestions below.
|
|
||
| @ParameterizedTest | ||
| @EnumSource(ListEncoding.class) | ||
| void testReadArrayOfRowWithFieldsMissingFromFile(ListEncoding encoding) throws IOException { |
There was a problem hiding this comment.
🤖 non-blocking: the null placeholder for a missing field is sized to batchSize, so once a batch holds more array elements than rows (e.g. batchSize=2 with two rows of two items each), isNullAt on added_scores/added_tag_field throws ArrayIndexOutOfBoundsException; this also fails before this PR, so would it make sense to track it as a follow-up, maybe with a small-batch variant of this test?
There was a problem hiding this comment.
Good catch, confirmed: with a batch size of 2 the evolved array-of-rows test fails with an ArrayIndexOutOfBoundsException in AbstractHeapVector.isNullAt, since inside an array the row has one position per element. Fixed in NestedColumnReader.readRow by growing the null placeholders of absent fields to the position count, and added testReadArrayOfRowWithFieldsMissingFromFileInBatchSmallerThanElementCount (batch size 2, all three list encodings), which fails without the fix.
| ArrayType arrayType = (ArrayType) fieldType; | ||
| // Locate the element type with the same LIST backward-compatibility rules as | ||
| // getArrayElementColumn, so that the vectors line up with the ParquetField tree. | ||
| Type elementType = physicalType; |
There was a problem hiding this comment.
🤖 nit: would it make sense to pull the element-type lookup in this ARRAY case into a small helper (e.g. unwrapListElementType)? It would keep createWritableColumnVector shorter and avoid the block braces on the case.
| int index = recordIndex[0]++; | ||
| if (index == 0) { | ||
| Group items = record.addGroup("items"); | ||
| addItem(encoding, items, "a", "x", "y"); |
There was a problem hiding this comment.
🤖 nit: itemsType just wraps new ArrayType(...) and rowType just wraps RowType.of(...); could you inline them at the call sites?
There was a problem hiding this comment.
Inlined itemsType since it only wrapped new ArrayType(...). I kept rowType, as it reads better with the field names listed before the types than building RowType.RowFields inline. Also pulled the evolved items type, now used by two tests, into evolvedItemsType().
…sted arrays createWritableColumnVector passed the LIST group itself as the physical type of an array's element. For an array of rows the element ROW then looked its fields up by name in the LIST wrapper, found none, and filled every slot with a constant placeholder vector. Reading array<row<..., array<row<...>>>> then failed with a NullPointerException in NestedColumnReader.readRow, because the placeholder for the inner array has no child vector. Locate the element type with the same Parquet LIST backward-compatibility rules as getArrayElementColumn (3-level, 2-level "array" / "<name>_tuple", multi-field and primitive repeated types), so the vector tree lines up with the ParquetField tree. Row fields are still resolved by name, preserving schema evolution. Inside an array, a row has one position per element, which can exceed the batch size that the null placeholders of fields absent from the file were sized with. Grow those placeholders to the position count in NestedColumnReader.readRow, so that reading an evolved array of rows doesn't fail with an ArrayIndexOutOfBoundsException.
bb17ac2 to
3ea6452
Compare
|
@danny0405 Do you mind taking a look at this bug fix as well? Fix an NPE issue in parquet reader. Tested internally in Uber. Once this is merged into flink 1.18 I will create PR for flink 1.19 and onwards. Thanks! |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #20337 +/- ##
============================================
+ Coverage 80.71% 80.75% +0.03%
- Complexity 35152 35313 +161
============================================
Files 2553 2562 +9
Lines 143556 144143 +587
Branches 17482 17629 +147
============================================
+ Hits 115875 116397 +522
- Misses 19737 19771 +34
- Partials 7944 7975 +31
Flags with carried forward coverage won't be shown. Click here to find out more. 🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This fixes the Flink 1.18 COW reader so the element vector of an ARRAY uses the unwrapped Parquet list element, and grows absent-field placeholders inside arrays to the element count; one note inline on which Flink module the fix and its tests land in. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of small readability suggestions on the reader change.
| return new HeapDecimalVector(batchSize); | ||
| case ARRAY: | ||
| ArrayType arrayType = (ArrayType) fieldType; | ||
| Pair<Type, Integer> element = unwrapListElementType(physicalType, depth); |
There was a problem hiding this comment.
🤖 non-blocking: hudi-flink2.2.x is the default profile and the only one CI runs (the flink1.18 matrix entries in bot.yml are commented out), and it has the same ARRAY branch bug, so the new tests won't run in CI and the default bundle still hits the NPE. Would it make sense to land this in hudi-flink2.2.x too (or in place of 1.18), and add 2.2.x to the follow-up list in the description?
There was a problem hiding this comment.
Sure I will create a new PR to cover other flink branches once this is merged
| // Inside an array the row has one position per element, which can exceed the batch size the | ||
| // placeholders of absent fields were sized with, so grow them to the position count. | ||
| int positionsCount = rowPosition.getPositionsCount(); | ||
| for (int i = 0; i < children.size(); i++) { |
There was a problem hiding this comment.
🤖 nit: could you hoist ((RowType) field.getType()) into a local variable outside the loop, and consider extracting this placeholder-growing loop into a small helper so readRow stays focused?
There was a problem hiding this comment.
Hoisted the RowType cast out of the loop. Disregard the helper one as it would have only 1 caller.
| */ | ||
| private static Pair<Type, Integer> unwrapListElementType(Type listType, int depth) { | ||
| Type elementType = listType; | ||
| String listName = null; |
There was a problem hiding this comment.
🤖 nit: unwrapListElementType duplicates the unwrapping logic of getArrayElementColumn; would it make sense to note (or share) which parts must stay in sync so they don't drift apart?
There was a problem hiding this comment.
The Javadoc of unwrapListElementType already says it applies the same backward-compatibility rules as getArrayElementColumn. Sharing the code doesn't fit well, since one walks the Parquet Type and the other builds the ParquetField tree, so I'll leave it as is.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR makes the Flink 1.18 COW reader's vector tree pick the same LIST element as the ParquetField tree, and resizes the placeholders for absent fields to the row's position count, so arrays of rows that hold nested arrays of rows can now be read. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
cshuo
left a comment
cshuo
left a comment
There was a problem hiding this comment.
Thks for the fix. Since this reader was ported from Flink, have you checked whether the latest upstream Flink implementation already handles this deep nested-array case? If so, could we backport the relevant changes while preserving Hudi’s schema-evolution support?
| batchSize, | ||
| createWritableColumnVector( | ||
| batchSize, arrayType.getElementType(), physicalType, descriptors, depth)); | ||
| batchSize, arrayType.getElementType(), element.getLeft(), descriptors, element.getRight())); |
There was a problem hiding this comment.
Before this change, ARRAY<ROW<...>> containing INT32/INT64-encoded decimals was readable because NestedPrimitiveColumnReader replaced the initial placeholder with a correctly decoded vector. Unwrapping now reaches the actual decimal field during allocation, where the existing DECIMAL branch accepts only BINARY/FIXED_LEN_BYTE_ARRAY, so those files fail during reader construction. Could we preserve support for both integer encodings and add regression tests? Current Hudi writers use fixed-length bytes, so this is a compatibility regression for other valid Parquet files, rather than a failure of the default Hudi write path.
There was a problem hiding this comment.
Great catch, thanks! The allocator has accepted only bytes for decimals since HUDI-2430, and arrays of rows only avoided that check because of the unwrapping bug fixed here. Fixed the allocation and added UTs to cover.
| && repeatedType.getLogicalTypeAnnotation() == null | ||
| && repeatedType.asGroupType().getFieldCount() == 1 | ||
| && !repeatedType.getName().equals("array") | ||
| && (listName == null || !repeatedType.getName().equals(listName + "_tuple")); |
There was a problem hiding this comment.
A bare repeated group named <parent-name>_tuple with a single primitive field previously read correctly as an array of rows. When this helper receives that repeated group directly, the loop is skipped and listName remains null. This condition then unwraps the actual element ROW to its primitive child, causing a ClassCastException in the ROW allocator. Meanwhile, getArrayElementColumn retains the actual parent name and correctly preserves the ROW. Could we retain that parent context so both paths select the same element type?
There was a problem hiding this comment.
Fixed and added test cases too
| int positionsCount = rowPosition.getPositionsCount(); | ||
| RowType rowType = (RowType) field.getType(); | ||
| for (int i = 0; i < children.size(); i++) { | ||
| if (children.get(i) == null && vectorLength(finalChildrenVectors[i]) < positionsCount) { |
There was a problem hiding this comment.
When a ROW's last existing child is an array and the parent rows are non-null, calculateRowOffsets returns the unfiltered leaf-level count, including deeper array elements. That counting bug predates this PR, but using it to resize missing-field vectors is new. With batchSize=2 and 100 nested elements per record, the missing-field capacity grows from 2 to 201 while the parent capacity remains 2, adding allocation and null-filling work. Could we use the owning ROW's actual required capacity, accounting for lookahead, rather than the deeper element count?
There was a problem hiding this comment.
Thanks, fixed. I found this problem exists in flink too, and was brought in hudi during the backport.
…, _tuple list elements
I checked the latest Flink master: there's no newer fix to backport. The last changes there are FLINK-35702 and FLINK-36709, and Hudi already has both. Flink handles an array of rows in its ROW allocator by always stepping down two levels and matching fields by position, so it only supports the standard three-level list and would break Hudi's name-based schema evolution. So this PR unwraps in the ARRAY allocator instead, using the same rules as getArrayElementColumn. |
|
@cshuo Do you mind taking another look? Thanks! |
This branch has not been deployed
4 participants
Describe the issue this Pull Request addresses
Reading a Hudi table through the Flink COW Parquet reader fails with a
NullPointerExceptionwhen a column is an array of rows that itself contains an array of rows, e.g.ARRAY<ROW<name STRING, items ARRAY<ROW<id STRING>>>>:This is a follow-up to the Dremel-style nested Parquet reader backport (FLINK-35702, #18636 / #18700 / #18809). We hit it streaming a production COW table with deeply nested schemas.
Root cause: in
ParquetSplitReaderUtil.createWritableColumnVector, theARRAYbranch passes the ParquetLISTgroup itself as the physical type of the array's element. For an array of rows, the elementROWthen looks its fields up by name in theLISTwrapper instead of in the element group, finds none, and fills every slot with a constant placeholder vector. For leaf fields this goes unnoticed (the array of rows silently reads as all-null fields), but the placeholder for an inner array has no child vector, soNestedColumnReader.readRowdereferences null.Summary and Changelog
Arrays of rows that contain nested arrays of rows can now be read by the Flink COW reader.
This PR fixes
hudi-flink1.18.xonly. The other Flink modules (1.19, 1.20, 2.0, 2.1) have the same code and will get the same change in a follow-up PR once this one is merged.hudi-flink1.18.xParquetSplitReaderUtil: theARRAYbranch ofcreateWritableColumnVectornow locates the element type using the same ParquetLISTbackward-compatibility rules asgetArrayElementColumn, so the vector tree lines up with theParquetFieldtree:LIST-> repeated group -> element),arrayor<list name>_tuple,The depth passed down is incremented at each unwrapped level, matching the
ParquetFieldconstruction. Row fields are still resolved by name, so schema evolution (fields missing from the file) keeps working. The unwrapping lives in a newunwrapListElementTypehelper.hudi-flink1.18.xNestedColumnReader: inside an array, a row has one position per element, which can exceed the batch size that the null placeholders of fields absent from the file were sized with.readRownow grows those placeholders to the position count; otherwise reading an evolved array of rows fails with anArrayIndexOutOfBoundsExceptiononce a batch holds more elements than rows.New
TestParquetSplitReaderUtilinhudi-flink1.18.x. The tests write real Parquet files withExampleParquetWriterand read them back throughParquetColumnarRowSplitReader:array, 2-level_tuple), including empty and null arrays,Impact
Fixes a read failure for nested array-of-row schemas, and arrays of rows without a nested array now return their actual field values instead of nulls. No public API or config changes.
Risk Level
Low. The change is confined to how the element type is located when building column vectors for arrays, reusing the rules
getArrayElementColumnalready applies when building the reader tree, plus resizing placeholder vectors that would otherwise be too small. Verification:ParquetSplitReaderUtilfix, the nested-array tests fail with theNullPointerExceptionabove (the primitive-array tests pass either way).NestedColumnReaderfix, the three small-batch tests fail with anArrayIndexOutOfBoundsExceptioninAbstractHeapVector.isNullAt.-Dflink1.18, as do the existingITTestHoodieDataSource#testParquet*integration tests (17).checkstyle:checkpasses forhudi-flink1.18.x.Documentation Update
none
Contributor's checklist