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