You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hudi.apache.org by vh...@apache.org on 2026/08/05 07:35:22 UTC
(hudi) 04/12: feat(flink): Backport Flink 2.1 Dremel nested Parquet reader rewrite to hudi-flink1.19.x (FLINK-35702) (#18809)
This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch release-1.2.1
in repository https://gitbox.apache.org/repos/asf/hudi.git
commit a15bc7bd882775d8316e16ccdb9de18f0d7bfe47
Author: Shihuan Liu <sk...@gmail.com>
AuthorDate: Mon May 25 19:23:15 2026 -0700
feat(flink): Backport Flink 2.1 Dremel nested Parquet reader rewrite to hudi-flink1.19.x (FLINK-35702) (#18809)
* feat(flink): Backport Flink 2.1 Dremel nested Parquet reader rewrite to hudi-flink1.19.x (FLINK-35702)
(cherry picked from commit 06dc09b1fedd6ff5640c95f28c8136e8a78bce98)
---
.../table/format/cow/utils/NestedPositionUtil.java | 4 +-
.../reader/ParquetDataColumnReaderFactory.java | 8 +-
.../table/format/cow/ParquetSplitReaderUtil.java | 688 ++++++++++++---------
.../table/format/cow/utils/BooleanArrayList.java | 69 +++
.../hudi/table/format/cow/utils/IntArrayList.java | 93 +++
.../hudi/table/format/cow/utils/LongArrayList.java | 83 +++
.../table/format/cow/utils/NestedPositionUtil.java | 4 +-
.../format/cow/vector/ColumnarGroupArrayData.java | 179 ------
.../format/cow/vector/ColumnarGroupMapData.java | 63 --
.../format/cow/vector/ColumnarGroupRowData.java | 138 -----
.../cow/vector/HeapArrayGroupColumnVector.java | 53 --
.../table/format/cow/vector/HeapArrayVector.java | 31 +
.../format/cow/vector/HeapMapColumnVector.java | 75 ++-
.../format/cow/vector/HeapRowColumnVector.java | 15 +
.../format/cow/vector/ParquetDecimalVector.java | 197 +++++-
.../cow/vector/position/CollectionPosition.java | 57 ++
.../cow/vector/position/LevelDelegation.java | 43 ++
.../format/cow/vector/position/RowPosition.java | 45 ++
.../cow/vector/reader/ArrayColumnReader.java | 473 --------------
.../format/cow/vector/reader/ArrayGroupReader.java | 45 --
.../format/cow/vector/reader/MapColumnReader.java | 56 --
.../cow/vector/reader/NestedColumnReader.java | 278 +++++++++
.../vector/reader/NestedPrimitiveColumnReader.java | 639 +++++++++++++++++++
.../reader/ParquetColumnarRowSplitReader.java | 34 +-
.../reader/ParquetDataColumnReaderFactory.java | 112 +++-
.../format/cow/vector/reader/RowColumnReader.java | 63 --
.../table/format/cow/vector/type/ParquetField.java | 72 +++
.../format/cow/vector/type/ParquetGroupField.java | 59 ++
.../cow/vector/type/ParquetPrimitiveField.java | 55 ++
.../cow/vector/TestHeapColumnVectorAccessors.java | 139 +++++
.../cow/vector/TestParquetDecimalVector.java | 188 ++++++
.../reader/TestParquetDataColumnReaderFactory.java | 272 ++++++++
.../cow/vector/type/TestParquetGroupField.java | 134 ++++
33 files changed, 3078 insertions(+), 1386 deletions(-)
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 a1fa3c8fa835..3f2f8976b69b 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
@@ -87,8 +87,8 @@ public class NestedPositionUtil {
*
* @param field field that contains array/map column message include max repetition level and
* definition level.
- * @param definitionLevels int array with each value's repetition level.
- * @param repetitionLevels int array with each value's definition level.
+ * @param definitionLevels int array with each value's definition level.
+ * @param repetitionLevels int array with each value's repetition level.
* @return {@link CollectionPosition} contains collections offset array, length array and isNull
* array.
*/
diff --git a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetDataColumnReaderFactory.java b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetDataColumnReaderFactory.java
index 3748e53fc284..1abc6ed56c0d 100644
--- a/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetDataColumnReaderFactory.java
+++ b/hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetDataColumnReaderFactory.java
@@ -359,9 +359,9 @@ public final class ParquetDataColumnReaderFactory {
}
private static TimestampData int64ToTimestamp(
- boolean utcTimestamp, long value, ChronoUnit unit) {
+ boolean isUtcTimestamp, long value, ChronoUnit unit) {
Instant instant = Instant.EPOCH.plus(value, unit);
- if (utcTimestamp) {
+ if (isUtcTimestamp) {
return TimestampData.fromInstant(instant);
}
return TimestampData.fromTimestamp(Timestamp.from(instant));
@@ -379,10 +379,10 @@ public final class ParquetDataColumnReaderFactory {
}
private static TimestampData int96ToTimestamp(
- boolean utcTimestamp, long nanosOfDay, int julianDay) {
+ boolean isUtcTimestamp, long nanosOfDay, int julianDay) {
long millisecond = julianDayToMillis(julianDay) + (nanosOfDay / NANOS_PER_MILLISECOND);
- if (utcTimestamp) {
+ if (isUtcTimestamp) {
int nanoOfMillisecond = (int) (nanosOfDay % NANOS_PER_MILLISECOND);
return TimestampData.fromEpochMillis(millisecond, nanoOfMillisecond);
} else {
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
index ab05beb87459..5468dc86a25a 100644
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/ParquetSplitReaderUtil.java
@@ -7,7 +7,7 @@
* "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
+ * 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,
@@ -19,19 +19,18 @@
package org.apache.hudi.table.format.cow;
import org.apache.hudi.common.util.ValidationUtils;
-import org.apache.hudi.table.format.cow.vector.HeapArrayGroupColumnVector;
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;
import org.apache.hudi.table.format.cow.vector.HeapRowColumnVector;
-import org.apache.hudi.table.format.cow.vector.reader.ArrayColumnReader;
-import org.apache.hudi.table.format.cow.vector.reader.ArrayGroupReader;
import org.apache.hudi.table.format.cow.vector.reader.EmptyColumnReader;
import org.apache.hudi.table.format.cow.vector.reader.FixedLenBytesColumnReader;
import org.apache.hudi.table.format.cow.vector.reader.Int64TimestampColumnReader;
-import org.apache.hudi.table.format.cow.vector.reader.MapColumnReader;
+import org.apache.hudi.table.format.cow.vector.reader.NestedColumnReader;
import org.apache.hudi.table.format.cow.vector.reader.ParquetColumnarRowSplitReader;
-import org.apache.hudi.table.format.cow.vector.reader.RowColumnReader;
+import org.apache.hudi.table.format.cow.vector.type.ParquetField;
+import org.apache.hudi.table.format.cow.vector.type.ParquetGroupField;
+import org.apache.hudi.table.format.cow.vector.type.ParquetPrimitiveField;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.parquet.vector.reader.BooleanColumnReader;
@@ -64,12 +63,14 @@ import org.apache.flink.table.types.logical.DecimalType;
import org.apache.flink.table.types.logical.IntType;
import org.apache.flink.table.types.logical.LocalZonedTimestampType;
import org.apache.flink.table.types.logical.LogicalType;
-import org.apache.flink.table.types.logical.LogicalTypeFamily;
-import org.apache.flink.table.types.logical.LogicalTypeRoot;
import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.MultisetType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.logical.TimestampType;
+import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.StringUtils;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.parquet.ParquetRuntimeException;
import org.apache.parquet.column.ColumnDescriptor;
@@ -77,12 +78,18 @@ import org.apache.parquet.column.page.PageReadStore;
import org.apache.parquet.column.page.PageReader;
import org.apache.parquet.filter.UnboundRecordFilter;
import org.apache.parquet.filter2.predicate.FilterPredicate;
+import org.apache.parquet.io.ColumnIO;
+import org.apache.parquet.io.GroupColumnIO;
+import org.apache.parquet.io.MessageColumnIO;
+import org.apache.parquet.io.PrimitiveColumnIO;
import org.apache.parquet.schema.GroupType;
import org.apache.parquet.schema.InvalidSchemaException;
import org.apache.parquet.schema.OriginalType;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Type;
+import javax.annotation.Nullable;
+
import java.io.IOException;
import java.math.BigDecimal;
import java.sql.Date;
@@ -90,25 +97,38 @@ import java.time.LocalDate;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.stream.Collectors;
import static org.apache.flink.table.utils.DateTimeUtils.toInternal;
import static org.apache.hudi.common.util.StringUtils.getUTF8Bytes;
import static org.apache.parquet.Preconditions.checkArgument;
+import static org.apache.parquet.schema.Type.Repetition.REPEATED;
+import static org.apache.parquet.schema.Type.Repetition.REQUIRED;
/**
* Util for generating {@link ParquetColumnarRowSplitReader}.
*
- * <p>NOTE: reference from Flink release 1.11.2 {@code ParquetSplitReaderUtil}, modify to support INT64
- * based TIMESTAMP_MILLIS as ConvertedType, should remove when Flink supports that.
+ * <p>Uses the Dremel-style nested reader ported from Apache Flink 2.1 (FLINK-35702). For primitive
+ * top-level columns we keep Hudi's specialized readers — {@link Int64TimestampColumnReader},
+ * {@link FixedLenBytesColumnReader}, and the Hudi {@link HeapDecimalVector} — unchanged. For
+ * nested types (ARRAY / MAP / MULTISET / ROW) we build a {@link ParquetField} tree once per
+ * split via {@link #buildFieldsList(List, List, MessageColumnIO)} and delegate reading to
+ * {@link NestedColumnReader}.
+ *
+ * <p>Schema evolution: missing top-level fields are still handled by the caller
+ * ({@link ParquetColumnarRowSplitReader} patches them with null vectors). Missing fields
+ * <em>inside</em> a Row are handled here — {@link #constructField} returns {@code null} for a
+ * child that isn't physically present, and the corresponding child in the pre-allocated vector
+ * is filled with nulls via {@link #createVectorFromConstant} so the Dremel assembler can
+ * passthrough the slot (see {@link NestedColumnReader#readToVector}).
*/
public class ParquetSplitReaderUtil {
- /**
- * Util for generating partitioned {@link ParquetColumnarRowSplitReader}.
- */
+ /** Util for generating partitioned {@link ParquetColumnarRowSplitReader}. */
public static ParquetColumnarRowSplitReader genPartColumnarRowReader(
boolean utcTimestamp,
boolean caseSensitive,
@@ -125,7 +145,7 @@ public class ParquetSplitReaderUtil {
UnboundRecordFilter recordFilter) throws IOException {
ValidationUtils.checkState(Arrays.stream(selectedFields).noneMatch(x -> x == -1),
- "One or more specified columns does not exist in the hudi table.");
+ "One or more specified columns does not exist in the hudi table.");
List<String> selNonPartNames = Arrays.stream(selectedFields)
.mapToObj(i -> fullFieldNames[i])
@@ -182,10 +202,13 @@ public class ParquetSplitReaderUtil {
return readVector;
}
- private static ColumnVector createVectorFromConstant(
- LogicalType type,
- Object value,
- int batchSize) {
+ /**
+ * Builds a constant-filled column vector for either a partition column (non-null value) or a
+ * missing-column slot (null value). Used both at the batch-generator level for partition
+ * injection and at the row-reader level for fields absent from the Parquet file.
+ */
+ public static ColumnVector createVectorFromConstant(
+ LogicalType type, Object value, int batchSize) {
switch (type.getTypeRoot()) {
case CHAR:
case VARCHAR:
@@ -278,6 +301,7 @@ public class ParquetSplitReaderUtil {
value == null ? null : toInternal((Date) value),
batchSize);
case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
HeapTimestampVector tv = new HeapTimestampVector(batchSize);
if (value == null) {
tv.fillWithNulls();
@@ -286,46 +310,41 @@ public class ParquetSplitReaderUtil {
}
return tv;
case ARRAY:
- ArrayType arrayType = (ArrayType) type;
- if (arrayType.getElementType().isAnyOf(LogicalTypeFamily.CONSTRUCTED)) {
- HeapArrayGroupColumnVector arrayGroup = new HeapArrayGroupColumnVector(batchSize);
- if (value == null) {
- arrayGroup.fillWithNulls();
- return arrayGroup;
- } else {
- throw new UnsupportedOperationException("Unsupported create array with default value.");
- }
- } else {
- HeapArrayVector arrayVector = new HeapArrayVector(batchSize);
- if (value == null) {
- arrayVector.fillWithNulls();
- return arrayVector;
- } else {
- throw new UnsupportedOperationException("Unsupported create array with default value.");
- }
+ if (value != null) {
+ throw new UnsupportedOperationException("Unsupported create array with default value.");
}
+ HeapArrayVector arrayVector = new HeapArrayVector(batchSize);
+ arrayVector.fillWithNulls();
+ return arrayVector;
case MAP:
- HeapMapColumnVector mapVector = new HeapMapColumnVector(batchSize, null, null);
- if (value == null) {
- mapVector.fillWithNulls();
- return mapVector;
- } else {
- throw new UnsupportedOperationException("Unsupported create map with default value.");
+ case MULTISET:
+ if (value != null) {
+ throw new UnsupportedOperationException(
+ "Unsupported create " + type.getTypeRoot() + " with default value.");
}
+ HeapMapColumnVector mapVector = new HeapMapColumnVector(batchSize, null, null);
+ mapVector.fillWithNulls();
+ return mapVector;
case ROW:
- HeapRowColumnVector rowVector = new HeapRowColumnVector(batchSize);
- if (value == null) {
- rowVector.fillWithNulls();
- return rowVector;
- } else {
+ if (value != null) {
throw new UnsupportedOperationException("Unsupported create row with default value.");
}
+ RowType rowType = (RowType) type;
+ WritableColumnVector[] childVectors = new WritableColumnVector[rowType.getFieldCount()];
+ for (int i = 0; i < childVectors.length; i++) {
+ childVectors[i] =
+ (WritableColumnVector) createVectorFromConstant(rowType.getTypeAt(i), null, batchSize);
+ }
+ HeapRowColumnVector rowVector = new HeapRowColumnVector(batchSize, childVectors);
+ rowVector.fillWithNulls();
+ return rowVector;
default:
throw new UnsupportedOperationException("Unsupported type: " + type);
}
}
- private static List<ColumnDescriptor> filterDescriptors(int depth, Type type, List<ColumnDescriptor> columns) throws ParquetRuntimeException {
+ private static List<ColumnDescriptor> filterDescriptors(
+ int depth, Type type, List<ColumnDescriptor> columns) throws ParquetRuntimeException {
List<ColumnDescriptor> filtered = new ArrayList<>();
for (ColumnDescriptor descriptor : columns) {
if (depth >= descriptor.getPath().length) {
@@ -339,24 +358,61 @@ public class ParquetSplitReaderUtil {
return filtered;
}
+ /**
+ * Creates a {@link ColumnReader} for one top-level requested field. For primitive types the
+ * Hudi-specialized reader path is used. For nested types ({@code ARRAY}, {@code MAP},
+ * {@code MULTISET}, {@code ROW}) the Dremel-style {@link NestedColumnReader} is used, driven by
+ * the supplied pre-built {@link ParquetField} tree.
+ *
+ * @param field the {@link ParquetField} tree for this column, built by
+ * {@link #buildFieldsList(List, List, MessageColumnIO)}. Required (non-null) for nested
+ * types; ignored for primitives.
+ */
+ public static ColumnReader createColumnReader(
+ boolean utcTimestamp,
+ LogicalType fieldType,
+ Type physicalType,
+ List<ColumnDescriptor> descriptors,
+ PageReadStore pages,
+ @Nullable ParquetField field) throws IOException {
+ switch (fieldType.getTypeRoot()) {
+ case ARRAY:
+ case MAP:
+ case MULTISET:
+ case ROW:
+ Preconditions.checkNotNull(
+ field, "ParquetField must be provided for nested type: %s", fieldType);
+ return new NestedColumnReader(utcTimestamp, pages, field);
+ default:
+ return createPrimitiveColumnReader(utcTimestamp, fieldType, physicalType, descriptors, pages);
+ }
+ }
+
+ /**
+ * Backward-compat entry point kept for callers that don't project nested types and therefore
+ * never need a {@link ParquetField} tree. Forwards to the {@link ParquetField}-aware overload
+ * with a null field; nested types now go through that overload directly.
+ *
+ * @deprecated use {@link #createColumnReader(boolean, LogicalType, Type, List, PageReadStore,
+ * ParquetField)} so nested types take the Dremel path.
+ */
+ @Deprecated
public static ColumnReader createColumnReader(
boolean utcTimestamp,
LogicalType fieldType,
Type physicalType,
List<ColumnDescriptor> descriptors,
PageReadStore pages) throws IOException {
- return createColumnReader(utcTimestamp, fieldType, physicalType, descriptors,
- pages, 0);
+ return createColumnReader(utcTimestamp, fieldType, physicalType, descriptors, pages, null);
}
- private static ColumnReader createColumnReader(
+ private static ColumnReader createPrimitiveColumnReader(
boolean utcTimestamp,
LogicalType fieldType,
Type physicalType,
List<ColumnDescriptor> columns,
- PageReadStore pages,
- int depth) throws IOException {
- List<ColumnDescriptor> descriptors = filterDescriptors(depth, physicalType, columns);
+ PageReadStore pages) throws IOException {
+ List<ColumnDescriptor> descriptors = filterDescriptors(0, physicalType, columns);
ColumnDescriptor descriptor = descriptors.get(0);
PageReader pageReader = pages.getPageReader(descriptor);
switch (fieldType.getTypeRoot()) {
@@ -392,7 +448,9 @@ public class ParquetSplitReaderUtil {
case INT96:
return new TimestampColumnReader(utcTimestamp, descriptor, pageReader);
default:
- throw new AssertionError();
+ throw new AssertionError(
+ "Unexpected physical type for TIMESTAMP: "
+ + descriptor.getPrimitiveType().getPrimitiveTypeName());
}
case DECIMAL:
switch (descriptor.getPrimitiveType().getPrimitiveTypeName()) {
@@ -403,106 +461,23 @@ public class ParquetSplitReaderUtil {
case BINARY:
return new BytesColumnReader(descriptor, pageReader);
case FIXED_LEN_BYTE_ARRAY:
- return new FixedLenBytesColumnReader(
- descriptor, pageReader);
+ return new FixedLenBytesColumnReader(descriptor, pageReader);
default:
- throw new AssertionError();
- }
- case ARRAY:
- ArrayType arrayType = (ArrayType) fieldType;
- if (arrayType.getElementType().isAnyOf(LogicalTypeFamily.CONSTRUCTED)) {
- boolean isThreeLevelList = isThreeLevelList(physicalType);
- // 3-level List structure, drill down 2 level to get type for `element`
- Type elementType = isThreeLevelList
- ? physicalType.asGroupType().getType(0).asGroupType().getType(0)
- : physicalType.asGroupType().getType(0);
- int elementDepth = isThreeLevelList ? depth + 2 : depth + 1;
- return new ArrayGroupReader(createColumnReader(
- utcTimestamp,
- arrayType.getElementType(),
- elementType,
- descriptors,
- pages,
- elementDepth));
- } else {
- return new ArrayColumnReader(
- descriptor,
- pageReader,
- utcTimestamp,
- descriptor.getPrimitiveType(),
- fieldType);
- }
- case MAP:
- MapType mapType = (MapType) fieldType;
- ArrayColumnReader keyReader =
- new ArrayColumnReader(
- descriptor,
- pageReader,
- utcTimestamp,
- descriptor.getPrimitiveType(),
- new ArrayType(mapType.getKeyType()));
- ColumnReader<WritableColumnVector> valueReader;
- if (mapType.getValueType().isAnyOf(LogicalTypeFamily.CONSTRUCTED)) {
- valueReader = new ArrayGroupReader(createColumnReader(
- utcTimestamp,
- mapType.getValueType(),
- physicalType.asGroupType().getType(0).asGroupType().getType(1), // Get the value physical type
- descriptors.subList(1, descriptors.size()), // remove the key descriptor
- pages,
- depth + 2)); // increase the depth by 2, because there's a key_value entry in the path
- } else {
- valueReader = new ArrayColumnReader(
- descriptors.get(1),
- pages.getPageReader(descriptors.get(1)),
- utcTimestamp,
- descriptors.get(1).getPrimitiveType(),
- new ArrayType(mapType.getValueType()));
+ throw new AssertionError(
+ "Unexpected physical type for DECIMAL: "
+ + descriptor.getPrimitiveType().getPrimitiveTypeName());
}
- return new MapColumnReader(keyReader, valueReader);
- case ROW:
- RowType rowType = (RowType) fieldType;
- GroupType groupType = physicalType.asGroupType();
- List<ColumnReader> fieldReaders = new ArrayList<>();
- for (int i = 0; i < rowType.getFieldCount(); i++) {
- // schema evolution: read the parquet file with a new extended field name.
- int fieldIndex = getFieldIndexInPhysicalType(rowType.getFields().get(i).getName(), groupType);
- if (fieldIndex < 0) {
- fieldReaders.add(new EmptyColumnReader());
- } else {
- // Check for nested row in array with atomic field type.
-
- // This is done to meet the Parquet field algorithm that pushes multiplicity and structures down to individual fields.
- // In Parquet, an array of rows is stored as separate arrays for each field.
-
- // Limitations: It won't work for multiple nested arrays and maps.
- // The main problem is that the Flink classes and interface don't follow that pattern.
- if (descriptors.get(fieldIndex).getMaxRepetitionLevel() > 0 && !rowType.getTypeAt(i).is(LogicalTypeRoot.ARRAY)) {
- fieldReaders.add(
- createColumnReader(
- utcTimestamp,
- new ArrayType(rowType.getTypeAt(i).isNullable(), rowType.getTypeAt(i)),
- groupType.getType(fieldIndex),
- descriptors,
- pages,
- depth + 1));
- } else {
- fieldReaders.add(
- createColumnReader(
- utcTimestamp,
- rowType.getTypeAt(i),
- groupType.getType(fieldIndex),
- descriptors,
- pages,
- depth + 1));
- }
- }
- }
- return new RowColumnReader(fieldReaders);
default:
throw new UnsupportedOperationException(fieldType + " is not supported now.");
}
}
+ /**
+ * Creates the writable column vector that the reader will write into. The returned vector shape
+ * 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.
+ */
public static WritableColumnVector createWritableColumnVector(
int batchSize,
LogicalType fieldType,
@@ -523,40 +498,48 @@ public class ParquetSplitReaderUtil {
switch (fieldType.getTypeRoot()) {
case BOOLEAN:
checkArgument(
- typeName == PrimitiveType.PrimitiveTypeName.BOOLEAN, getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+ typeName == PrimitiveType.PrimitiveTypeName.BOOLEAN,
+ getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
return new HeapBooleanVector(batchSize);
case TINYINT:
checkArgument(
- typeName == PrimitiveType.PrimitiveTypeName.INT32, getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+ typeName == PrimitiveType.PrimitiveTypeName.INT32,
+ getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
return new HeapByteVector(batchSize);
case DOUBLE:
checkArgument(
- typeName == PrimitiveType.PrimitiveTypeName.DOUBLE, getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+ typeName == PrimitiveType.PrimitiveTypeName.DOUBLE,
+ getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
return new HeapDoubleVector(batchSize);
case FLOAT:
checkArgument(
- typeName == PrimitiveType.PrimitiveTypeName.FLOAT, getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+ typeName == PrimitiveType.PrimitiveTypeName.FLOAT,
+ getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
return new HeapFloatVector(batchSize);
case INTEGER:
case DATE:
case TIME_WITHOUT_TIME_ZONE:
checkArgument(
- typeName == PrimitiveType.PrimitiveTypeName.INT32, getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+ typeName == PrimitiveType.PrimitiveTypeName.INT32,
+ getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
return new HeapIntVector(batchSize);
case BIGINT:
checkArgument(
- typeName == PrimitiveType.PrimitiveTypeName.INT64, getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+ typeName == PrimitiveType.PrimitiveTypeName.INT64,
+ getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
return new HeapLongVector(batchSize);
case SMALLINT:
checkArgument(
- typeName == PrimitiveType.PrimitiveTypeName.INT32, getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+ typeName == PrimitiveType.PrimitiveTypeName.INT32,
+ getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
return new HeapShortVector(batchSize);
case CHAR:
case VARCHAR:
case BINARY:
case VARBINARY:
checkArgument(
- typeName == PrimitiveType.PrimitiveTypeName.BINARY, getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
+ typeName == PrimitiveType.PrimitiveTypeName.BINARY,
+ getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
return new HeapBytesVector(batchSize);
case TIMESTAMP_WITHOUT_TIME_ZONE:
case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
@@ -566,112 +549,64 @@ public class ParquetSplitReaderUtil {
case DECIMAL:
checkArgument(
(typeName == PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY
- || typeName == PrimitiveType.PrimitiveTypeName.BINARY)
+ || typeName == PrimitiveType.PrimitiveTypeName.BINARY)
&& primitiveType.getOriginalType() == OriginalType.DECIMAL,
getPrimitiveTypeCheckFailureMessage(typeName, fieldType));
return new HeapDecimalVector(batchSize);
case ARRAY:
ArrayType arrayType = (ArrayType) fieldType;
- if (arrayType.getElementType().isAnyOf(LogicalTypeFamily.CONSTRUCTED)) {
- boolean isThreeLevelList = isThreeLevelList(physicalType);
- // 3-level List structure, drill down 2 level to get type for `element`
- Type elementType = isThreeLevelList
- ? physicalType.asGroupType().getType(0).asGroupType().getType(0)
- : physicalType.asGroupType().getType(0);
- int elementDepth = isThreeLevelList ? depth + 2 : depth + 1;
- return new HeapArrayGroupColumnVector(
- batchSize,
- createWritableColumnVector(
- batchSize,
- arrayType.getElementType(),
- elementType,
- descriptors,
- elementDepth));
- } else {
- return new HeapArrayVector(
- batchSize,
- createWritableColumnVector(
- batchSize,
- arrayType.getElementType(),
- physicalType,
- descriptors,
- depth));
- }
- case MAP:
+ return new HeapArrayVector(
+ batchSize,
+ createWritableColumnVector(
+ batchSize, arrayType.getElementType(), physicalType, descriptors, depth));
+ case MAP: {
MapType mapType = (MapType) fieldType;
- GroupType repeatedType = physicalType.asGroupType().getType(0).asGroupType();
- // the map column has three level paths.
- WritableColumnVector keyColumnVector = createWritableColumnVector(
+ GroupType repeatedType = unwrapMapRepeatedType(physicalType);
+ return new HeapMapColumnVector(
batchSize,
- new ArrayType(mapType.getKeyType().isNullable(), mapType.getKeyType()),
- repeatedType.getType(0),
- descriptors,
- depth + 2);
- WritableColumnVector valueColumnVector;
- if (mapType.getValueType().isAnyOf(LogicalTypeFamily.CONSTRUCTED)) {
- valueColumnVector = new HeapArrayGroupColumnVector(
- batchSize,
- createWritableColumnVector(
- batchSize,
- mapType.getValueType(),
- repeatedType.getType(1).asGroupType(),
- descriptors,
- depth + 2));
- } else {
- valueColumnVector = createWritableColumnVector(
- batchSize,
- new ArrayType(mapType.getValueType().isNullable(), mapType.getValueType()),
- repeatedType.getType(1),
- descriptors,
- depth + 2);
- }
- return new HeapMapColumnVector(batchSize, keyColumnVector, valueColumnVector);
+ createWritableColumnVector(
+ batchSize, mapType.getKeyType(), repeatedType.getType(0), descriptors, depth + 2),
+ createWritableColumnVector(
+ batchSize, mapType.getValueType(), repeatedType.getType(1), descriptors, depth + 2));
+ }
+ case MULTISET: {
+ MultisetType multisetType = (MultisetType) fieldType;
+ GroupType repeatedType = unwrapMapRepeatedType(physicalType);
+ return new HeapMapColumnVector(
+ batchSize,
+ createWritableColumnVector(
+ batchSize,
+ multisetType.getElementType(),
+ repeatedType.getType(0),
+ descriptors,
+ depth + 2),
+ createWritableColumnVector(
+ batchSize,
+ new IntType(false),
+ repeatedType.getType(1),
+ descriptors,
+ depth + 2));
+ }
case ROW:
RowType rowType = (RowType) fieldType;
GroupType groupType = physicalType.asGroupType();
WritableColumnVector[] columnVectors = new WritableColumnVector[rowType.getFieldCount()];
for (int i = 0; i < columnVectors.length; i++) {
- // schema evolution: read the file with a new extended field name.
int fieldIndex = getFieldIndexInPhysicalType(rowType.getFields().get(i).getName(), groupType);
if (fieldIndex < 0) {
- // Check for nested row in array with atomic field type.
-
- // This is done to meet the Parquet field algorithm that pushes multiplicity and structures down to individual fields.
- // In Parquet, an array of rows is stored as separate arrays for each field.
-
- // Limitations: It won't work for multiple nested arrays and maps.
- // The main problem is that the Flink classes and interface don't follow that pattern.
- if (groupType.getRepetition().equals(Type.Repetition.REPEATED) && !rowType.getTypeAt(i).is(LogicalTypeRoot.ARRAY)) {
- columnVectors[i] = (WritableColumnVector) createVectorFromConstant(
- new ArrayType(rowType.getTypeAt(i).isNullable(), rowType.getTypeAt(i)), null, batchSize);
- } else {
- columnVectors[i] = (WritableColumnVector) createVectorFromConstant(rowType.getTypeAt(i), null, batchSize);
- }
+ // Schema evolution: logical field is absent from the Parquet file. Slot a null-filled
+ // vector of the correct shape; NestedColumnReader.readRow will pass it through when the
+ // matching ParquetField child is null.
+ columnVectors[i] =
+ (WritableColumnVector) createVectorFromConstant(rowType.getTypeAt(i), null, batchSize);
} else {
- // Check for nested row in array with atomic field type.
-
- // This is done to meet the Parquet field algorithm that pushes multiplicity and structures down to individual fields.
- // In Parquet, an array of rows is stored as separate arrays for each field.
-
- // Limitations: It won't work for multiple nested arrays and maps.
- // The main problem is that the Flink classes and interface don't follow that pattern.
- if (descriptors.get(fieldIndex).getMaxRepetitionLevel() > 0 && !rowType.getTypeAt(i).is(LogicalTypeRoot.ARRAY)) {
- columnVectors[i] =
- createWritableColumnVector(
- batchSize,
- new ArrayType(rowType.getTypeAt(i).isNullable(), rowType.getTypeAt(i)),
- groupType.getType(fieldIndex),
- descriptors,
- depth + 1);
- } else {
- columnVectors[i] =
- createWritableColumnVector(
- batchSize,
- rowType.getTypeAt(i),
- groupType.getType(fieldIndex),
- descriptors,
- depth + 1);
- }
+ columnVectors[i] =
+ createWritableColumnVector(
+ batchSize,
+ rowType.getTypeAt(i),
+ groupType.getType(fieldIndex),
+ descriptors,
+ depth + 1);
}
}
return new HeapRowColumnVector(batchSize, columnVectors);
@@ -681,56 +616,245 @@ public class ParquetSplitReaderUtil {
}
/**
- * Returns the field index with given physical row type {@code groupType} and field name {@code fieldName}.
- *
- * @return The physical field index or -1 if the field does not exist
+ * Peels one {@code repeated group key_value} wrapper off a MAP / MULTISET physical type, matching
+ * Parquet's canonical 3-level map encoding.
*/
- private static int getFieldIndexInPhysicalType(String fieldName, GroupType groupType) {
- // get index from fileSchema type, else, return -1
- return groupType.containsField(fieldName) ? groupType.getFieldIndex(fieldName) : -1;
+ private static GroupType unwrapMapRepeatedType(Type physicalType) {
+ return physicalType.asGroupType().getType(0).asGroupType();
}
+ // ------------------------------------------------------------------------------------------
+ // ParquetField tree construction (vendored from Apache Flink 2.1 ParquetSplitReaderUtil)
+ //
+ // The only Hudi-specific divergence is in `constructField`: the ROW branch tolerates children
+ // missing from the Parquet file by emitting a null ParquetField child (upstream throws). This
+ // matches the Hudi schema-evolution contract and is the companion to the null-child branch in
+ // `NestedColumnReader#readRow` and the null-vector slot in `createWritableColumnVector#ROW`.
+ // ------------------------------------------------------------------------------------------
+
/**
- * Check whether the given list type is a three-level list type.
- * <p>
- * <list-repetition> group <name> (LIST) {
- * repeated group list {
- * <element-repetition> <element-type> element;
- * }
- * }
- *
- * @param type list type
- * @return true if the list type is a three-level list type
+ * Builds {@link ParquetField} trees — one per top-level projected logical column — that feed
+ * {@link NestedColumnReader}. The returned list mirrors the input {@code children} positionally;
+ * primitive top-level fields produce {@code null} entries (callers don't need a tree for those).
+ */
+ public static List<ParquetField> buildFieldsList(
+ List<RowType.RowField> children, List<String> fieldNames, MessageColumnIO columnIO) {
+ List<ParquetField> list = new ArrayList<>();
+ for (int i = 0; i < children.size(); i++) {
+ RowType.RowField child = children.get(i);
+ if (isNestedType(child.getType())) {
+ list.add(constructField(child, lookupColumnByName(columnIO, fieldNames.get(i))));
+ } else {
+ list.add(null);
+ }
+ }
+ return list;
+ }
+
+ private static boolean isNestedType(LogicalType type) {
+ return type instanceof RowType
+ || type instanceof ArrayType
+ || type instanceof MapType
+ || type instanceof MultisetType;
+ }
+
+ @Nullable
+ private static ParquetField constructField(RowType.RowField rowField, ColumnIO columnIO) {
+ boolean required = columnIO.getType().getRepetition() == REQUIRED;
+ int repetitionLevel = columnIO.getRepetitionLevel();
+ int definitionLevel = columnIO.getDefinitionLevel();
+ LogicalType type = rowField.getType();
+ String fieldName = rowField.getName();
+ if (type instanceof RowType) {
+ GroupColumnIO groupColumnIO = (GroupColumnIO) columnIO;
+ RowType rowType = (RowType) type;
+ List<RowType.RowField> childFields = rowType.getFields();
+ List<ParquetField> fieldsList = new ArrayList<>(childFields.size());
+ for (RowType.RowField childField : childFields) {
+ // Hudi schema evolution: a logical child may be absent from the Parquet file. In that
+ // case we emit a null ParquetField so that NestedColumnReader.readRow passes through the
+ // pre-filled null vector instead of recursing.
+ ColumnIO childIo = lookupColumnByNameOrNull(groupColumnIO, childField.getName());
+ if (childIo == null) {
+ fieldsList.add(null);
+ } else {
+ fieldsList.add(constructField(childField, childIo));
+ }
+ }
+ return new ParquetGroupField(
+ type,
+ repetitionLevel,
+ definitionLevel,
+ required,
+ Collections.unmodifiableList(fieldsList));
+ }
+
+ if (type instanceof MapType) {
+ GroupColumnIO groupColumnIO = (GroupColumnIO) columnIO;
+ GroupColumnIO keyValueColumnIO = getMapKeyValueColumn(groupColumnIO);
+ MapType mapType = (MapType) type;
+ ParquetField keyField =
+ constructField(
+ new RowType.RowField("", mapType.getKeyType()), keyValueColumnIO.getChild(0));
+ ParquetField valueField =
+ constructField(
+ new RowType.RowField("", mapType.getValueType()), keyValueColumnIO.getChild(1));
+ return new ParquetGroupField(
+ type,
+ repetitionLevel,
+ definitionLevel,
+ required,
+ Collections.unmodifiableList(Arrays.asList(keyField, valueField)));
+ }
+
+ if (type instanceof MultisetType) {
+ GroupColumnIO groupColumnIO = (GroupColumnIO) columnIO;
+ GroupColumnIO keyValueColumnIO = getMapKeyValueColumn(groupColumnIO);
+ MultisetType multisetType = (MultisetType) type;
+ ParquetField keyField =
+ constructField(
+ new RowType.RowField("", multisetType.getElementType()),
+ keyValueColumnIO.getChild(0));
+ ParquetField valueField =
+ constructField(
+ new RowType.RowField("", new IntType()), keyValueColumnIO.getChild(1));
+ return new ParquetGroupField(
+ type,
+ repetitionLevel,
+ definitionLevel,
+ required,
+ Collections.unmodifiableList(Arrays.asList(keyField, valueField)));
+ }
+
+ if (type instanceof ArrayType) {
+ ArrayType arrayType = (ArrayType) type;
+ ColumnIO elementTypeColumnIO;
+ if (columnIO instanceof GroupColumnIO) {
+ GroupColumnIO groupColumnIO = (GroupColumnIO) columnIO;
+ if (!StringUtils.isNullOrWhitespaceOnly(fieldName)) {
+ while (!Objects.equals(groupColumnIO.getName(), fieldName)) {
+ groupColumnIO = (GroupColumnIO) groupColumnIO.getChild(0);
+ }
+ elementTypeColumnIO = groupColumnIO;
+ } else {
+ if (arrayType.getElementType() instanceof RowType) {
+ elementTypeColumnIO = groupColumnIO;
+ } else {
+ elementTypeColumnIO = groupColumnIO.getChild(0);
+ }
+ }
+ } else if (columnIO instanceof PrimitiveColumnIO) {
+ elementTypeColumnIO = columnIO;
+ } else {
+ throw new FlinkRuntimeException(String.format("Unknown ColumnIO, %s", columnIO));
+ }
+
+ ParquetField elementField =
+ constructField(
+ new RowType.RowField("", arrayType.getElementType()),
+ getArrayElementColumn(elementTypeColumnIO));
+ if (repetitionLevel == elementField.getRepetitionLevel()) {
+ repetitionLevel = columnIO.getParent().getRepetitionLevel();
+ }
+ return new ParquetGroupField(
+ type,
+ repetitionLevel,
+ definitionLevel,
+ required,
+ Collections.singletonList(elementField));
+ }
+
+ PrimitiveColumnIO primitiveColumnIO = (PrimitiveColumnIO) columnIO;
+ return new ParquetPrimitiveField(
+ type, required, primitiveColumnIO.getColumnDescriptor(), primitiveColumnIO.getId());
+ }
+
+ /**
+ * Parquet column names are case-insensitive in Flink's lookup. Matches upstream
+ * {@code ParquetSplitReaderUtil.lookupColumnByName}; throws when absent.
*/
- private static boolean isThreeLevelList(Type type) {
- if (type.isPrimitive()) {
- return false;
+ public static ColumnIO lookupColumnByName(GroupColumnIO groupColumnIO, String columnName) {
+ ColumnIO columnIO = lookupColumnByNameOrNull(groupColumnIO, columnName);
+ if (columnIO != null) {
+ return columnIO;
}
- GroupType groupType = type.asGroupType();
- OriginalType originalType = groupType.getOriginalType();
- return originalType == OriginalType.LIST
- && groupType.getType(0).getName().equals("list");
+ throw new FlinkRuntimeException(
+ "Can not find column io for parquet reader. Column name: " + columnName);
}
/**
- * Construct the error message when primitive type mismatches.
- *
- * @param primitiveType Primitive type
- * @param fieldType Logical field type
- * @return The error message
+ * Case-insensitive column lookup that returns {@code null} when no match is found — the
+ * Hudi-specific companion to {@link #lookupColumnByName}, used by {@link #constructField} to
+ * emit null {@link ParquetField} children for fields absent from the Parquet file.
*/
- private static String getPrimitiveTypeCheckFailureMessage(PrimitiveType.PrimitiveTypeName primitiveType, LogicalType fieldType) {
- return String.format("Unexpected type exception. Primitive type: %s. Field type: %s.", primitiveType, fieldType.getTypeRoot().name());
+ @Nullable
+ private static ColumnIO lookupColumnByNameOrNull(
+ GroupColumnIO groupColumnIO, String columnName) {
+ ColumnIO columnIO = groupColumnIO.getChild(columnName);
+ if (columnIO != null) {
+ return columnIO;
+ }
+ for (int i = 0; i < groupColumnIO.getChildrenCount(); i++) {
+ if (groupColumnIO.getChild(i).getName().equalsIgnoreCase(columnName)) {
+ return groupColumnIO.getChild(i);
+ }
+ }
+ return null;
+ }
+
+ public static GroupColumnIO getMapKeyValueColumn(GroupColumnIO groupColumnIO) {
+ while (groupColumnIO.getChildrenCount() == 1) {
+ groupColumnIO = (GroupColumnIO) groupColumnIO.getChild(0);
+ }
+ return groupColumnIO;
+ }
+
+ public static ColumnIO getArrayElementColumn(ColumnIO columnIO) {
+ while (columnIO instanceof GroupColumnIO && !columnIO.getType().isRepetition(REPEATED)) {
+ columnIO = ((GroupColumnIO) columnIO).getChild(0);
+ }
+
+ // 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")) {
+ return ((GroupColumnIO) columnIO).getChild(0);
+ }
+ return columnIO;
}
/**
- * Construct the error message when original type mismatches.
+ * Returns the field index with given physical row type {@code groupType} and field name
+ * {@code fieldName}.
*
- * @param originalType Original type
- * @param fieldType Logical field type
- * @return The error message
+ * @return the physical field index or -1 if the field does not exist
+ */
+ private static int getFieldIndexInPhysicalType(String fieldName, GroupType groupType) {
+ return groupType.containsField(fieldName) ? groupType.getFieldIndex(fieldName) : -1;
+ }
+
+ private static String getPrimitiveTypeCheckFailureMessage(
+ PrimitiveType.PrimitiveTypeName primitiveType, LogicalType fieldType) {
+ return String.format(
+ "Unexpected type exception. Primitive type: %s. Field type: %s.",
+ primitiveType, fieldType.getTypeRoot().name());
+ }
+
+ private static String getOriginalTypeCheckFailureMessage(
+ OriginalType originalType, LogicalType fieldType) {
+ return String.format(
+ "Unexpected type exception. Original type: %s. Field type: %s.",
+ originalType, fieldType.getTypeRoot().name());
+ }
+
+ /**
+ * Returns a synthetic null-column reader to fill missing top-level fields. Kept as a convenience
+ * for callers that need to mirror Hudi's original behaviour where a missing column produces an
+ * explicit null-valued reader rather than being omitted from the batch.
*/
- private static String getOriginalTypeCheckFailureMessage(OriginalType originalType, LogicalType fieldType) {
- return String.format("Unexpected type exception. Original type: %s. Field type: %s.", originalType, fieldType.getTypeRoot().name());
+ public static ColumnReader<WritableColumnVector> emptyColumnReader() {
+ return new EmptyColumnReader();
}
}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/BooleanArrayList.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/BooleanArrayList.java
new file mode 100644
index 000000000000..d51d7ee754b8
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/BooleanArrayList.java
@@ -0,0 +1,69 @@
+/*
+ * 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 java.util.Arrays;
+
+/**
+ * Minimal implementation of an array-backed list of booleans.
+ *
+ * <p>Note: Vendored from Apache Flink (FLINK-35702, {@code
+ * org.apache.flink.runtime.util.BooleanArrayList}) because Flink 1.18 does not ship this helper.
+ */
+public class BooleanArrayList {
+ private int size;
+ private boolean[] array;
+
+ public BooleanArrayList(int capacity) {
+ this.size = 0;
+ this.array = new boolean[capacity];
+ }
+
+ public int size() {
+ return size;
+ }
+
+ public boolean add(boolean element) {
+ grow(size + 1);
+ array[size++] = element;
+ return true;
+ }
+
+ public void clear() {
+ size = 0;
+ }
+
+ public boolean isEmpty() {
+ return (size == 0);
+ }
+
+ public boolean[] toArray() {
+ return Arrays.copyOf(array, size);
+ }
+
+ private void grow(int length) {
+ if (length > array.length) {
+ final int newLength =
+ (int) Math.max(Math.min(2L * array.length, Integer.MAX_VALUE - 8), length);
+ final boolean[] t = new boolean[newLength];
+ System.arraycopy(array, 0, t, 0, size);
+ array = t;
+ }
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/IntArrayList.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/IntArrayList.java
new file mode 100644
index 000000000000..4787dbb5b9dd
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/IntArrayList.java
@@ -0,0 +1,93 @@
+/*
+ * 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 java.util.Arrays;
+import java.util.NoSuchElementException;
+
+/**
+ * Minimal implementation of an array-backed list of ints.
+ *
+ * <p>Note: Vendored from Apache Flink ({@code org.apache.flink.runtime.util.IntArrayList}) to
+ * avoid depending on {@code @Internal} Flink runtime classes from Hudi's parquet reader.
+ */
+public class IntArrayList {
+
+ private int size;
+ private int[] array;
+
+ public IntArrayList(final int capacity) {
+ this.size = 0;
+ this.array = new int[capacity];
+ }
+
+ public int size() {
+ return size;
+ }
+
+ public boolean add(final int number) {
+ grow(size + 1);
+ array[size++] = number;
+ return true;
+ }
+
+ public int removeLast() {
+ if (size == 0) {
+ throw new NoSuchElementException();
+ }
+ --size;
+ return array[size];
+ }
+
+ public void clear() {
+ size = 0;
+ }
+
+ public boolean isEmpty() {
+ return size == 0;
+ }
+
+ private void grow(final int length) {
+ if (length > array.length) {
+ final int newLength =
+ (int) Math.max(Math.min(2L * array.length, Integer.MAX_VALUE - 8), length);
+ final int[] t = new int[newLength];
+ System.arraycopy(array, 0, t, 0, size);
+ array = t;
+ }
+ }
+
+ public int[] toArray() {
+ return Arrays.copyOf(array, size);
+ }
+
+ public static final IntArrayList EMPTY =
+ new IntArrayList(0) {
+
+ @Override
+ public boolean add(int number) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int removeLast() {
+ throw new UnsupportedOperationException();
+ }
+ };
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/LongArrayList.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/LongArrayList.java
new file mode 100644
index 000000000000..a51291f9d844
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/LongArrayList.java
@@ -0,0 +1,83 @@
+/*
+ * 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 java.util.Arrays;
+
+/**
+ * Minimal implementation of an array-backed list of longs.
+ *
+ * <p>Note: Vendored from Apache Flink ({@code org.apache.flink.runtime.util.LongArrayList}) to
+ * avoid depending on {@code @Internal} Flink runtime classes from Hudi's parquet reader.
+ */
+public class LongArrayList {
+
+ private int size;
+ private long[] array;
+
+ public LongArrayList(int capacity) {
+ this.size = 0;
+ this.array = new long[capacity];
+ }
+
+ public int size() {
+ return size;
+ }
+
+ public boolean add(long number) {
+ grow(size + 1);
+ array[size++] = number;
+ return true;
+ }
+
+ public long removeLong(int index) {
+ if (index >= size) {
+ throw new IndexOutOfBoundsException(
+ "Index (" + index + ") is greater than or equal to list size (" + size + ")");
+ }
+ final long old = array[index];
+ size--;
+ if (index != size) {
+ System.arraycopy(array, index + 1, array, index, size - index);
+ }
+ return old;
+ }
+
+ public void clear() {
+ size = 0;
+ }
+
+ public boolean isEmpty() {
+ return (size == 0);
+ }
+
+ public long[] toArray() {
+ return Arrays.copyOf(array, size);
+ }
+
+ private void grow(int length) {
+ if (length > array.length) {
+ final int newLength =
+ (int) Math.max(Math.min(2L * array.length, Integer.MAX_VALUE - 8), length);
+ final long[] t = new long[newLength];
+ System.arraycopy(array, 0, t, 0, size);
+ array = t;
+ }
+ }
+}
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.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
similarity index 98%
copy from hudi-flink-datasource/hudi-flink1.18.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
copy to hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
index a1fa3c8fa835..3f2f8976b69b 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.19.x/src/main/java/org/apache/hudi/table/format/cow/utils/NestedPositionUtil.java
@@ -87,8 +87,8 @@ public class NestedPositionUtil {
*
* @param field field that contains array/map column message include max repetition level and
* definition level.
- * @param definitionLevels int array with each value's repetition level.
- * @param repetitionLevels int array with each value's definition level.
+ * @param definitionLevels int array with each value's definition level.
+ * @param repetitionLevels int array with each value's repetition level.
* @return {@link CollectionPosition} contains collections offset array, length array and isNull
* array.
*/
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ColumnarGroupArrayData.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ColumnarGroupArrayData.java
deleted file mode 100644
index 4c9275f3b093..000000000000
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ColumnarGroupArrayData.java
+++ /dev/null
@@ -1,179 +0,0 @@
-/*
- * 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.vector;
-
-import org.apache.flink.table.data.ArrayData;
-import org.apache.flink.table.data.DecimalData;
-import org.apache.flink.table.data.MapData;
-import org.apache.flink.table.data.RawValueData;
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.StringData;
-import org.apache.flink.table.data.TimestampData;
-import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
-
-public class ColumnarGroupArrayData implements ArrayData {
-
- WritableColumnVector vector;
- int rowId;
-
- public ColumnarGroupArrayData(WritableColumnVector vector, int rowId) {
- this.vector = vector;
- this.rowId = rowId;
- }
-
- @Override
- public int size() {
- if (vector == null) {
- return 0;
- }
-
- if (vector instanceof HeapRowColumnVector) {
- // assume all fields have the same size
- if (((HeapRowColumnVector) vector).vectors == null || ((HeapRowColumnVector) vector).vectors.length == 0) {
- return 0;
- }
- return ((HeapArrayVector) ((HeapRowColumnVector) vector).vectors[0]).getArray(rowId).size();
- }
- throw new UnsupportedOperationException(vector.getClass().getName() + " is not supported. Supported vector types: HeapRowColumnVector");
- }
-
- @Override
- public boolean isNullAt(int index) {
- if (vector == null) {
- return true;
- }
-
- if (vector instanceof HeapRowColumnVector) {
- return ((HeapRowColumnVector) vector).vectors == null;
- }
-
- throw new UnsupportedOperationException(vector.getClass().getName() + " is not supported. Supported vector types: HeapRowColumnVector");
- }
-
- @Override
- public boolean getBoolean(int index) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public byte getByte(int index) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public short getShort(int index) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public int getInt(int index) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public long getLong(int index) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public float getFloat(int index) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public double getDouble(int index) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public StringData getString(int index) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public DecimalData getDecimal(int index, int precision, int scale) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public TimestampData getTimestamp(int index, int precision) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public <T> RawValueData<T> getRawValue(int index) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public byte[] getBinary(int index) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public ArrayData getArray(int index) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public MapData getMap(int index) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public RowData getRow(int index, int numFields) {
- return new ColumnarGroupRowData((HeapRowColumnVector) vector, rowId, index);
- }
-
- @Override
- public boolean[] toBooleanArray() {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public byte[] toByteArray() {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public short[] toShortArray() {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public int[] toIntArray() {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public long[] toLongArray() {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public float[] toFloatArray() {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public double[] toDoubleArray() {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
-}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ColumnarGroupMapData.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ColumnarGroupMapData.java
deleted file mode 100644
index 69cb6feca13e..000000000000
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ColumnarGroupMapData.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * 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.vector;
-
-import org.apache.flink.table.data.ArrayData;
-import org.apache.flink.table.data.MapData;
-import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
-
-public class ColumnarGroupMapData implements MapData {
-
- WritableColumnVector keyVector;
- WritableColumnVector valueVector;
- int rowId;
-
- public ColumnarGroupMapData(WritableColumnVector keyVector, WritableColumnVector valueVector, int rowId) {
- this.keyVector = keyVector;
- this.valueVector = valueVector;
- this.rowId = rowId;
- }
-
- @Override
- public int size() {
- if (keyVector == null) {
- return 0;
- }
-
- if (keyVector instanceof HeapArrayVector) {
- return ((HeapArrayVector) keyVector).getArray(rowId).size();
- }
- throw new UnsupportedOperationException(keyVector.getClass().getName() + " is not supported. Supported vector types: HeapArrayVector");
- }
-
- @Override
- public ArrayData keyArray() {
- return ((HeapArrayVector) keyVector).getArray(rowId);
- }
-
- @Override
- public ArrayData valueArray() {
- if (valueVector instanceof HeapArrayVector) {
- return ((HeapArrayVector) valueVector).getArray(rowId);
- } else if (valueVector instanceof HeapArrayGroupColumnVector) {
- return ((HeapArrayGroupColumnVector) valueVector).getArray(rowId);
- }
- throw new UnsupportedOperationException(valueVector.getClass().getName() + " is not supported. Supported vector types: HeapArrayVector, HeapArrayGroupColumnVector");
- }
-}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ColumnarGroupRowData.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ColumnarGroupRowData.java
deleted file mode 100644
index 439c1880823f..000000000000
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ColumnarGroupRowData.java
+++ /dev/null
@@ -1,138 +0,0 @@
-/*
- * 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.vector;
-
-import org.apache.flink.table.data.ArrayData;
-import org.apache.flink.table.data.DecimalData;
-import org.apache.flink.table.data.MapData;
-import org.apache.flink.table.data.RawValueData;
-import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.StringData;
-import org.apache.flink.table.data.TimestampData;
-import org.apache.flink.types.RowKind;
-
-public class ColumnarGroupRowData implements RowData {
-
- HeapRowColumnVector vector;
- int rowId;
- int index;
-
- public ColumnarGroupRowData(HeapRowColumnVector vector, int rowId, int index) {
- this.vector = vector;
- this.rowId = rowId;
- this.index = index;
- }
-
- @Override
- public int getArity() {
- return vector.vectors.length;
- }
-
- @Override
- public RowKind getRowKind() {
- return RowKind.INSERT;
- }
-
- @Override
- public void setRowKind(RowKind rowKind) {
- throw new UnsupportedOperationException("Not support the operation!");
- }
-
- @Override
- public boolean isNullAt(int pos) {
- return
- vector.vectors[pos].isNullAt(rowId)
- || ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).isNullAt(index);
- }
-
- @Override
- public boolean getBoolean(int pos) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getBoolean(index);
- }
-
- @Override
- public byte getByte(int pos) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getByte(index);
- }
-
- @Override
- public short getShort(int pos) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getShort(index);
- }
-
- @Override
- public int getInt(int pos) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getInt(index);
- }
-
- @Override
- public long getLong(int pos) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getLong(index);
- }
-
- @Override
- public float getFloat(int pos) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getFloat(index);
- }
-
- @Override
- public double getDouble(int pos) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getDouble(index);
- }
-
- @Override
- public StringData getString(int pos) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getString(index);
- }
-
- @Override
- public DecimalData getDecimal(int pos, int i1, int i2) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getDecimal(index, i1, i2);
- }
-
- @Override
- public TimestampData getTimestamp(int pos, int i1) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getTimestamp(index, i1);
- }
-
- @Override
- public <T> RawValueData<T> getRawValue(int pos) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getRawValue(index);
- }
-
- @Override
- public byte[] getBinary(int pos) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getBinary(index);
- }
-
- @Override
- public ArrayData getArray(int pos) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getArray(index);
- }
-
- @Override
- public MapData getMap(int pos) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getMap(index);
- }
-
- @Override
- public RowData getRow(int pos, int numFields) {
- return ((HeapArrayVector) (vector.vectors[pos])).getArray(rowId).getRow(index, numFields);
- }
-}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapArrayGroupColumnVector.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapArrayGroupColumnVector.java
deleted file mode 100644
index 3d7d8b1f0de0..000000000000
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapArrayGroupColumnVector.java
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
- * 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.vector;
-
-import org.apache.flink.table.data.ArrayData;
-import org.apache.flink.table.data.columnar.vector.ArrayColumnVector;
-import org.apache.flink.table.data.columnar.vector.heap.AbstractHeapVector;
-import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
-
-/**
- * This class represents a nullable heap row column vector.
- */
-public class HeapArrayGroupColumnVector extends AbstractHeapVector
- implements WritableColumnVector, ArrayColumnVector {
-
- public WritableColumnVector vector;
-
- public HeapArrayGroupColumnVector(int len) {
- super(len);
- }
-
- public HeapArrayGroupColumnVector(int len, WritableColumnVector vector) {
- super(len);
- this.vector = vector;
- }
-
- @Override
- public ArrayData getArray(int rowId) {
- return new ColumnarGroupArrayData(vector, rowId);
- }
-
- @Override
- public void reset() {
- super.reset();
- vector.reset();
- }
-}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapArrayVector.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapArrayVector.java
index a0dced01e5e8..2f21a323302f 100644
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapArrayVector.java
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapArrayVector.java
@@ -57,6 +57,37 @@ public class HeapArrayVector extends AbstractHeapVector
return this.isNull.length;
}
+ // ---------------------------------------------------------------------------------------------
+ // Flink 2.1-compatible accessors. Backed by the existing public {@code offsets}, {@code lengths}
+ // and {@code child} fields so legacy callers continue to work; the new {@link
+ // org.apache.hudi.table.format.cow.vector.reader.NestedColumnReader} (FLINK-35702 port) and any
+ // future Flink-2.1-style caller use these accessors.
+ // ---------------------------------------------------------------------------------------------
+
+ public long[] getOffsets() {
+ return offsets;
+ }
+
+ public void setOffsets(long[] offsets) {
+ this.offsets = offsets;
+ }
+
+ public long[] getLengths() {
+ return lengths;
+ }
+
+ public void setLengths(long[] lengths) {
+ this.lengths = lengths;
+ }
+
+ public ColumnVector getChild() {
+ return child;
+ }
+
+ public void setChild(ColumnVector child) {
+ this.child = child;
+ }
+
@Override
public ArrayData getArray(int i) {
long offset = offsets[i];
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapMapColumnVector.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapMapColumnVector.java
index 0f088df55c1a..14aad22039e0 100644
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapMapColumnVector.java
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapMapColumnVector.java
@@ -20,30 +20,97 @@ package org.apache.hudi.table.format.cow.vector;
import lombok.Getter;
import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.columnar.ColumnarMapData;
+import org.apache.flink.table.data.columnar.vector.ColumnVector;
import org.apache.flink.table.data.columnar.vector.MapColumnVector;
import org.apache.flink.table.data.columnar.vector.heap.AbstractHeapVector;
import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
/**
* This class represents a nullable heap map column vector.
+ *
+ * <p>Mirrors {@code org.apache.flink.table.data.columnar.vector.heap.HeapMapVector} from
+ * Flink 2.1 (FLINK-35702). One deliberate divergence from upstream is preserved for backward
+ * compatibility: the {@code keys} / {@code values} fields are typed
+ * {@link WritableColumnVector} rather than upstream's {@link ColumnVector}, so the existing
+ * Lombok-generated {@code getKeys()} / {@code getValues()} accessors keep their original
+ * signature. Callers wanting the Flink-2.1 contract (a {@code ColumnVector}) use
+ * {@link #getKeyColumnVector()} / {@link #getValueColumnVector()}.
*/
public class HeapMapColumnVector extends AbstractHeapVector
implements WritableColumnVector, MapColumnVector {
@Getter
- private final WritableColumnVector keys;
+ private WritableColumnVector keys;
@Getter
- private final WritableColumnVector values;
+ private WritableColumnVector values;
+
+ // ---------------------------------------------------------------------------------------------
+ // Flink 2.1 Dremel-style state. Populated by {@link
+ // org.apache.hudi.table.format.cow.vector.reader.NestedColumnReader} (FLINK-35702 port) and
+ // consumed by {@link #getMap(int)}.
+ // ---------------------------------------------------------------------------------------------
+ private long[] offsets;
+ private long[] lengths;
+ private int size;
public HeapMapColumnVector(int len, WritableColumnVector keys, WritableColumnVector values) {
super(len);
+ this.offsets = new long[len];
+ this.lengths = new long[len];
this.keys = keys;
this.values = values;
}
+ public long[] getOffsets() {
+ return offsets;
+ }
+
+ public void setOffsets(long[] offsets) {
+ this.offsets = offsets;
+ }
+
+ public long[] getLengths() {
+ return lengths;
+ }
+
+ public void setLengths(long[] lengths) {
+ this.lengths = lengths;
+ }
+
+ public int getSize() {
+ return size;
+ }
+
+ public void setSize(int size) {
+ this.size = size;
+ }
+
+ public void setKeys(WritableColumnVector keys) {
+ this.keys = keys;
+ }
+
+ public void setValues(WritableColumnVector values) {
+ this.values = values;
+ }
+
+ /**
+ * Returns the keys child vector typed as {@link ColumnVector}, matching the Flink 2.1 contract
+ * consumed by {@code NestedColumnReader}. Functionally equivalent to {@link #getKeys()}.
+ */
+ public ColumnVector getKeyColumnVector() {
+ return keys;
+ }
+
+ /** Counterpart of {@link #getKeyColumnVector()} for the values child vector. */
+ public ColumnVector getValueColumnVector() {
+ return values;
+ }
+
@Override
public MapData getMap(int rowId) {
- return new ColumnarGroupMapData(keys, values, rowId);
+ long offset = offsets[rowId];
+ long length = lengths[rowId];
+ return new ColumnarMapData(keys, values, (int) offset, (int) length);
}
}
-
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapRowColumnVector.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapRowColumnVector.java
index ae194e4e6ab0..0c640ce92ee4 100644
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapRowColumnVector.java
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/HeapRowColumnVector.java
@@ -37,6 +37,21 @@ public class HeapRowColumnVector extends AbstractHeapVector
this.vectors = vectors;
}
+ /**
+ * Flink 2.1-compatible accessor for the children vectors. Backed by the existing public {@code
+ * vectors} field so legacy callers continue to work; the new {@link
+ * org.apache.hudi.table.format.cow.vector.reader.NestedColumnReader} (FLINK-35702 port) and any
+ * future Flink-2.1-style caller use this accessor.
+ */
+ public WritableColumnVector[] getFields() {
+ return vectors;
+ }
+
+ /** Counterpart of {@link #getFields()}. */
+ public void setFields(WritableColumnVector[] fields) {
+ this.vectors = fields;
+ }
+
@Override
public ColumnarRowData getRow(int i) {
ColumnarRowData columnarRowData = new ColumnarRowData(new VectorizedColumnBatch(vectors));
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ParquetDecimalVector.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ParquetDecimalVector.java
index 98b5e6105089..a37b88352cf5 100644
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ParquetDecimalVector.java
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/ParquetDecimalVector.java
@@ -18,21 +18,29 @@
package org.apache.hudi.table.format.cow.vector;
+import org.apache.flink.formats.parquet.utils.ParquetSchemaConverter;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.columnar.vector.BytesColumnVector;
import org.apache.flink.table.data.columnar.vector.ColumnVector;
import org.apache.flink.table.data.columnar.vector.DecimalColumnVector;
+import org.apache.flink.table.data.columnar.vector.Dictionary;
+import org.apache.flink.table.data.columnar.vector.IntColumnVector;
+import org.apache.flink.table.data.columnar.vector.LongColumnVector;
+import org.apache.flink.table.data.columnar.vector.writable.WritableBytesVector;
+import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
+import org.apache.flink.table.data.columnar.vector.writable.WritableIntVector;
+import org.apache.flink.table.data.columnar.vector.writable.WritableLongVector;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
/**
- * Parquet write decimal as int32 and int64 and binary, this class wrap the real vector to
- * provide {@link DecimalColumnVector} interface.
- *
- * <p>Reference Flink release 1.11.2 {@link org.apache.flink.formats.parquet.vector.ParquetDecimalVector}
- * because it is not public.
+ * Parquet write decimal as int32 and int64 and binary, this class wrap the real vector to provide
+ * {@link DecimalColumnVector} interface.
*/
-public class ParquetDecimalVector implements DecimalColumnVector {
+public class ParquetDecimalVector
+ implements DecimalColumnVector, WritableLongVector, WritableIntVector, WritableBytesVector {
- public final ColumnVector vector;
+ private final ColumnVector vector;
public ParquetDecimalVector(ColumnVector vector) {
this.vector = vector;
@@ -40,15 +48,180 @@ public class ParquetDecimalVector implements DecimalColumnVector {
@Override
public DecimalData getDecimal(int i, int precision, int scale) {
- return DecimalData.fromUnscaledBytes(
- ((BytesColumnVector) vector).getBytes(i).getBytes(),
- precision,
- scale);
+ if (ParquetSchemaConverter.is32BitDecimal(precision) && vector instanceof IntColumnVector) {
+ return DecimalData.fromUnscaledLong(((IntColumnVector) vector).getInt(i), precision, scale);
+ } else if (ParquetSchemaConverter.is64BitDecimal(precision)
+ && vector instanceof LongColumnVector) {
+ return DecimalData.fromUnscaledLong(((LongColumnVector) vector).getLong(i), precision, scale);
+ } else {
+ checkArgument(
+ vector instanceof BytesColumnVector,
+ "Reading decimal type occur unsupported vector type: %s",
+ vector.getClass());
+ return DecimalData.fromUnscaledBytes(
+ ((BytesColumnVector) vector).getBytes(i).getBytes(), precision, scale);
+ }
+ }
+
+ public ColumnVector getVector() {
+ return vector;
}
@Override
public boolean isNullAt(int i) {
return vector.isNullAt(i);
}
-}
+ @Override
+ public void reset() {
+ if (vector instanceof WritableColumnVector) {
+ ((WritableColumnVector) vector).reset();
+ }
+ }
+
+ @Override
+ public void setNullAt(int rowId) {
+ if (vector instanceof WritableColumnVector) {
+ ((WritableColumnVector) vector).setNullAt(rowId);
+ }
+ }
+
+ @Override
+ public void setNulls(int rowId, int count) {
+ if (vector instanceof WritableColumnVector) {
+ ((WritableColumnVector) vector).setNulls(rowId, count);
+ }
+ }
+
+ @Override
+ public void fillWithNulls() {
+ if (vector instanceof WritableColumnVector) {
+ ((WritableColumnVector) vector).fillWithNulls();
+ }
+ }
+
+ @Override
+ public void setDictionary(Dictionary dictionary) {
+ if (vector instanceof WritableColumnVector) {
+ ((WritableColumnVector) vector).setDictionary(dictionary);
+ }
+ }
+
+ @Override
+ public boolean hasDictionary() {
+ if (vector instanceof WritableColumnVector) {
+ return ((WritableColumnVector) vector).hasDictionary();
+ }
+ return false;
+ }
+
+ @Override
+ public WritableIntVector reserveDictionaryIds(int capacity) {
+ if (vector instanceof WritableColumnVector) {
+ return ((WritableColumnVector) vector).reserveDictionaryIds(capacity);
+ }
+ throw new RuntimeException("Child vector must be instance of WritableColumnVector");
+ }
+
+ @Override
+ public WritableIntVector getDictionaryIds() {
+ if (vector instanceof WritableColumnVector) {
+ return ((WritableColumnVector) vector).getDictionaryIds();
+ }
+ throw new RuntimeException("Child vector must be instance of WritableColumnVector");
+ }
+
+ @Override
+ public Bytes getBytes(int i) {
+ if (vector instanceof WritableBytesVector) {
+ return ((WritableBytesVector) vector).getBytes(i);
+ }
+ throw new RuntimeException("Child vector must be instance of WritableColumnVector");
+ }
+
+ @Override
+ public void appendBytes(int rowId, byte[] value, int offset, int length) {
+ if (vector instanceof WritableBytesVector) {
+ ((WritableBytesVector) vector).appendBytes(rowId, value, offset, length);
+ }
+ }
+
+ @Override
+ public void fill(byte[] value) {
+ if (vector instanceof WritableBytesVector) {
+ ((WritableBytesVector) vector).fill(value);
+ }
+ }
+
+ @Override
+ public int getInt(int i) {
+ if (vector instanceof WritableIntVector) {
+ return ((WritableIntVector) vector).getInt(i);
+ }
+ throw new RuntimeException("Child vector must be instance of WritableColumnVector");
+ }
+
+ @Override
+ public void setInt(int rowId, int value) {
+ if (vector instanceof WritableIntVector) {
+ ((WritableIntVector) vector).setInt(rowId, value);
+ }
+ }
+
+ @Override
+ public void setIntsFromBinary(int rowId, int count, byte[] src, int srcIndex) {
+ if (vector instanceof WritableIntVector) {
+ ((WritableIntVector) vector).setIntsFromBinary(rowId, count, src, srcIndex);
+ }
+ }
+
+ @Override
+ public void setInts(int rowId, int count, int value) {
+ if (vector instanceof WritableIntVector) {
+ ((WritableIntVector) vector).setInts(rowId, count, value);
+ }
+ }
+
+ @Override
+ public void setInts(int rowId, int count, int[] src, int srcIndex) {
+ if (vector instanceof WritableIntVector) {
+ ((WritableIntVector) vector).setInts(rowId, count, src, srcIndex);
+ }
+ }
+
+ @Override
+ public void fill(int value) {
+ if (vector instanceof WritableIntVector) {
+ ((WritableIntVector) vector).fill(value);
+ }
+ }
+
+ @Override
+ public long getLong(int i) {
+ if (vector instanceof WritableLongVector) {
+ return ((WritableLongVector) vector).getLong(i);
+ }
+ throw new RuntimeException("Child vector must be instance of WritableColumnVector");
+ }
+
+ @Override
+ public void setLong(int rowId, long value) {
+ if (vector instanceof WritableLongVector) {
+ ((WritableLongVector) vector).setLong(rowId, value);
+ }
+ }
+
+ @Override
+ public void setLongsFromBinary(int rowId, int count, byte[] src, int srcIndex) {
+ if (vector instanceof WritableLongVector) {
+ ((WritableLongVector) vector).setLongsFromBinary(rowId, count, src, srcIndex);
+ }
+ }
+
+ @Override
+ public void fill(long value) {
+ if (vector instanceof WritableLongVector) {
+ ((WritableLongVector) vector).fill(value);
+ }
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/position/CollectionPosition.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/position/CollectionPosition.java
new file mode 100644
index 000000000000..fcdedfbc9d71
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/position/CollectionPosition.java
@@ -0,0 +1,57 @@
+/*
+ * 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.vector.position;
+
+import javax.annotation.Nullable;
+
+/**
+ * To represent collection's position in repeated type.
+ *
+ * <p>Note: Vendored from Apache Flink (FLINK-35702, {@code
+ * org.apache.flink.formats.parquet.vector.position.CollectionPosition}).
+ */
+public class CollectionPosition {
+ @Nullable private final boolean[] isNull;
+ private final long[] offsets;
+ private final long[] length;
+ private final int valueCount;
+
+ public CollectionPosition(boolean[] isNull, long[] offsets, long[] length, int valueCount) {
+ this.isNull = isNull;
+ this.offsets = offsets;
+ this.length = length;
+ this.valueCount = valueCount;
+ }
+
+ public boolean[] getIsNull() {
+ return isNull;
+ }
+
+ public long[] getOffsets() {
+ return offsets;
+ }
+
+ public long[] getLength() {
+ return length;
+ }
+
+ public int getValueCount() {
+ return valueCount;
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/position/LevelDelegation.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/position/LevelDelegation.java
new file mode 100644
index 000000000000..fe95419ac321
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/position/LevelDelegation.java
@@ -0,0 +1,43 @@
+/*
+ * 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.vector.position;
+
+/**
+ * To delegate repetition level and definition level.
+ *
+ * <p>Note: Vendored from Apache Flink (FLINK-35702, {@code
+ * org.apache.flink.formats.parquet.vector.position.LevelDelegation}).
+ */
+public class LevelDelegation {
+ private final int[] repetitionLevel;
+ private final int[] definitionLevel;
+
+ public LevelDelegation(int[] repetitionLevel, int[] definitionLevel) {
+ this.repetitionLevel = repetitionLevel;
+ this.definitionLevel = definitionLevel;
+ }
+
+ public int[] getRepetitionLevel() {
+ return repetitionLevel;
+ }
+
+ public int[] getDefinitionLevel() {
+ return definitionLevel;
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/position/RowPosition.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/position/RowPosition.java
new file mode 100644
index 000000000000..5438b6797323
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/position/RowPosition.java
@@ -0,0 +1,45 @@
+/*
+ * 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.vector.position;
+
+import javax.annotation.Nullable;
+
+/**
+ * To represent struct's position in repeated type.
+ *
+ * <p>Note: Vendored from Apache Flink (FLINK-35702, {@code
+ * org.apache.flink.formats.parquet.vector.position.RowPosition}).
+ */
+public class RowPosition {
+ @Nullable private final boolean[] isNull;
+ private final int positionsCount;
+
+ public RowPosition(boolean[] isNull, int positionsCount) {
+ this.isNull = isNull;
+ this.positionsCount = positionsCount;
+ }
+
+ public boolean[] getIsNull() {
+ return isNull;
+ }
+
+ public int getPositionsCount() {
+ return positionsCount;
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ArrayColumnReader.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ArrayColumnReader.java
deleted file mode 100644
index 6a8a01b74946..000000000000
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ArrayColumnReader.java
+++ /dev/null
@@ -1,473 +0,0 @@
-/*
- * 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.vector.reader;
-
-import org.apache.hudi.table.format.cow.vector.HeapArrayVector;
-import org.apache.hudi.table.format.cow.vector.ParquetDecimalVector;
-
-import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
-import org.apache.flink.table.data.TimestampData;
-import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
-import org.apache.flink.table.data.columnar.vector.heap.HeapBooleanVector;
-import org.apache.flink.table.data.columnar.vector.heap.HeapByteVector;
-import org.apache.flink.table.data.columnar.vector.heap.HeapBytesVector;
-import org.apache.flink.table.data.columnar.vector.heap.HeapDoubleVector;
-import org.apache.flink.table.data.columnar.vector.heap.HeapFloatVector;
-import org.apache.flink.table.data.columnar.vector.heap.HeapIntVector;
-import org.apache.flink.table.data.columnar.vector.heap.HeapLongVector;
-import org.apache.flink.table.data.columnar.vector.heap.HeapShortVector;
-import org.apache.flink.table.data.columnar.vector.heap.HeapTimestampVector;
-import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
-import org.apache.flink.table.types.logical.ArrayType;
-import org.apache.flink.table.types.logical.LogicalType;
-import org.apache.parquet.column.ColumnDescriptor;
-import org.apache.parquet.column.page.PageReader;
-import org.apache.parquet.schema.PrimitiveType;
-import org.apache.parquet.schema.Type;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * Array {@link ColumnReader}.
- */
-public class ArrayColumnReader extends BaseVectorizedColumnReader {
-
- // The value read in last time
- private Object lastValue;
-
- // flag to indicate if there is no data in parquet data page
- private boolean eof = false;
-
- // flag to indicate if it's the first time to read parquet data page with this instance
- boolean isFirstRow = true;
-
- public ArrayColumnReader(
- ColumnDescriptor descriptor,
- PageReader pageReader,
- boolean isUtcTimestamp,
- Type type,
- LogicalType logicalType)
- throws IOException {
- super(descriptor, pageReader, isUtcTimestamp, type, logicalType);
- }
-
- @Override
- public void readToVector(int readNumber, WritableColumnVector vector) throws IOException {
- HeapArrayVector lcv = (HeapArrayVector) vector;
- // before readBatch, initial the size of offsets & lengths as the default value,
- // the actual size will be assigned in setChildrenInfo() after reading complete.
- lcv.offsets = new long[VectorizedColumnBatch.DEFAULT_SIZE];
- lcv.lengths = new long[VectorizedColumnBatch.DEFAULT_SIZE];
- // Because the length of ListColumnVector.child can't be known now,
- // the valueList will save all data for ListColumnVector temporary.
- List<Object> valueList = new ArrayList<>();
-
- LogicalType category = ((ArrayType) logicalType).getElementType();
-
- // read the first row in parquet data page, this will be only happened once for this
- // instance
- if (isFirstRow) {
- if (!fetchNextValue(category)) {
- return;
- }
- isFirstRow = false;
- }
-
- int index = collectDataFromParquetPage(readNumber, lcv, valueList, category);
-
- // Convert valueList to array for the ListColumnVector.child
- fillColumnVector(category, lcv, valueList, index);
- }
-
- /**
- * Reads a single value from parquet page, puts it into lastValue. Returns a boolean indicating
- * if there is more values to read (true).
- *
- * @param category
- * @return boolean
- * @throws IOException
- */
- private boolean fetchNextValue(LogicalType category) throws IOException {
- int left = readPageIfNeed();
- if (left > 0) {
- // get the values of repetition and definitionLevel
- readRepetitionAndDefinitionLevels();
- // read the data if it isn't null
- if (definitionLevel == maxDefLevel) {
- if (isCurrentPageDictionaryEncoded) {
- lastValue = dataColumn.readValueDictionaryId();
- } else {
- lastValue = readPrimitiveTypedRow(category);
- }
- } else {
- lastValue = null;
- }
- return true;
- } else {
- eof = true;
- return false;
- }
- }
-
- private int readPageIfNeed() throws IOException {
- // Compute the number of values we want to read in this page.
- int leftInPage = (int) (endOfPageValueCount - valuesRead);
- if (leftInPage == 0) {
- // no data left in current page, load data from new page
- readPage();
- leftInPage = (int) (endOfPageValueCount - valuesRead);
- }
- return leftInPage;
- }
-
- // Need to be in consistent with that VectorizedPrimitiveColumnReader#readBatchHelper
- // TODO Reduce the duplicated code
- private Object readPrimitiveTypedRow(LogicalType category) {
- switch (category.getTypeRoot()) {
- case CHAR:
- case VARCHAR:
- case BINARY:
- case VARBINARY:
- return dataColumn.readString();
- case BOOLEAN:
- return dataColumn.readBoolean();
- case TIME_WITHOUT_TIME_ZONE:
- case DATE:
- case INTEGER:
- return dataColumn.readInteger();
- case TINYINT:
- return dataColumn.readTinyInt();
- case SMALLINT:
- return dataColumn.readSmallInt();
- case BIGINT:
- return dataColumn.readLong();
- case FLOAT:
- return dataColumn.readFloat();
- case DOUBLE:
- return dataColumn.readDouble();
- case DECIMAL:
- switch (descriptor.getPrimitiveType().getPrimitiveTypeName()) {
- case INT32:
- return dataColumn.readInteger();
- case INT64:
- return dataColumn.readLong();
- case BINARY:
- case FIXED_LEN_BYTE_ARRAY:
- return dataColumn.readString();
- default:
- throw new AssertionError();
- }
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- return dataColumn.readTimestamp();
- default:
- throw new RuntimeException("Unsupported type in the list: " + type);
- }
- }
-
- private Object dictionaryDecodeValue(LogicalType category, Integer dictionaryValue) {
- if (dictionaryValue == null) {
- return null;
- }
-
- switch (category.getTypeRoot()) {
- case CHAR:
- case VARCHAR:
- case BINARY:
- case VARBINARY:
- return dictionary.readString(dictionaryValue);
- case DATE:
- case TIME_WITHOUT_TIME_ZONE:
- case INTEGER:
- return dictionary.readInteger(dictionaryValue);
- case BOOLEAN:
- return dictionary.readBoolean(dictionaryValue) ? 1 : 0;
- case DOUBLE:
- return dictionary.readDouble(dictionaryValue);
- case FLOAT:
- return dictionary.readFloat(dictionaryValue);
- case TINYINT:
- return dictionary.readTinyInt(dictionaryValue);
- case SMALLINT:
- return dictionary.readSmallInt(dictionaryValue);
- case BIGINT:
- return dictionary.readLong(dictionaryValue);
- case DECIMAL:
- switch (descriptor.getPrimitiveType().getPrimitiveTypeName()) {
- case INT32:
- return dictionary.readInteger(dictionaryValue);
- case INT64:
- return dictionary.readLong(dictionaryValue);
- case FIXED_LEN_BYTE_ARRAY:
- case BINARY:
- return dictionary.readString(dictionaryValue);
- default:
- throw new AssertionError();
- }
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- return dictionary.readTimestamp(dictionaryValue);
- default:
- throw new RuntimeException("Unsupported type in the list: " + type);
- }
- }
-
- /**
- * Collects data from a parquet page and returns the final row index where it stopped. The
- * returned index can be equal to or less than total.
- *
- * @param total maximum number of rows to collect
- * @param lcv column vector to do initial setup in data collection time
- * @param valueList collection of values that will be fed into the vector later
- * @param category
- * @return int
- * @throws IOException
- */
- private int collectDataFromParquetPage(
- int total, HeapArrayVector lcv, List<Object> valueList, LogicalType category)
- throws IOException {
- int index = 0;
- /*
- * Here is a nested loop for collecting all values from a parquet page.
- * A column of array type can be considered as a list of lists, so the two loops are as below:
- * 1. The outer loop iterates on rows (index is a row index, so points to a row in the batch), e.g.:
- * [0, 2, 3] <- index: 0
- * [NULL, 3, 4] <- index: 1
- *
- * 2. The inner loop iterates on values within a row (sets all data from parquet data page
- * for an element in ListColumnVector), so fetchNextValue returns values one-by-one:
- * 0, 2, 3, NULL, 3, 4
- *
- * As described below, the repetition level (repetitionLevel != 0)
- * can be used to decide when we'll start to read values for the next list.
- */
- while (!eof && index < total) {
- // add element to ListColumnVector one by one
- lcv.offsets[index] = valueList.size();
- /*
- * Let's collect all values for a single list.
- * Repetition level = 0 means that a new list started there in the parquet page,
- * in that case, let's exit from the loop, and start to collect value for a new list.
- */
- do {
- /*
- * Definition level = 0 when a NULL value was returned instead of a list
- * (this is not the same as a NULL value in of a list).
- */
- if (definitionLevel == 0) {
- lcv.setNullAt(index);
- }
- valueList.add(
- isCurrentPageDictionaryEncoded
- ? dictionaryDecodeValue(category, (Integer) lastValue)
- : lastValue);
- } while (fetchNextValue(category) && (repetitionLevel != 0));
-
- lcv.lengths[index] = valueList.size() - lcv.offsets[index];
- index++;
- }
- return index;
- }
-
- /**
- * The lengths & offsets will be initialized as default size (1024), it should be set to the
- * actual size according to the element number.
- */
- private void setChildrenInfo(HeapArrayVector lcv, int itemNum, int elementNum) {
- lcv.setSize(itemNum);
- long[] lcvLength = new long[elementNum];
- long[] lcvOffset = new long[elementNum];
- System.arraycopy(lcv.lengths, 0, lcvLength, 0, elementNum);
- System.arraycopy(lcv.offsets, 0, lcvOffset, 0, elementNum);
- lcv.lengths = lcvLength;
- lcv.offsets = lcvOffset;
- }
-
- private void fillColumnVector(
- LogicalType category, HeapArrayVector lcv, List valueList, int elementNum) {
- int total = valueList.size();
- setChildrenInfo(lcv, total, elementNum);
- switch (category.getTypeRoot()) {
- case CHAR:
- case VARCHAR:
- case BINARY:
- case VARBINARY:
- lcv.child = new HeapBytesVector(total);
- ((HeapBytesVector) lcv.child).reset();
- for (int i = 0; i < valueList.size(); i++) {
- byte[] src = ((List<byte[]>) valueList).get(i);
- if (src == null) {
- ((HeapBytesVector) lcv.child).setNullAt(i);
- } else {
- ((HeapBytesVector) lcv.child).appendBytes(i, src, 0, src.length);
- }
- }
- break;
- case BOOLEAN:
- lcv.child = new HeapBooleanVector(total);
- ((HeapBooleanVector) lcv.child).reset();
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapBooleanVector) lcv.child).setNullAt(i);
- } else {
- ((HeapBooleanVector) lcv.child).vector[i] =
- ((List<Boolean>) valueList).get(i);
- }
- }
- break;
- case TINYINT:
- lcv.child = new HeapByteVector(total);
- ((HeapByteVector) lcv.child).reset();
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapByteVector) lcv.child).setNullAt(i);
- } else {
- ((HeapByteVector) lcv.child).vector[i] =
- (byte) ((List<Integer>) valueList).get(i).intValue();
- }
- }
- break;
- case SMALLINT:
- lcv.child = new HeapShortVector(total);
- ((HeapShortVector) lcv.child).reset();
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapShortVector) lcv.child).setNullAt(i);
- } else {
- ((HeapShortVector) lcv.child).vector[i] =
- (short) ((List<Integer>) valueList).get(i).intValue();
- }
- }
- break;
- case INTEGER:
- case DATE:
- case TIME_WITHOUT_TIME_ZONE:
- lcv.child = new HeapIntVector(total);
- ((HeapIntVector) lcv.child).reset();
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapIntVector) lcv.child).setNullAt(i);
- } else {
- ((HeapIntVector) lcv.child).vector[i] = ((List<Integer>) valueList).get(i);
- }
- }
- break;
- case FLOAT:
- lcv.child = new HeapFloatVector(total);
- ((HeapFloatVector) lcv.child).reset();
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapFloatVector) lcv.child).setNullAt(i);
- } else {
- ((HeapFloatVector) lcv.child).vector[i] = ((List<Float>) valueList).get(i);
- }
- }
- break;
- case BIGINT:
- lcv.child = new HeapLongVector(total);
- ((HeapLongVector) lcv.child).reset();
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapLongVector) lcv.child).setNullAt(i);
- } else {
- ((HeapLongVector) lcv.child).vector[i] = ((List<Long>) valueList).get(i);
- }
- }
- break;
- case DOUBLE:
- lcv.child = new HeapDoubleVector(total);
- ((HeapDoubleVector) lcv.child).reset();
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapDoubleVector) lcv.child).setNullAt(i);
- } else {
- ((HeapDoubleVector) lcv.child).vector[i] =
- ((List<Double>) valueList).get(i);
- }
- }
- break;
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- lcv.child = new HeapTimestampVector(total);
- ((HeapTimestampVector) lcv.child).reset();
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapTimestampVector) lcv.child).setNullAt(i);
- } else {
- ((HeapTimestampVector) lcv.child)
- .setTimestamp(i, ((List<TimestampData>) valueList).get(i));
- }
- }
- break;
- case DECIMAL:
- PrimitiveType.PrimitiveTypeName primitiveTypeName =
- descriptor.getPrimitiveType().getPrimitiveTypeName();
- switch (primitiveTypeName) {
- case INT32:
- lcv.child = new ParquetDecimalVector(new HeapIntVector(total));
- ((HeapIntVector) ((ParquetDecimalVector) lcv.child).vector).reset();
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapIntVector) ((ParquetDecimalVector) lcv.child).vector)
- .setNullAt(i);
- } else {
- ((HeapIntVector) ((ParquetDecimalVector) lcv.child).vector)
- .vector[i] =
- ((List<Integer>) valueList).get(i);
- }
- }
- break;
- case INT64:
- lcv.child = new ParquetDecimalVector(new HeapLongVector(total));
- ((HeapLongVector) ((ParquetDecimalVector) lcv.child).vector).reset();
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapLongVector) ((ParquetDecimalVector) lcv.child).vector)
- .setNullAt(i);
- } else {
- ((HeapLongVector) ((ParquetDecimalVector) lcv.child).vector)
- .vector[i] =
- ((List<Long>) valueList).get(i);
- }
- }
- break;
- default:
- lcv.child = new ParquetDecimalVector(new HeapBytesVector(total));
- ((HeapBytesVector) ((ParquetDecimalVector) lcv.child).vector).reset();
- for (int i = 0; i < valueList.size(); i++) {
- byte[] src = ((List<byte[]>) valueList).get(i);
- if (valueList.get(i) == null) {
- ((HeapBytesVector) ((ParquetDecimalVector) lcv.child).vector)
- .setNullAt(i);
- } else {
- ((HeapBytesVector) ((ParquetDecimalVector) lcv.child).vector)
- .appendBytes(i, src, 0, src.length);
- }
- }
- break;
- }
- break;
- default:
- throw new RuntimeException("Unsupported type in the list: " + type);
- }
- }
-}
-
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ArrayGroupReader.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ArrayGroupReader.java
deleted file mode 100644
index df7c5d85bc4a..000000000000
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ArrayGroupReader.java
+++ /dev/null
@@ -1,45 +0,0 @@
-/*
- * 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.vector.reader;
-
-import org.apache.hudi.table.format.cow.vector.HeapArrayGroupColumnVector;
-
-import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
-import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
-
-import java.io.IOException;
-
-/**
- * Array of a Group type (Array, Map, Row, etc.) {@link ColumnReader}.
- */
-public class ArrayGroupReader implements ColumnReader<WritableColumnVector> {
-
- private final ColumnReader<WritableColumnVector> fieldReader;
-
- public ArrayGroupReader(ColumnReader<WritableColumnVector> fieldReader) {
- this.fieldReader = fieldReader;
- }
-
- @Override
- public void readToVector(int readNumber, WritableColumnVector vector) throws IOException {
- HeapArrayGroupColumnVector rowColumnVector = (HeapArrayGroupColumnVector) vector;
-
- fieldReader.readToVector(readNumber, rowColumnVector.vector);
- }
-}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/MapColumnReader.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/MapColumnReader.java
deleted file mode 100644
index ee65dd22c436..000000000000
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/MapColumnReader.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * 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.vector.reader;
-
-import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector;
-
-import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
-import org.apache.flink.table.data.columnar.vector.heap.AbstractHeapVector;
-import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
-
-import java.io.IOException;
-
-/**
- * Map {@link ColumnReader}.
- */
-public class MapColumnReader implements ColumnReader<WritableColumnVector> {
-
- private final ArrayColumnReader keyReader;
- private final ColumnReader<WritableColumnVector> valueReader;
-
- public MapColumnReader(
- ArrayColumnReader keyReader, ColumnReader<WritableColumnVector> valueReader) {
- this.keyReader = keyReader;
- this.valueReader = valueReader;
- }
-
- @Override
- public void readToVector(int readNumber, WritableColumnVector vector) throws IOException {
- HeapMapColumnVector mapColumnVector = (HeapMapColumnVector) vector;
- AbstractHeapVector keyArrayColumnVector = (AbstractHeapVector) (mapColumnVector.getKeys());
- keyReader.readToVector(readNumber, mapColumnVector.getKeys());
- valueReader.readToVector(readNumber, mapColumnVector.getValues());
- for (int i = 0; i < keyArrayColumnVector.getLen(); i++) {
- if (keyArrayColumnVector.isNullAt(i)) {
- mapColumnVector.setNullAt(i);
- }
- }
- }
-}
-
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
new file mode 100644
index 000000000000..b4f04b20c477
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedColumnReader.java
@@ -0,0 +1,278 @@
+/*
+ * 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.vector.reader;
+
+import org.apache.hudi.table.format.cow.utils.NestedPositionUtil;
+import org.apache.hudi.table.format.cow.vector.HeapArrayVector;
+import org.apache.hudi.table.format.cow.vector.HeapMapColumnVector;
+import org.apache.hudi.table.format.cow.vector.HeapRowColumnVector;
+import org.apache.hudi.table.format.cow.vector.position.CollectionPosition;
+import org.apache.hudi.table.format.cow.vector.position.LevelDelegation;
+import org.apache.hudi.table.format.cow.vector.position.RowPosition;
+import org.apache.hudi.table.format.cow.vector.type.ParquetField;
+import org.apache.hudi.table.format.cow.vector.type.ParquetGroupField;
+import org.apache.hudi.table.format.cow.vector.type.ParquetPrimitiveField;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.columnar.vector.ColumnVector;
+import org.apache.flink.table.data.columnar.vector.heap.AbstractHeapVector;
+import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.ArrayType;
+import org.apache.flink.table.types.logical.MapType;
+import org.apache.flink.table.types.logical.MultisetType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 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
+ * href="https://github.com/julienledem/redelm/wiki/The-striping-and-assembly-algorithms-from-the-Dremel-paper">the
+ * striping and assembly algorithms from the Dremel paper</a>.
+ *
+ * <p>Vendored from Apache Flink 2.1 (FLINK-35702, {@code
+ * org.apache.flink.formats.parquet.vector.reader.NestedColumnReader}). Differences vs. upstream:
+ *
+ * <ul>
+ * <li>Uses Hudi-local {@code HeapRowColumnVector}/{@code HeapMapColumnVector}/{@code
+ * HeapArrayVector} instead of the Flink-private {@code HeapRowVector}/{@code
+ * HeapMapVector}/{@code HeapArrayVector}.
+ * <li>Supports Hudi's schema-evolution contract: a {@code ParquetGroupField} representing a
+ * {@link RowType} may contain {@code null} children — meaning the corresponding logical
+ * field is absent from the Parquet file. Those slots are passed through unchanged and do
+ * not contribute to the row's repetition/definition-level stream.
+ * </ul>
+ */
+public class NestedColumnReader implements ColumnReader<WritableColumnVector> {
+
+ private final Map<ColumnDescriptor, NestedPrimitiveColumnReader> columnReaders;
+ private final boolean isUtcTimestamp;
+
+ private final PageReadStore pages;
+
+ private final ParquetField field;
+
+ public NestedColumnReader(boolean isUtcTimestamp, PageReadStore pages, ParquetField field) {
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.pages = pages;
+ this.field = field;
+ this.columnReaders = new HashMap<>();
+ }
+
+ @Override
+ public void readToVector(int readNumber, WritableColumnVector vector) throws IOException {
+ readData(field, readNumber, vector, false);
+ }
+
+ private Tuple2<LevelDelegation, WritableColumnVector> readData(
+ ParquetField field, int readNumber, ColumnVector vector, boolean inside) throws IOException {
+ if (field.getType() instanceof RowType) {
+ return readRow((ParquetGroupField) field, readNumber, vector, inside);
+ } else if (field.getType() instanceof MapType || field.getType() instanceof MultisetType) {
+ return readMap((ParquetGroupField) field, readNumber, vector, inside);
+ } else if (field.getType() instanceof ArrayType) {
+ return readArray((ParquetGroupField) field, readNumber, vector, inside);
+ } else {
+ return readPrimitive((ParquetPrimitiveField) field, readNumber, vector);
+ }
+ }
+
+ private Tuple2<LevelDelegation, WritableColumnVector> readRow(
+ ParquetGroupField field, int readNumber, ColumnVector vector, boolean inside)
+ throws IOException {
+ HeapRowColumnVector heapRowVector = (HeapRowColumnVector) vector;
+ LevelDelegation levelDelegation = null;
+ List<ParquetField> children = field.getChildren();
+ WritableColumnVector[] childrenVectors = heapRowVector.getFields();
+ WritableColumnVector[] finalChildrenVectors = new WritableColumnVector[childrenVectors.length];
+ for (int i = 0; i < children.size(); i++) {
+ ParquetField child = children.get(i);
+ if (child == null) {
+ // Hudi schema-evolution: the logical field is not present in the Parquet file. The slot
+ // vector was pre-populated with nulls by ParquetSplitReaderUtil#createWritableColumnVector
+ // (ROW branch); keep it as is and skip contributing to the level stream.
+ finalChildrenVectors[i] = childrenVectors[i];
+ continue;
+ }
+ Tuple2<LevelDelegation, WritableColumnVector> tuple =
+ readData(child, readNumber, childrenVectors[i], true);
+ levelDelegation = tuple.f0;
+ finalChildrenVectors[i] = tuple.f1;
+ }
+ if (levelDelegation == null) {
+ throw new FlinkRuntimeException(
+ String.format("Row field does not have any non-null children: %s.", field));
+ }
+
+ RowPosition rowPosition =
+ NestedPositionUtil.calculateRowOffsets(
+ field,
+ levelDelegation.getDefinitionLevel(),
+ levelDelegation.getRepetitionLevel());
+
+ // If row was inside the structure, then we need to renew the vector to reset the
+ // capacity.
+ if (inside) {
+ heapRowVector = new HeapRowColumnVector(rowPosition.getPositionsCount(), finalChildrenVectors);
+ } else {
+ heapRowVector.setFields(finalChildrenVectors);
+ }
+
+ if (rowPosition.getIsNull() != null) {
+ setFieldNullFlag(rowPosition.getIsNull(), heapRowVector);
+ }
+
+ // Hudi-specific: collapse a present row whose every child is null into a null row, so that a
+ // SQL value like `row(null, null)` round-trips to NULL on read. This was the behaviour of the
+ // legacy RowColumnReader (deleted alongside the Dremel rewire) and existing Hudi tables rely
+ // on it. Diverges from Flink 2.1, which would surface it as Row(null, null). Pinned by the
+ // integration test ITTestHoodieDataSource#testParquetNullChildColumnsRowTypes.
+ int rowCount = rowPosition.getPositionsCount();
+ for (int j = 0; j < rowCount; j++) {
+ if (heapRowVector.isNullAt(j)) {
+ continue;
+ }
+ boolean allChildrenNull = true;
+ for (WritableColumnVector child : finalChildrenVectors) {
+ if (!child.isNullAt(j)) {
+ allChildrenNull = false;
+ break;
+ }
+ }
+ if (allChildrenNull) {
+ heapRowVector.setNullAt(j);
+ }
+ }
+ return Tuple2.of(levelDelegation, heapRowVector);
+ }
+
+ private Tuple2<LevelDelegation, WritableColumnVector> readMap(
+ ParquetGroupField field, int readNumber, ColumnVector vector, boolean inside)
+ throws IOException {
+ HeapMapColumnVector mapVector = (HeapMapColumnVector) vector;
+ mapVector.reset();
+ List<ParquetField> children = field.getChildren();
+ Preconditions.checkArgument(
+ children.size() == 2,
+ "Maps must have two type parameters, found %s",
+ children.size());
+ Tuple2<LevelDelegation, WritableColumnVector> keyTuple =
+ readData(children.get(0), readNumber, mapVector.getKeyColumnVector(), true);
+ Tuple2<LevelDelegation, WritableColumnVector> valueTuple =
+ readData(children.get(1), readNumber, mapVector.getValueColumnVector(), true);
+
+ LevelDelegation levelDelegation = keyTuple.f0;
+
+ CollectionPosition collectionPosition =
+ NestedPositionUtil.calculateCollectionOffsets(
+ field,
+ levelDelegation.getDefinitionLevel(),
+ levelDelegation.getRepetitionLevel());
+
+ // If map was inside the structure, then we need to renew the vector to reset the
+ // capacity.
+ if (inside) {
+ mapVector = new HeapMapColumnVector(collectionPosition.getValueCount(), keyTuple.f1, valueTuple.f1);
+ } else {
+ mapVector.setKeys(keyTuple.f1);
+ mapVector.setValues(valueTuple.f1);
+ }
+
+ if (collectionPosition.getIsNull() != null) {
+ setFieldNullFlag(collectionPosition.getIsNull(), mapVector);
+ }
+
+ mapVector.setLengths(collectionPosition.getLength());
+ mapVector.setOffsets(collectionPosition.getOffsets());
+
+ return Tuple2.of(levelDelegation, mapVector);
+ }
+
+ private Tuple2<LevelDelegation, WritableColumnVector> readArray(
+ ParquetGroupField field, int readNumber, ColumnVector vector, boolean inside)
+ throws IOException {
+ HeapArrayVector arrayVector = (HeapArrayVector) vector;
+ arrayVector.reset();
+ List<ParquetField> children = field.getChildren();
+ Preconditions.checkArgument(
+ children.size() == 1,
+ "Arrays must have a single type parameter, found %s",
+ children.size());
+ Tuple2<LevelDelegation, WritableColumnVector> tuple =
+ readData(children.get(0), readNumber, arrayVector.getChild(), true);
+
+ LevelDelegation levelDelegation = tuple.f0;
+ CollectionPosition collectionPosition =
+ NestedPositionUtil.calculateCollectionOffsets(
+ field,
+ levelDelegation.getDefinitionLevel(),
+ levelDelegation.getRepetitionLevel());
+
+ // If array was inside the structure, then we need to renew the vector to reset the
+ // capacity.
+ if (inside) {
+ arrayVector = new HeapArrayVector(collectionPosition.getValueCount(), tuple.f1);
+ } else {
+ arrayVector.setChild(tuple.f1);
+ }
+
+ if (collectionPosition.getIsNull() != null) {
+ setFieldNullFlag(collectionPosition.getIsNull(), arrayVector);
+ }
+ arrayVector.setLengths(collectionPosition.getLength());
+ arrayVector.setOffsets(collectionPosition.getOffsets());
+ return Tuple2.of(levelDelegation, arrayVector);
+ }
+
+ private Tuple2<LevelDelegation, WritableColumnVector> readPrimitive(
+ ParquetPrimitiveField field, int readNumber, ColumnVector vector) throws IOException {
+ ColumnDescriptor descriptor = field.getDescriptor();
+ NestedPrimitiveColumnReader reader = columnReaders.get(descriptor);
+ if (reader == null) {
+ reader =
+ new NestedPrimitiveColumnReader(
+ descriptor,
+ pages.getPageReader(descriptor),
+ isUtcTimestamp,
+ descriptor.getPrimitiveType(),
+ field.getType());
+ columnReaders.put(descriptor, reader);
+ }
+ WritableColumnVector writableColumnVector =
+ reader.readAndNewVector(readNumber, (WritableColumnVector) vector);
+ return Tuple2.of(reader.getLevelDelegation(), writableColumnVector);
+ }
+
+ private static void setFieldNullFlag(boolean[] nullFlags, AbstractHeapVector vector) {
+ for (int index = 0; index < vector.getLen() && index < nullFlags.length; index++) {
+ if (nullFlags[index]) {
+ vector.setNullAt(index);
+ }
+ }
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedPrimitiveColumnReader.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedPrimitiveColumnReader.java
new file mode 100644
index 000000000000..72809db1b2ce
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/NestedPrimitiveColumnReader.java
@@ -0,0 +1,639 @@
+/*
+ * 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.vector.reader;
+
+import org.apache.hudi.table.format.cow.utils.IntArrayList;
+import org.apache.hudi.table.format.cow.vector.ParquetDecimalVector;
+import org.apache.hudi.table.format.cow.vector.position.LevelDelegation;
+
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.TimestampData;
+import org.apache.flink.table.data.columnar.vector.heap.HeapBooleanVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapByteVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapBytesVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapDoubleVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapFloatVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapIntVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapLongVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapShortVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapTimestampVector;
+import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.bytes.BytesUtils;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.page.DataPage;
+import org.apache.parquet.column.page.DataPageV1;
+import org.apache.parquet.column.page.DataPageV2;
+import org.apache.parquet.column.page.DictionaryPage;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridDecoder;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.parquet.column.ValuesType.DEFINITION_LEVEL;
+import static org.apache.parquet.column.ValuesType.REPETITION_LEVEL;
+import static org.apache.parquet.column.ValuesType.VALUES;
+
+/**
+ * Reader to read a single primitive leaf column that participates in a nested (Dremel) structure.
+ *
+ * <p>Vendored from Apache Flink 2.1 (FLINK-35702, {@code
+ * org.apache.flink.formats.parquet.vector.reader.NestedPrimitiveColumnReader}). Only the package
+ * and the Hudi-local {@link ParquetDecimalVector} / {@link LevelDelegation} / {@link IntArrayList}
+ * imports are changed; the algorithm is untouched. The companion Hudi-specific {@code
+ * Int64TimestampColumnReader} / {@code FixedLenBytesColumnReader} behaviours stay at the leaf-
+ * reader creation boundary in {@code ParquetSplitReaderUtil}, not inside this class — keeping it
+ * a faithful copy of upstream.
+ */
+public class NestedPrimitiveColumnReader implements ColumnReader<WritableColumnVector> {
+ private static final Logger LOG = LoggerFactory.getLogger(NestedPrimitiveColumnReader.class);
+
+ private final IntArrayList repetitionLevelList = new IntArrayList(0);
+ private final IntArrayList definitionLevelList = new IntArrayList(0);
+
+ private final PageReader pageReader;
+ private final ColumnDescriptor descriptor;
+ private final Type type;
+ private final LogicalType logicalType;
+
+ /** The dictionary, if this column has dictionary encoding. */
+ private final ParquetDataColumnReader dictionary;
+
+ /** Maximum definition level for this column. */
+ private final int maxDefLevel;
+
+ private boolean isUtcTimestamp;
+
+ /** Total number of values read. */
+ private long valuesRead;
+
+ /**
+ * value that indicates the end of the current page. That is, if valuesRead ==
+ * endOfPageValueCount, we are at the end of the page.
+ */
+ private long endOfPageValueCount;
+
+ /** If true, the current page is dictionary encoded. */
+ private boolean isCurrentPageDictionaryEncoded;
+
+ private int definitionLevel;
+ private int repetitionLevel;
+
+ /** Repetition/Definition/Value readers. */
+ private IntIterator repetitionLevelColumn;
+
+ private IntIterator definitionLevelColumn;
+ private ParquetDataColumnReader dataColumn;
+
+ /** Total values in the current page. */
+ private int pageValueCount;
+
+ // flag to indicate if there is no data in parquet data page
+ private boolean eof = false;
+
+ private boolean isFirstRow = true;
+
+ private Object lastValue;
+
+ public NestedPrimitiveColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader,
+ boolean isUtcTimestamp,
+ Type parquetType,
+ LogicalType logicalType)
+ throws IOException {
+ this.descriptor = descriptor;
+ this.type = parquetType;
+ this.pageReader = pageReader;
+ this.maxDefLevel = descriptor.getMaxDefinitionLevel();
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.logicalType = logicalType;
+
+ DictionaryPage dictionaryPage = pageReader.readDictionaryPage();
+ if (dictionaryPage != null) {
+ try {
+ this.dictionary =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByTypeOnDictionary(
+ parquetType.asPrimitiveType(),
+ dictionaryPage.getEncoding().initDictionary(descriptor, dictionaryPage),
+ isUtcTimestamp);
+ this.isCurrentPageDictionaryEncoded = true;
+ } catch (IOException e) {
+ throw new IOException(
+ String.format("Could not decode the dictionary for %s", descriptor), e);
+ }
+ } else {
+ this.dictionary = null;
+ this.isCurrentPageDictionaryEncoded = false;
+ }
+ }
+
+ // Not invoked directly; callers use readAndNewVector instead.
+ @Override
+ public void readToVector(int readNumber, WritableColumnVector vector) throws IOException {
+ throw new UnsupportedOperationException("This function should not be called.");
+ }
+
+ public WritableColumnVector readAndNewVector(int readNumber, WritableColumnVector vector)
+ throws IOException {
+ if (isFirstRow) {
+ if (!readValue()) {
+ return vector;
+ }
+ isFirstRow = false;
+ }
+
+ // index to set value.
+ int index = 0;
+ int valueIndex = 0;
+ List<Object> valueList = new ArrayList<>();
+
+ // repeated type need two loops to read data.
+ while (!eof && index < readNumber) {
+ do {
+ valueList.add(lastValue);
+ valueIndex++;
+ } while (readValue() && (repetitionLevel != 0));
+ index++;
+ }
+
+ return fillColumnVector(valueIndex, valueList);
+ }
+
+ public LevelDelegation getLevelDelegation() {
+ int[] repetition = repetitionLevelList.toArray();
+ int[] definition = definitionLevelList.toArray();
+ repetitionLevelList.clear();
+ definitionLevelList.clear();
+ repetitionLevelList.add(repetitionLevel);
+ definitionLevelList.add(definitionLevel);
+ return new LevelDelegation(repetition, definition);
+ }
+
+ private boolean readValue() throws IOException {
+ int left = readPageIfNeed();
+ if (left > 0) {
+ // get the values of repetition and definitionLevel
+ readAndSaveRepetitionAndDefinitionLevels();
+ // read the data if it isn't null
+ if (definitionLevel == maxDefLevel) {
+ if (isCurrentPageDictionaryEncoded) {
+ int dictionaryId = dataColumn.readValueDictionaryId();
+ lastValue = dictionaryDecodeValue(logicalType, dictionaryId);
+ } else {
+ lastValue = readPrimitiveTypedRow(logicalType);
+ }
+ } else {
+ lastValue = null;
+ }
+ return true;
+ } else {
+ eof = true;
+ return false;
+ }
+ }
+
+ private void readAndSaveRepetitionAndDefinitionLevels() {
+ // get the values of repetition and definitionLevel
+ repetitionLevel = repetitionLevelColumn.nextInt();
+ definitionLevel = definitionLevelColumn.nextInt();
+ valuesRead++;
+ repetitionLevelList.add(repetitionLevel);
+ definitionLevelList.add(definitionLevel);
+ }
+
+ private int readPageIfNeed() throws IOException {
+ // Compute the number of values we want to read in this page.
+ int leftInPage = (int) (endOfPageValueCount - valuesRead);
+ if (leftInPage == 0) {
+ // no data left in current page, load data from new page
+ readPage();
+ leftInPage = (int) (endOfPageValueCount - valuesRead);
+ }
+ return leftInPage;
+ }
+
+ private Object readPrimitiveTypedRow(LogicalType category) {
+ switch (category.getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ case BINARY:
+ case VARBINARY:
+ return dataColumn.readBytes();
+ case BOOLEAN:
+ return dataColumn.readBoolean();
+ case TIME_WITHOUT_TIME_ZONE:
+ case DATE:
+ case INTEGER:
+ return dataColumn.readInteger();
+ case TINYINT:
+ return dataColumn.readTinyInt();
+ case SMALLINT:
+ return dataColumn.readSmallInt();
+ case BIGINT:
+ return dataColumn.readLong();
+ case FLOAT:
+ return dataColumn.readFloat();
+ case DOUBLE:
+ return dataColumn.readDouble();
+ case DECIMAL:
+ switch (descriptor.getPrimitiveType().getPrimitiveTypeName()) {
+ case INT32:
+ return dataColumn.readInteger();
+ case INT64:
+ return dataColumn.readLong();
+ case BINARY:
+ case FIXED_LEN_BYTE_ARRAY:
+ return dataColumn.readBytes();
+ default:
+ throw new RuntimeException(
+ "Unsupported physical type for DECIMAL: " + descriptor.getPrimitiveType());
+ }
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return dataColumn.readTimestamp();
+ default:
+ throw new RuntimeException("Unsupported type in the list: " + type);
+ }
+ }
+
+ private Object dictionaryDecodeValue(LogicalType category, Integer dictionaryValue) {
+ if (dictionaryValue == null) {
+ return null;
+ }
+
+ switch (category.getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ case BINARY:
+ case VARBINARY:
+ return dictionary.readBytes(dictionaryValue);
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ case INTEGER:
+ return dictionary.readInteger(dictionaryValue);
+ case BOOLEAN:
+ return dictionary.readBoolean(dictionaryValue) ? 1 : 0;
+ case DOUBLE:
+ return dictionary.readDouble(dictionaryValue);
+ case FLOAT:
+ return dictionary.readFloat(dictionaryValue);
+ case TINYINT:
+ return dictionary.readTinyInt(dictionaryValue);
+ case SMALLINT:
+ return dictionary.readSmallInt(dictionaryValue);
+ case BIGINT:
+ return dictionary.readLong(dictionaryValue);
+ case DECIMAL:
+ switch (descriptor.getPrimitiveType().getPrimitiveTypeName()) {
+ case INT32:
+ return dictionary.readInteger(dictionaryValue);
+ case INT64:
+ return dictionary.readLong(dictionaryValue);
+ case FIXED_LEN_BYTE_ARRAY:
+ case BINARY:
+ return dictionary.readBytes(dictionaryValue);
+ default:
+ throw new RuntimeException(
+ "Unsupported physical type for DECIMAL: " + descriptor.getPrimitiveType());
+ }
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return dictionary.readTimestamp(dictionaryValue);
+ default:
+ throw new RuntimeException("Unsupported type in the list: " + type);
+ }
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private WritableColumnVector fillColumnVector(int total, List valueList) {
+ switch (logicalType.getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ case BINARY:
+ case VARBINARY:
+ HeapBytesVector heapBytesVector = new HeapBytesVector(total);
+ for (int i = 0; i < valueList.size(); i++) {
+ byte[] src = ((List<byte[]>) valueList).get(i);
+ if (src == null) {
+ heapBytesVector.setNullAt(i);
+ } else {
+ heapBytesVector.appendBytes(i, src, 0, src.length);
+ }
+ }
+ return heapBytesVector;
+ case BOOLEAN:
+ HeapBooleanVector heapBooleanVector = new HeapBooleanVector(total);
+ for (int i = 0; i < valueList.size(); i++) {
+ if (valueList.get(i) == null) {
+ heapBooleanVector.setNullAt(i);
+ } else {
+ heapBooleanVector.vector[i] = ((List<Boolean>) valueList).get(i);
+ }
+ }
+ return heapBooleanVector;
+ case TINYINT:
+ HeapByteVector heapByteVector = new HeapByteVector(total);
+ for (int i = 0; i < valueList.size(); i++) {
+ if (valueList.get(i) == null) {
+ heapByteVector.setNullAt(i);
+ } else {
+ heapByteVector.vector[i] = (byte) ((List<Integer>) valueList).get(i).intValue();
+ }
+ }
+ return heapByteVector;
+ case SMALLINT:
+ HeapShortVector heapShortVector = new HeapShortVector(total);
+ for (int i = 0; i < valueList.size(); i++) {
+ if (valueList.get(i) == null) {
+ heapShortVector.setNullAt(i);
+ } else {
+ heapShortVector.vector[i] = (short) ((List<Integer>) valueList).get(i).intValue();
+ }
+ }
+ return heapShortVector;
+ case INTEGER:
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ HeapIntVector heapIntVector = new HeapIntVector(total);
+ for (int i = 0; i < valueList.size(); i++) {
+ if (valueList.get(i) == null) {
+ heapIntVector.setNullAt(i);
+ } else {
+ heapIntVector.vector[i] = ((List<Integer>) valueList).get(i);
+ }
+ }
+ return heapIntVector;
+ case FLOAT:
+ HeapFloatVector heapFloatVector = new HeapFloatVector(total);
+ for (int i = 0; i < valueList.size(); i++) {
+ if (valueList.get(i) == null) {
+ heapFloatVector.setNullAt(i);
+ } else {
+ heapFloatVector.vector[i] = ((List<Float>) valueList).get(i);
+ }
+ }
+ return heapFloatVector;
+ case BIGINT:
+ HeapLongVector heapLongVector = new HeapLongVector(total);
+ for (int i = 0; i < valueList.size(); i++) {
+ if (valueList.get(i) == null) {
+ heapLongVector.setNullAt(i);
+ } else {
+ heapLongVector.vector[i] = ((List<Long>) valueList).get(i);
+ }
+ }
+ return heapLongVector;
+ case DOUBLE:
+ HeapDoubleVector heapDoubleVector = new HeapDoubleVector(total);
+ for (int i = 0; i < valueList.size(); i++) {
+ if (valueList.get(i) == null) {
+ heapDoubleVector.setNullAt(i);
+ } else {
+ heapDoubleVector.vector[i] = ((List<Double>) valueList).get(i);
+ }
+ }
+ return heapDoubleVector;
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ HeapTimestampVector heapTimestampVector = new HeapTimestampVector(total);
+ for (int i = 0; i < valueList.size(); i++) {
+ if (valueList.get(i) == null) {
+ heapTimestampVector.setNullAt(i);
+ } else {
+ heapTimestampVector.setTimestamp(i, ((List<TimestampData>) valueList).get(i));
+ }
+ }
+ return heapTimestampVector;
+ case DECIMAL:
+ PrimitiveType.PrimitiveTypeName primitiveTypeName =
+ descriptor.getPrimitiveType().getPrimitiveTypeName();
+ switch (primitiveTypeName) {
+ case INT32:
+ HeapIntVector phiv = new HeapIntVector(total);
+ for (int i = 0; i < valueList.size(); i++) {
+ if (valueList.get(i) == null) {
+ phiv.setNullAt(i);
+ } else {
+ phiv.vector[i] = ((List<Integer>) valueList).get(i);
+ }
+ }
+ return new ParquetDecimalVector(phiv);
+ case INT64:
+ HeapLongVector phlv = new HeapLongVector(total);
+ for (int i = 0; i < valueList.size(); i++) {
+ if (valueList.get(i) == null) {
+ phlv.setNullAt(i);
+ } else {
+ phlv.vector[i] = ((List<Long>) valueList).get(i);
+ }
+ }
+ return new ParquetDecimalVector(phlv);
+ default:
+ HeapBytesVector phbv = getHeapBytesVector(total, valueList);
+ return new ParquetDecimalVector(phbv);
+ }
+ default:
+ throw new RuntimeException("Unsupported type in the list: " + type);
+ }
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static HeapBytesVector getHeapBytesVector(int total, List valueList) {
+ HeapBytesVector phbv = new HeapBytesVector(total);
+ for (int i = 0; i < valueList.size(); i++) {
+ byte[] src = ((List<byte[]>) valueList).get(i);
+ if (valueList.get(i) == null) {
+ phbv.setNullAt(i);
+ } else {
+ phbv.appendBytes(i, src, 0, src.length);
+ }
+ }
+ return phbv;
+ }
+
+ protected void readPage() {
+ DataPage page = pageReader.readPage();
+
+ if (page == null) {
+ return;
+ }
+
+ page.accept(
+ new DataPage.Visitor<Void>() {
+ @Override
+ public Void visit(DataPageV1 dataPageV1) {
+ readPageV1(dataPageV1);
+ return null;
+ }
+
+ @Override
+ public Void visit(DataPageV2 dataPageV2) {
+ readPageV2(dataPageV2);
+ return null;
+ }
+ });
+ }
+
+ private void initDataReader(Encoding dataEncoding, ByteBufferInputStream in, int valueCount)
+ throws IOException {
+ this.pageValueCount = valueCount;
+ this.endOfPageValueCount = valuesRead + pageValueCount;
+ if (dataEncoding.usesDictionary()) {
+ this.dataColumn = null;
+ if (dictionary == null) {
+ throw new IOException(
+ String.format(
+ "Could not read page in col %s because the dictionary was missing for encoding %s.",
+ descriptor, dataEncoding));
+ }
+ dataColumn =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(
+ type.asPrimitiveType(),
+ dataEncoding.getDictionaryBasedValuesReader(
+ descriptor, VALUES, dictionary.getDictionary()),
+ isUtcTimestamp);
+ this.isCurrentPageDictionaryEncoded = true;
+ } else {
+ dataColumn =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(
+ type.asPrimitiveType(),
+ dataEncoding.getValuesReader(descriptor, VALUES),
+ isUtcTimestamp);
+ this.isCurrentPageDictionaryEncoded = false;
+ }
+
+ try {
+ dataColumn.initFromPage(pageValueCount, in);
+ } catch (IOException e) {
+ throw new IOException(String.format("Could not read page in col %s.", descriptor), e);
+ }
+ }
+
+ private void readPageV1(DataPageV1 page) {
+ ValuesReader rlReader = page.getRlEncoding().getValuesReader(descriptor, REPETITION_LEVEL);
+ ValuesReader dlReader = page.getDlEncoding().getValuesReader(descriptor, DEFINITION_LEVEL);
+ this.repetitionLevelColumn = new ValuesReaderIntIterator(rlReader);
+ this.definitionLevelColumn = new ValuesReaderIntIterator(dlReader);
+ try {
+ BytesInput bytes = page.getBytes();
+ LOG.debug("Page size {} bytes and {} records.", bytes.size(), pageValueCount);
+ ByteBufferInputStream in = bytes.toInputStream();
+ LOG.debug("Reading repetition levels at {}.", in.position());
+ rlReader.initFromPage(pageValueCount, in);
+ LOG.debug("Reading definition levels at {}.", in.position());
+ dlReader.initFromPage(pageValueCount, in);
+ LOG.debug("Reading data at {}.", in.position());
+ initDataReader(page.getValueEncoding(), in, page.getValueCount());
+ } catch (IOException e) {
+ throw new ParquetDecodingException(
+ String.format("Could not read page %s in col %s.", page, descriptor), e);
+ }
+ }
+
+ private void readPageV2(DataPageV2 page) {
+ this.pageValueCount = page.getValueCount();
+ this.repetitionLevelColumn =
+ newRLEIterator(descriptor.getMaxRepetitionLevel(), page.getRepetitionLevels());
+ this.definitionLevelColumn =
+ newRLEIterator(descriptor.getMaxDefinitionLevel(), page.getDefinitionLevels());
+ try {
+ LOG.debug(
+ "Page data size {} bytes and {} records.", page.getData().size(), pageValueCount);
+ initDataReader(
+ page.getDataEncoding(), page.getData().toInputStream(), page.getValueCount());
+ } catch (IOException e) {
+ throw new ParquetDecodingException(
+ String.format("Could not read page %s in col %s.", page, descriptor), e);
+ }
+ }
+
+ private IntIterator newRLEIterator(int maxLevel, BytesInput bytes) {
+ try {
+ if (maxLevel == 0) {
+ return new NullIntIterator();
+ }
+ return new RLEIntIterator(
+ new RunLengthBitPackingHybridDecoder(
+ BytesUtils.getWidthFromMaxInt(maxLevel),
+ new ByteArrayInputStream(bytes.toByteArray())));
+ } catch (IOException e) {
+ throw new ParquetDecodingException(
+ String.format("Could not read levels in page for col %s.", descriptor), e);
+ }
+ }
+
+ /** Utility interface to abstract over different way to read ints with different encodings. */
+ interface IntIterator {
+ int nextInt();
+ }
+
+ /** Reading int from {@link ValuesReader}. */
+ protected static final class ValuesReaderIntIterator implements IntIterator {
+ ValuesReader delegate;
+
+ public ValuesReaderIntIterator(ValuesReader delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public int nextInt() {
+ return delegate.readInteger();
+ }
+ }
+
+ /** Reading int from {@link RunLengthBitPackingHybridDecoder}. */
+ protected static final class RLEIntIterator implements IntIterator {
+ RunLengthBitPackingHybridDecoder delegate;
+
+ public RLEIntIterator(RunLengthBitPackingHybridDecoder delegate) {
+ this.delegate = delegate;
+ }
+
+ @Override
+ public int nextInt() {
+ try {
+ return delegate.readInt();
+ } catch (IOException e) {
+ throw new ParquetDecodingException(e);
+ }
+ }
+ }
+
+ /** Reading zero always. */
+ protected static final class NullIntIterator implements IntIterator {
+ @Override
+ public int nextInt() {
+ return 0;
+ }
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
index 3572b117a631..1826419db5d4 100644
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetColumnarRowSplitReader.java
@@ -18,7 +18,9 @@
package org.apache.hudi.table.format.cow.vector.reader;
+import org.apache.hudi.table.format.cow.ParquetSplitReaderUtil;
import org.apache.hudi.table.format.cow.vector.ParquetDecimalVector;
+import org.apache.hudi.table.format.cow.vector.type.ParquetField;
import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
import org.apache.flink.table.data.RowData;
@@ -28,6 +30,7 @@ import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.util.FlinkRuntimeException;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
@@ -39,6 +42,8 @@ import org.apache.parquet.filter2.predicate.FilterPredicate;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.BlockMetaData;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.io.ColumnIOFactory;
+import org.apache.parquet.io.MessageColumnIO;
import org.apache.parquet.schema.GroupType;
import org.apache.parquet.schema.MessageType;
import org.apache.parquet.schema.Type;
@@ -46,6 +51,7 @@ import org.apache.parquet.schema.Types;
import java.io.Closeable;
import java.io.IOException;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
@@ -53,7 +59,6 @@ import java.util.Locale;
import java.util.Map;
import java.util.stream.IntStream;
-import static org.apache.hudi.table.format.cow.ParquetSplitReaderUtil.createColumnReader;
import static org.apache.hudi.table.format.cow.ParquetSplitReaderUtil.createWritableColumnVector;
import static org.apache.parquet.filter2.compat.FilterCompat.get;
import static org.apache.parquet.filter2.compat.RowGroupFilter.filterRowGroups;
@@ -77,6 +82,14 @@ public class ParquetColumnarRowSplitReader implements Closeable {
private final MessageType requestedSchema;
+ /**
+ * {@link ParquetField} tree per top-level requested column, used by
+ * {@link ParquetSplitReaderUtil#createColumnReader(boolean, LogicalType, Type, List,
+ * PageReadStore, ParquetField)} to drive the Dremel-style {@link NestedColumnReader} for
+ * nested types. Entries are {@code null} for primitive top-level fields. Built once per split.
+ */
+ private final List<ParquetField> requestedFields;
+
/**
* The total number of rows this RecordReader will eventually read. The sum of the rows of all
* the row groups.
@@ -158,6 +171,20 @@ public class ParquetColumnarRowSplitReader implements Closeable {
checkSchema();
+ // Build the ParquetField tree once per split (the Dremel-style nested reader reuses it across
+ // row groups). Only columns with nested logical type get a non-null entry — primitive columns
+ // still use Hudi's specialized ColumnReaders.
+ MessageColumnIO messageColumnIO = new ColumnIOFactory().getColumnIO(requestedSchema);
+ List<RowType.RowField> requestedRowFields = new ArrayList<>(requestedTypes.length);
+ List<String> requestedFieldNames = new ArrayList<>(requestedTypes.length);
+ for (int i = 0; i < requestedTypes.length; i++) {
+ String name = requestedSchema.getFieldName(i);
+ requestedRowFields.add(new RowType.RowField(name, requestedTypes[i]));
+ requestedFieldNames.add(name);
+ }
+ this.requestedFields = ParquetSplitReaderUtil.buildFieldsList(
+ requestedRowFields, requestedFieldNames, messageColumnIO);
+
this.writableVectors = createWritableVectors();
ColumnVector[] columnVectors = patchedVector(selectedFieldNames.length, createReadableVectors(), requestedIndices);
this.columnarBatch = generator.generate(columnVectors);
@@ -340,12 +367,13 @@ public class ParquetColumnarRowSplitReader implements Closeable {
List<ColumnDescriptor> columns = requestedSchema.getColumns();
columnReaders = new ColumnReader[types.size()];
for (int i = 0; i < types.size(); ++i) {
- columnReaders[i] = createColumnReader(
+ columnReaders[i] = ParquetSplitReaderUtil.createColumnReader(
utcTimestamp,
requestedTypes[i],
types.get(i),
columns,
- pages);
+ pages,
+ requestedFields.get(i));
}
totalCountLoadedSoFar += pages.getRowCount();
}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetDataColumnReaderFactory.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetDataColumnReaderFactory.java
index fdfe5d6fa3a3..1abc6ed56c0d 100644
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetDataColumnReaderFactory.java
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/ParquetDataColumnReaderFactory.java
@@ -26,12 +26,16 @@ import org.apache.parquet.bytes.ByteBufferInputStream;
import org.apache.parquet.column.Dictionary;
import org.apache.parquet.column.values.ValuesReader;
import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.OriginalType;
import org.apache.parquet.schema.PrimitiveType;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
import static org.apache.flink.formats.parquet.vector.reader.TimestampColumnReader.JULIAN_EPOCH_OFFSET_DAYS;
import static org.apache.flink.formats.parquet.vector.reader.TimestampColumnReader.MILLIS_IN_DAY;
@@ -252,21 +256,115 @@ public final class ParquetDataColumnReaderFactory {
}
}
+ /**
+ * Reader for Parquet INT64 timestamp values (MILLIS / MICROS / NANOS), i.e. the standard
+ * timestamp encoding defined by Parquet's
+ * {@link LogicalTypeAnnotation.TimestampLogicalTypeAnnotation} and the legacy
+ * {@link OriginalType#TIMESTAMP_MILLIS} / {@link OriginalType#TIMESTAMP_MICROS} annotations.
+ * (The older INT96 encoding is marked deprecated by the Parquet format spec — see
+ * <a href="https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#timestamp">
+ * LogicalTypes.md</a> — but is still supported here via {@link TypesFromInt96PageReader} for
+ * backwards compatibility with files written by older Hive / Spark / Impala versions.)
+ *
+ * <p>Used by {@link NestedPrimitiveColumnReader} when a TIMESTAMP column sits inside a
+ * {@code Row}, {@code Array} or {@code Map}; the top-level path continues to use
+ * {@link Int64TimestampColumnReader} for batched-vector efficiency.
+ */
+ public static class TypesFromInt64PageReader extends DefaultParquetDataColumnReader {
+ private final boolean isUtcTimestamp;
+ private final ChronoUnit chronoUnit;
+
+ public TypesFromInt64PageReader(
+ ValuesReader realReader, boolean isUtcTimestamp, ChronoUnit chronoUnit) {
+ super(realReader);
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.chronoUnit = chronoUnit;
+ }
+
+ public TypesFromInt64PageReader(
+ Dictionary dict, boolean isUtcTimestamp, ChronoUnit chronoUnit) {
+ super(dict);
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.chronoUnit = chronoUnit;
+ }
+
+ @Override
+ public TimestampData readTimestamp() {
+ return int64ToTimestamp(isUtcTimestamp, valuesReader.readLong(), chronoUnit);
+ }
+
+ @Override
+ public TimestampData readTimestamp(int id) {
+ return int64ToTimestamp(isUtcTimestamp, dict.decodeToLong(id), chronoUnit);
+ }
+ }
+
private static ParquetDataColumnReader getDataColumnReaderByTypeHelper(
boolean isDictionary,
PrimitiveType parquetType,
Dictionary dictionary,
ValuesReader valuesReader,
boolean isUtcTimestamp) {
- if (parquetType.getPrimitiveTypeName() == PrimitiveType.PrimitiveTypeName.INT96) {
+ PrimitiveType.PrimitiveTypeName typeName = parquetType.getPrimitiveTypeName();
+ if (typeName == PrimitiveType.PrimitiveTypeName.INT96) {
return isDictionary
? new TypesFromInt96PageReader(dictionary, isUtcTimestamp)
: new TypesFromInt96PageReader(valuesReader, isUtcTimestamp);
- } else {
- return isDictionary
- ? new DefaultParquetDataColumnReader(dictionary)
- : new DefaultParquetDataColumnReader(valuesReader);
}
+ if (typeName == PrimitiveType.PrimitiveTypeName.INT64) {
+ ChronoUnit unit = resolveInt64TimestampUnit(parquetType);
+ if (unit != null) {
+ return isDictionary
+ ? new TypesFromInt64PageReader(dictionary, isUtcTimestamp, unit)
+ : new TypesFromInt64PageReader(valuesReader, isUtcTimestamp, unit);
+ }
+ }
+ return isDictionary
+ ? new DefaultParquetDataColumnReader(dictionary)
+ : new DefaultParquetDataColumnReader(valuesReader);
+ }
+
+ /**
+ * Returns the {@link ChronoUnit} for a Parquet INT64 TIMESTAMP column, or {@code null} if the
+ * column is a plain INT64 (not a timestamp).
+ *
+ * <p>Supports both the modern {@link LogicalTypeAnnotation.TimestampLogicalTypeAnnotation} and
+ * the legacy {@link OriginalType#TIMESTAMP_MILLIS} / {@link OriginalType#TIMESTAMP_MICROS}
+ * encodings.
+ */
+ private static ChronoUnit resolveInt64TimestampUnit(PrimitiveType parquetType) {
+ LogicalTypeAnnotation annotation = parquetType.getLogicalTypeAnnotation();
+ if (annotation instanceof LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) {
+ LogicalTypeAnnotation.TimeUnit unit =
+ ((LogicalTypeAnnotation.TimestampLogicalTypeAnnotation) annotation).getUnit();
+ switch (unit) {
+ case MILLIS:
+ return ChronoUnit.MILLIS;
+ case MICROS:
+ return ChronoUnit.MICROS;
+ case NANOS:
+ return ChronoUnit.NANOS;
+ default:
+ return null;
+ }
+ }
+ OriginalType originalType = parquetType.getOriginalType();
+ if (originalType == OriginalType.TIMESTAMP_MILLIS) {
+ return ChronoUnit.MILLIS;
+ }
+ if (originalType == OriginalType.TIMESTAMP_MICROS) {
+ return ChronoUnit.MICROS;
+ }
+ return null;
+ }
+
+ private static TimestampData int64ToTimestamp(
+ boolean isUtcTimestamp, long value, ChronoUnit unit) {
+ Instant instant = Instant.EPOCH.plus(value, unit);
+ if (isUtcTimestamp) {
+ return TimestampData.fromInstant(instant);
+ }
+ return TimestampData.fromTimestamp(Timestamp.from(instant));
}
public static ParquetDataColumnReader getDataColumnReaderByTypeOnDictionary(
@@ -281,10 +379,10 @@ public final class ParquetDataColumnReaderFactory {
}
private static TimestampData int96ToTimestamp(
- boolean utcTimestamp, long nanosOfDay, int julianDay) {
+ boolean isUtcTimestamp, long nanosOfDay, int julianDay) {
long millisecond = julianDayToMillis(julianDay) + (nanosOfDay / NANOS_PER_MILLISECOND);
- if (utcTimestamp) {
+ if (isUtcTimestamp) {
int nanoOfMillisecond = (int) (nanosOfDay % NANOS_PER_MILLISECOND);
return TimestampData.fromEpochMillis(millisecond, nanoOfMillisecond);
} else {
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/RowColumnReader.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/RowColumnReader.java
deleted file mode 100644
index 79b50487f13c..000000000000
--- a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/reader/RowColumnReader.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * 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.vector.reader;
-
-import org.apache.hudi.table.format.cow.vector.HeapRowColumnVector;
-
-import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
-import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
-
-import java.io.IOException;
-import java.util.List;
-
-/**
- * Row {@link ColumnReader}.
- */
-public class RowColumnReader implements ColumnReader<WritableColumnVector> {
-
- private final List<ColumnReader> fieldReaders;
-
- public RowColumnReader(List<ColumnReader> fieldReaders) {
- this.fieldReaders = fieldReaders;
- }
-
- @Override
- public void readToVector(int readNumber, WritableColumnVector vector) throws IOException {
- HeapRowColumnVector rowColumnVector = (HeapRowColumnVector) vector;
- WritableColumnVector[] vectors = rowColumnVector.vectors;
- // row vector null array
- boolean[] isNulls = new boolean[readNumber];
- for (int i = 0; i < vectors.length; i++) {
- fieldReaders.get(i).readToVector(readNumber, vectors[i]);
-
- for (int j = 0; j < readNumber; j++) {
- if (i == 0) {
- isNulls[j] = vectors[i].isNullAt(j);
- } else {
- isNulls[j] = isNulls[j] && vectors[i].isNullAt(j);
- }
- if (i == vectors.length - 1 && isNulls[j]) {
- // rowColumnVector[j] is null only when all fields[j] of rowColumnVector[j] is
- // null
- rowColumnVector.setNullAt(j);
- }
- }
- }
- }
-}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/type/ParquetField.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/type/ParquetField.java
new file mode 100644
index 000000000000..0f5e00779a2f
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/type/ParquetField.java
@@ -0,0 +1,72 @@
+/*
+ * 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.vector.type;
+
+import org.apache.flink.table.types.logical.LogicalType;
+
+/**
+ * Field that represent parquet's field type.
+ *
+ * <p>Note: Vendored from Apache Flink (FLINK-35702, {@code
+ * org.apache.flink.formats.parquet.vector.type.ParquetField}).
+ */
+public abstract class ParquetField {
+ private final LogicalType type;
+ private final int repetitionLevel;
+ private final int definitionLevel;
+ private final boolean required;
+
+ public ParquetField(
+ LogicalType type, int repetitionLevel, int definitionLevel, boolean required) {
+ this.type = type;
+ this.repetitionLevel = repetitionLevel;
+ this.definitionLevel = definitionLevel;
+ this.required = required;
+ }
+
+ public LogicalType getType() {
+ return type;
+ }
+
+ public int getRepetitionLevel() {
+ return repetitionLevel;
+ }
+
+ public int getDefinitionLevel() {
+ return definitionLevel;
+ }
+
+ public boolean isRequired() {
+ return required;
+ }
+
+ @Override
+ public String toString() {
+ return "Field{"
+ + "type="
+ + type
+ + ", repetitionLevel="
+ + repetitionLevel
+ + ", definitionLevel="
+ + definitionLevel
+ + ", required="
+ + required
+ + '}';
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/type/ParquetGroupField.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/type/ParquetGroupField.java
new file mode 100644
index 000000000000..f91dcca965d6
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/type/ParquetGroupField.java
@@ -0,0 +1,59 @@
+/*
+ * 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.vector.type;
+
+import org.apache.flink.table.types.logical.LogicalType;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Field that represent parquet's Group Field.
+ *
+ * <p>Note: Vendored from Apache Flink (FLINK-35702, {@code
+ * org.apache.flink.formats.parquet.vector.type.ParquetGroupField}) with a Hudi-specific extension:
+ * entries in the {@code children} list may be {@code null} to denote a Row child that is absent
+ * from the parquet file but present in the requested logical schema (schema evolution). This
+ * replaces Hudi's previous {@code EmptyColumnReader} branch for Row subtrees.
+ */
+public class ParquetGroupField extends ParquetField {
+
+ private final List<ParquetField> children;
+
+ public ParquetGroupField(
+ LogicalType type,
+ int repetitionLevel,
+ int definitionLevel,
+ boolean required,
+ List<ParquetField> children) {
+ super(type, repetitionLevel, definitionLevel, required);
+ // Use a plain unmodifiable list (not ImmutableList) so that null entries are allowed for
+ // schema-evolution missing children in ROW types.
+ this.children =
+ Collections.unmodifiableList(new ArrayList<>(requireNonNull(children, "children is null")));
+ }
+
+ /** Children of this group. Entries may be {@code null} for absent-in-file Row fields. */
+ public List<ParquetField> getChildren() {
+ return children;
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/type/ParquetPrimitiveField.java b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/type/ParquetPrimitiveField.java
new file mode 100644
index 000000000000..f6af6f9ff479
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/main/java/org/apache/hudi/table/format/cow/vector/type/ParquetPrimitiveField.java
@@ -0,0 +1,55 @@
+/*
+ * 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.vector.type;
+
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.parquet.column.ColumnDescriptor;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * Field that represent parquet's primitive field.
+ *
+ * <p>Note: Vendored from Apache Flink (FLINK-35702, {@code
+ * org.apache.flink.formats.parquet.vector.type.ParquetPrimitiveField}).
+ */
+public class ParquetPrimitiveField extends ParquetField {
+
+ private final ColumnDescriptor descriptor;
+ private final int id;
+
+ public ParquetPrimitiveField(
+ LogicalType type, boolean required, ColumnDescriptor descriptor, int id) {
+ super(
+ type,
+ descriptor.getMaxRepetitionLevel(),
+ descriptor.getMaxDefinitionLevel(),
+ required);
+ this.descriptor = requireNonNull(descriptor, "descriptor is required");
+ this.id = id;
+ }
+
+ public ColumnDescriptor getDescriptor() {
+ return descriptor;
+ }
+
+ public int getId() {
+ return id;
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/vector/TestHeapColumnVectorAccessors.java b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/vector/TestHeapColumnVectorAccessors.java
new file mode 100644
index 000000000000..7cb62824e854
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/vector/TestHeapColumnVectorAccessors.java
@@ -0,0 +1,139 @@
+/*
+ * 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.vector;
+
+import org.apache.flink.table.data.columnar.vector.heap.HeapIntVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapLongVector;
+import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+/**
+ * Tests for the Flink 2.1-compatible accessors added on {@link HeapArrayVector},
+ * {@link HeapMapColumnVector} and {@link HeapRowColumnVector} when vendoring Flink 2.1's
+ * nested-Parquet reader (FLINK-35702).
+ *
+ * <p>The accessors are wrappers over the existing public fields so legacy callers continue to
+ * work. These tests exist solely to pin down that wrapper contract — runtime correctness of the
+ * Dremel-style read path is exercised end-to-end by integration tests in
+ * {@code ITTestHoodieDataSource} (testParquetComplexTypes / testParquetComplexNestedRowTypes /
+ * testParquetArrayMapOfRowTypes / testParquetNullChildColumnsRowTypes).
+ */
+class TestHeapColumnVectorAccessors {
+
+ // -----------------------------------------------------------------------------------------------
+ // HeapArrayVector
+ // -----------------------------------------------------------------------------------------------
+
+ @Test
+ void heapArrayVectorAccessorsReflectPublicFields() {
+ HeapIntVector child = new HeapIntVector(4);
+ HeapArrayVector vector = new HeapArrayVector(2, child);
+
+ long[] offsets = {0L, 2L};
+ long[] lengths = {2L, 2L};
+ HeapLongVector replacementChild = new HeapLongVector(4);
+
+ vector.setOffsets(offsets);
+ vector.setLengths(lengths);
+ vector.setChild(replacementChild);
+ vector.setSize(2);
+
+ assertArrayEquals(offsets, vector.getOffsets());
+ assertArrayEquals(lengths, vector.getLengths());
+ assertSame(replacementChild, vector.getChild());
+ assertEquals(2, vector.getSize());
+
+ // Backing public fields are kept in sync — preserves backward compatibility.
+ assertSame(offsets, vector.offsets);
+ assertSame(lengths, vector.lengths);
+ assertSame(replacementChild, vector.child);
+ }
+
+ // -----------------------------------------------------------------------------------------------
+ // HeapMapColumnVector
+ // -----------------------------------------------------------------------------------------------
+
+ @Test
+ void heapMapColumnVectorConstructorInitializesOffsetsAndLengths() {
+ HeapIntVector keys = new HeapIntVector(4);
+ HeapIntVector values = new HeapIntVector(4);
+
+ HeapMapColumnVector vector = new HeapMapColumnVector(3, keys, values);
+
+ assertEquals(3, vector.getOffsets().length);
+ assertEquals(3, vector.getLengths().length);
+ }
+
+ @Test
+ void heapMapColumnVectorAccessorsReflectInternalState() {
+ HeapIntVector keys = new HeapIntVector(4);
+ HeapIntVector values = new HeapIntVector(4);
+ HeapMapColumnVector vector = new HeapMapColumnVector(2, keys, values);
+
+ long[] offsets = {0L, 2L};
+ long[] lengths = {2L, 2L};
+ HeapLongVector newKeys = new HeapLongVector(4);
+ HeapLongVector newValues = new HeapLongVector(4);
+
+ vector.setOffsets(offsets);
+ vector.setLengths(lengths);
+ vector.setKeys(newKeys);
+ vector.setValues(newValues);
+ vector.setSize(2);
+
+ assertArrayEquals(offsets, vector.getOffsets());
+ assertArrayEquals(lengths, vector.getLengths());
+ assertSame(newKeys, vector.getKeys());
+ assertSame(newValues, vector.getValues());
+ // The Flink-2.1-style ColumnVector accessors return the same underlying child.
+ assertSame(newKeys, vector.getKeyColumnVector());
+ assertSame(newValues, vector.getValueColumnVector());
+ assertEquals(2, vector.getSize());
+ }
+
+ // -----------------------------------------------------------------------------------------------
+ // HeapRowColumnVector
+ // -----------------------------------------------------------------------------------------------
+
+ @Test
+ void heapRowColumnVectorFieldsAccessorsReflectPublicVectors() {
+ HeapIntVector intField = new HeapIntVector(2);
+ HeapLongVector longField = new HeapLongVector(2);
+ HeapRowColumnVector vector = new HeapRowColumnVector(2, intField, longField);
+
+ WritableColumnVector[] originalFields = vector.getFields();
+ assertEquals(2, originalFields.length);
+ assertSame(intField, originalFields[0]);
+ assertSame(longField, originalFields[1]);
+ // Backing public field is kept in sync — preserves backward compatibility.
+ assertSame(originalFields, vector.vectors);
+
+ HeapIntVector replacement = new HeapIntVector(2);
+ WritableColumnVector[] replacementFields = {replacement, longField};
+ vector.setFields(replacementFields);
+
+ assertSame(replacementFields, vector.getFields());
+ assertSame(replacementFields, vector.vectors);
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/vector/TestParquetDecimalVector.java b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/vector/TestParquetDecimalVector.java
new file mode 100644
index 000000000000..02fe1e61ccb0
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/vector/TestParquetDecimalVector.java
@@ -0,0 +1,188 @@
+/*
+ * 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.vector;
+
+import org.apache.flink.table.data.DecimalData;
+import org.apache.flink.table.data.columnar.vector.BytesColumnVector;
+import org.apache.flink.table.data.columnar.vector.ColumnVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapBytesVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapIntVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapLongVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapShortVector;
+
+import org.junit.jupiter.api.Test;
+
+import java.math.BigDecimal;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link ParquetDecimalVector}.
+ */
+public class TestParquetDecimalVector {
+
+ @Test
+ void testGetDecimalFromInt32Vector() {
+ // precision <= 9 => ParquetSchemaConverter.is32BitDecimal(precision) == true
+ HeapIntVector intVector = new HeapIntVector(1);
+ intVector.vector[0] = 12345;
+ ParquetDecimalVector wrapped = new ParquetDecimalVector(intVector);
+
+ DecimalData decoded = wrapped.getDecimal(0, 5, 2);
+
+ assertEquals(new BigDecimal("123.45"), decoded.toBigDecimal());
+ }
+
+ @Test
+ void testGetDecimalFromInt64Vector() {
+ // 9 < precision <= 18 => ParquetSchemaConverter.is64BitDecimal(precision) == true
+ HeapLongVector longVector = new HeapLongVector(1);
+ longVector.vector[0] = 1234567890123456L;
+ ParquetDecimalVector wrapped = new ParquetDecimalVector(longVector);
+
+ DecimalData decoded = wrapped.getDecimal(0, 18, 4);
+
+ assertEquals(new BigDecimal("123456789012.3456"), decoded.toBigDecimal());
+ }
+
+ @Test
+ void testGetDecimalFromBytesVectorAtLargePrecision() {
+ // precision > 18 => BINARY / FIXED_LEN_BYTE_ARRAY path
+ BigDecimal original = new BigDecimal("12345678901234567890.1234567890");
+ byte[] unscaled = original.unscaledValue().toByteArray();
+ HeapBytesVector bytesVector = new HeapBytesVector(1);
+ bytesVector.appendBytes(0, unscaled, 0, unscaled.length);
+ ParquetDecimalVector wrapped = new ParquetDecimalVector(bytesVector);
+
+ DecimalData decoded = wrapped.getDecimal(0, 30, 10);
+
+ assertEquals(original, decoded.toBigDecimal());
+ }
+
+ @Test
+ void testGetDecimalFromBytesVectorAtSmallPrecision() {
+ // A Parquet file can legally encode a small-precision decimal as BINARY. In that case the
+ // dispatch must fall through to the bytes branch rather than require an IntColumnVector.
+ BigDecimal original = new BigDecimal("123.45");
+ byte[] unscaled = original.unscaledValue().toByteArray();
+ HeapBytesVector bytesVector = new HeapBytesVector(1);
+ bytesVector.appendBytes(0, unscaled, 0, unscaled.length);
+ ParquetDecimalVector wrapped = new ParquetDecimalVector(bytesVector);
+
+ DecimalData decoded = wrapped.getDecimal(0, 5, 2);
+
+ assertEquals(original, decoded.toBigDecimal());
+ }
+
+ @Test
+ void testGetDecimalThrowsOnUnsupportedVectorType() {
+ // A large-precision request must have a bytes-backed child; any other writable child is an
+ // illegal combination and must be surfaced via Preconditions.checkArgument.
+ ColumnVector unsupported = new HeapShortVector(1);
+ ParquetDecimalVector wrapped = new ParquetDecimalVector(unsupported);
+
+ assertThrows(IllegalArgumentException.class, () -> wrapped.getDecimal(0, 30, 10));
+ }
+
+ @Test
+ void testIsNullAtDelegatesToChild() {
+ HeapIntVector intVector = new HeapIntVector(2);
+ intVector.vector[0] = 1;
+ intVector.setNullAt(1);
+ ParquetDecimalVector wrapped = new ParquetDecimalVector(intVector);
+
+ assertFalse(wrapped.isNullAt(0));
+ assertTrue(wrapped.isNullAt(1));
+ }
+
+ @Test
+ void testWritableIntRoundTrip() {
+ HeapIntVector intVector = new HeapIntVector(1);
+ ParquetDecimalVector wrapped = new ParquetDecimalVector(intVector);
+
+ wrapped.setInt(0, 42);
+
+ assertEquals(42, wrapped.getInt(0));
+ assertEquals(42, intVector.vector[0]);
+ }
+
+ @Test
+ void testWritableLongRoundTrip() {
+ HeapLongVector longVector = new HeapLongVector(1);
+ ParquetDecimalVector wrapped = new ParquetDecimalVector(longVector);
+
+ wrapped.setLong(0, 9876543210L);
+
+ assertEquals(9876543210L, wrapped.getLong(0));
+ assertEquals(9876543210L, longVector.vector[0]);
+ }
+
+ @Test
+ void testWritableBytesRoundTrip() {
+ HeapBytesVector bytesVector = new HeapBytesVector(1);
+ ParquetDecimalVector wrapped = new ParquetDecimalVector(bytesVector);
+ byte[] payload = new byte[] {0x01, 0x02, 0x03};
+
+ wrapped.appendBytes(0, payload, 0, payload.length);
+
+ BytesColumnVector.Bytes out = wrapped.getBytes(0);
+ assertEquals(payload.length, out.len);
+ assertEquals(0x01, out.data[out.offset]);
+ assertEquals(0x02, out.data[out.offset + 1]);
+ assertEquals(0x03, out.data[out.offset + 2]);
+ }
+
+ @Test
+ void testResetDelegatesToChild() {
+ HeapIntVector intVector = new HeapIntVector(1);
+ intVector.setNullAt(0);
+ ParquetDecimalVector wrapped = new ParquetDecimalVector(intVector);
+ assertTrue(wrapped.isNullAt(0));
+
+ wrapped.reset();
+
+ assertFalse(wrapped.isNullAt(0));
+ }
+
+ @Test
+ void testFillWithNullsDelegatesToChild() {
+ HeapIntVector intVector = new HeapIntVector(2);
+ ParquetDecimalVector wrapped = new ParquetDecimalVector(intVector);
+
+ wrapped.fillWithNulls();
+
+ assertTrue(wrapped.isNullAt(0));
+ assertTrue(wrapped.isNullAt(1));
+ }
+
+ @Test
+ void testSetNullAtDelegatesToChild() {
+ HeapIntVector intVector = new HeapIntVector(2);
+ ParquetDecimalVector wrapped = new ParquetDecimalVector(intVector);
+
+ wrapped.setNullAt(0);
+ wrapped.setNulls(1, 1);
+
+ assertTrue(wrapped.isNullAt(0));
+ assertTrue(wrapped.isNullAt(1));
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/vector/reader/TestParquetDataColumnReaderFactory.java b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/vector/reader/TestParquetDataColumnReaderFactory.java
new file mode 100644
index 000000000000..9d6607d03feb
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/vector/reader/TestParquetDataColumnReaderFactory.java
@@ -0,0 +1,272 @@
+/*
+ * 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.vector.reader;
+
+import org.apache.flink.table.data.TimestampData;
+
+import org.apache.parquet.column.Dictionary;
+import org.apache.parquet.column.values.ValuesReader;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.LogicalTypeAnnotation;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Types;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+
+/**
+ * Tests for the {@link ParquetDataColumnReaderFactory} INT64 timestamp dispatch added when
+ * vendoring Flink 2.1's nested-Parquet reader (FLINK-35702).
+ *
+ * <p>The factory is exercised end-to-end by integration tests through
+ * {@link NestedPrimitiveColumnReader}; this unit test focuses on the small, deterministic piece
+ * that was added by this PR — selecting the right {@code ParquetDataColumnReader} for each
+ * supported INT64 TIMESTAMP encoding (modern {@link LogicalTypeAnnotation.TimestampLogicalTypeAnnotation}
+ * MILLIS / MICROS / NANOS plus the legacy {@link OriginalType} encodings) and decoding values
+ * using both the values-reader and dictionary code paths.
+ */
+class TestParquetDataColumnReaderFactory {
+
+ // -----------------------------------------------------------------------------------------------
+ // Type dispatch
+ // -----------------------------------------------------------------------------------------------
+
+ @Test
+ void valuesReaderDispatchInt96TimestampUsesInt96Reader() {
+ PrimitiveType type = Types.required(PrimitiveType.PrimitiveTypeName.INT96).named("ts");
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(type, new StubValuesReader(), true);
+ assertInstanceOf(ParquetDataColumnReaderFactory.TypesFromInt96PageReader.class, reader);
+ }
+
+ @Test
+ void valuesReaderDispatchInt64WithoutAnnotationUsesDefaultReader() {
+ PrimitiveType type = Types.required(PrimitiveType.PrimitiveTypeName.INT64).named("plainLong");
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(type, new StubValuesReader(), true);
+ assertInstanceOf(ParquetDataColumnReaderFactory.DefaultParquetDataColumnReader.class, reader);
+ }
+
+ @Test
+ void valuesReaderDispatchInt64TimestampMillisLogicalUsesInt64Reader() {
+ PrimitiveType type =
+ Types.required(PrimitiveType.PrimitiveTypeName.INT64)
+ .as(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS))
+ .named("ts");
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(type, new StubValuesReader(), true);
+ assertInstanceOf(ParquetDataColumnReaderFactory.TypesFromInt64PageReader.class, reader);
+ }
+
+ @Test
+ void valuesReaderDispatchInt64TimestampMicrosLogicalUsesInt64Reader() {
+ PrimitiveType type =
+ Types.required(PrimitiveType.PrimitiveTypeName.INT64)
+ .as(LogicalTypeAnnotation.timestampType(false, LogicalTypeAnnotation.TimeUnit.MICROS))
+ .named("ts");
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(type, new StubValuesReader(), true);
+ assertInstanceOf(ParquetDataColumnReaderFactory.TypesFromInt64PageReader.class, reader);
+ }
+
+ @Test
+ void valuesReaderDispatchInt64TimestampNanosLogicalUsesInt64Reader() {
+ PrimitiveType type =
+ Types.required(PrimitiveType.PrimitiveTypeName.INT64)
+ .as(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.NANOS))
+ .named("ts");
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(type, new StubValuesReader(), true);
+ assertInstanceOf(ParquetDataColumnReaderFactory.TypesFromInt64PageReader.class, reader);
+ }
+
+ @Test
+ void valuesReaderDispatchInt64LegacyTimestampMillisOriginalUsesInt64Reader() {
+ PrimitiveType type =
+ Types.required(PrimitiveType.PrimitiveTypeName.INT64)
+ .as(OriginalType.TIMESTAMP_MILLIS)
+ .named("ts");
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(type, new StubValuesReader(), true);
+ assertInstanceOf(ParquetDataColumnReaderFactory.TypesFromInt64PageReader.class, reader);
+ }
+
+ @Test
+ void valuesReaderDispatchInt64LegacyTimestampMicrosOriginalUsesInt64Reader() {
+ PrimitiveType type =
+ Types.required(PrimitiveType.PrimitiveTypeName.INT64)
+ .as(OriginalType.TIMESTAMP_MICROS)
+ .named("ts");
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(type, new StubValuesReader(), true);
+ assertInstanceOf(ParquetDataColumnReaderFactory.TypesFromInt64PageReader.class, reader);
+ }
+
+ @Test
+ void valuesReaderDispatchInt32DoesNotUseTimestampReader() {
+ PrimitiveType type = Types.required(PrimitiveType.PrimitiveTypeName.INT32).named("i");
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(type, new StubValuesReader(), true);
+ assertInstanceOf(ParquetDataColumnReaderFactory.DefaultParquetDataColumnReader.class, reader);
+ }
+
+ @Test
+ void dictionaryReaderDispatchInt64TimestampMillisLogicalUsesInt64Reader() {
+ PrimitiveType type =
+ Types.required(PrimitiveType.PrimitiveTypeName.INT64)
+ .as(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS))
+ .named("ts");
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByTypeOnDictionary(
+ type, new StubDictionary(), true);
+ assertInstanceOf(ParquetDataColumnReaderFactory.TypesFromInt64PageReader.class, reader);
+ }
+
+ // -----------------------------------------------------------------------------------------------
+ // INT64 → TimestampData decoding (per ChronoUnit, both UTC and local-time-zone branches)
+ // -----------------------------------------------------------------------------------------------
+
+ @Test
+ void int64ReaderReadsTimestampMillisFromValuesReaderInUtc() {
+ PrimitiveType type =
+ Types.required(PrimitiveType.PrimitiveTypeName.INT64)
+ .as(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS))
+ .named("ts");
+ long epochMillis = 1_700_000_000_123L;
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(
+ type, new StubValuesReader(epochMillis), true);
+
+ TimestampData ts = reader.readTimestamp();
+ assertNotNull(ts);
+ assertEquals(epochMillis, ts.getMillisecond());
+ assertEquals(0, ts.getNanoOfMillisecond());
+ }
+
+ @Test
+ void int64ReaderReadsTimestampMicrosFromValuesReaderInUtc() {
+ PrimitiveType type =
+ Types.required(PrimitiveType.PrimitiveTypeName.INT64)
+ .as(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MICROS))
+ .named("ts");
+ long epochMicros = 1_700_000_000_123_456L;
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(
+ type, new StubValuesReader(epochMicros), true);
+
+ TimestampData ts = reader.readTimestamp();
+ assertNotNull(ts);
+ assertEquals(epochMicros / 1_000L, ts.getMillisecond());
+ // 456 microseconds remain → 456_000 nanoseconds within the millisecond
+ assertEquals(456_000, ts.getNanoOfMillisecond());
+ }
+
+ @Test
+ void int64ReaderReadsTimestampNanosFromValuesReaderInUtc() {
+ PrimitiveType type =
+ Types.required(PrimitiveType.PrimitiveTypeName.INT64)
+ .as(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.NANOS))
+ .named("ts");
+ long epochNanos = 1_700_000_000_123_456_789L;
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByType(
+ type, new StubValuesReader(epochNanos), true);
+
+ TimestampData ts = reader.readTimestamp();
+ assertNotNull(ts);
+ assertEquals(epochNanos / 1_000_000L, ts.getMillisecond());
+ assertEquals(456_789, ts.getNanoOfMillisecond());
+ }
+
+ @Test
+ void int64ReaderReadsTimestampMillisFromDictionaryInUtc() {
+ PrimitiveType type =
+ Types.required(PrimitiveType.PrimitiveTypeName.INT64)
+ .as(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS))
+ .named("ts");
+ long epochMillis = 1_700_000_000_456L;
+ ParquetDataColumnReader reader =
+ ParquetDataColumnReaderFactory.getDataColumnReaderByTypeOnDictionary(
+ type, new StubDictionary(epochMillis), true);
+
+ TimestampData ts = reader.readTimestamp(0);
+ assertNotNull(ts);
+ assertEquals(epochMillis, ts.getMillisecond());
+ }
+
+ // -----------------------------------------------------------------------------------------------
+ // Stubs (only the methods exercised by the dispatch + decoding tests above)
+ // -----------------------------------------------------------------------------------------------
+
+ /** Minimal {@link ValuesReader} returning a fixed long; other methods throw. */
+ private static final class StubValuesReader extends ValuesReader {
+ private final long fixedLong;
+
+ StubValuesReader() {
+ this(0L);
+ }
+
+ StubValuesReader(long fixedLong) {
+ this.fixedLong = fixedLong;
+ }
+
+ @Override
+ public long readLong() {
+ return fixedLong;
+ }
+
+ @Override
+ public void skip() {
+ // unused
+ }
+ }
+
+ /** Minimal {@link Dictionary} returning a fixed long for any id; other methods throw. */
+ private static final class StubDictionary extends Dictionary {
+ private final long fixedLong;
+
+ StubDictionary() {
+ this(0L);
+ }
+
+ StubDictionary(long fixedLong) {
+ super(null);
+ this.fixedLong = fixedLong;
+ }
+
+ @Override
+ public Binary decodeToBinary(int id) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public long decodeToLong(int id) {
+ return fixedLong;
+ }
+
+ @Override
+ public int getMaxId() {
+ return 0;
+ }
+ }
+}
diff --git a/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/vector/type/TestParquetGroupField.java b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/vector/type/TestParquetGroupField.java
new file mode 100644
index 000000000000..a071ef937611
--- /dev/null
+++ b/hudi-flink-datasource/hudi-flink1.19.x/src/test/java/org/apache/hudi/table/format/cow/vector/type/TestParquetGroupField.java
@@ -0,0 +1,134 @@
+/*
+ * 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.vector.type;
+
+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.parquet.column.ColumnDescriptor;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName;
+import org.apache.parquet.schema.Types;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * Tests for {@link ParquetGroupField}.
+ */
+public class TestParquetGroupField {
+
+ @Test
+ void testChildrenWithAllNonNullEntriesAreRetained() {
+ ParquetField c0 = new ParquetPrimitiveField(new IntType(), true, descriptor(), 0);
+ ParquetField c1 = new ParquetPrimitiveField(new VarCharType(), true, descriptor(), 1);
+ List<ParquetField> children = Arrays.asList(c0, c1);
+
+ ParquetGroupField group = new ParquetGroupField(rowType(), 0, 1, true, children);
+
+ assertEquals(2, group.getChildren().size());
+ assertSame(c0, group.getChildren().get(0));
+ assertSame(c1, group.getChildren().get(1));
+ }
+
+ @Test
+ void testChildrenMayContainNullForSchemaEvolution() {
+ // A ROW field present in the requested Flink schema but absent from the Parquet file is
+ // represented by a null slot in `children`. The group must allow this (Hudi-specific
+ // extension over Flink's ImmutableList-backed equivalent).
+ ParquetField present = new ParquetPrimitiveField(new IntType(), true, descriptor(), 0);
+ List<ParquetField> children = Arrays.asList(present, null);
+
+ ParquetGroupField group = new ParquetGroupField(rowType(), 0, 1, true, children);
+
+ assertEquals(2, group.getChildren().size());
+ assertNotNull(group.getChildren().get(0));
+ assertNull(group.getChildren().get(1));
+ }
+
+ @Test
+ void testChildrenListIsUnmodifiable() {
+ ParquetField child = new ParquetPrimitiveField(new IntType(), true, descriptor(), 0);
+ ParquetGroupField group =
+ new ParquetGroupField(rowType(), 0, 1, true, Collections.singletonList(child));
+
+ assertThrows(UnsupportedOperationException.class, () -> group.getChildren().add(null));
+ assertThrows(UnsupportedOperationException.class, () -> group.getChildren().remove(0));
+ }
+
+ @Test
+ void testChildrenListIsDefensivelyCopied() {
+ // Mutations to the caller-supplied list must not be visible through the group.
+ ParquetField child = new ParquetPrimitiveField(new IntType(), true, descriptor(), 0);
+ List<ParquetField> mutable = new ArrayList<>();
+ mutable.add(child);
+
+ ParquetGroupField group = new ParquetGroupField(rowType(), 0, 1, true, mutable);
+ mutable.add(null);
+
+ assertEquals(1, group.getChildren().size());
+ }
+
+ @Test
+ void testNullChildrenListThrows() {
+ assertThrows(
+ NullPointerException.class,
+ () -> new ParquetGroupField(rowType(), 0, 1, true, null));
+ }
+
+ @Test
+ void testEmptyChildrenListIsAllowed() {
+ ParquetGroupField group =
+ new ParquetGroupField(rowType(), 0, 1, true, Collections.emptyList());
+
+ assertEquals(0, group.getChildren().size());
+ }
+
+ @Test
+ void testFieldMetadataIsExposed() {
+ ParquetGroupField group =
+ new ParquetGroupField(rowType(), 2, 5, false, Collections.emptyList());
+
+ assertEquals(2, group.getRepetitionLevel());
+ assertEquals(5, group.getDefinitionLevel());
+ assertFalse(group.isRequired());
+ }
+
+ private static LogicalType rowType() {
+ return RowType.of(new IntType());
+ }
+
+ private static ColumnDescriptor descriptor() {
+ PrimitiveType primitive = Types.required(PrimitiveTypeName.INT32).named("f");
+ return new ColumnDescriptor(new String[] {"f"}, primitive, 0, 0);
+ }
+}