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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
