You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by "skywalker0618 (via GitHub)" <gi...@apache.org> on 2026/10/14 21:51:11 UTC
[PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
skywalker0618 opened a new pull request, #20337:
URL: https://github.com/apache/hudi/pull/20337
### Describe the issue this Pull Request addresses
Reading a Hudi table through the Flink COW Parquet reader fails with a `NullPointerException` when 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>>>>`:
```
java.lang.NullPointerException
at org.apache.hudi.table.format.cow.vector.reader.NestedColumnReader.readRow(NestedColumnReader.java:112)
```
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`, the `ARRAY` branch passes the Parquet `LIST` group itself as the physical type of the array's element. For an array of rows, the element `ROW` then looks its fields up by name in the `LIST` wrapper instead of in the element group, finds none, and fills every slot with a constant placeholder vector. That's harmless for leaf fields, but the placeholder for an inner array has no child vector, so `NestedColumnReader.readRow` dereferences null.
### Summary and Changelog
Arrays of rows that contain nested arrays of rows can now be read by the Flink COW reader.
- `ParquetSplitReaderUtil` (all five Flink modules: 1.18, 1.19, 1.20, 2.0, 2.1): the `ARRAY` branch of `createWritableColumnVector` now locates the element type using the same Parquet `LIST` backward-compatibility rules as `getArrayElementColumn`, so the vector tree lines up with the `ParquetField` tree:
- standard 3-level lists (`LIST` -> repeated group -> element),
- 2-level lists whose repeated group is named `array` or `<list name>_tuple`,
- repeated groups with multiple fields and repeated primitives (the repeated type is the element).
The depth passed down is incremented at each unwrapped level, matching the `ParquetField` construction. Row fields are still resolved by name, so schema evolution (fields missing from the file) keeps working.
- New `TestParquetSplitReaderUtil` in each Flink module. The tests write real Parquet files with `ExampleParquetWriter` and read them back through `ParquetColumnarRowSplitReader`:
- array of rows containing an array of rows, parameterized over the three list encodings (3-level, 2-level `array`, 2-level `_tuple`), including empty and null arrays,
- the same shape with row fields missing from the file (schema evolution),
- 3-level and 2-level arrays of primitives.
### Impact
Fixes a read failure for nested array-of-row schemas; no public API or config changes, and no behavior change for schemas that were already readable.
### Risk Level
Low. The change is confined to how the element type is located when building column vectors for arrays, and it reuses the rules `getArrayElementColumn` already applies when building the reader tree. Verification:
- Without the fix, 6 of the 8 new tests fail with the `NullPointerException` above (the two primitive-array tests pass either way).
- With the fix, all 8 pass on each of `-Dflink1.18`, `-Dflink1.19`, `-Dflink1.20`, `-Dflink2.0`, `-Dflink2.1`.
- `checkstyle:check` passes for all five modules.
- The same change has been running in a production Flink streaming job reading a ~10B-row COW table with these nested schemas.
### Documentation Update
none
### Contributor's checklist
- [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Enough context is provided in the sections above
- [x] Adequate tests were added if applicable
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "skywalker0618 (via GitHub)" <gi...@apache.org>.
skywalker0618 commented on code in PR #20337:
URL: https://github.com/apache/hudi/pull/20337#discussion_r4328056816
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -622,6 +640,44 @@ private static WritableColumnVector createWritableColumnVector(
}
}
+ /**
+ * Returns the element type of a LIST physical type, the name of the group that contains it and its
+ * depth, unwrapping it with the same backward-compatibility rules as {@link #getArrayElementColumn}
+ * so that the vectors line up with the {@code ParquetField} tree.
+ *
+ * @param parentName name of the group that contains {@code listType}
+ */
+ private static Triple<Type, String, Integer> unwrapListElementType(Type listType, String parentName, int depth) {
+ Type elementType = listType;
+ String elementParentName = parentName;
+ int elementDepth = depth;
+ while (!elementType.isPrimitive() && !elementType.isRepetition(REPEATED)) {
+ elementParentName = elementType.getName();
+ elementType = elementType.asGroupType().getType(0);
+ elementDepth++;
+ }
+ if (isThreeLevelListRepeatedType(elementType, elementParentName)) {
+ elementParentName = elementType.getName();
+ elementType = elementType.asGroupType().getType(0);
+ elementDepth++;
+ }
+ return Triple.of(elementType, elementParentName, elementDepth);
+ }
+
+ /**
+ * Returns whether the repeated type of a list is the synthetic wrapper of a three-level list,
+ * rather than the element itself. Type-based counterpart of the check in
+ * {@link #getArrayElementColumn}, where {@code parentName} is the name of the group that contains
+ * the repeated type, as {@code columnIO.getParent().getName()} is there.
+ */
+ private static boolean isThreeLevelListRepeatedType(Type repeatedType, String parentName) {
Review Comment:
Sure, done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "skywalker0618 (via GitHub)" <gi...@apache.org>.
skywalker0618 commented on code in PR #20337:
URL: https://github.com/apache/hudi/pull/20337#discussion_r4286774050
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java:
##########
@@ -139,6 +141,16 @@ private Tuple2<LevelDelegation, WritableColumnVector> readRow(
levelDelegation.getDefinitionLevel(),
levelDelegation.getRepetitionLevel());
+ // 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++) {
Review Comment:
Hoisted the RowType cast out of the loop. Disregard the helper one as it would have only 1 caller.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "skywalker0618 (via GitHub)" <gi...@apache.org>.
skywalker0618 commented on PR #20337:
URL: https://github.com/apache/hudi/pull/20337#issuecomment-6180131644
@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!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "skywalker0618 (via GitHub)" <gi...@apache.org>.
skywalker0618 commented on PR #20337:
URL: https://github.com/apache/hudi/pull/20337#issuecomment-6222377899
> 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?
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "skywalker0618 (via GitHub)" <gi...@apache.org>.
skywalker0618 commented on code in PR #20337:
URL: https://github.com/apache/hudi/pull/20337#discussion_r4286759232
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -562,10 +563,11 @@ private static WritableColumnVector createWritableColumnVector(
return new HeapDecimalVector(batchSize);
case ARRAY:
ArrayType arrayType = (ArrayType) fieldType;
+ Pair<Type, Integer> element = unwrapListElementType(physicalType, depth);
Review Comment:
Sure I will create a new PR to cover other flink branches once this is merged
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "skywalker0618 (via GitHub)" <gi...@apache.org>.
skywalker0618 commented on code in PR #20337:
URL: https://github.com/apache/hudi/pull/20337#discussion_r4276949512
##########
hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table.format.cow;
+
+import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests for reading nested ARRAY columns through {@link ParquetSplitReaderUtil} and
+ * {@link ParquetColumnarRowSplitReader}, covering the list encodings allowed by the Parquet
+ * backward-compatibility rules.
+ */
+public class TestParquetSplitReaderUtil {
+
+ private static final LogicalType STRING = new VarCharType(VarCharType.MAX_LENGTH);
+
+ @TempDir
+ File tempDir;
+
+ /**
+ * Physical encodings of a list of single- or multi-field element groups.
+ */
+ enum ListEncoding {
+ THREE_LEVEL {
+ @Override
+ String schema(String name, String elementFields) {
+ return "optional group " + name + " (LIST) { repeated group list { optional group element { "
+ + elementFields + " } } }";
+ }
+
+ @Override
+ Group addElement(Group list, String name) {
+ return list.addGroup("list").addGroup("element");
+ }
+ },
+ TWO_LEVEL_ARRAY {
+ @Override
+ String schema(String name, String elementFields) {
+ return "optional group " + name + " (LIST) { repeated group array { " + elementFields + " } }";
+ }
+
+ @Override
+ Group addElement(Group list, String name) {
+ return list.addGroup("array");
+ }
+ },
+ TWO_LEVEL_TUPLE {
+ @Override
+ String schema(String name, String elementFields) {
+ return "optional group " + name + " (LIST) { repeated group " + name + "_tuple { "
+ + elementFields + " } }";
+ }
+
+ @Override
+ Group addElement(Group list, String name) {
+ return list.addGroup(name + "_tuple");
+ }
+ };
+
+ abstract String schema(String name, String elementFields);
+
+ abstract Group addElement(Group list, String name);
+ }
+
+ @ParameterizedTest
+ @EnumSource(ListEncoding.class)
+ void testReadArrayOfRowWithNestedArrayOfRow(ListEncoding encoding) throws IOException {
+ MessageType schema = itemsSchema(encoding);
+ LogicalType itemsType = itemsType(
+ rowType(new String[] {"name", "tags"}, STRING, new ArrayType(rowType(new String[] {"value"}, STRING))));
+
+ List<String> values = writeAndRead(schema, itemsWriter(encoding), "items", itemsType,
+ row -> formatItems(row, false));
+
+ assertEquals(Arrays.asList("[a:[x, y], b:[z]]", "[c:[]]", "null"), values);
+ }
+
+ @ParameterizedTest
+ @EnumSource(ListEncoding.class)
+ void testReadArrayOfRowWithFieldsMissingFromFile(ListEncoding encoding) throws IOException {
Review 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.
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -560,12 +560,27 @@ private static WritableColumnVector createWritableColumnVector(
&& primitiveType.getOriginalType() == OriginalType.DECIMAL,
getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
return new HeapDecimalVector(batchSize);
- case ARRAY:
+ case ARRAY: {
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;
Review Comment:
Done, extracted into `unwrapListElementType`, which returns the element type and its depth.
##########
hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table.format.cow;
+
+import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests for reading nested ARRAY columns through {@link ParquetSplitReaderUtil} and
+ * {@link ParquetColumnarRowSplitReader}, covering the list encodings allowed by the Parquet
+ * backward-compatibility rules.
+ */
+public class TestParquetSplitReaderUtil {
+
+ private static final LogicalType STRING = new VarCharType(VarCharType.MAX_LENGTH);
+
+ @TempDir
+ File tempDir;
+
+ /**
+ * Physical encodings of a list of single- or multi-field element groups.
+ */
+ enum ListEncoding {
+ THREE_LEVEL {
+ @Override
+ String schema(String name, String elementFields) {
+ return "optional group " + name + " (LIST) { repeated group list { optional group element { "
+ + elementFields + " } } }";
+ }
+
+ @Override
+ Group addElement(Group list, String name) {
+ return list.addGroup("list").addGroup("element");
+ }
+ },
+ TWO_LEVEL_ARRAY {
+ @Override
+ String schema(String name, String elementFields) {
+ return "optional group " + name + " (LIST) { repeated group array { " + elementFields + " } }";
+ }
+
+ @Override
+ Group addElement(Group list, String name) {
+ return list.addGroup("array");
+ }
+ },
+ TWO_LEVEL_TUPLE {
+ @Override
+ String schema(String name, String elementFields) {
+ return "optional group " + name + " (LIST) { repeated group " + name + "_tuple { "
+ + elementFields + " } }";
+ }
+
+ @Override
+ Group addElement(Group list, String name) {
+ return list.addGroup(name + "_tuple");
+ }
+ };
+
+ abstract String schema(String name, String elementFields);
+
+ abstract Group addElement(Group list, String name);
+ }
+
+ @ParameterizedTest
+ @EnumSource(ListEncoding.class)
+ void testReadArrayOfRowWithNestedArrayOfRow(ListEncoding encoding) throws IOException {
+ MessageType schema = itemsSchema(encoding);
+ LogicalType itemsType = itemsType(
+ rowType(new String[] {"name", "tags"}, STRING, new ArrayType(rowType(new String[] {"value"}, STRING))));
+
+ List<String> values = writeAndRead(schema, itemsWriter(encoding), "items", itemsType,
+ row -> formatItems(row, false));
+
+ assertEquals(Arrays.asList("[a:[x, y], b:[z]]", "[c:[]]", "null"), values);
+ }
+
+ @ParameterizedTest
+ @EnumSource(ListEncoding.class)
+ void testReadArrayOfRowWithFieldsMissingFromFile(ListEncoding encoding) throws IOException {
+ MessageType schema = itemsSchema(encoding);
+ RowType tagType = rowType(new String[] {"value", "added_tag_field"}, STRING, STRING);
+ LogicalType itemsType = itemsType(
+ rowType(new String[] {"name", "added_scores", "tags"}, STRING, new ArrayType(new IntType()),
+ new ArrayType(tagType)));
+
+ List<String> values = writeAndRead(schema, itemsWriter(encoding), "items", itemsType,
+ row -> formatItems(row, true));
+
+ assertEquals(
+ Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]", "[c(null):[]]", "null"), values);
+ }
+
+ @Test
+ void testReadThreeLevelArrayOfPrimitive() throws IOException {
+ MessageType schema = MessageTypeParser.parseMessageType(
+ "message m { optional group nums (LIST) { repeated group list { required int32 element; } } }");
+ Consumer<Group> writer = record -> {
+ Group list = record.addGroup("nums");
+ list.addGroup("list").append("element", 1);
+ list.addGroup("list").append("element", 2);
+ };
+
+ List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+ row -> formatInts(row.getArray(0)));
+
+ assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+ }
+
+ @Test
+ void testReadTwoLevelArrayOfPrimitive() throws IOException {
+ MessageType schema = MessageTypeParser.parseMessageType(
+ "message m { optional group nums (LIST) { repeated int32 array; } }");
+ Consumer<Group> writer = record -> {
+ Group list = record.addGroup("nums");
+ list.append("array", 1);
+ list.append("array", 2);
+ };
+
+ List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+ row -> formatInts(row.getArray(0)));
+
+ assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+ }
+
+ private static MessageType itemsSchema(ListEncoding encoding) {
+ String tags = encoding.schema("tags", "optional binary value (UTF8);");
+ String items = encoding.schema("items", "optional binary name (UTF8); " + tags);
+ return MessageTypeParser.parseMessageType("message m { " + items + " }");
+ }
+
+ /**
+ * Writes three records: two non-empty item lists (one with an empty nested list) and a null list.
+ */
+ private static Consumer<Group> itemsWriter(ListEncoding encoding) {
+ int[] recordIndex = {0};
+ return record -> {
+ int index = recordIndex[0]++;
+ if (index == 0) {
+ Group items = record.addGroup("items");
+ addItem(encoding, items, "a", "x", "y");
Review 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.RowField`s inline. Also pulled the evolved items type, now used by two tests, into `evolvedItemsType()`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "skywalker0618 (via GitHub)" <gi...@apache.org>.
skywalker0618 commented on PR #20337:
URL: https://github.com/apache/hudi/pull/20337#issuecomment-6272877230
@cshuo Do you mind taking another look? Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "skywalker0618 (via GitHub)" <gi...@apache.org>.
skywalker0618 commented on code in PR #20337:
URL: https://github.com/apache/hudi/pull/20337#discussion_r4276949943
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -560,12 +560,27 @@ private static WritableColumnVector createWritableColumnVector(
&& primitiveType.getOriginalType() == OriginalType.DECIMAL,
getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
return new HeapDecimalVector(batchSize);
- case ARRAY:
+ case ARRAY: {
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;
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #20337:
URL: https://github.com/apache/hudi/pull/20337#discussion_r4279736033
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -622,6 +624,40 @@ private static WritableColumnVector createWritableColumnVector(
}
}
+ /**
+ * Returns the element type of a LIST physical type and its depth, unwrapping it with the same
+ * backward-compatibility rules as {@link #getArrayElementColumn} so that the vectors line up with
+ * the {@code ParquetField} tree.
+ */
+ private static Pair<Type, Integer> unwrapListElementType(Type listType, int depth) {
+ Type elementType = listType;
+ String listName = null;
Review 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?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java:
##########
@@ -139,6 +141,16 @@ private Tuple2<LevelDelegation, WritableColumnVector> readRow(
levelDelegation.getDefinitionLevel(),
levelDelegation.getRepetitionLevel());
+ // 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++) {
Review 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?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -562,10 +563,11 @@ private static WritableColumnVector createWritableColumnVector(
return new HeapDecimalVector(batchSize);
case ARRAY:
ArrayType arrayType = (ArrayType) fieldType;
+ Pair<Type, Integer> element = unwrapListElementType(physicalType, depth);
Review 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?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "cshuo (via GitHub)" <gi...@apache.org>.
cshuo commented on code in PR #20337:
URL: https://github.com/apache/hudi/pull/20337#discussion_r4290326390
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -562,10 +563,11 @@ private static WritableColumnVector createWritableColumnVector(
return new HeapDecimalVector(batchSize);
case ARRAY:
ArrayType arrayType = (ArrayType) fieldType;
+ Pair<Type, Integer> element = unwrapListElementType(physicalType, depth);
return new HeapArrayVector(
batchSize,
createWritableColumnVector(
- batchSize, arrayType.getElementType(), physicalType, descriptors, depth));
+ batchSize, arrayType.getElementType(), element.getLeft(), descriptors, element.getRight()));
Review 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.
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java:
##########
@@ -139,6 +141,17 @@ private Tuple2<LevelDelegation, WritableColumnVector> readRow(
levelDelegation.getDefinitionLevel(),
levelDelegation.getRepetitionLevel());
+ // 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();
+ RowType rowType = (RowType) field.getType();
+ for (int i = 0; i < children.size(); i++) {
+ if (children.get(i) == null && vectorLength(finalChildrenVectors[i]) < positionsCount) {
Review 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?
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -622,6 +624,40 @@ private static WritableColumnVector createWritableColumnVector(
}
}
+ /**
+ * Returns the element type of a LIST physical type and its depth, unwrapping it with the same
+ * backward-compatibility rules as {@link #getArrayElementColumn} so that the vectors line up with
+ * the {@code ParquetField} tree.
+ */
+ private static Pair<Type, Integer> unwrapListElementType(Type listType, int depth) {
+ Type elementType = listType;
+ String listName = null;
+ int elementDepth = depth;
+ while (!elementType.isPrimitive() && !elementType.isRepetition(REPEATED)) {
+ listName = elementType.getName();
+ elementType = elementType.asGroupType().getType(0);
+ elementDepth++;
+ }
+ if (isThreeLevelListRepeatedType(elementType, listName)) {
+ elementType = elementType.asGroupType().getType(0);
+ elementDepth++;
+ }
+ return Pair.of(elementType, elementDepth);
+ }
+
+ /**
+ * Returns whether the repeated type of a list is the synthetic wrapper of a three-level list,
+ * rather than the element itself. Type-based counterpart of the check in
+ * {@link #getArrayElementColumn}.
+ */
+ private static boolean isThreeLevelListRepeatedType(Type repeatedType, @Nullable String listName) {
+ return !repeatedType.isPrimitive()
+ && repeatedType.getLogicalTypeAnnotation() == null
+ && repeatedType.asGroupType().getFieldCount() == 1
+ && !repeatedType.getName().equals("array")
+ && (listName == null || !repeatedType.getName().equals(listName + "_tuple"));
Review 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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "hudi-agent (via GitHub)" <gi...@apache.org>.
hudi-agent commented on code in PR #20337:
URL: https://github.com/apache/hudi/pull/20337#discussion_r4276800525
##########
hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table.format.cow;
+
+import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests for reading nested ARRAY columns through {@link ParquetSplitReaderUtil} and
+ * {@link ParquetColumnarRowSplitReader}, covering the list encodings allowed by the Parquet
+ * backward-compatibility rules.
+ */
+public class TestParquetSplitReaderUtil {
+
+ private static final LogicalType STRING = new VarCharType(VarCharType.MAX_LENGTH);
+
+ @TempDir
+ File tempDir;
+
+ /**
+ * Physical encodings of a list of single- or multi-field element groups.
+ */
+ enum ListEncoding {
+ THREE_LEVEL {
+ @Override
+ String schema(String name, String elementFields) {
+ return "optional group " + name + " (LIST) { repeated group list { optional group element { "
+ + elementFields + " } } }";
+ }
+
+ @Override
+ Group addElement(Group list, String name) {
+ return list.addGroup("list").addGroup("element");
+ }
+ },
+ TWO_LEVEL_ARRAY {
+ @Override
+ String schema(String name, String elementFields) {
+ return "optional group " + name + " (LIST) { repeated group array { " + elementFields + " } }";
+ }
+
+ @Override
+ Group addElement(Group list, String name) {
+ return list.addGroup("array");
+ }
+ },
+ TWO_LEVEL_TUPLE {
+ @Override
+ String schema(String name, String elementFields) {
+ return "optional group " + name + " (LIST) { repeated group " + name + "_tuple { "
+ + elementFields + " } }";
+ }
+
+ @Override
+ Group addElement(Group list, String name) {
+ return list.addGroup(name + "_tuple");
+ }
+ };
+
+ abstract String schema(String name, String elementFields);
+
+ abstract Group addElement(Group list, String name);
+ }
+
+ @ParameterizedTest
+ @EnumSource(ListEncoding.class)
+ void testReadArrayOfRowWithNestedArrayOfRow(ListEncoding encoding) throws IOException {
+ MessageType schema = itemsSchema(encoding);
+ LogicalType itemsType = itemsType(
+ rowType(new String[] {"name", "tags"}, STRING, new ArrayType(rowType(new String[] {"value"}, STRING))));
+
+ List<String> values = writeAndRead(schema, itemsWriter(encoding), "items", itemsType,
+ row -> formatItems(row, false));
+
+ assertEquals(Arrays.asList("[a:[x, y], b:[z]]", "[c:[]]", "null"), values);
+ }
+
+ @ParameterizedTest
+ @EnumSource(ListEncoding.class)
+ void testReadArrayOfRowWithFieldsMissingFromFile(ListEncoding encoding) throws IOException {
Review 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?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -560,12 +560,27 @@ private static WritableColumnVector createWritableColumnVector(
&& primitiveType.getOriginalType() == OriginalType.DECIMAL,
getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
return new HeapDecimalVector(batchSize);
- case ARRAY:
+ case ARRAY: {
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;
Review 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.
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
##########
hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java:
##########
@@ -0,0 +1,289 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hudi.table.format.cow;
+
+import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
+
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.MessageTypeParser;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests for reading nested ARRAY columns through {@link ParquetSplitReaderUtil} and
+ * {@link ParquetColumnarRowSplitReader}, covering the list encodings allowed by the Parquet
+ * backward-compatibility rules.
+ */
+public class TestParquetSplitReaderUtil {
+
+ private static final LogicalType STRING = new VarCharType(VarCharType.MAX_LENGTH);
+
+ @TempDir
+ File tempDir;
+
+ /**
+ * Physical encodings of a list of single- or multi-field element groups.
+ */
+ enum ListEncoding {
+ THREE_LEVEL {
+ @Override
+ String schema(String name, String elementFields) {
+ return "optional group " + name + " (LIST) { repeated group list { optional group element { "
+ + elementFields + " } } }";
+ }
+
+ @Override
+ Group addElement(Group list, String name) {
+ return list.addGroup("list").addGroup("element");
+ }
+ },
+ TWO_LEVEL_ARRAY {
+ @Override
+ String schema(String name, String elementFields) {
+ return "optional group " + name + " (LIST) { repeated group array { " + elementFields + " } }";
+ }
+
+ @Override
+ Group addElement(Group list, String name) {
+ return list.addGroup("array");
+ }
+ },
+ TWO_LEVEL_TUPLE {
+ @Override
+ String schema(String name, String elementFields) {
+ return "optional group " + name + " (LIST) { repeated group " + name + "_tuple { "
+ + elementFields + " } }";
+ }
+
+ @Override
+ Group addElement(Group list, String name) {
+ return list.addGroup(name + "_tuple");
+ }
+ };
+
+ abstract String schema(String name, String elementFields);
+
+ abstract Group addElement(Group list, String name);
+ }
+
+ @ParameterizedTest
+ @EnumSource(ListEncoding.class)
+ void testReadArrayOfRowWithNestedArrayOfRow(ListEncoding encoding) throws IOException {
+ MessageType schema = itemsSchema(encoding);
+ LogicalType itemsType = itemsType(
+ rowType(new String[] {"name", "tags"}, STRING, new ArrayType(rowType(new String[] {"value"}, STRING))));
+
+ List<String> values = writeAndRead(schema, itemsWriter(encoding), "items", itemsType,
+ row -> formatItems(row, false));
+
+ assertEquals(Arrays.asList("[a:[x, y], b:[z]]", "[c:[]]", "null"), values);
+ }
+
+ @ParameterizedTest
+ @EnumSource(ListEncoding.class)
+ void testReadArrayOfRowWithFieldsMissingFromFile(ListEncoding encoding) throws IOException {
+ MessageType schema = itemsSchema(encoding);
+ RowType tagType = rowType(new String[] {"value", "added_tag_field"}, STRING, STRING);
+ LogicalType itemsType = itemsType(
+ rowType(new String[] {"name", "added_scores", "tags"}, STRING, new ArrayType(new IntType()),
+ new ArrayType(tagType)));
+
+ List<String> values = writeAndRead(schema, itemsWriter(encoding), "items", itemsType,
+ row -> formatItems(row, true));
+
+ assertEquals(
+ Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]", "[c(null):[]]", "null"), values);
+ }
+
+ @Test
+ void testReadThreeLevelArrayOfPrimitive() throws IOException {
+ MessageType schema = MessageTypeParser.parseMessageType(
+ "message m { optional group nums (LIST) { repeated group list { required int32 element; } } }");
+ Consumer<Group> writer = record -> {
+ Group list = record.addGroup("nums");
+ list.addGroup("list").append("element", 1);
+ list.addGroup("list").append("element", 2);
+ };
+
+ List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+ row -> formatInts(row.getArray(0)));
+
+ assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+ }
+
+ @Test
+ void testReadTwoLevelArrayOfPrimitive() throws IOException {
+ MessageType schema = MessageTypeParser.parseMessageType(
+ "message m { optional group nums (LIST) { repeated int32 array; } }");
+ Consumer<Group> writer = record -> {
+ Group list = record.addGroup("nums");
+ list.append("array", 1);
+ list.append("array", 2);
+ };
+
+ List<String> values = writeAndRead(schema, writer, "nums", new ArrayType(new IntType()),
+ row -> formatInts(row.getArray(0)));
+
+ assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+ }
+
+ private static MessageType itemsSchema(ListEncoding encoding) {
+ String tags = encoding.schema("tags", "optional binary value (UTF8);");
+ String items = encoding.schema("items", "optional binary name (UTF8); " + tags);
+ return MessageTypeParser.parseMessageType("message m { " + items + " }");
+ }
+
+ /**
+ * Writes three records: two non-empty item lists (one with an empty nested list) and a null list.
+ */
+ private static Consumer<Group> itemsWriter(ListEncoding encoding) {
+ int[] recordIndex = {0};
+ return record -> {
+ int index = recordIndex[0]++;
+ if (index == 0) {
+ Group items = record.addGroup("items");
+ addItem(encoding, items, "a", "x", "y");
Review Comment:
🤖 nit: `itemsType` just wraps `new ArrayType(...)` and `rowType` just wraps `RowType.of(...)`; could you inline them at the call sites?
<sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "skywalker0618 (via GitHub)" <gi...@apache.org>.
skywalker0618 commented on code in PR #20337:
URL: https://github.com/apache/hudi/pull/20337#discussion_r4297997671
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java:
##########
@@ -139,6 +141,17 @@ private Tuple2<LevelDelegation, WritableColumnVector> readRow(
levelDelegation.getDefinitionLevel(),
levelDelegation.getRepetitionLevel());
+ // 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();
+ RowType rowType = (RowType) field.getType();
+ for (int i = 0; i < children.size(); i++) {
+ if (children.get(i) == null && vectorLength(finalChildrenVectors[i]) < positionsCount) {
Review Comment:
Thanks, fixed. I found this problem exists in flink too, and was brought in hudi during the backport.
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -562,10 +563,11 @@ private static WritableColumnVector createWritableColumnVector(
return new HeapDecimalVector(batchSize);
case ARRAY:
ArrayType arrayType = (ArrayType) fieldType;
+ Pair<Type, Integer> element = unwrapListElementType(physicalType, depth);
return new HeapArrayVector(
batchSize,
createWritableColumnVector(
- batchSize, arrayType.getElementType(), physicalType, descriptors, depth));
+ batchSize, arrayType.getElementType(), element.getLeft(), descriptors, element.getRight()));
Review 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.
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -622,6 +624,40 @@ private static WritableColumnVector createWritableColumnVector(
}
}
+ /**
+ * Returns the element type of a LIST physical type and its depth, unwrapping it with the same
+ * backward-compatibility rules as {@link #getArrayElementColumn} so that the vectors line up with
+ * the {@code ParquetField} tree.
+ */
+ private static Pair<Type, Integer> unwrapListElementType(Type listType, int depth) {
+ Type elementType = listType;
+ String listName = null;
+ int elementDepth = depth;
+ while (!elementType.isPrimitive() && !elementType.isRepetition(REPEATED)) {
+ listName = elementType.getName();
+ elementType = elementType.asGroupType().getType(0);
+ elementDepth++;
+ }
+ if (isThreeLevelListRepeatedType(elementType, listName)) {
+ elementType = elementType.asGroupType().getType(0);
+ elementDepth++;
+ }
+ return Pair.of(elementType, elementDepth);
+ }
+
+ /**
+ * Returns whether the repeated type of a list is the synthetic wrapper of a three-level list,
+ * rather than the element itself. Type-based counterpart of the check in
+ * {@link #getArrayElementColumn}.
+ */
+ private static boolean isThreeLevelListRepeatedType(Type repeatedType, @Nullable String listName) {
+ return !repeatedType.isPrimitive()
+ && repeatedType.getLogicalTypeAnnotation() == null
+ && repeatedType.asGroupType().getFieldCount() == 1
+ && !repeatedType.getName().equals("array")
+ && (listName == null || !repeatedType.getName().equals(listName + "_tuple"));
Review Comment:
Fixed and added test cases too
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #20337:
URL: https://github.com/apache/hudi/pull/20337#issuecomment-6180847636
## [Codecov](https://app.codecov.io/gh/apache/hudi/pull/20337?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 80.71%. Comparing base ([`6bbf798`](https://app.codecov.io/gh/apache/hudi/commit/6bbf798f88d468d1ca3ce774aae991145b0cd0a2?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`3ea6452`](https://app.codecov.io/gh/apache/hudi/commit/3ea6452d43cac34857bdfeffc55cdefdf6e68d81?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #20337 +/- ##
=========================================
Coverage 80.71% 80.71%
Complexity 35152 35152
=========================================
Files 2553 2553
Lines 143556 143556
Branches 17482 17482
=========================================
Hits 115875 115875
Misses 19737 19737
Partials 7944 7944
```
| [Components](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hudi-common](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `84.18% <ø> (ø)` | |
| [hudi-client](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.94% <ø> (ø)` | |
| [hudi-flink](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `86.08% <ø> (ø)` | |
| [hudi-spark-datasource](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `74.20% <ø> (ø)` | |
| [hudi-utilities](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `78.18% <ø> (ø)` | |
| [hudi-cli](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `70.80% <ø> (ø)` | |
| [hudi-hadoop](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `71.17% <ø> (ø)` | |
| [hudi-sync](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `76.62% <ø> (ø)` | |
| [hudi-io](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.52% <ø> (ø)` | |
| [hudi-timeline-service](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `83.41% <ø> (ø)` | |
| [hudi-cloud](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `81.00% <ø> (ø)` | |
| [hudi-kafka-connect](https://app.codecov.io/gh/apache/hudi/pull/20337/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `54.48% <ø> (ø)` | |
| [Flag](https://app.codecov.io/gh/apache/hudi/pull/20337/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [hadoop-mr-java-client](https://app.codecov.io/gh/apache/hudi/pull/20337/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `43.96% <ø> (+0.02%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "skywalker0618 (via GitHub)" <gi...@apache.org>.
skywalker0618 commented on code in PR #20337:
URL: https://github.com/apache/hudi/pull/20337#discussion_r4286778929
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -622,6 +624,40 @@ private static WritableColumnVector createWritableColumnVector(
}
}
+ /**
+ * Returns the element type of a LIST physical type and its depth, unwrapping it with the same
+ * backward-compatibility rules as {@link #getArrayElementColumn} so that the vectors line up with
+ * the {@code ParquetField} tree.
+ */
+ private static Pair<Type, Integer> unwrapListElementType(Type listType, int depth) {
+ Type elementType = listType;
+ String listName = null;
Review 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "danny0405 (via GitHub)" <gi...@apache.org>.
danny0405 merged PR #20337:
URL: https://github.com/apache/hudi/pull/20337
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] fix(flink): unwrap the list element type when creating vectors for nested arrays [hudi]
Posted by "cshuo (via GitHub)" <gi...@apache.org>.
cshuo commented on code in PR #20337:
URL: https://github.com/apache/hudi/pull/20337#discussion_r4321137599
##########
hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java:
##########
@@ -622,6 +640,44 @@ private static WritableColumnVector createWritableColumnVector(
}
}
+ /**
+ * Returns the element type of a LIST physical type, the name of the group that contains it and its
+ * depth, unwrapping it with the same backward-compatibility rules as {@link #getArrayElementColumn}
+ * so that the vectors line up with the {@code ParquetField} tree.
+ *
+ * @param parentName name of the group that contains {@code listType}
+ */
+ private static Triple<Type, String, Integer> unwrapListElementType(Type listType, String parentName, int depth) {
+ Type elementType = listType;
+ String elementParentName = parentName;
+ int elementDepth = depth;
+ while (!elementType.isPrimitive() && !elementType.isRepetition(REPEATED)) {
+ elementParentName = elementType.getName();
+ elementType = elementType.asGroupType().getType(0);
+ elementDepth++;
+ }
+ if (isThreeLevelListRepeatedType(elementType, elementParentName)) {
+ elementParentName = elementType.getName();
+ elementType = elementType.asGroupType().getType(0);
+ elementDepth++;
+ }
+ return Triple.of(elementType, elementParentName, elementDepth);
+ }
+
+ /**
+ * Returns whether the repeated type of a list is the synthetic wrapper of a three-level list,
+ * rather than the element itself. Type-based counterpart of the check in
+ * {@link #getArrayElementColumn}, where {@code parentName} is the name of the group that contains
+ * the repeated type, as {@code columnIO.getParent().getName()} is there.
+ */
+ private static boolean isThreeLevelListRepeatedType(Type repeatedType, String parentName) {
Review Comment:
[`isThreeLevelListRepeatedType()` (lines 673–678)](https://github.com/apache/hudi/blob/f9f2acfae2e7819e398a12a0dee9ba99e4f4d68b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java#L673-L678) duplicates the checks already in [`getArrayElementColumn()` (lines 881–885)](https://github.com/apache/hudi/blob/f9f2acfae2e7819e398a12a0dee9ba99e4f4d68b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java#L881-L885): a group with no logical annotation, exactly one child, and neither legacy name (`array` or `<parent>_tuple`). Could we have `getArrayElementColumn()` call this helper so both paths share the same LIST compatibility rules?
```java
if (columnIO instanceof GroupColumnIO
&& isThreeLevelListRepeatedType(
columnIO.getType(), columnIO.getParent().getName())) {
return ((GroupColumnIO) columnIO).getChild(0);
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@hudi.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org