This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new b5b05d1cafdd fix(flink): unwrap the list element type when creating
vectors for nested arrays (#20337)
b5b05d1cafdd is described below
commit b5b05d1cafdda4549ca15b9d068b847033097c4f
Author: Shihuan Liu <[email protected]>
AuthorDate: Tue Oct 20 19:35:05 2026 -0700
fix(flink): unwrap the list element type when creating vectors for nested
arrays (#20337)
* fix(flink): unwrap the list element type when creating vectors for nested
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.
* Fix review comments
* Address review comments: row position count, integer-encoded decimals,
_tuple list elements
* Share the three-level list check between the vector and ParquetField paths
---
.../table/format/cow/ParquetSplitReaderUtil.java | 85 ++++-
.../table/format/cow/utils/NestedPositionUtil.java | 4 +-
.../cow/vector/reader/NestedColumnReader.java | 13 +
.../reader/ParquetColumnarRowSplitReader.java | 1 +
.../format/cow/TestParquetSplitReaderUtil.java | 415 +++++++++++++++++++++
.../format/cow/utils/TestNestedPositionUtil.java | 66 ++++
6 files changed, 568 insertions(+), 16 deletions(-)
diff --git
a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
index 33da58dd3ad5..e702461b5203 100644
---
a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
+++
b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
@@ -19,6 +19,7 @@
package org.apache.hudi.table.format.cow;
import org.apache.hudi.common.util.ValidationUtils;
+import org.apache.hudi.common.util.collection.Triple;
import org.apache.hudi.table.format.cow.vector.HeapArrayVector;
import org.apache.hudi.table.format.cow.vector.HeapDecimalVector;
import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector;
@@ -483,19 +484,24 @@ public class ParquetSplitReaderUtil {
* matches {@code fieldType}; for ROW types missing physical fields are
slotted with null-filled
* vectors (sourced from {@link #createVectorFromConstant}) so that the
Dremel assembler in
* {@link NestedColumnReader} can pass them through unchanged.
+ *
+ * @param parentName name of the group that contains {@code physicalType},
which decides whether a
+ * repeated group named {@code <parentName>_tuple} is a list element
*/
public static WritableColumnVector createWritableColumnVector(
int batchSize,
LogicalType fieldType,
Type physicalType,
+ String parentName,
List<ColumnDescriptor> descriptors) {
- return createWritableColumnVector(batchSize, fieldType, physicalType,
descriptors, 0);
+ return createWritableColumnVector(batchSize, fieldType, physicalType,
parentName, descriptors, 0);
}
private static WritableColumnVector createWritableColumnVector(
int batchSize,
LogicalType fieldType,
Type physicalType,
+ String parentName,
List<ColumnDescriptor> columns,
int depth) {
List<ColumnDescriptor> descriptors = filterDescriptors(depth,
physicalType, columns);
@@ -555,26 +561,35 @@ public class ParquetSplitReaderUtil {
return new HeapTimestampVector(batchSize);
case DECIMAL:
checkArgument(
- (typeName == PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY
- || typeName == PrimitiveType.PrimitiveTypeName.BINARY)
- && primitiveType.getOriginalType() == OriginalType.DECIMAL,
+ primitiveType.getOriginalType() == OriginalType.DECIMAL,
getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
- return new HeapDecimalVector(batchSize);
+ switch (typeName) {
+ case INT32:
+ return new HeapIntVector(batchSize);
+ case INT64:
+ return new HeapLongVector(batchSize);
+ case BINARY:
+ case FIXED_LEN_BYTE_ARRAY:
+ return new HeapDecimalVector(batchSize);
+ default:
+ throw new
IllegalArgumentException(getPrimitiveTypeCheckFailureMessage(typeName,
fieldType));
+ }
case ARRAY:
ArrayType arrayType = (ArrayType) fieldType;
+ Triple<Type, String, Integer> element =
unwrapListElementType(physicalType, parentName, depth);
return new HeapArrayVector(
batchSize,
- createWritableColumnVector(
- batchSize, arrayType.getElementType(), physicalType,
descriptors, depth));
+ createWritableColumnVector(batchSize, arrayType.getElementType(),
element.getLeft(),
+ element.getMiddle(), descriptors, element.getRight()));
case MAP: {
MapType mapType = (MapType) fieldType;
GroupType repeatedType = unwrapMapRepeatedType(physicalType);
return new HeapMapColumnVector(
batchSize,
- createWritableColumnVector(
- batchSize, mapType.getKeyType(), repeatedType.getType(0),
descriptors, depth + 2),
- createWritableColumnVector(
- batchSize, mapType.getValueType(), repeatedType.getType(1),
descriptors, depth + 2));
+ createWritableColumnVector(batchSize, mapType.getKeyType(),
repeatedType.getType(0),
+ repeatedType.getName(), descriptors, depth + 2),
+ createWritableColumnVector(batchSize, mapType.getValueType(),
repeatedType.getType(1),
+ repeatedType.getName(), descriptors, depth + 2));
}
case MULTISET: {
MultisetType multisetType = (MultisetType) fieldType;
@@ -585,12 +600,14 @@ public class ParquetSplitReaderUtil {
batchSize,
multisetType.getElementType(),
repeatedType.getType(0),
+ repeatedType.getName(),
descriptors,
depth + 2),
createWritableColumnVector(
batchSize,
new IntType(false),
repeatedType.getType(1),
+ repeatedType.getName(),
descriptors,
depth + 2));
}
@@ -612,6 +629,7 @@ public class ParquetSplitReaderUtil {
batchSize,
rowType.getTypeAt(i),
groupType.getType(fieldIndex),
+ groupType.getName(),
descriptors,
depth + 1);
}
@@ -622,6 +640,46 @@ public class ParquetSplitReaderUtil {
}
}
+ /**
+ * 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. Shared by {@link #unwrapListElementType}
and
+ * {@link #getArrayElementColumn}, so that the vectors and the {@code
ParquetField} tree pick the
+ * same element.
+ *
+ * @param parentName name of the group that contains {@code repeatedType}
+ */
+ private static boolean isThreeLevelListRepeatedType(Type repeatedType,
String parentName) {
+ return !repeatedType.isPrimitive()
+ && repeatedType.getLogicalTypeAnnotation() == null
+ && repeatedType.asGroupType().getFieldCount() == 1
+ && !repeatedType.getName().equals("array")
+ && !repeatedType.getName().equals(parentName + "_tuple");
+ }
+
/**
* Peels one {@code repeated group key_value} wrapper off a MAP / MULTISET
physical type, matching
* Parquet's canonical 3-level map encoding.
@@ -823,10 +881,7 @@ public class ParquetSplitReaderUtil {
// Three-level list: skip the synthetic `element` / `list` wrapper when
present.
if (columnIO instanceof GroupColumnIO
- && columnIO.getType().getLogicalTypeAnnotation() == null
- && ((GroupColumnIO) columnIO).getChildrenCount() == 1
- && !columnIO.getName().equals("array")
- && !columnIO.getName().equals(columnIO.getParent().getName() +
"_tuple")) {
+ && isThreeLevelListRepeatedType(columnIO.getType(),
columnIO.getParent().getName())) {
return ((GroupColumnIO) columnIO).getChild(0);
}
return columnIO;
diff --git
a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
index 3f2f8976b69b..52745f747d25 100644
---
a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
+++
b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
@@ -69,7 +69,9 @@ public class NestedPositionUtil {
}
}
if (nullValuesCount == 0) {
- return new RowPosition(null, fieldDefinitionLevels.length);
+ // Diverges from Flink, which returns fieldDefinitionLevels.length here:
that also counts the
+ // skipped values of a last field that is an array, one per element
instead of one per row.
+ return new RowPosition(null, nullRowFlags.size());
}
return new RowPosition(nullRowFlags.toArray(), nullRowFlags.size());
}
diff --git
a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
index 60575f148cc4..bbfe5fcff31f 100644
---
a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
+++
b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
@@ -49,6 +49,8 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import static
org.apache.hudi.table.format.cow.ParquetSplitReaderUtil.createVectorFromConstant;
+
/**
* ColumnReader used to read a {@code Group} type in Parquet ({@code Map},
{@code Array}, {@code
* Row}). Resolves nested structures using Dremel striping/assembly; see <a
@@ -139,6 +141,17 @@ public class NestedColumnReader implements
ColumnReader<WritableColumnVector> {
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) {
+ finalChildrenVectors[i] = (WritableColumnVector)
createVectorFromConstant(
+ rowType.getTypeAt(i), null, positionsCount);
+ }
+ }
+
// If row was inside the structure, then we need to renew the vector to
reset the
// capacity.
if (inside) {
diff --git
a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
index 1826419db5d4..b9f7527a91a8 100644
---
a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
+++
b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
@@ -250,6 +250,7 @@ public class ParquetColumnarRowSplitReader implements
Closeable {
batchSize,
requestedTypes[i],
types.get(i),
+ requestedSchema.getName(),
descriptors);
} catch (IllegalArgumentException e) {
String fieldName = requestedSchema.getFieldName(i);
diff --git
a/hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
b/hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
new file mode 100644
index 000000000000..3763003e554d
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/table/format/cow/TestParquetSplitReaderUtil.java
@@ -0,0 +1,415 @@
+/*
+ * 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.DecimalType;
+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.CsvSource;
+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);
+
+ private static final int DEFAULT_BATCH_SIZE = 2048;
+
+ @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 = new ArrayType(
+ 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 {
+ List<String> values = writeAndRead(itemsSchema(encoding),
itemsWriter(encoding), "items",
+ evolvedItemsType(), row -> formatItems(row, true), DEFAULT_BATCH_SIZE);
+
+ assertEquals(
+ Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]",
"[c(null):[]]", "null"), values);
+ }
+
+ @ParameterizedTest
+ @EnumSource(ListEncoding.class)
+ void
testReadArrayOfRowWithFieldsMissingFromFileInBatchSmallerThanElementCount(ListEncoding
encoding)
+ throws IOException {
+ // The first batch holds two rows with three items and three tags, so the
placeholders of the
+ // absent fields must grow beyond the batch size.
+ List<String> values = writeAndRead(itemsSchema(encoding),
itemsWriter(encoding), "items",
+ evolvedItemsType(), row -> formatItems(row, true), 2);
+
+ assertEquals(
+ Arrays.asList("[a(null):[x(null), y(null)], b(null):[z(null)]]",
"[c(null):[]]", "null"), values);
+ }
+
+ @Test
+ void testReadBareRepeatedGroupNamedAfterItsParentAsArrayOfRow() throws
IOException {
+ // A repeated group with one field named <parent>_tuple is the element
itself, not a wrapper.
+ MessageType schema = MessageTypeParser.parseMessageType(
+ "message m { optional group info { repeated group info_tuple {
optional int32 x; } } }");
+ Consumer<Group> writer = record -> {
+ Group info = record.addGroup("info");
+ info.addGroup("info_tuple").append("x", 1);
+ info.addGroup("info_tuple").append("x", 2);
+ };
+ LogicalType infoType = rowType(new String[] {"info_tuple"}, new
ArrayType(rowType(new String[] {"x"}, new IntType())));
+
+ List<String> values = writeAndRead(schema, writer, "info", infoType, row
-> {
+ ArrayData tuples = row.getRow(0, 1).getArray(0);
+ List<Integer> xs = new ArrayList<>();
+ for (int i = 0; i < tuples.size(); i++) {
+ xs.add(tuples.getRow(i, 1).getInt(0));
+ }
+ return xs.toString();
+ });
+
+ assertEquals(Arrays.asList("[1, 2]", "[1, 2]", "[1, 2]"), values);
+ }
+
+ @Test
+ void testReadRowWithFieldMissingFromFileAfterLongArray() throws IOException {
+ // The row's last field in the file is an array with more elements than
the batch size, which
+ // must not count as more rows when sizing the placeholder of the absent
field.
+ MessageType schema = MessageTypeParser.parseMessageType(
+ "message m { optional group info { optional binary name (UTF8); "
+ + "optional group nums (LIST) { repeated group list { optional
int32 element; } } } }");
+ int[] recordIndex = {0};
+ Consumer<Group> writer = record -> {
+ Group info = record.addGroup("info");
+ info.append("name", "r" + recordIndex[0]++);
+ Group nums = info.addGroup("nums");
+ for (int i = 0; i < 100; i++) {
+ nums.addGroup("list").append("element", i);
+ }
+ };
+ LogicalType infoType = rowType(new String[] {"name", "nums", "added"},
STRING, new ArrayType(new IntType()), STRING);
+
+ List<String> values = writeAndRead(schema, writer, "info", infoType, row
-> {
+ RowData info = row.getRow(0, 3);
+ return info.getString(0) + ":" + info.getArray(1).size() + ":" +
(info.isNullAt(2) ? "null" : info.getString(2));
+ }, 2);
+
+ assertEquals(Arrays.asList("r0:100:null", "r1:100:null", "r2:100:null"),
values);
+ }
+
+ @ParameterizedTest
+ @CsvSource({"int32, 9", "int64, 18"})
+ void testReadArrayOfRowWithIntegerEncodedDecimal(String physicalType, int
precision) throws IOException {
+ MessageType schema = MessageTypeParser.parseMessageType(
+ "message m { optional group items (LIST) { repeated group list {
optional group element { "
+ + "optional " + physicalType + " amount (DECIMAL(" + precision +
",2)); } } } }");
+ Consumer<Group> writer = record -> {
+ Group items = record.addGroup("items");
+ appendUnscaled(items.addGroup("list").addGroup("element"), physicalType,
12345);
+ appendUnscaled(items.addGroup("list").addGroup("element"), physicalType,
-1);
+ };
+ LogicalType itemsType = new ArrayType(rowType(new String[] {"amount"}, new
DecimalType(precision, 2)));
+
+ List<String> values = writeAndRead(schema, writer, "items", itemsType, row
-> {
+ ArrayData items = row.getArray(0);
+ List<String> amounts = new ArrayList<>();
+ for (int i = 0; i < items.size(); i++) {
+ amounts.add(items.getRow(i, 1).getDecimal(0, precision, 2).toString());
+ }
+ return amounts.toString();
+ });
+
+ assertEquals(Arrays.asList("[123.45, -0.01]", "[123.45, -0.01]", "[123.45,
-0.01]"), values);
+ }
+
+ @ParameterizedTest
+ @CsvSource({"int32, 9", "int64, 18"})
+ void testReadIntegerEncodedDecimal(String physicalType, int precision)
throws IOException {
+ MessageType schema = MessageTypeParser.parseMessageType(
+ "message m { optional " + physicalType + " amount (DECIMAL(" +
precision + ",2)); }");
+
+ List<String> values = writeAndRead(schema, record ->
appendUnscaled(record, physicalType, 12345), "amount",
+ new DecimalType(precision, 2), row -> row.getDecimal(0, precision,
2).toString());
+
+ assertEquals(Arrays.asList("123.45", "123.45", "123.45"), 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);
+ }
+
+ /**
+ * Appends the unscaled value of a decimal to the {@code amount} field, as
the given physical type.
+ */
+ private static void appendUnscaled(Group group, String physicalType, int
unscaled) {
+ if (physicalType.equals("int32")) {
+ group.append("amount", unscaled);
+ } else {
+ group.append("amount", (long) unscaled);
+ }
+ }
+
+ 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");
+ addItem(encoding, items, "b", "z");
+ } else if (index == 1) {
+ addItem(encoding, record.addGroup("items"), "c");
+ }
+ };
+ }
+
+ private static void addItem(ListEncoding encoding, Group items, String name,
String... tags) {
+ Group item = encoding.addElement(items, "items");
+ item.append("name", name);
+ Group tagList = item.addGroup("tags");
+ for (String tag : tags) {
+ encoding.addElement(tagList, "tags").append("value", tag);
+ }
+ }
+
+ /**
+ * Returns the items type with fields that the written files don't have:
{@code added_scores} on
+ * the item and {@code added_tag_field} on the nested tag.
+ */
+ private static LogicalType evolvedItemsType() {
+ RowType tagType = rowType(new String[] {"value", "added_tag_field"},
STRING, STRING);
+ return new ArrayType(
+ rowType(new String[] {"name", "added_scores", "tags"}, STRING, new
ArrayType(new IntType()),
+ new ArrayType(tagType)));
+ }
+
+ private static RowType rowType(String[] names, LogicalType... types) {
+ return RowType.of(types, names);
+ }
+
+ private List<String> writeAndRead(
+ MessageType schema,
+ Consumer<Group> recordWriter,
+ String columnName,
+ LogicalType columnType,
+ Function<RowData, String> formatter) throws IOException {
+ return writeAndRead(schema, recordWriter, columnName, columnType,
formatter, DEFAULT_BATCH_SIZE);
+ }
+
+ private List<String> writeAndRead(
+ MessageType schema,
+ Consumer<Group> recordWriter,
+ String columnName,
+ LogicalType columnType,
+ Function<RowData, String> formatter,
+ int batchSize) throws IOException {
+ Path path = new Path(new File(tempDir, "data.parquet").getAbsolutePath());
+ Configuration conf = new Configuration();
+ SimpleGroupFactory factory = new SimpleGroupFactory(schema);
+ try (ParquetWriter<Group> writer =
ExampleParquetWriter.builder(path).withType(schema).withConf(conf).build()) {
+ for (int i = 0; i < 3; i++) {
+ Group record = factory.newGroup();
+ recordWriter.accept(record);
+ writer.write(record);
+ }
+ }
+
+ List<String> values = new ArrayList<>();
+ try (ParquetColumnarRowSplitReader reader = new
ParquetColumnarRowSplitReader(
+ true,
+ true,
+ conf,
+ new LogicalType[] {columnType},
+ new String[] {columnName},
+ VectorizedColumnBatch::new,
+ batchSize,
+ path,
+ 0,
+ new File(tempDir, "data.parquet").length(),
+ null,
+ null)) {
+ while (!reader.reachedEnd()) {
+ values.add(formatter.apply(reader.nextRecord()));
+ }
+ }
+ return values;
+ }
+
+ /**
+ * Formats {@code array<row<name, [added_scores,] array<row<value[,
added_tag_field]>>>>}. With
+ * {@code evolved}, the added fields are printed in parentheses after the
existing ones.
+ */
+ private static String formatItems(RowData row, boolean evolved) {
+ if (row.isNullAt(0)) {
+ return "null";
+ }
+ ArrayData items = row.getArray(0);
+ int tagsPos = evolved ? 2 : 1;
+ List<String> formatted = new ArrayList<>();
+ for (int i = 0; i < items.size(); i++) {
+ RowData item = items.getRow(i, tagsPos + 1);
+ String name = item.getString(0).toString();
+ if (evolved) {
+ name += "(" + (item.isNullAt(1) ? "null" :
formatInts(item.getArray(1))) + ")";
+ }
+ ArrayData tags = item.getArray(tagsPos);
+ List<String> tagValues = new ArrayList<>();
+ for (int j = 0; j < tags.size(); j++) {
+ RowData tag = tags.getRow(j, evolved ? 2 : 1);
+ String value = tag.getString(0).toString();
+ if (evolved) {
+ value += "(" + (tag.isNullAt(1) ? "null" :
tag.getString(1).toString()) + ")";
+ }
+ tagValues.add(value);
+ }
+ formatted.add(name + ":" + tagValues);
+ }
+ return formatted.toString();
+ }
+
+ private static String formatInts(ArrayData array) {
+ List<Integer> ints = new ArrayList<>();
+ for (int i = 0; i < array.size(); i++) {
+ ints.add(array.getInt(i));
+ }
+ return ints.toString();
+ }
+}
diff --git
a/hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
b/hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
new file mode 100644
index 000000000000..0bfb1a8aa668
--- /dev/null
+++
b/hudi-flink-datasource/hudi-flink1.18.x/src/test/java/org/apache/hudi/table/format/cow/utils/TestNestedPositionUtil.java
@@ -0,0 +1,66 @@
+/*
+ * 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.utils;
+
+import org.apache.hudi.table.format.cow.vector.position.RowPosition;
+import org.apache.hudi.table.format.cow.vector.type.ParquetGroupField;
+
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.RowType;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+/**
+ * Tests for {@link NestedPositionUtil}.
+ */
+class TestNestedPositionUtil {
+
+ /** An optional top-level row whose last field is an optional list of
optional ints. */
+ private static final ParquetGroupField ROW =
+ new ParquetGroupField(RowType.of(new IntType()), 0, 1, false,
Collections.emptyList());
+
+ @Test
+ void testRowOffsetsCountOneValuePerRowWhenTheLastFieldIsAnArray() {
+ // two rows of three elements each, then the first element of the next row
+ int[] definitionLevels = {4, 4, 4, 4, 4, 4, 4};
+ int[] repetitionLevels = {0, 1, 1, 0, 1, 1, 0};
+
+ RowPosition position = NestedPositionUtil.calculateRowOffsets(ROW,
definitionLevels, repetitionLevels);
+
+ assertEquals(3, position.getPositionsCount());
+ assertNull(position.getIsNull());
+ }
+
+ @Test
+ void testRowOffsetsWithNullRows() {
+ // a row of three elements, a null row, then the first element of the next
row
+ int[] definitionLevels = {4, 4, 4, 0, 4};
+ int[] repetitionLevels = {0, 1, 1, 0, 0};
+
+ RowPosition position = NestedPositionUtil.calculateRowOffsets(ROW,
definitionLevels, repetitionLevels);
+
+ assertEquals(3, position.getPositionsCount());
+ assertArrayEquals(new boolean[] {false, true, false},
position.getIsNull());
+ }
+}
