You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2024/06/30 11:38:54 UTC
(flink) branch master updated: [FLINK-35702][parquet] support parquet read nested in nested (#24991)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 8c899b02ab2 [FLINK-35702][parquet] support parquet read nested in nested (#24991)
8c899b02ab2 is described below
commit 8c899b02ab25a34e4d35c73a8fc0299894110aca
Author: Wenchao Wu <60...@users.noreply.github.com>
AuthorDate: Sun Jun 30 19:38:46 2024 +0800
[FLINK-35702][parquet] support parquet read nested in nested (#24991)
---
.../connectors/hive/HiveTableSourceITCase.java | 14 +-
flink-formats/flink-parquet/pom.xml | 8 +-
.../formats/parquet/ParquetFileFormatFactory.java | 9 +-
.../parquet/ParquetVectorizedInputFormat.java | 19 +-
.../formats/parquet/utils/NestedPositionUtil.java | 208 +++++++
.../vector/ParquetColumnarRowSplitReader.java | 13 +
.../parquet/vector/ParquetDecimalVector.java | 161 +++++-
.../parquet/vector/ParquetSplitReaderUtil.java | 268 +++++++--
.../vector/position/CollectionPosition.java | 54 ++
.../parquet/vector/position/LevelDelegation.java | 38 ++
.../parquet/vector/position/RowPosition.java | 40 ++
.../parquet/vector/reader/ArrayColumnReader.java | 483 ----------------
.../vector/reader/BaseVectorizedColumnReader.java | 290 ----------
.../parquet/vector/reader/MapColumnReader.java | 64 ---
.../parquet/vector/reader/NestedColumnReader.java | 256 +++++++++
.../vector/reader/NestedPrimitiveColumnReader.java | 624 +++++++++++++++++++++
.../parquet/vector/reader/RowColumnReader.java | 58 --
.../formats/parquet/vector/type/ParquetField.java | 67 +++
.../parquet/vector/type/ParquetGroupField.java | 47 ++
.../parquet/vector/type/ParquetPrimitiveField.java | 51 ++
.../vector/ParquetColumnarRowSplitReaderTest.java | 234 ++++++++
.../flink/runtime/util/BooleanArrayList.java | 65 +++
.../data/columnar/vector/heap/HeapMapVector.java | 8 +
.../data/columnar/vector/heap/HeapRowVector.java | 4 +
24 files changed, 2114 insertions(+), 969 deletions(-)
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
index 402ca54e50f..8d6e2bab95e 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveTableSourceITCase.java
@@ -1035,19 +1035,7 @@ public class HiveTableSourceITCase extends BatchAbstractTestBase {
sortedRows.addAll(rows);
sortedRows.addAll(rows);
sortedRows.sort(Comparator.comparingInt(o -> (Integer) o.getField(0)));
-
- List<Row> expectedRows = new ArrayList<>();
- for (int i = 0; i < sortedRows.size(); i++) {
- Row rowExpect = Row.copy(sortedRows.get(i));
- Row nestedRow = (Row) rowExpect.getField(2);
- if (nestedRow.getField(0) == null
- && nestedRow.getField(1) == null
- && nestedRow.getField(2) == null) {
- rowExpect.setField(2, null);
- }
- expectedRows.add(rowExpect);
- }
- return expectedRows;
+ return sortedRows;
}
private static CloseableIterator<Row> executeAndGetResult(StreamTableEnvironment tEnv)
diff --git a/flink-formats/flink-parquet/pom.xml b/flink-formats/flink-parquet/pom.xml
index 2d4db0a6669..cb7452f607f 100644
--- a/flink-formats/flink-parquet/pom.xml
+++ b/flink-formats/flink-parquet/pom.xml
@@ -273,7 +273,13 @@ under the License.
<scope>test</scope>
<type>test-jar</type>
</dependency>
- </dependencies>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-runtime</artifactId>
+ <version>2.0-SNAPSHOT</version>
+ <scope>compile</scope>
+ </dependency>
+ </dependencies>
<dependencyManagement>
<dependencies>
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetFileFormatFactory.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetFileFormatFactory.java
index 8be727c7947..2c7ccda2259 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetFileFormatFactory.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetFileFormatFactory.java
@@ -38,7 +38,6 @@ import org.apache.flink.table.connector.format.ProjectableDecodingFormat;
import org.apache.flink.table.connector.sink.DynamicTableSink;
import org.apache.flink.table.connector.source.DynamicTableSource;
import org.apache.flink.table.data.RowData;
-import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
import org.apache.flink.table.factories.DynamicTableFactory;
import org.apache.flink.table.plan.stats.TableStats;
import org.apache.flink.table.types.DataType;
@@ -83,6 +82,12 @@ public class ParquetFileFormatFactory implements BulkReaderFormatFactory, BulkWr
"Write parquet timestamp as int64/LogicalTypes instead of int96/OriginalTypes. "
+ "Note: Timestamp will be time zone agnostic (NEVER converted to a different time zone).");
+ public static final ConfigOption<Integer> BATCH_SIZE =
+ key("batch-size")
+ .intType()
+ .defaultValue(2048)
+ .withDescription("Determine the batch size when reading parquet files.");
+
@Override
public BulkDecodingFormat<RowData> createDecodingFormat(
DynamicTableFactory.Context context, ReadableConfig formatOptions) {
@@ -159,7 +164,7 @@ public class ParquetFileFormatFactory implements BulkReaderFormatFactory, BulkWr
sourceContext.createTypeInformation(producedDataType),
Collections.emptyList(),
null,
- VectorizedColumnBatch.DEFAULT_SIZE,
+ formatOptions.get(BATCH_SIZE),
formatOptions.get(UTC_TIMEZONE),
true);
}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
index badd86072d8..b8bafbe8439 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
@@ -31,6 +31,7 @@ import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.formats.parquet.vector.type.ParquetField;
import org.apache.flink.table.data.columnar.vector.ColumnVector;
import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
import org.apache.flink.table.data.columnar.vector.writable.WritableColumnVector;
@@ -46,6 +47,8 @@ import org.apache.parquet.column.page.PageReadStore;
import org.apache.parquet.filter2.compat.FilterCompat;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.BlockMetaData;
+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;
@@ -65,6 +68,7 @@ import java.util.Locale;
import java.util.Map;
import java.util.Set;
+import static org.apache.flink.formats.parquet.vector.ParquetSplitReaderUtil.buildFieldsList;
import static org.apache.flink.formats.parquet.vector.ParquetSplitReaderUtil.createColumnReader;
import static org.apache.flink.formats.parquet.vector.ParquetSplitReaderUtil.createWritableColumnVector;
import static org.apache.parquet.hadoop.ParquetInputFormat.getFilter;
@@ -140,12 +144,18 @@ public abstract class ParquetVectorizedInputFormat<T, SplitT extends FileSourceS
final Pool<ParquetReaderBatch<T>> poolOfBatches =
createPoolOfBatches(split, requestedSchema, numBatchesToCirculate(config));
+ RowType projectedType = RowType.of(projectedTypes, projectedFields);
+ MessageColumnIO columnIO = new ColumnIOFactory().getColumnIO(requestedSchema);
+ List<ParquetField> fields =
+ buildFieldsList(projectedType.getFields(), projectedType.getFieldNames(), columnIO);
+
return new ParquetReader(
parquetFileReader,
requestedSchema,
unknownFieldsIndices,
totalRowCount,
- poolOfBatches);
+ poolOfBatches,
+ fields);
}
protected int numBatchesToCirculate(Configuration config) {
@@ -342,12 +352,15 @@ public abstract class ParquetVectorizedInputFormat<T, SplitT extends FileSourceS
private long recordsToSkip;
+ private final List<ParquetField> fields;
+
private ParquetReader(
ParquetFileReader reader,
MessageType requestedSchema,
Set<Integer> unknownFieldsIndices,
long totalRowCount,
- Pool<ParquetReaderBatch<T>> pool) {
+ Pool<ParquetReaderBatch<T>> pool,
+ List<ParquetField> fields) {
this.reader = reader;
this.requestedSchema = requestedSchema;
this.unknownFieldsIndices = unknownFieldsIndices;
@@ -356,6 +369,7 @@ public abstract class ParquetVectorizedInputFormat<T, SplitT extends FileSourceS
this.rowsReturned = 0;
this.totalCountLoadedSoFar = 0;
this.recordsToSkip = 0;
+ this.fields = fields;
}
@Nullable
@@ -425,6 +439,7 @@ public abstract class ParquetVectorizedInputFormat<T, SplitT extends FileSourceS
types.get(i),
requestedSchema.getColumns(),
pages,
+ fields.get(i),
0);
}
}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/NestedPositionUtil.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/NestedPositionUtil.java
new file mode 100644
index 00000000000..4e392ebadd9
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/NestedPositionUtil.java
@@ -0,0 +1,208 @@
+/*
+ * 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.flink.formats.parquet.utils;
+
+import org.apache.flink.formats.parquet.vector.position.CollectionPosition;
+import org.apache.flink.formats.parquet.vector.position.RowPosition;
+import org.apache.flink.formats.parquet.vector.type.ParquetField;
+import org.apache.flink.runtime.util.BooleanArrayList;
+import org.apache.flink.runtime.util.LongArrayList;
+
+import static java.lang.String.format;
+
+/** Utils to calculate nested type position. */
+public class NestedPositionUtil {
+
+ /**
+ * Calculate row offsets according to column's max repetition level, definition level, value's
+ * repetition level and definition level. Each row has three situation:
+ * <li>Row is not defined,because it's optional parent fields is null, this is decided by its
+ * parent's repetition level
+ * <li>Row is null
+ * <li>Row is defined and not empty.
+ *
+ * @param field field that contains the row column message include max repetition level and
+ * definition level.
+ * @param fieldRepetitionLevels int array with each value's repetition level.
+ * @param fieldDefinitionLevels int array with each value's definition level.
+ * @return {@link RowPosition} contains collections row count and isNull array.
+ */
+ public static RowPosition calculateRowOffsets(
+ ParquetField field, int[] fieldDefinitionLevels, int[] fieldRepetitionLevels) {
+ int rowDefinitionLevel = field.getDefinitionLevel();
+ int rowRepetitionLevel = field.getRepetitionLevel();
+ int nullValuesCount = 0;
+ BooleanArrayList nullRowFlags = new BooleanArrayList(0);
+ for (int i = 0; i < fieldDefinitionLevels.length; i++) {
+ if (fieldRepetitionLevels[i] > rowRepetitionLevel) {
+ throw new IllegalStateException(
+ format(
+ "In parquet's row type field repetition level should not larger than row's repetition level. "
+ + "Row repetition level is %s, row field repetition level is %s.",
+ rowRepetitionLevel, fieldRepetitionLevels[i]));
+ }
+
+ if (fieldDefinitionLevels[i] >= rowDefinitionLevel) {
+ // current row is defined and not empty
+ nullRowFlags.add(false);
+ } else {
+ // current row is null
+ nullRowFlags.add(true);
+ nullValuesCount++;
+ }
+ }
+ if (nullValuesCount == 0) {
+ return new RowPosition(null, fieldDefinitionLevels.length);
+ }
+ return new RowPosition(nullRowFlags.toArray(), nullRowFlags.size());
+ }
+
+ /**
+ * Calculate the collection's offsets according to column's max repetition level, definition
+ * level, value's repetition level and definition level. Each collection (Array or Map) has four
+ * situation:
+ * <li>Collection is not defined, because optional parent fields is null, this is decided by its
+ * parent's repetition level
+ * <li>Collection is null
+ * <li>Collection is defined but empty
+ * <li>Collection is defined and not empty. In this case offset value is increased by the number
+ * of elements in that collection
+ *
+ * @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.
+ * @return {@link CollectionPosition} contains collections offset array, length array and isNull
+ * array.
+ */
+ public static CollectionPosition calculateCollectionOffsets(
+ ParquetField field, int[] definitionLevels, int[] repetitionLevels) {
+ int collectionDefinitionLevel = field.getDefinitionLevel();
+ int collectionRepetitionLevel = field.getRepetitionLevel() + 1;
+ int offset = 0;
+ int valueCount = 0;
+ LongArrayList offsets = new LongArrayList(0);
+ offsets.add(offset);
+ BooleanArrayList emptyCollectionFlags = new BooleanArrayList(0);
+ BooleanArrayList nullCollectionFlags = new BooleanArrayList(0);
+ int nullValuesCount = 0;
+ for (int i = 0;
+ i < definitionLevels.length;
+ i = getNextCollectionStartIndex(repetitionLevels, collectionRepetitionLevel, i)) {
+ valueCount++;
+ if (definitionLevels[i] >= collectionDefinitionLevel - 1) {
+ boolean isNull =
+ isOptionalFieldValueNull(definitionLevels[i], collectionDefinitionLevel);
+ nullCollectionFlags.add(isNull);
+ nullValuesCount += isNull ? 1 : 0;
+ // definitionLevels[i] > collectionDefinitionLevel => Collection is defined and not
+ // empty
+ // definitionLevels[i] == collectionDefinitionLevel => Collection is defined but
+ // empty
+ if (definitionLevels[i] > collectionDefinitionLevel) {
+ emptyCollectionFlags.add(false);
+ offset += getCollectionSize(repetitionLevels, collectionRepetitionLevel, i + 1);
+ } else if (definitionLevels[i] == collectionDefinitionLevel) {
+ offset++;
+ emptyCollectionFlags.add(true);
+ } else {
+ offset++;
+ emptyCollectionFlags.add(false);
+ }
+ offsets.add(offset);
+ } else {
+ // when definitionLevels[i] < collectionDefinitionLevel - 1, it means the collection
+ // is
+ // not defined, but we need to regard it as null to avoid getting value wrong.
+ nullCollectionFlags.add(true);
+ nullValuesCount++;
+ offsets.add(++offset);
+ emptyCollectionFlags.add(false);
+ }
+ }
+ long[] offsetsArray = offsets.toArray();
+ long[] length = calculateLengthByOffsets(emptyCollectionFlags.toArray(), offsetsArray);
+ if (nullValuesCount == 0) {
+ return new CollectionPosition(null, offsetsArray, length, valueCount);
+ }
+ return new CollectionPosition(
+ nullCollectionFlags.toArray(), offsetsArray, length, valueCount);
+ }
+
+ public static boolean isOptionalFieldValueNull(int definitionLevel, int maxDefinitionLevel) {
+ return definitionLevel == maxDefinitionLevel - 1;
+ }
+
+ public static long[] calculateLengthByOffsets(
+ boolean[] collectionIsEmpty, long[] arrayOffsets) {
+ LongArrayList lengthList = new LongArrayList(arrayOffsets.length);
+ for (int i = 0; i < arrayOffsets.length - 1; i++) {
+ long offset = arrayOffsets[i];
+ long length = arrayOffsets[i + 1] - offset;
+ if (length < 0) {
+ throw new IllegalArgumentException(
+ format(
+ "Offset is not monotonically ascending. offsets[%s]=%s, offsets[%s]=%s",
+ i, arrayOffsets[i], i + 1, arrayOffsets[i + 1]));
+ }
+ if (collectionIsEmpty[i]) {
+ length = 0;
+ }
+ lengthList.add(length);
+ }
+ return lengthList.toArray();
+ }
+
+ private static int getNextCollectionStartIndex(
+ int[] repetitionLevels, int maxRepetitionLevel, int elementIndex) {
+ do {
+ elementIndex++;
+ } while (hasMoreElements(repetitionLevels, elementIndex)
+ && isNotCollectionBeginningMarker(
+ repetitionLevels, maxRepetitionLevel, elementIndex));
+ return elementIndex;
+ }
+
+ /** This method is only called for non-empty collections. */
+ private static int getCollectionSize(
+ int[] repetitionLevels, int maxRepetitionLevel, int nextIndex) {
+ int size = 1;
+ while (hasMoreElements(repetitionLevels, nextIndex)
+ && isNotCollectionBeginningMarker(
+ repetitionLevels, maxRepetitionLevel, nextIndex)) {
+ // Collection elements cannot only be primitive, but also can have nested structure
+ // Counting only elements which belong to current collection, skipping inner elements of
+ // nested collections/structs
+ if (repetitionLevels[nextIndex] <= maxRepetitionLevel) {
+ size++;
+ }
+ nextIndex++;
+ }
+ return size;
+ }
+
+ private static boolean isNotCollectionBeginningMarker(
+ int[] repetitionLevels, int maxRepetitionLevel, int nextIndex) {
+ return repetitionLevels[nextIndex] >= maxRepetitionLevel;
+ }
+
+ private static boolean hasMoreElements(int[] repetitionLevels, int nextIndex) {
+ return nextIndex < repetitionLevels.length;
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReader.java
index 2611dfbb364..7634978e916 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReader.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReader.java
@@ -19,12 +19,14 @@
package org.apache.flink.formats.parquet.vector;
import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.formats.parquet.vector.type.ParquetField;
import org.apache.flink.table.data.columnar.ColumnarRowData;
import org.apache.flink.table.data.columnar.vector.ColumnVector;
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;
@@ -35,6 +37,8 @@ import org.apache.parquet.filter2.compat.FilterCompat;
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;
@@ -48,6 +52,7 @@ import java.util.List;
import java.util.Locale;
import java.util.Map;
+import static org.apache.flink.formats.parquet.vector.ParquetSplitReaderUtil.buildFieldsList;
import static org.apache.flink.formats.parquet.vector.ParquetSplitReaderUtil.createColumnReader;
import static org.apache.flink.formats.parquet.vector.ParquetSplitReaderUtil.createWritableColumnVector;
import static org.apache.parquet.filter2.compat.RowGroupFilter.filterRowGroups;
@@ -80,6 +85,8 @@ public class ParquetColumnarRowSplitReader implements Closeable {
private final int batchSize;
+ private final List<ParquetField> fieldList;
+
private ParquetFileReader reader;
/**
@@ -142,6 +149,11 @@ public class ParquetColumnarRowSplitReader implements Closeable {
this.writableVectors = createWritableVectors();
this.columnarBatch = generator.generate(createReadableVectors());
this.row = new ColumnarRowData(columnarBatch);
+
+ MessageColumnIO columnIO = new ColumnIOFactory().getColumnIO(requestedSchema);
+ RowType selectedType = RowType.of(selectedTypes, selectedFieldNames);
+ this.fieldList =
+ buildFieldsList(selectedType.getFields(), selectedType.getFieldNames(), columnIO);
}
/** Clips `parquetSchema` according to `fieldNames`. */
@@ -321,6 +333,7 @@ public class ParquetColumnarRowSplitReader implements Closeable {
types.get(i),
requestedSchema.getColumns(),
pages,
+ fieldList.get(i),
0);
}
totalCountLoadedSoFar += pages.getRowCount();
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetDecimalVector.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetDecimalVector.java
index 0c62dbb0460..9d5f020eeb3 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetDecimalVector.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetDecimalVector.java
@@ -24,8 +24,13 @@ 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 org.apache.parquet.Preconditions;
@@ -34,7 +39,8 @@ import org.apache.parquet.Preconditions;
* {@link DecimalColumnVector} interface.
*/
@Internal
-public class ParquetDecimalVector implements DecimalColumnVector {
+public class ParquetDecimalVector
+ implements DecimalColumnVector, WritableLongVector, WritableIntVector, WritableBytesVector {
private final ColumnVector vector;
@@ -69,4 +75,157 @@ public class ParquetDecimalVector implements DecimalColumnVector {
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/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
index 77fa87c7e26..145120aea1f 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
@@ -20,7 +20,6 @@ package org.apache.flink.formats.parquet.vector;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.parquet.utils.ParquetSchemaConverter;
-import org.apache.flink.formats.parquet.vector.reader.ArrayColumnReader;
import org.apache.flink.formats.parquet.vector.reader.BooleanColumnReader;
import org.apache.flink.formats.parquet.vector.reader.ByteColumnReader;
import org.apache.flink.formats.parquet.vector.reader.BytesColumnReader;
@@ -30,10 +29,12 @@ import org.apache.flink.formats.parquet.vector.reader.FixedLenBytesColumnReader;
import org.apache.flink.formats.parquet.vector.reader.FloatColumnReader;
import org.apache.flink.formats.parquet.vector.reader.IntColumnReader;
import org.apache.flink.formats.parquet.vector.reader.LongColumnReader;
-import org.apache.flink.formats.parquet.vector.reader.MapColumnReader;
-import org.apache.flink.formats.parquet.vector.reader.RowColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.NestedColumnReader;
import org.apache.flink.formats.parquet.vector.reader.ShortColumnReader;
import org.apache.flink.formats.parquet.vector.reader.TimestampColumnReader;
+import org.apache.flink.formats.parquet.vector.type.ParquetField;
+import org.apache.flink.formats.parquet.vector.type.ParquetGroupField;
+import org.apache.flink.formats.parquet.vector.type.ParquetPrimitiveField;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.data.columnar.vector.ColumnVector;
@@ -61,14 +62,23 @@ 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.VarBinaryType;
+import org.apache.flink.util.FlinkRuntimeException;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.StringUtils;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableList;
import org.apache.hadoop.conf.Configuration;
import org.apache.parquet.ParquetRuntimeException;
import org.apache.parquet.column.ColumnDescriptor;
import org.apache.parquet.column.page.PageReadStore;
+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.LogicalTypeAnnotation;
import org.apache.parquet.schema.OriginalType;
import org.apache.parquet.schema.PrimitiveType;
import org.apache.parquet.schema.Type;
@@ -83,10 +93,13 @@ import java.util.ArrayList;
import java.util.Arrays;
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.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}. */
public class ParquetSplitReaderUtil {
@@ -298,6 +311,7 @@ public class ParquetSplitReaderUtil {
Type type,
List<ColumnDescriptor> columnDescriptors,
PageReadStore pages,
+ ParquetField field,
int depth)
throws IOException {
List<ColumnDescriptor> descriptors =
@@ -356,61 +370,10 @@ public class ParquetSplitReaderUtil {
((DecimalType) fieldType).getPrecision());
}
case ARRAY:
- return new ArrayColumnReader(
- descriptors.get(0),
- pages.getPageReader(descriptors.get(0)),
- isUtcTimestamp,
- descriptors.get(0).getPrimitiveType(),
- fieldType);
case MAP:
- MapType mapType = (MapType) fieldType;
- ArrayColumnReader mapKeyReader =
- new ArrayColumnReader(
- descriptors.get(0),
- pages.getPageReader(descriptors.get(0)),
- isUtcTimestamp,
- descriptors.get(0).getPrimitiveType(),
- new ArrayType(mapType.getKeyType()));
- ArrayColumnReader mapValueReader =
- new ArrayColumnReader(
- descriptors.get(1),
- pages.getPageReader(descriptors.get(1)),
- isUtcTimestamp,
- descriptors.get(1).getPrimitiveType(),
- new ArrayType(mapType.getValueType()));
- return new MapColumnReader(mapKeyReader, mapValueReader);
case MULTISET:
- MultisetType multisetType = (MultisetType) fieldType;
- ArrayColumnReader multisetKeyReader =
- new ArrayColumnReader(
- descriptors.get(0),
- pages.getPageReader(descriptors.get(0)),
- isUtcTimestamp,
- descriptors.get(0).getPrimitiveType(),
- new ArrayType(multisetType.getElementType()));
- ArrayColumnReader multisetValueReader =
- new ArrayColumnReader(
- descriptors.get(1),
- pages.getPageReader(descriptors.get(1)),
- isUtcTimestamp,
- descriptors.get(1).getPrimitiveType(),
- new ArrayType(new IntType(false)));
- return new MapColumnReader(multisetKeyReader, multisetValueReader);
case ROW:
- RowType rowType = (RowType) fieldType;
- GroupType groupType = type.asGroupType();
- List<ColumnReader> fieldReaders = new ArrayList<>();
- for (int i = 0; i < rowType.getFieldCount(); i++) {
- fieldReaders.add(
- createColumnReader(
- isUtcTimestamp,
- rowType.getTypeAt(i),
- groupType.getType(i),
- descriptors,
- pages,
- depth + 1));
- }
- return new RowColumnReader(fieldReaders);
+ return new NestedColumnReader(isUtcTimestamp, pages, field);
default:
throw new UnsupportedOperationException(fieldType + " is not supported now.");
}
@@ -527,7 +490,18 @@ public class ParquetSplitReaderUtil {
depth));
case MAP:
MapType mapType = (MapType) fieldType;
+ LogicalTypeAnnotation mapTypeAnnotation = type.getLogicalTypeAnnotation();
GroupType mapRepeatedType = type.asGroupType().getType(0).asGroupType();
+ if (mapTypeAnnotation.equals(LogicalTypeAnnotation.listType())) {
+ mapRepeatedType = mapRepeatedType.getType(0).asGroupType();
+ depth++;
+ if (mapRepeatedType
+ .getLogicalTypeAnnotation()
+ .equals(LogicalTypeAnnotation.mapType())) {
+ mapRepeatedType = mapRepeatedType.getType(0).asGroupType();
+ depth++;
+ }
+ }
return new HeapMapVector(
batchSize,
createWritableColumnVector(
@@ -544,7 +518,18 @@ public class ParquetSplitReaderUtil {
depth + 2));
case MULTISET:
MultisetType multisetType = (MultisetType) fieldType;
+ LogicalTypeAnnotation multisetTypeAnnotation = type.getLogicalTypeAnnotation();
GroupType multisetRepeatedType = type.asGroupType().getType(0).asGroupType();
+ if (multisetTypeAnnotation.equals(LogicalTypeAnnotation.listType())) {
+ multisetRepeatedType = multisetRepeatedType.getType(0).asGroupType();
+ depth++;
+ if (multisetRepeatedType
+ .getLogicalTypeAnnotation()
+ .equals(LogicalTypeAnnotation.mapType())) {
+ multisetRepeatedType = multisetRepeatedType.getType(0).asGroupType();
+ depth++;
+ }
+ }
return new HeapMapVector(
batchSize,
createWritableColumnVector(
@@ -562,6 +547,12 @@ public class ParquetSplitReaderUtil {
case ROW:
RowType rowType = (RowType) fieldType;
GroupType groupType = type.asGroupType();
+ if (LogicalTypeAnnotation.listType().equals(groupType.getLogicalTypeAnnotation())) {
+ // this means there was two outside struct, need to get group twice.
+ groupType = groupType.getType(0).asGroupType();
+ groupType = groupType.getType(0).asGroupType();
+ depth = depth + 2;
+ }
WritableColumnVector[] columnVectors =
new WritableColumnVector[rowType.getFieldCount()];
for (int i = 0; i < columnVectors.length; i++) {
@@ -578,4 +569,171 @@ public class ParquetSplitReaderUtil {
throw new UnsupportedOperationException(fieldType + " is not supported now.");
}
}
+
+ public static List<ParquetField> buildFieldsList(
+ List<RowType.RowField> childrens, List<String> fieldNames, MessageColumnIO columnIO) {
+ List<ParquetField> list = new ArrayList<>();
+ for (int i = 0; i < childrens.size(); i++) {
+ list.add(
+ constructField(
+ childrens.get(i), lookupColumnByName(columnIO, fieldNames.get(i))));
+ }
+ return list;
+ }
+
+ 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 filedName = rowField.getName();
+ if (type instanceof RowType) {
+ GroupColumnIO groupColumnIO = (GroupColumnIO) columnIO;
+ RowType rowType = (RowType) type;
+ ImmutableList.Builder<ParquetField> fieldsBuilder = ImmutableList.builder();
+ List<String> fieldNames = rowType.getFieldNames();
+ List<RowType.RowField> childrens = rowType.getFields();
+ for (int i = 0; i < childrens.size(); i++) {
+ fieldsBuilder.add(
+ constructField(
+ childrens.get(i),
+ lookupColumnByName(groupColumnIO, fieldNames.get(i))));
+ }
+
+ return new ParquetGroupField(
+ type, repetitionLevel, definitionLevel, required, fieldsBuilder.build());
+ }
+
+ 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,
+ ImmutableList.of(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,
+ ImmutableList.of(keyField, valueField));
+ }
+
+ if (type instanceof ArrayType) {
+ ArrayType arrayType = (ArrayType) type;
+ ColumnIO elementTypeColumnIO;
+ if (columnIO instanceof GroupColumnIO) {
+ GroupColumnIO groupColumnIO = (GroupColumnIO) columnIO;
+ if (!StringUtils.isNullOrWhitespaceOnly(filedName)) {
+ while (!Objects.equals(groupColumnIO.getName(), filedName)) {
+ 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 RuntimeException(String.format("Unknown ColumnIO, %s", columnIO));
+ }
+
+ ParquetField field =
+ constructField(
+ new RowType.RowField("", arrayType.getElementType()),
+ getArrayElementColumn(elementTypeColumnIO));
+ if (repetitionLevel == field.getRepetitionLevel()) {
+ repetitionLevel = columnIO.getParent().getRepetitionLevel();
+ }
+ return new ParquetGroupField(
+ type, repetitionLevel, definitionLevel, required, ImmutableList.of(field));
+ }
+
+ PrimitiveColumnIO primitiveColumnIO = (PrimitiveColumnIO) columnIO;
+ return new ParquetPrimitiveField(
+ type, required, primitiveColumnIO.getColumnDescriptor(), primitiveColumnIO.getId());
+ }
+
+ /**
+ * Parquet's column names are case in sensitive. So when we look up columns we first check for
+ * exact match, and if that can not find we look for a case-insensitive match.
+ */
+ public static ColumnIO lookupColumnByName(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);
+ }
+ }
+
+ throw new FlinkRuntimeException("Can not find column io for parquet reader.");
+ }
+
+ 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);
+ }
+
+ /* Compatible with array has a standard 3-level structure:
+ * optional group my_list (LIST) {
+ * repeated group element {
+ * required binary str (UTF8);
+ * };
+ * }
+ */
+ 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);
+ }
+
+ /* Compatible with array for 2-level arrays where a repeated field is not a group:
+ * optional group my_list (LIST) {
+ * repeated int32 element;
+ * }
+ */
+ return columnIO;
+ }
}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/position/CollectionPosition.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/position/CollectionPosition.java
new file mode 100644
index 00000000000..d1f22fd727e
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/position/CollectionPosition.java
@@ -0,0 +1,54 @@
+/*
+ * 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.flink.formats.parquet.vector.position;
+
+import javax.annotation.Nullable;
+
+/** To represent collection's position in repeated type. */
+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/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/position/LevelDelegation.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/position/LevelDelegation.java
new file mode 100644
index 00000000000..6b686980c00
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/position/LevelDelegation.java
@@ -0,0 +1,38 @@
+/*
+ * 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.flink.formats.parquet.vector.position;
+
+/** To delegate repetition level and definition level. */
+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/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/position/RowPosition.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/position/RowPosition.java
new file mode 100644
index 00000000000..0e4db0c5054
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/position/RowPosition.java
@@ -0,0 +1,40 @@
+/*
+ * 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.flink.formats.parquet.vector.position;
+
+import javax.annotation.Nullable;
+
+/** To represent struct's position in repeated type. */
+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/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/ArrayColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/ArrayColumnReader.java
deleted file mode 100644
index 72b0be87dd5..00000000000
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/ArrayColumnReader.java
+++ /dev/null
@@ -1,483 +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.flink.formats.parquet.vector.reader;
-
-import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
-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.HeapArrayVector;
-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}. TODO Currently ARRAY type only support non nested case. */
-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) {
- 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.setOffsets(new long[VectorizedColumnBatch.DEFAULT_SIZE]);
- lcv.setLengths(new long[VectorizedColumnBatch.DEFAULT_SIZE]);
-
- LogicalType elementType = ((ArrayType) logicalType).getElementType();
-
- // read the first row in parquet data page, this will be only happened once for this
- // instance
- if (isFirstRow) {
- if (!fetchNextValue(elementType)) {
- return;
- }
- isFirstRow = false;
- }
-
- // 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<>();
-
- int index = collectDataFromParquetPage(readNumber, lcv, valueList, elementType);
- // Convert valueList to array for the ListColumnVector.child
- fillColumnVector(elementType, 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 type the element type of array
- * @return boolean
- */
- private boolean fetchNextValue(LogicalType type) {
- 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(type);
- }
- } else {
- lastValue = null;
- }
- return true;
- } else {
- eof = true;
- return false;
- }
- }
-
- private int readPageIfNeed() {
- // 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 type) {
- switch (type.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();
- }
- 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 type, Integer dictionaryValue) {
- if (dictionaryValue == null) {
- return null;
- }
-
- switch (type.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);
- }
- 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 type the element type of array
- * @return int
- */
- private int collectDataFromParquetPage(
- int total, HeapArrayVector lcv, List<Object> valueList, LogicalType type) {
- 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.getOffsets()[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(type, (Integer) lastValue)
- : lastValue);
- } while (fetchNextValue(type) && (repetitionLevel != 0));
-
- lcv.getLengths()[index] = valueList.size() - lcv.getOffsets()[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.getLengths(), 0, lcvLength, 0, elementNum);
- System.arraycopy(lcv.getOffsets(), 0, lcvOffset, 0, elementNum);
- lcv.setLengths(lcvLength);
- lcv.setOffsets(lcvOffset);
- }
-
- private void fillColumnVector(
- LogicalType type, HeapArrayVector lcv, List valueList, int elementNum) {
- int total = valueList.size();
- setChildrenInfo(lcv, total, elementNum);
- switch (type.getTypeRoot()) {
- case CHAR:
- case VARCHAR:
- case BINARY:
- case VARBINARY:
- HeapBytesVector bytesVector = new HeapBytesVector(total);
- bytesVector.reset();
- lcv.setChild(bytesVector);
- for (int i = 0; i < valueList.size(); i++) {
- byte[] src = (byte[]) valueList.get(i);
- if (src == null) {
- ((HeapBytesVector) lcv.getChild()).setNullAt(i);
- } else {
- ((HeapBytesVector) lcv.getChild()).appendBytes(i, src, 0, src.length);
- }
- }
- break;
- case BOOLEAN:
- HeapBooleanVector booleanVector = new HeapBooleanVector(total);
- booleanVector.reset();
- lcv.setChild(booleanVector);
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapBooleanVector) lcv.getChild()).setNullAt(i);
- } else {
- ((HeapBooleanVector) lcv.getChild()).vector[i] = (boolean) valueList.get(i);
- }
- }
- break;
- case TINYINT:
- HeapByteVector byteVector = new HeapByteVector(total);
- byteVector.reset();
- lcv.setChild(byteVector);
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapByteVector) lcv.getChild()).setNullAt(i);
- } else {
- ((HeapByteVector) lcv.getChild()).vector[i] =
- ((List<Integer>) valueList).get(i).byteValue();
- }
- }
- break;
- case SMALLINT:
- HeapShortVector shortVector = new HeapShortVector(total);
- shortVector.reset();
- lcv.setChild(shortVector);
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapShortVector) lcv.getChild()).setNullAt(i);
- } else {
- ((HeapShortVector) lcv.getChild()).vector[i] =
- ((List<Integer>) valueList).get(i).shortValue();
- }
- }
- break;
- case INTEGER:
- case DATE:
- case TIME_WITHOUT_TIME_ZONE:
- HeapIntVector intVector = new HeapIntVector(total);
- intVector.reset();
- lcv.setChild(intVector);
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapIntVector) lcv.getChild()).setNullAt(i);
- } else {
- ((HeapIntVector) lcv.getChild()).vector[i] =
- ((List<Integer>) valueList).get(i);
- }
- }
- break;
- case FLOAT:
- HeapFloatVector floatVector = new HeapFloatVector(total);
- floatVector.reset();
- lcv.setChild(floatVector);
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapFloatVector) lcv.getChild()).setNullAt(i);
- } else {
- ((HeapFloatVector) lcv.getChild()).vector[i] =
- ((List<Float>) valueList).get(i);
- }
- }
- break;
- case BIGINT:
- HeapLongVector longVector = new HeapLongVector(total);
- longVector.reset();
- lcv.setChild(longVector);
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapLongVector) lcv.getChild()).setNullAt(i);
- } else {
- ((HeapLongVector) lcv.getChild()).vector[i] =
- ((List<Long>) valueList).get(i);
- }
- }
- break;
- case DOUBLE:
- HeapDoubleVector doubleVector = new HeapDoubleVector(total);
- doubleVector.reset();
- lcv.setChild(doubleVector);
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapDoubleVector) lcv.getChild()).setNullAt(i);
- } else {
- ((HeapDoubleVector) lcv.getChild()).vector[i] =
- ((List<Double>) valueList).get(i);
- }
- }
- break;
- case TIMESTAMP_WITHOUT_TIME_ZONE:
- case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
- HeapTimestampVector timestampVector = new HeapTimestampVector(total);
- timestampVector.reset();
- lcv.setChild(timestampVector);
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapTimestampVector) lcv.getChild()).setNullAt(i);
- } else {
- ((HeapTimestampVector) lcv.getChild())
- .setTimestamp(i, ((List<TimestampData>) valueList).get(i));
- }
- }
- break;
- case DECIMAL:
- PrimitiveType.PrimitiveTypeName primitiveTypeName =
- descriptor.getPrimitiveType().getPrimitiveTypeName();
- switch (primitiveTypeName) {
- case INT32:
- HeapIntVector heapIntVector = new HeapIntVector(total);
- heapIntVector.reset();
- lcv.setChild(new ParquetDecimalVector(heapIntVector));
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapIntVector)
- ((ParquetDecimalVector) lcv.getChild()).getVector())
- .setNullAt(i);
- } else {
- ((HeapIntVector)
- ((ParquetDecimalVector) lcv.getChild())
- .getVector())
- .vector[i] =
- ((List<Integer>) valueList).get(i);
- }
- }
- break;
- case INT64:
- HeapLongVector heapLongVector = new HeapLongVector(total);
- heapLongVector.reset();
- lcv.setChild(new ParquetDecimalVector(heapLongVector));
- for (int i = 0; i < valueList.size(); i++) {
- if (valueList.get(i) == null) {
- ((HeapLongVector)
- ((ParquetDecimalVector) lcv.getChild()).getVector())
- .setNullAt(i);
- } else {
- ((HeapLongVector)
- ((ParquetDecimalVector) lcv.getChild())
- .getVector())
- .vector[i] =
- ((List<Long>) valueList).get(i);
- }
- }
- break;
- default:
- HeapBytesVector heapBytesVector = new HeapBytesVector(total);
- heapBytesVector.reset();
- lcv.setChild(new ParquetDecimalVector(heapBytesVector));
- for (int i = 0; i < valueList.size(); i++) {
- byte[] src = (byte[]) valueList.get(i);
- if (valueList.get(i) == null) {
- ((HeapBytesVector)
- ((ParquetDecimalVector) lcv.getChild()).getVector())
- .setNullAt(i);
- } else {
- ((HeapBytesVector)
- ((ParquetDecimalVector) lcv.getChild()).getVector())
- .appendBytes(i, src, 0, src.length);
- }
- }
- break;
- }
- break;
- default:
- throw new RuntimeException("Unsupported type in the list: " + type);
- }
- }
-}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/BaseVectorizedColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/BaseVectorizedColumnReader.java
deleted file mode 100644
index 3fe859fb849..00000000000
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/BaseVectorizedColumnReader.java
+++ /dev/null
@@ -1,290 +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.flink.formats.parquet.vector.reader;
-
-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.Type;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.ByteArrayInputStream;
-import java.io.IOException;
-
-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;
-
-/**
- * It's column level Parquet reader which is used to read a batch of records for a column, part of
- * the code is referred from Apache Hive and Apache Parquet.
- */
-public abstract class BaseVectorizedColumnReader implements ColumnReader<WritableColumnVector> {
-
- private static final Logger LOG = LoggerFactory.getLogger(BaseVectorizedColumnReader.class);
-
- protected boolean isUtcTimestamp;
-
- /** Total number of values read. */
- protected long valuesRead;
-
- /**
- * value that indicates the end of the current page. That is, if valuesRead ==
- * endOfPageValueCount, we are at the end of the page.
- */
- protected long endOfPageValueCount;
-
- /** The dictionary, if this column has dictionary encoding. */
- protected final ParquetDataColumnReader dictionary;
-
- /** If true, the current page is dictionary encoded. */
- protected boolean isCurrentPageDictionaryEncoded;
-
- /** Maximum definition level for this column. */
- protected final int maxDefLevel;
-
- protected int definitionLevel;
- protected int repetitionLevel;
-
- /** Repetition/Definition/Value readers. */
- protected IntIterator repetitionLevelColumn;
-
- protected IntIterator definitionLevelColumn;
- protected ParquetDataColumnReader dataColumn;
-
- /** Total values in the current page. */
- protected int pageValueCount;
-
- protected final PageReader pageReader;
- protected final ColumnDescriptor descriptor;
- protected final Type type;
- protected final LogicalType logicalType;
-
- public BaseVectorizedColumnReader(
- 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;
- }
- }
-
- protected void readRepetitionAndDefinitionLevels() {
- repetitionLevel = repetitionLevelColumn.nextInt();
- definitionLevel = definitionLevelColumn.nextInt();
- valuesRead++;
- }
-
- 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/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/MapColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/MapColumnReader.java
deleted file mode 100644
index 4a685ac6e9c..00000000000
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/MapColumnReader.java
+++ /dev/null
@@ -1,64 +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.flink.formats.parquet.vector.reader;
-
-import org.apache.flink.table.data.columnar.vector.ColumnVector;
-import org.apache.flink.table.data.columnar.vector.heap.HeapArrayVector;
-import org.apache.flink.table.data.columnar.vector.heap.HeapMapVector;
-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 ArrayColumnReader valueReader;
-
- public MapColumnReader(ArrayColumnReader keyReader, ArrayColumnReader valueReader) {
- this.keyReader = keyReader;
- this.valueReader = valueReader;
- }
-
- public void readBatch(int total, ColumnVector column) throws IOException {
- HeapMapVector mapVector = (HeapMapVector) column;
- // initialize 2 ListColumnVector for keys and values
- HeapArrayVector keyArrayColumnVector = new HeapArrayVector(total);
- HeapArrayVector valueArrayColumnVector = new HeapArrayVector(total);
- // read the keys and values
- keyReader.readToVector(total, keyArrayColumnVector);
- valueReader.readToVector(total, valueArrayColumnVector);
-
- // set the related attributes according to the keys and values
- mapVector.setKeys(keyArrayColumnVector.getChild());
- mapVector.setValues(valueArrayColumnVector.getChild());
- mapVector.setOffsets(keyArrayColumnVector.getOffsets());
- mapVector.setLengths(keyArrayColumnVector.getLengths());
- mapVector.setSize(keyArrayColumnVector.getSize());
- for (int i = 0; i < keyArrayColumnVector.getLen(); i++) {
- if (keyArrayColumnVector.isNullAt(i)) {
- mapVector.setNullAt(i);
- }
- }
- }
-
- @Override
- public void readToVector(int readNumber, WritableColumnVector vector) throws IOException {
- readBatch(readNumber, vector);
- }
-}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/NestedColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/NestedColumnReader.java
new file mode 100644
index 00000000000..0db4b402582
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/NestedColumnReader.java
@@ -0,0 +1,256 @@
+/*
+ * 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.flink.formats.parquet.vector.reader;
+
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.formats.parquet.utils.NestedPositionUtil;
+import org.apache.flink.formats.parquet.vector.position.CollectionPosition;
+import org.apache.flink.formats.parquet.vector.position.LevelDelegation;
+import org.apache.flink.formats.parquet.vector.position.RowPosition;
+import org.apache.flink.formats.parquet.vector.type.ParquetField;
+import org.apache.flink.formats.parquet.vector.type.ParquetGroupField;
+import org.apache.flink.formats.parquet.vector.type.ParquetPrimitiveField;
+import org.apache.flink.table.data.columnar.vector.ColumnVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapArrayVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapMapVector;
+import org.apache.flink.table.data.columnar.vector.heap.HeapRowVector;
+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;
+
+/**
+ * This ColumnReader mainly used to read `Group` type in parquet such as `Map`, `Array`, `Row`. The
+ * method about how to resolve nested struct mainly refer to : <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>Brief explanation of reading repetition and definition levels: Repetition level equal to 0
+ * means that this is the beginning of a new row. Other value means that we should add data to the
+ * current row.
+ *
+ * <p>For example, if we have the following data: repetition levels: 0,1,1,0,0,1,[0] (last 0 is
+ * implicit, normally will be the end of the page) values: a,b,c,d,e,f will consist of the sets of:
+ * (a, b, c), (d), (e, f). <br>
+ *
+ * <p>Definition levels contains 3 situations: level = maxDefLevel means value exist and is not null
+ * level = maxDefLevel - 1 means value is null level < maxDefLevel - 1 means value doesn't exist For
+ * non-nullable (REQUIRED) fields the (level = maxDefLevel - 1) condition means non-existing value
+ * as well. <br>
+ *
+ * <p>Quick example (maxDefLevel is 2): Read 3 rows out of: repetition levels: 0,1,0,1,1,0,0,...
+ * definition levels: 2,1,0,2,1,2,... values: a,b,c,d,e,f,... Resulting buffer: a,n, ,d,n,f that
+ * result is (a,n),(d,n),(f) where n means null
+ */
+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 {
+ HeapRowVector heapRowVector = (HeapRowVector) 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++) {
+ Tuple2<LevelDelegation, WritableColumnVector> tuple =
+ readData(children.get(i), 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 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 HeapRowVector(rowPosition.getPositionsCount(), finalChildrenVectors);
+ } else {
+ heapRowVector.setFields(finalChildrenVectors);
+ }
+
+ if (rowPosition.getIsNull() != null) {
+ setFieldNullFalg(rowPosition.getIsNull(), heapRowVector);
+ }
+ return Tuple2.of(levelDelegation, heapRowVector);
+ }
+
+ private Tuple2<LevelDelegation, WritableColumnVector> readMap(
+ ParquetGroupField field, int readNumber, ColumnVector vector, boolean inside)
+ throws IOException {
+ HeapMapVector mapVector = (HeapMapVector) 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 HeapMapVector(
+ collectionPosition.getValueCount(), keyTuple.f1, valueTuple.f1);
+ } else {
+ mapVector.setKeys(keyTuple.f1);
+ mapVector.setValues(valueTuple.f1);
+ }
+
+ if (collectionPosition.getIsNull() != null) {
+ setFieldNullFalg(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) {
+ setFieldNullFalg(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 setFieldNullFalg(boolean[] rowPosition, WritableColumnVector vector) {
+ int index = 0;
+ for (boolean isNullIndex : rowPosition) {
+ if (isNullIndex) {
+ vector.setNullAt(index);
+ }
+ index++;
+ }
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/NestedPrimitiveColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/NestedPrimitiveColumnReader.java
new file mode 100644
index 00000000000..efb71eeb7da
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/NestedPrimitiveColumnReader.java
@@ -0,0 +1,624 @@
+/*
+ * 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.flink.formats.parquet.vector.reader;
+
+import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
+import org.apache.flink.formats.parquet.vector.position.LevelDelegation;
+import org.apache.flink.runtime.util.IntArrayList;
+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 nested primitive column. */
+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;
+ }
+ }
+
+ // This won't call, will actually call readAndNewVector
+ @Override
+ public void readToVector(int readNumber, WritableColumnVector vector) throws IOException {
+ throw new UnsupportedOperationException("This function should no 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();
+ }
+ 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);
+ }
+ 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);
+ }
+ }
+
+ 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);
+ }
+ }
+
+ 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/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/RowColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/RowColumnReader.java
deleted file mode 100644
index 7fec09e568e..00000000000
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/RowColumnReader.java
+++ /dev/null
@@ -1,58 +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.flink.formats.parquet.vector.reader;
-
-import org.apache.flink.table.data.columnar.vector.heap.HeapRowVector;
-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 {
- HeapRowVector rowVector = (HeapRowVector) vector;
- WritableColumnVector[] vectors = rowVector.getFields();
- // 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
- rowVector.setNullAt(j);
- }
- }
- }
- }
-}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/type/ParquetField.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/type/ParquetField.java
new file mode 100644
index 00000000000..fd483daa45e
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/type/ParquetField.java
@@ -0,0 +1,67 @@
+/*
+ * 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.flink.formats.parquet.vector.type;
+
+import org.apache.flink.table.types.logical.LogicalType;
+
+/** Field that represent parquet's field type. */
+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/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/type/ParquetGroupField.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/type/ParquetGroupField.java
new file mode 100644
index 00000000000..ebe656ad014
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/type/ParquetGroupField.java
@@ -0,0 +1,47 @@
+/*
+ * 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.flink.formats.parquet.vector.type;
+
+import org.apache.flink.table.types.logical.LogicalType;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
+import static java.util.Objects.requireNonNull;
+
+/** Field that represent parquet's Group Field. */
+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);
+ this.children = ImmutableList.copyOf(requireNonNull(children, "children is null"));
+ }
+
+ public List<ParquetField> getChildren() {
+ return children;
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/type/ParquetPrimitiveField.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/type/ParquetPrimitiveField.java
new file mode 100644
index 00000000000..9393fc9276c
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/type/ParquetPrimitiveField.java
@@ -0,0 +1,51 @@
+/*
+ * 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.flink.formats.parquet.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. */
+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/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReaderTest.java b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReaderTest.java
index 96530186084..0e544759a95 100644
--- a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReaderTest.java
+++ b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReaderTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.parquet.ParquetWriterFactory;
import org.apache.flink.formats.parquet.row.ParquetRowDataBuilder;
+import org.apache.flink.formats.parquet.utils.ParquetSchemaConverter;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.GenericArrayData;
import org.apache.flink.table.data.GenericMapData;
@@ -32,6 +33,8 @@ import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.TimestampData;
import org.apache.flink.table.data.columnar.ColumnarRowData;
import org.apache.flink.table.data.columnar.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.conversion.RowRowConverter;
+import org.apache.flink.table.data.util.DataFormatConverters;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.ArrayType;
import org.apache.flink.table.types.logical.BigIntType;
@@ -51,10 +54,21 @@ import org.apache.flink.table.types.logical.TinyIntType;
import org.apache.flink.table.types.logical.VarCharType;
import org.apache.flink.table.types.utils.TypeConversions;
import org.apache.flink.table.utils.DateTimeUtils;
+import org.apache.flink.types.Row;
+
+import org.apache.flink.shaded.guava31.com.google.common.collect.Lists;
import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.example.data.Group;
+import org.apache.parquet.example.data.simple.SimpleGroupFactory;
+import org.apache.parquet.hadoop.ParquetFileWriter;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.example.ExampleParquetWriter;
+import org.apache.parquet.schema.MessageType;
+import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.io.TempDir;
import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.params.provider.MethodSource;
import java.io.File;
@@ -65,6 +79,7 @@ import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -119,6 +134,42 @@ class ParquetColumnarRowSplitReaderTest {
new MultisetType(new VarCharType(VarCharType.MAX_LENGTH)),
RowType.of(new VarCharType(VarCharType.MAX_LENGTH), new IntType()));
+ private static final RowType NESTED_ARRAY_MAP_TYPE =
+ RowType.of(
+ new IntType(),
+ new ArrayType(true, new IntType()),
+ new ArrayType(true, new ArrayType(true, new IntType())),
+ new ArrayType(
+ true,
+ new MapType(
+ true,
+ new VarCharType(VarCharType.MAX_LENGTH),
+ new VarCharType(VarCharType.MAX_LENGTH))),
+ new ArrayType(
+ true,
+ new RowType(
+ Collections.singletonList(
+ new RowType.RowField("a", new IntType())))),
+ RowType.of(
+ new ArrayType(
+ true,
+ new RowType(
+ Lists.newArrayList(
+ new RowType.RowField(
+ "b",
+ new ArrayType(
+ true,
+ new ArrayType(
+ true, new IntType()))),
+ new RowType.RowField("c", new IntType())))),
+ new IntType()));
+
+ @SuppressWarnings("unchecked")
+ private static final DataFormatConverters.DataFormatConverter<RowData, Row>
+ NESTED_ARRAY_MAP_CONVERTER =
+ DataFormatConverters.getConverterForDataType(
+ TypeConversions.fromLogicalToDataType(NESTED_ARRAY_MAP_TYPE));
+
@TempDir File tmpDir;
public static Collection<Integer> parameters() {
@@ -592,6 +643,45 @@ class ParquetColumnarRowSplitReaderTest {
innerTestPartitionValues(testPath, partSpec, true, rowGroupSize);
}
+ @ParameterizedTest
+ @CsvSource({"10, flink", "1000, flink", "10, origin", "1000, origin"})
+ public void testNestedRead(int rowGroupSize, String writerType) throws Exception {
+ List<Row> rows = prepareNestedData(1283);
+ Path path;
+ if ("flink".equals(writerType)) {
+ path = createNestedDataByFlinkWriter(rows, tmpDir, rowGroupSize);
+ } else if ("origin".equals(writerType)) {
+ path = createNestedDataByOriginWriter(1283, tmpDir, rowGroupSize);
+ } else {
+ throw new RuntimeException("Unknown writer type.");
+ }
+
+ ParquetColumnarRowSplitReader reader =
+ new ParquetColumnarRowSplitReader(
+ false,
+ true,
+ new Configuration(),
+ NESTED_ARRAY_MAP_TYPE.getChildren().toArray(new LogicalType[0]),
+ NESTED_ARRAY_MAP_TYPE.getFieldNames().toArray(new String[0]),
+ VectorizedColumnBatch::new,
+ 1000,
+ new org.apache.hadoop.fs.Path(path.getPath()),
+ 0,
+ path.getFileSystem().getFileStatus(path).getLen());
+
+ List<Row> results = new ArrayList<>(1000);
+ while (!reader.reachedEnd()) {
+ ColumnarRowData row = reader.nextRecord();
+ RowRowConverter rowRowConverter =
+ RowRowConverter.create(
+ TypeConversions.fromLogicalToDataType(NESTED_ARRAY_MAP_TYPE));
+ Row external = rowRowConverter.toExternal(row);
+ results.add(external);
+ }
+ reader.close();
+ Assertions.assertEquals(rows, results);
+ }
+
private void innerTestPartitionValues(
Path testPath, Map<String, Object> partSpec, boolean nullPartValue, int rowGroupSize)
throws IOException {
@@ -700,4 +790,148 @@ class ParquetColumnarRowSplitReaderTest {
}
reader.close();
}
+
+ private Path createNestedDataByFlinkWriter(List<Row> rows, File tmpDir, int rowGroupSize)
+ throws IOException {
+ Path path = new Path(tmpDir.getPath(), UUID.randomUUID().toString());
+ Configuration conf = new Configuration();
+ conf.setInt("parquet.block.size", rowGroupSize);
+
+ ParquetWriterFactory<RowData> factory =
+ ParquetRowDataBuilder.createWriterFactory(NESTED_ARRAY_MAP_TYPE, conf, false);
+ BulkWriter<RowData> writer =
+ factory.create(path.getFileSystem().create(path, FileSystem.WriteMode.OVERWRITE));
+ for (int i = 0; i < rows.size(); i++) {
+ writer.addElement(NESTED_ARRAY_MAP_CONVERTER.toInternal(rows.get(i)));
+ }
+ writer.flush();
+ writer.finish();
+
+ return path;
+ }
+
+ private List<Row> prepareNestedData(int rowNum) {
+ List<Row> rows = new ArrayList<>(rowNum);
+
+ for (int i = 0; i < rowNum; i++) {
+ Integer v = i;
+ Map<String, String> mp1 = new HashMap<>();
+ mp1.put(null, "val_" + i);
+ Map<String, String> mp2 = new HashMap<>();
+ mp2.put("key_" + i, null);
+ mp2.put("key@" + i, "val@" + i);
+
+ rows.add(
+ Row.of(
+ v,
+ new Integer[] {v, v + 1, null},
+ new Integer[][] {{i, i + 1, null}, {i, i + 2, null}, {}, null},
+ new Map[] {null, mp1, mp2},
+ new Row[] {Row.of(i), Row.of(i + 1), null},
+ Row.of(
+ new Row[] {
+ Row.of(
+ new Integer[][] {
+ {i, i + 1, null}, {i, i + 2, null}, {}, null
+ },
+ i),
+ null
+ },
+ i)));
+ }
+ return rows;
+ }
+
+ private Path createNestedDataByOriginWriter(int rowNum, File tmpDir, int rowGroupSize) {
+ Path path = new Path(tmpDir.getPath(), UUID.randomUUID().toString());
+ Configuration conf = new Configuration();
+ conf.setInt("parquet.block.size", rowGroupSize);
+ MessageType schema =
+ ParquetSchemaConverter.convertToParquetMessageType(
+ "flink-parquet", NESTED_ARRAY_MAP_TYPE, conf);
+ try (ParquetWriter<Group> writer =
+ ExampleParquetWriter.builder(new org.apache.hadoop.fs.Path(path.getPath()))
+ .withWriteMode(ParquetFileWriter.Mode.OVERWRITE)
+ .withConf(new Configuration())
+ .withType(schema)
+ .build()) {
+ SimpleGroupFactory simpleGroupFactory = new SimpleGroupFactory(schema);
+ for (int i = 0; i < rowNum; i++) {
+ Group row = simpleGroupFactory.newGroup();
+ // add int
+ row.append("f0", i);
+
+ // add array<int>
+ Group f1 = row.addGroup("f1");
+ createParquetArrayGroup(f1, i, i + 1);
+
+ // add array<array<int>>
+ Group f2 = row.addGroup("f2");
+ createParquetDoubleNestedArray(f2, i);
+
+ // add array<map>
+ Group f3 = row.addGroup("f3");
+ f3.addGroup(0);
+ Group mapList = f3.addGroup(0);
+ Group map1 = mapList.addGroup(0);
+ createParquetMapGroup(map1, null, "val_" + i);
+ Group map2 = mapList.addGroup(0);
+ createParquetMapGroup(map2, "key_" + i, null);
+ createParquetMapGroup(map2, "key@" + i, "val@" + i);
+
+ // add array<row>
+ Group f4 = row.addGroup("f4");
+ Group rowList = f4.addGroup(0);
+ Group row1 = rowList.addGroup(0);
+ row1.add(0, i);
+ Group row2 = rowList.addGroup(0);
+ row2.add(0, i + 1);
+ f4.addGroup(0);
+
+ // add ROW<`f0` ARRAY<ROW<`b` ARRAY<ARRAY<INT>>, `c` INT>>, `f1` INT>>
+ Group f5 = row.addGroup("f5");
+ Group arrayRow = f5.addGroup(0);
+ Group insideRow = arrayRow.addGroup(0).addGroup(0);
+ Group insideArray = insideRow.addGroup(0);
+ createParquetDoubleNestedArray(insideArray, i);
+ insideRow.add(1, i);
+ arrayRow.addGroup(0);
+ f5.add(1, i);
+ writer.write(row);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException("Create nested data by parquet origin writer failed.");
+ }
+ return path;
+ }
+
+ private void createParquetDoubleNestedArray(Group group, int i) {
+ Group outside = group.addGroup(0);
+ Group inside = outside.addGroup(0);
+ createParquetArrayGroup(inside, i, i + 1);
+ Group inside2 = outside.addGroup(0);
+ createParquetArrayGroup(inside2, i, i + 2);
+ // create empty array []
+ outside.addGroup(0);
+ // create null
+ group.addGroup(0);
+ }
+
+ private void createParquetArrayGroup(Group group, int i, int j) {
+ Group element = group.addGroup(0);
+ element.add(0, i);
+ element = group.addGroup(0);
+ element.add(0, j);
+ group.addGroup(0);
+ }
+
+ private void createParquetMapGroup(Group map, String key, String value) {
+ Group entry = map.addGroup(0);
+ if (key != null) {
+ entry.append("key", key);
+ }
+ if (value != null) {
+ entry.append("value", value);
+ }
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/util/BooleanArrayList.java b/flink-runtime/src/main/java/org/apache/flink/runtime/util/BooleanArrayList.java
new file mode 100644
index 00000000000..587d4ede68a
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/util/BooleanArrayList.java
@@ -0,0 +1,65 @@
+/*
+ * 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.flink.runtime.util;
+
+import java.util.Arrays;
+
+/** Minimal implementation of an array-backed list of booleans */
+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/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/vector/heap/HeapMapVector.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/vector/heap/HeapMapVector.java
index 8ec4216ea99..bb7b662d37e 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/vector/heap/HeapMapVector.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/vector/heap/HeapMapVector.java
@@ -74,4 +74,12 @@ public class HeapMapVector extends AbstractHeapVector
long length = lengths[i];
return new ColumnarMapData(keys, values, (int) offset, (int) length);
}
+
+ public ColumnVector getKeyColumnVector() {
+ return keys;
+ }
+
+ public ColumnVector getValueColumnVector() {
+ return values;
+ }
}
diff --git a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/vector/heap/HeapRowVector.java b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/vector/heap/HeapRowVector.java
index ab26fcf85fc..6a604c4dcd0 100644
--- a/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/vector/heap/HeapRowVector.java
+++ b/flink-table/flink-table-common/src/main/java/org/apache/flink/table/data/columnar/vector/heap/HeapRowVector.java
@@ -40,6 +40,10 @@ public class HeapRowVector extends AbstractHeapVector
return fields;
}
+ public void setFields(WritableColumnVector[] fields) {
+ this.fields = fields;
+ }
+
@Override
public ColumnarRowData getRow(int i) {
ColumnarRowData columnarRowData = new ColumnarRowData(new VectorizedColumnBatch(fields));