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 2020/03/06 01:30:35 UTC
[flink] branch master updated: [FLINK-11899][parquet] Introduce
parquet ColumnarRow split reader
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 1aabab8 [FLINK-11899][parquet] Introduce parquet ColumnarRow split reader
1aabab8 is described below
commit 1aabab86f9461ff557ef9ffe506bd67556988fc7
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Fri Mar 6 09:29:25 2020 +0800
[FLINK-11899][parquet] Introduce parquet ColumnarRow split reader
This closes #10922
---
flink-formats/flink-parquet/pom.xml | 8 +
.../flink/formats/parquet/ParquetTableSource.java | 4 +-
.../vector/ParquetColumnarRowSplitReader.java | 307 ++++++++++++
.../parquet/vector/ParquetDecimalVector.java | 58 +++
.../formats/parquet/vector/ParquetDictionary.java | 66 +++
.../parquet/vector/ParquetSplitReaderUtil.java | 384 +++++++++++++++
.../vector/reader/AbstractColumnReader.java | 318 ++++++++++++
.../parquet/vector/reader/BooleanColumnReader.java | 115 +++++
.../parquet/vector/reader/ByteColumnReader.java | 99 ++++
.../parquet/vector/reader/BytesColumnReader.java | 98 ++++
.../parquet/vector/reader/ColumnReader.java | 34 ++
.../parquet/vector/reader/DoubleColumnReader.java | 101 ++++
.../vector/reader/FixedLenBytesColumnReader.java | 144 ++++++
.../parquet/vector/reader/FloatColumnReader.java | 101 ++++
.../parquet/vector/reader/IntColumnReader.java | 100 ++++
.../parquet/vector/reader/LongColumnReader.java | 101 ++++
.../parquet/vector/reader/RunLengthDecoder.java | 293 +++++++++++
.../parquet/vector/reader/ShortColumnReader.java | 84 ++++
.../vector/reader/TimestampColumnReader.java | 120 +++++
.../formats/parquet/utils/ParquetWriterUtil.java | 166 +++++++
.../flink/formats/parquet/utils/TestUtil.java | 2 +-
.../vector/ParquetColumnarRowSplitReaderTest.java | 545 +++++++++++++++++++++
22 files changed, 3245 insertions(+), 3 deletions(-)
diff --git a/flink-formats/flink-parquet/pom.xml b/flink-formats/flink-parquet/pom.xml
index 5e0171d..b8a7c60 100644
--- a/flink-formats/flink-parquet/pom.xml
+++ b/flink-formats/flink-parquet/pom.xml
@@ -73,6 +73,14 @@ under the License.
<optional>true</optional>
</dependency>
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-table-runtime-blink_${scala.binary.version}</artifactId>
+ <version>${project.version}</version>
+ <scope>provided</scope>
+ <optional>true</optional>
+ </dependency>
+
<!-- Parquet Dependencies -->
<dependency>
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java
index a8b90f4..dd853ad 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetTableSource.java
@@ -81,12 +81,12 @@ import java.util.List;
*
* <pre>
* {@code
- * ParquetTableSource orcSrc = ParquetTableSource.builder()
+ * ParquetTableSource parquetSrc = ParquetTableSource.builder()
* .path("file:///my/data/file.parquet")
* .schema(messageType)
* .build();
*
- * tEnv.registerTableSource("parquetTable", orcSrc);
+ * tEnv.registerTableSource("parquetTable", parquetSrc);
* Table res = tableEnv.sqlQuery("SELECT * FROM parquetTable");
* }
* </pre>
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
new file mode 100644
index 0000000..e7bf0a7
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReader.java
@@ -0,0 +1,307 @@
+/*
+ * 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;
+
+import org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.dataformat.ColumnarRow;
+import org.apache.flink.table.dataformat.vector.ColumnVector;
+import org.apache.flink.table.dataformat.vector.VectorizedColumnBatch;
+import org.apache.flink.table.dataformat.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.parquet.column.ColumnDescriptor;
+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.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Types;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+
+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;
+import static org.apache.parquet.format.converter.ParquetMetadataConverter.range;
+import static org.apache.parquet.hadoop.ParquetFileReader.readFooter;
+import static org.apache.parquet.hadoop.ParquetInputFormat.getFilter;
+
+/**
+ * This reader is used to read a {@link VectorizedColumnBatch} from input split.
+ */
+public class ParquetColumnarRowSplitReader implements Closeable {
+
+ private final boolean utcTimestamp;
+
+ private final MessageType fileSchema;
+
+ private final MessageType requestedSchema;
+
+ /**
+ * The total number of rows this RecordReader will eventually read. The sum of the rows of all
+ * the row groups.
+ */
+ private final long totalRowCount;
+
+ private final WritableColumnVector[] writableVectors;
+
+ private final VectorizedColumnBatch columnarBatch;
+
+ private final ColumnarRow row;
+
+ private final LogicalType[] selectedTypes;
+
+ private final int batchSize;
+
+ private ParquetFileReader reader;
+
+ /**
+ * For each request column, the reader to read this column. This is NULL if this column is
+ * missing from the file, in which case we populate the attribute with NULL.
+ */
+ private ColumnReader[] columnReaders;
+
+ /**
+ * The number of rows that have been returned.
+ */
+ private long rowsReturned;
+
+ /**
+ * The number of rows that have been reading, including the current in flight row group.
+ */
+ private long totalCountLoadedSoFar;
+
+ // the index of the next row to return
+ private int nextRow;
+
+ // the number of rows in the current batch
+ private int rowsInBatch;
+
+ public ParquetColumnarRowSplitReader(
+ boolean utcTimestamp,
+ Configuration conf,
+ LogicalType[] selectedTypes,
+ String[] selectedFieldNames,
+ ColumnBatchGenerator generator,
+ int batchSize,
+ Path path,
+ long splitStart,
+ long splitLength) throws IOException {
+ this.utcTimestamp = utcTimestamp;
+ this.selectedTypes = selectedTypes;
+ this.batchSize = batchSize;
+ // then we need to apply the predicate push down filter
+ ParquetMetadata footer = readFooter(conf, path, range(splitStart, splitLength));
+ MessageType fileSchema = footer.getFileMetaData().getSchema();
+ FilterCompat.Filter filter = getFilter(conf);
+ List<BlockMetaData> blocks = filterRowGroups(filter, footer.getBlocks(), fileSchema);
+
+ this.fileSchema = footer.getFileMetaData().getSchema();
+ this.requestedSchema = clipParquetSchema(fileSchema, selectedFieldNames);
+ this.reader = new ParquetFileReader(
+ conf, footer.getFileMetaData(), path, blocks, requestedSchema.getColumns());
+
+ long totalRowCount = 0;
+ for (BlockMetaData block : blocks) {
+ totalRowCount += block.getRowCount();
+ }
+ this.totalRowCount = totalRowCount;
+ this.nextRow = 0;
+ this.rowsInBatch = 0;
+ this.rowsReturned = 0;
+
+ checkSchema();
+
+ this.writableVectors = createWritableVectors();
+ this.columnarBatch = generator.generate(createReadableVectors());
+ this.row = new ColumnarRow(columnarBatch);
+ }
+
+ /**
+ * Clips `parquetSchema` according to `fieldNames`.
+ */
+ private static MessageType clipParquetSchema(GroupType parquetSchema, String[] fieldNames) {
+ Type[] types = new Type[fieldNames.length];
+ for (int i = 0; i < fieldNames.length; ++i) {
+ String fieldName = fieldNames[i];
+ if (parquetSchema.getFieldIndex(fieldName) < 0) {
+ throw new IllegalArgumentException(fieldName + " does not exist");
+ }
+ types[i] = parquetSchema.getType(fieldName);
+ }
+ return Types.buildMessage().addFields(types).named("flink-parquet");
+ }
+
+ private WritableColumnVector[] createWritableVectors() {
+ WritableColumnVector[] columns = new WritableColumnVector[selectedTypes.length];
+ for (int i = 0; i < selectedTypes.length; i++) {
+ columns[i] = createWritableColumnVector(
+ batchSize,
+ selectedTypes[i],
+ requestedSchema.getColumns().get(i).getPrimitiveType());
+ }
+ return columns;
+ }
+
+ /**
+ * Create readable vectors from writable vectors.
+ * Especially for decimal, see {@link ParquetDecimalVector}.
+ */
+ private ColumnVector[] createReadableVectors() {
+ ColumnVector[] vectors = new ColumnVector[writableVectors.length];
+ for (int i = 0; i < writableVectors.length; i++) {
+ vectors[i] = selectedTypes[i].getTypeRoot() == LogicalTypeRoot.DECIMAL ?
+ new ParquetDecimalVector(writableVectors[i]) :
+ writableVectors[i];
+ }
+ return vectors;
+ }
+
+ private void checkSchema() throws IOException, UnsupportedOperationException {
+ if (selectedTypes.length != requestedSchema.getFieldCount()) {
+ throw new RuntimeException("The quality of field type is incompatible with the request schema!");
+ }
+
+ /*
+ * Check that the requested schema is supported.
+ */
+ for (int i = 0; i < requestedSchema.getFieldCount(); ++i) {
+ Type t = requestedSchema.getFields().get(i);
+ if (!t.isPrimitive() || t.isRepetition(Type.Repetition.REPEATED)) {
+ throw new UnsupportedOperationException("Complex types not supported.");
+ }
+
+ String[] colPath = requestedSchema.getPaths().get(i);
+ if (fileSchema.containsPath(colPath)) {
+ ColumnDescriptor fd = fileSchema.getColumnDescription(colPath);
+ if (!fd.equals(requestedSchema.getColumns().get(i))) {
+ throw new UnsupportedOperationException("Schema evolution not supported.");
+ }
+ } else {
+ if (requestedSchema.getColumns().get(i).getMaxDefinitionLevel() == 0) {
+ // Column is missing in data but the required data is non-nullable. This file is invalid.
+ throw new IOException("Required column is missing in data file. Col: " + Arrays.toString(colPath));
+ }
+ }
+ }
+ }
+
+ /**
+ * Method used to check if the end of the input is reached.
+ *
+ * @return True if the end is reached, otherwise false.
+ * @throws IOException Thrown, if an I/O error occurred.
+ */
+ public boolean reachedEnd() throws IOException {
+ return !ensureBatch();
+ }
+
+ public ColumnarRow nextRecord() {
+ // return the next row
+ row.setRowId(this.nextRow++);
+ return row;
+ }
+
+ /**
+ * Checks if there is at least one row left in the batch to return. If no more row are
+ * available, it reads another batch of rows.
+ *
+ * @return Returns true if there is one more row to return, false otherwise.
+ * @throws IOException throw if an exception happens while reading a batch.
+ */
+ private boolean ensureBatch() throws IOException {
+ if (nextRow >= rowsInBatch) {
+ // No more rows available in the Rows array.
+ nextRow = 0;
+ // Try to read the next batch if rows from the file.
+ return nextBatch();
+ }
+ // there is at least one Row left in the Rows array.
+ return true;
+ }
+
+ /**
+ * Advances to the next batch of rows. Returns false if there are no more.
+ */
+ private boolean nextBatch() throws IOException {
+ for (WritableColumnVector v : writableVectors) {
+ v.reset();
+ }
+ columnarBatch.setNumRows(0);
+ if (rowsReturned >= totalRowCount) {
+ return false;
+ }
+ if (rowsReturned == totalCountLoadedSoFar) {
+ readNextRowGroup();
+ }
+
+ int num = (int) Math.min(batchSize, totalCountLoadedSoFar - rowsReturned);
+ for (int i = 0; i < columnReaders.length; ++i) {
+ //noinspection unchecked
+ columnReaders[i].readToVector(num, writableVectors[i]);
+ }
+ rowsReturned += num;
+ columnarBatch.setNumRows(num);
+ rowsInBatch = num;
+ return true;
+ }
+
+ private void readNextRowGroup() throws IOException {
+ PageReadStore pages = reader.readNextRowGroup();
+ if (pages == null) {
+ throw new IOException("expecting more rows but reached last block. Read "
+ + rowsReturned + " out of " + totalRowCount);
+ }
+ List<ColumnDescriptor> columns = requestedSchema.getColumns();
+ columnReaders = new AbstractColumnReader[columns.size()];
+ for (int i = 0; i < columns.size(); ++i) {
+ columnReaders[i] = createColumnReader(
+ utcTimestamp,
+ selectedTypes[i],
+ columns.get(i),
+ pages.getPageReader(columns.get(i)));
+ }
+ totalCountLoadedSoFar += pages.getRowCount();
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (reader != null) {
+ reader.close();
+ reader = null;
+ }
+ }
+
+ /**
+ * Interface to gen {@link VectorizedColumnBatch}.
+ */
+ public interface ColumnBatchGenerator {
+ VectorizedColumnBatch generate(ColumnVector[] readVectors);
+ }
+}
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
new file mode 100644
index 0000000..288aed0
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetDecimalVector.java
@@ -0,0 +1,58 @@
+/*
+ * 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;
+
+import org.apache.flink.table.dataformat.Decimal;
+import org.apache.flink.table.dataformat.vector.BytesColumnVector;
+import org.apache.flink.table.dataformat.vector.ColumnVector;
+import org.apache.flink.table.dataformat.vector.DecimalColumnVector;
+import org.apache.flink.table.dataformat.vector.IntColumnVector;
+import org.apache.flink.table.dataformat.vector.LongColumnVector;
+
+/**
+ * Parquet write decimal as int32 and int64 and binary, this class wrap the real vector to
+ * provide {@link DecimalColumnVector} interface.
+ */
+public class ParquetDecimalVector implements DecimalColumnVector {
+
+ private final ColumnVector vector;
+
+ ParquetDecimalVector(ColumnVector vector) {
+ this.vector = vector;
+ }
+
+ @Override
+ public Decimal getDecimal(int i, int precision, int scale) {
+ if (Decimal.is32BitDecimal(precision)) {
+ return Decimal.fromUnscaledLong(
+ precision, scale, ((IntColumnVector) vector).getInt(i));
+ } else if (Decimal.is64BitDecimal(precision)) {
+ return Decimal.fromUnscaledLong(
+ precision, scale, ((LongColumnVector) vector).getLong(i));
+ } else {
+ return Decimal.fromUnscaledBytes(
+ precision, scale, ((BytesColumnVector) vector).getBytes(i).getBytes());
+ }
+ }
+
+ @Override
+ public boolean isNullAt(int i) {
+ return vector.isNullAt(i);
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetDictionary.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetDictionary.java
new file mode 100644
index 0000000..c93025e
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetDictionary.java
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.formats.parquet.vector;
+
+import org.apache.flink.table.dataformat.SqlTimestamp;
+import org.apache.flink.table.dataformat.vector.Dictionary;
+
+import static org.apache.flink.formats.parquet.vector.reader.TimestampColumnReader.decodeInt96ToTimestamp;
+
+/**
+ * Parquet dictionary.
+ */
+public final class ParquetDictionary implements Dictionary {
+
+ private org.apache.parquet.column.Dictionary dictionary;
+
+ public ParquetDictionary(org.apache.parquet.column.Dictionary dictionary) {
+ this.dictionary = dictionary;
+ }
+
+ @Override
+ public int decodeToInt(int id) {
+ return dictionary.decodeToInt(id);
+ }
+
+ @Override
+ public long decodeToLong(int id) {
+ return dictionary.decodeToLong(id);
+ }
+
+ @Override
+ public float decodeToFloat(int id) {
+ return dictionary.decodeToFloat(id);
+ }
+
+ @Override
+ public double decodeToDouble(int id) {
+ return dictionary.decodeToDouble(id);
+ }
+
+ @Override
+ public byte[] decodeToBinary(int id) {
+ return dictionary.decodeToBinary(id).getBytes();
+ }
+
+ @Override
+ public SqlTimestamp decodeToTimestamp(int id) {
+ return decodeInt96ToTimestamp(true, dictionary, id);
+ }
+}
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
new file mode 100644
index 0000000..dab7ff9
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/ParquetSplitReaderUtil.java
@@ -0,0 +1,384 @@
+/*
+ * 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;
+
+import org.apache.flink.core.fs.Path;
+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;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.DoubleColumnReader;
+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.ShortColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.TimestampColumnReader;
+import org.apache.flink.table.dataformat.Decimal;
+import org.apache.flink.table.dataformat.SqlTimestamp;
+import org.apache.flink.table.dataformat.vector.ColumnVector;
+import org.apache.flink.table.dataformat.vector.VectorizedColumnBatch;
+import org.apache.flink.table.dataformat.vector.heap.HeapBooleanVector;
+import org.apache.flink.table.dataformat.vector.heap.HeapByteVector;
+import org.apache.flink.table.dataformat.vector.heap.HeapBytesVector;
+import org.apache.flink.table.dataformat.vector.heap.HeapDoubleVector;
+import org.apache.flink.table.dataformat.vector.heap.HeapFloatVector;
+import org.apache.flink.table.dataformat.vector.heap.HeapIntVector;
+import org.apache.flink.table.dataformat.vector.heap.HeapLongVector;
+import org.apache.flink.table.dataformat.vector.heap.HeapShortVector;
+import org.apache.flink.table.dataformat.vector.heap.HeapTimestampVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.VarBinaryType;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.sql.Date;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.table.runtime.functions.SqlDateTimeUtils.dateToInternal;
+import static org.apache.parquet.Preconditions.checkArgument;
+
+/**
+ * Util for generating {@link ParquetColumnarRowSplitReader}.
+ */
+public class ParquetSplitReaderUtil {
+
+ /**
+ * Util for generating partitioned {@link ParquetColumnarRowSplitReader}.
+ */
+ public static ParquetColumnarRowSplitReader genPartColumnarRowReader(
+ boolean utcTimestamp,
+ Configuration conf,
+ String[] fullFieldNames,
+ DataType[] fullFieldTypes,
+ Map<String, Object> partitionSpec,
+ int[] selectedFields,
+ int batchSize,
+ Path path,
+ long splitStart,
+ long splitLength) throws IOException {
+ List<String> nonPartNames = Arrays.stream(fullFieldNames)
+ .filter(n -> !partitionSpec.containsKey(n))
+ .collect(Collectors.toList());
+
+ List<String> selNonPartNames = Arrays.stream(selectedFields)
+ .mapToObj(i -> fullFieldNames[i])
+ .filter(nonPartNames::contains).collect(Collectors.toList());
+
+ int[] selParquetFields = selNonPartNames.stream()
+ .mapToInt(nonPartNames::indexOf)
+ .toArray();
+
+ ParquetColumnarRowSplitReader.ColumnBatchGenerator gen = readVectors -> {
+ // create and initialize the row batch
+ ColumnVector[] vectors = new ColumnVector[selectedFields.length];
+ for (int i = 0; i < vectors.length; i++) {
+ String name = fullFieldNames[selectedFields[i]];
+ LogicalType type = fullFieldTypes[selectedFields[i]].getLogicalType();
+ vectors[i] = partitionSpec.containsKey(name) ?
+ createVectorFromConstant(type, partitionSpec.get(name), batchSize) :
+ readVectors[i];
+ }
+ return new VectorizedColumnBatch(vectors);
+ };
+
+ return new ParquetColumnarRowSplitReader(
+ utcTimestamp,
+ conf,
+ Arrays.stream(selParquetFields)
+ .mapToObj(i -> fullFieldTypes[i].getLogicalType())
+ .toArray(LogicalType[]::new),
+ selNonPartNames.toArray(new String[0]),
+ gen,
+ batchSize,
+ new org.apache.hadoop.fs.Path(path.toUri()),
+ splitStart,
+ splitLength);
+ }
+
+ private static ColumnVector createVectorFromConstant(
+ LogicalType type,
+ Object value,
+ int batchSize) {
+ switch (type.getTypeRoot()) {
+ case CHAR:
+ case VARCHAR:
+ case BINARY:
+ case VARBINARY:
+ HeapBytesVector bsv = new HeapBytesVector(batchSize);
+ if (value == null) {
+ bsv.fillWithNulls();
+ } else {
+ bsv.fill(value instanceof byte[] ?
+ (byte[]) value :
+ value.toString().getBytes(StandardCharsets.UTF_8));
+ }
+ return bsv;
+ case BOOLEAN:
+ HeapBooleanVector bv = new HeapBooleanVector(batchSize);
+ if (value == null) {
+ bv.fillWithNulls();
+ } else {
+ bv.fill((boolean) value);
+ }
+ return bv;
+ case TINYINT:
+ HeapByteVector byteVector = new HeapByteVector(batchSize);
+ if (value == null) {
+ byteVector.fillWithNulls();
+ } else {
+ byteVector.fill(((Number) value).byteValue());
+ }
+ return byteVector;
+ case SMALLINT:
+ HeapShortVector sv = new HeapShortVector(batchSize);
+ if (value == null) {
+ sv.fillWithNulls();
+ } else {
+ sv.fill(((Number) value).shortValue());
+ }
+ return sv;
+ case INTEGER:
+ HeapIntVector iv = new HeapIntVector(batchSize);
+ if (value == null) {
+ iv.fillWithNulls();
+ } else {
+ iv.fill(((Number) value).intValue());
+ }
+ return iv;
+ case BIGINT:
+ HeapLongVector lv = new HeapLongVector(batchSize);
+ if (value == null) {
+ lv.fillWithNulls();
+ } else {
+ lv.fill(((Number) value).longValue());
+ }
+ return lv;
+ case DECIMAL:
+ DecimalType decimalType = (DecimalType) type;
+ int precision = decimalType.getPrecision();
+ int scale = decimalType.getScale();
+ Decimal decimal = value == null ? null : Preconditions.checkNotNull(
+ Decimal.fromBigDecimal((BigDecimal) value, precision, scale));
+ ColumnVector internalVector;
+ if (Decimal.is32BitDecimal(precision)) {
+ internalVector = createVectorFromConstant(
+ new IntType(),
+ decimal == null ? null : (int) decimal.toUnscaledLong(),
+ batchSize);
+ } else if (Decimal.is64BitDecimal(precision)) {
+ internalVector = createVectorFromConstant(
+ new BigIntType(),
+ decimal == null ? null : decimal.toUnscaledLong(),
+ batchSize);
+ } else {
+ internalVector = createVectorFromConstant(
+ new VarBinaryType(),
+ decimal == null ? null : decimal.toUnscaledBytes(),
+ batchSize);
+ }
+ return new ParquetDecimalVector(internalVector);
+ case FLOAT:
+ HeapFloatVector fv = new HeapFloatVector(batchSize);
+ if (value == null) {
+ fv.fillWithNulls();
+ } else {
+ fv.fill(((Number) value).floatValue());
+ }
+ return fv;
+ case DOUBLE:
+ HeapDoubleVector dv = new HeapDoubleVector(batchSize);
+ if (value == null) {
+ dv.fillWithNulls();
+ } else {
+ dv.fill(((Number) value).doubleValue());
+ }
+ return dv;
+ case DATE:
+ if (value instanceof LocalDate) {
+ value = Date.valueOf((LocalDate) value);
+ }
+ return createVectorFromConstant(
+ new IntType(),
+ value == null ? null : dateToInternal((Date) value),
+ batchSize);
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ HeapTimestampVector tv = new HeapTimestampVector(batchSize);
+ if (value == null) {
+ tv.fillWithNulls();
+ } else {
+ tv.fill(SqlTimestamp.fromLocalDateTime((LocalDateTime) value));
+ }
+ return tv;
+ default:
+ throw new UnsupportedOperationException("Unsupported type: " + type);
+ }
+ }
+
+ public static ColumnReader createColumnReader(
+ boolean utcTimestamp,
+ LogicalType fieldType,
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ switch (fieldType.getTypeRoot()) {
+ case BOOLEAN:
+ return new BooleanColumnReader(descriptor, pageReader);
+ case TINYINT:
+ return new ByteColumnReader(descriptor, pageReader);
+ case DOUBLE:
+ return new DoubleColumnReader(descriptor, pageReader);
+ case FLOAT:
+ return new FloatColumnReader(descriptor, pageReader);
+ case INTEGER:
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ return new IntColumnReader(descriptor, pageReader);
+ case BIGINT:
+ return new LongColumnReader(descriptor, pageReader);
+ case SMALLINT:
+ return new ShortColumnReader(descriptor, pageReader);
+ case CHAR:
+ case VARCHAR:
+ case BINARY:
+ case VARBINARY:
+ return new BytesColumnReader(descriptor, pageReader);
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ return new TimestampColumnReader(utcTimestamp, descriptor, pageReader);
+ case DECIMAL:
+ switch (descriptor.getPrimitiveType().getPrimitiveTypeName()) {
+ case INT32:
+ return new IntColumnReader(descriptor, pageReader);
+ case INT64:
+ return new LongColumnReader(descriptor, pageReader);
+ case BINARY:
+ return new BytesColumnReader(descriptor, pageReader);
+ case FIXED_LEN_BYTE_ARRAY:
+ return new FixedLenBytesColumnReader(
+ descriptor, pageReader, ((DecimalType) fieldType).getPrecision());
+ }
+ default:
+ throw new UnsupportedOperationException(fieldType + " is not supported now.");
+ }
+ }
+
+ public static WritableColumnVector createWritableColumnVector(
+ int batchSize,
+ LogicalType fieldType,
+ PrimitiveType primitiveType) {
+ PrimitiveType.PrimitiveTypeName typeName = primitiveType.getPrimitiveTypeName();
+ switch (fieldType.getTypeRoot()) {
+ case BOOLEAN:
+ checkArgument(
+ typeName == PrimitiveType.PrimitiveTypeName.BOOLEAN,
+ "Unexpected type: %s", typeName);
+ return new HeapBooleanVector(batchSize);
+ case TINYINT:
+ checkArgument(
+ typeName == PrimitiveType.PrimitiveTypeName.INT32,
+ "Unexpected type: %s", typeName);
+ return new HeapByteVector(batchSize);
+ case DOUBLE:
+ checkArgument(
+ typeName == PrimitiveType.PrimitiveTypeName.DOUBLE,
+ "Unexpected type: %s", typeName);
+ return new HeapDoubleVector(batchSize);
+ case FLOAT:
+ checkArgument(
+ typeName == PrimitiveType.PrimitiveTypeName.FLOAT,
+ "Unexpected type: %s", typeName);
+ return new HeapFloatVector(batchSize);
+ case INTEGER:
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ checkArgument(
+ typeName == PrimitiveType.PrimitiveTypeName.INT32,
+ "Unexpected type: %s", typeName);
+ return new HeapIntVector(batchSize);
+ case BIGINT:
+ checkArgument(
+ typeName == PrimitiveType.PrimitiveTypeName.INT64,
+ "Unexpected type: %s", typeName);
+ return new HeapLongVector(batchSize);
+ case SMALLINT:
+ checkArgument(
+ typeName == PrimitiveType.PrimitiveTypeName.INT32,
+ "Unexpected type: %s", typeName);
+ return new HeapShortVector(batchSize);
+ case CHAR:
+ case VARCHAR:
+ case BINARY:
+ case VARBINARY:
+ checkArgument(
+ typeName == PrimitiveType.PrimitiveTypeName.BINARY,
+ "Unexpected type: %s", typeName);
+ return new HeapBytesVector(batchSize);
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case TIMESTAMP_WITH_LOCAL_TIME_ZONE:
+ checkArgument(
+ typeName == PrimitiveType.PrimitiveTypeName.INT96,
+ "Unexpected type: %s", typeName);
+ return new HeapTimestampVector(batchSize);
+ case DECIMAL:
+ DecimalType decimalType = (DecimalType) fieldType;
+ if (Decimal.is32BitDecimal(decimalType.getPrecision())) {
+ checkArgument(
+ (typeName == PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY ||
+ typeName == PrimitiveType.PrimitiveTypeName.INT32) &&
+ primitiveType.getOriginalType() == OriginalType.DECIMAL,
+ "Unexpected type: %s", typeName);
+ return new HeapIntVector(batchSize);
+ } else if (Decimal.is64BitDecimal(decimalType.getPrecision())) {
+ checkArgument(
+ (typeName == PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY ||
+ typeName == PrimitiveType.PrimitiveTypeName.INT64) &&
+ primitiveType.getOriginalType() == OriginalType.DECIMAL,
+ "Unexpected type: %s", typeName);
+ return new HeapLongVector(batchSize);
+ } else {
+ checkArgument(
+ (typeName == PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY ||
+ typeName == PrimitiveType.PrimitiveTypeName.BINARY) &&
+ primitiveType.getOriginalType() == OriginalType.DECIMAL,
+ "Unexpected type: %s", typeName);
+ return new HeapBytesVector(batchSize);
+ }
+ default:
+ throw new UnsupportedOperationException(fieldType + " is not supported now.");
+ }
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/AbstractColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/AbstractColumnReader.java
new file mode 100644
index 0000000..786f546
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/AbstractColumnReader.java
@@ -0,0 +1,318 @@
+/*
+ * 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.ParquetDictionary;
+import org.apache.flink.table.dataformat.vector.writable.WritableColumnVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableIntVector;
+
+import org.apache.parquet.Preconditions;
+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.Dictionary;
+import org.apache.parquet.column.Encoding;
+import org.apache.parquet.column.impl.ColumnReaderImpl;
+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.io.ParquetDecodingException;
+import org.apache.parquet.schema.PrimitiveType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+
+import static org.apache.parquet.column.ValuesType.REPETITION_LEVEL;
+
+/**
+ * Abstract {@link ColumnReader}.
+ * See {@link ColumnReaderImpl}, part of the code is referred from Apache Spark and Apache Parquet.
+ */
+public abstract class AbstractColumnReader<VECTOR extends WritableColumnVector>
+ implements ColumnReader<VECTOR> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AbstractColumnReader.class);
+
+ private final PageReader pageReader;
+
+ /**
+ * The dictionary, if this column has dictionary encoding.
+ */
+ protected final Dictionary dictionary;
+
+ /**
+ * Maximum definition level for this column.
+ */
+ protected final int maxDefLevel;
+
+ protected final ColumnDescriptor descriptor;
+
+ /**
+ * 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;
+
+ /**
+ * Total values in the current page.
+ */
+ private int pageValueCount;
+
+ /*
+ * Input streams:
+ * 1.Run length encoder to encode every data, so we have run length stream to get
+ * run length information.
+ * 2.Data maybe is real data, maybe is dictionary ids which need be decode to real
+ * data from Dictionary.
+ *
+ * Run length stream ------> Data stream
+ * |
+ * ------> Dictionary ids stream
+ */
+
+ /**
+ * Run length decoder for data and dictionary.
+ */
+ protected RunLengthDecoder runLenDecoder;
+
+ /**
+ * Data input stream.
+ */
+ ByteBufferInputStream dataInputStream;
+
+ /**
+ * Dictionary decoder to wrap dictionary ids input stream.
+ */
+ private RunLengthDecoder dictionaryIdsDecoder;
+
+ public AbstractColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ this.descriptor = descriptor;
+ this.pageReader = pageReader;
+ this.maxDefLevel = descriptor.getMaxDefinitionLevel();
+
+ DictionaryPage dictionaryPage = pageReader.readDictionaryPage();
+ if (dictionaryPage != null) {
+ try {
+ this.dictionary = dictionaryPage.getEncoding().initDictionary(descriptor, dictionaryPage);
+ this.isCurrentPageDictionaryEncoded = true;
+ } catch (IOException e) {
+ throw new IOException("could not decode the dictionary for " + descriptor, e);
+ }
+ } else {
+ this.dictionary = null;
+ this.isCurrentPageDictionaryEncoded = false;
+ }
+ /*
+ * Total number of values in this column (in this row group).
+ */
+ long totalValueCount = pageReader.getTotalValueCount();
+ if (totalValueCount == 0) {
+ throw new IOException("totalValueCount == 0");
+ }
+ }
+
+ protected void checkTypeName(PrimitiveType.PrimitiveTypeName expectedName) {
+ PrimitiveType.PrimitiveTypeName actualName = descriptor.getPrimitiveType().getPrimitiveTypeName();
+ Preconditions.checkArgument(
+ actualName == expectedName,
+ "Expected type name: %s, actual type name: %s",
+ expectedName,
+ actualName);
+ }
+
+ /**
+ * Reads `total` values from this columnReader into column.
+ */
+ @Override
+ public final void readToVector(int readNumber, VECTOR vector) throws IOException {
+ int rowId = 0;
+ WritableIntVector dictionaryIds = null;
+ if (dictionary != null) {
+ dictionaryIds = vector.reserveDictionaryIds(readNumber);
+ }
+ while (readNumber > 0) {
+ // Compute the number of values we want to read in this page.
+ int leftInPage = (int) (endOfPageValueCount - valuesRead);
+ if (leftInPage == 0) {
+ DataPage page = pageReader.readPage();
+ if (page instanceof DataPageV1) {
+ readPageV1((DataPageV1) page);
+ } else if (page instanceof DataPageV2) {
+ readPageV2((DataPageV2) page);
+ } else {
+ throw new RuntimeException("Unsupported page type: " + page.getClass());
+ }
+ leftInPage = (int) (endOfPageValueCount - valuesRead);
+ }
+ int num = Math.min(readNumber, leftInPage);
+ if (isCurrentPageDictionaryEncoded) {
+ // Read and decode dictionary ids.
+ runLenDecoder.readDictionaryIds(
+ num, dictionaryIds, vector, rowId, maxDefLevel, this.dictionaryIdsDecoder);
+
+ if (vector.hasDictionary() || (rowId == 0 && supportLazyDecode())) {
+ // Column vector supports lazy decoding of dictionary values so just set the dictionary.
+ // We can't do this if rowId != 0 AND the column doesn't have a dictionary (i.e. some
+ // non-dictionary encoded values have already been added).
+ vector.setDictionary(new ParquetDictionary(dictionary));
+ } else {
+ readBatchFromDictionaryIds(rowId, num, vector, dictionaryIds);
+ }
+ } else {
+ if (vector.hasDictionary() && rowId != 0) {
+ // This batch already has dictionary encoded values but this new page is not. The batch
+ // does not support a mix of dictionary and not so we will decode the dictionary.
+ readBatchFromDictionaryIds(0, rowId, vector, vector.getDictionaryIds());
+ }
+ vector.setDictionary(null);
+ readBatch(rowId, num, vector);
+ }
+
+ valuesRead += num;
+ rowId += num;
+ readNumber -= num;
+ }
+ }
+
+ private void readPageV1(DataPageV1 page) throws IOException {
+ this.pageValueCount = page.getValueCount();
+ ValuesReader rlReader = page.getRlEncoding().getValuesReader(descriptor, REPETITION_LEVEL);
+
+ // Initialize the decoders.
+ if (page.getDlEncoding() != Encoding.RLE && descriptor.getMaxDefinitionLevel() != 0) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + page.getDlEncoding());
+ }
+ int bitWidth = BytesUtils.getWidthFromMaxInt(descriptor.getMaxDefinitionLevel());
+ this.runLenDecoder = new RunLengthDecoder(bitWidth);
+ try {
+ BytesInput bytes = page.getBytes();
+ ByteBufferInputStream in = bytes.toInputStream();
+ rlReader.initFromPage(pageValueCount, in);
+ this.runLenDecoder.initFromStream(pageValueCount, in);
+ prepareNewPage(page.getValueEncoding(), in);
+ } catch (IOException e) {
+ throw new IOException("could not read page " + page + " in col " + descriptor, e);
+ }
+ }
+
+ private void readPageV2(DataPageV2 page) throws IOException {
+ this.pageValueCount = page.getValueCount();
+
+ int bitWidth = BytesUtils.getWidthFromMaxInt(descriptor.getMaxDefinitionLevel());
+ // do not read the length from the stream. v2 pages handle dividing the page bytes.
+ this.runLenDecoder = new RunLengthDecoder(bitWidth, false);
+ this.runLenDecoder.initFromStream(
+ this.pageValueCount, page.getDefinitionLevels().toInputStream());
+ try {
+ prepareNewPage(page.getDataEncoding(), page.getData().toInputStream());
+ } catch (IOException e) {
+ throw new IOException("could not read page " + page + " in col " + descriptor, e);
+ }
+ }
+
+ private void prepareNewPage(
+ Encoding dataEncoding,
+ ByteBufferInputStream in) throws IOException {
+ this.endOfPageValueCount = valuesRead + pageValueCount;
+ if (dataEncoding.usesDictionary()) {
+ if (dictionary == null) {
+ throw new IOException(
+ "could not read page in col " + descriptor +
+ " as the dictionary was missing for encoding " + dataEncoding);
+ }
+ @SuppressWarnings("deprecation")
+ Encoding plainDict = Encoding.PLAIN_DICTIONARY; // var to allow warning suppression
+ if (dataEncoding != plainDict && dataEncoding != Encoding.RLE_DICTIONARY) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + dataEncoding);
+ }
+ this.dataInputStream = null;
+ this.dictionaryIdsDecoder = new RunLengthDecoder();
+ try {
+ this.dictionaryIdsDecoder.initFromStream(pageValueCount, in);
+ } catch (IOException e) {
+ throw new IOException("could not read dictionary in col " + descriptor, e);
+ }
+ this.isCurrentPageDictionaryEncoded = true;
+ } else {
+ if (dataEncoding != Encoding.PLAIN) {
+ throw new UnsupportedOperationException("Unsupported encoding: " + dataEncoding);
+ }
+ this.dictionaryIdsDecoder = null;
+ LOG.debug("init from page at offset {} for length {}", in.position(), in.available());
+ this.dataInputStream = in.remainingStream();
+ this.isCurrentPageDictionaryEncoded = false;
+ }
+
+ afterReadPage();
+ }
+
+ final ByteBuffer readDataBuffer(int length) {
+ try {
+ return dataInputStream.slice(length).order(ByteOrder.LITTLE_ENDIAN);
+ } catch (IOException e) {
+ throw new ParquetDecodingException("Failed to read " + length + " bytes", e);
+ }
+ }
+
+ /**
+ * After read a page, we may need some initialization.
+ */
+ protected void afterReadPage() {}
+
+ /**
+ * Support lazy dictionary ids decode. See more in {@link ParquetDictionary}.
+ * If return false, we will decode all the data first.
+ */
+ protected boolean supportLazyDecode() {
+ return true;
+ }
+
+ /**
+ * Read batch from {@link #runLenDecoder} and {@link #dataInputStream}.
+ */
+ protected abstract void readBatch(int rowId, int num, VECTOR column);
+
+ /**
+ * Decode dictionary ids to data.
+ * From {@link #runLenDecoder} and {@link #dictionaryIdsDecoder}.
+ */
+ protected abstract void readBatchFromDictionaryIds(
+ int rowId,
+ int num,
+ VECTOR column,
+ WritableIntVector dictionaryIds);
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/BooleanColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/BooleanColumnReader.java
new file mode 100644
index 0000000..504a504
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/BooleanColumnReader.java
@@ -0,0 +1,115 @@
+/*
+ * 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.dataformat.vector.writable.WritableBooleanVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableIntVector;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.io.ParquetDecodingException;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.IOException;
+
+/**
+ * Boolean {@link ColumnReader}.
+ */
+public class BooleanColumnReader extends AbstractColumnReader<WritableBooleanVector> {
+
+ /**
+ * Parquet use a bit to store booleans, so we need split a byte to 8 boolean.
+ */
+ private int bitOffset;
+ private byte currentByte = 0;
+
+ public BooleanColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ super(descriptor, pageReader);
+ checkTypeName(PrimitiveType.PrimitiveTypeName.BOOLEAN);
+ }
+
+ @Override
+ protected boolean supportLazyDecode() {
+ return true;
+ }
+
+ @Override
+ protected void afterReadPage() {
+ bitOffset = 0;
+ currentByte = 0;
+ }
+
+ @Override
+ protected void readBatchFromDictionaryIds(int rowId, int num, WritableBooleanVector column,
+ WritableIntVector dictionaryIds) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ protected void readBatch(int rowId, int num, WritableBooleanVector column) {
+ int left = num;
+ while (left > 0) {
+ if (runLenDecoder.currentCount == 0) {
+ runLenDecoder.readNextGroup();
+ }
+ int n = Math.min(left, runLenDecoder.currentCount);
+ switch (runLenDecoder.mode) {
+ case RLE:
+ if (runLenDecoder.currentValue == maxDefLevel) {
+ for (int i = 0; i < n; i++) {
+ column.setBoolean(rowId + i, readBoolean());
+ }
+ } else {
+ column.setNulls(rowId, n);
+ }
+ break;
+ case PACKED:
+ for (int i = 0; i < n; ++i) {
+ if (runLenDecoder.currentBuffer[runLenDecoder.currentBufferIdx++] == maxDefLevel) {
+ column.setBoolean(rowId + i, readBoolean());
+ } else {
+ column.setNullAt(rowId + i);
+ }
+ }
+ break;
+ }
+ rowId += n;
+ left -= n;
+ runLenDecoder.currentCount -= n;
+ }
+ }
+
+ private boolean readBoolean() {
+ if (bitOffset == 0) {
+ try {
+ currentByte = (byte) dataInputStream.read();
+ } catch (IOException e) {
+ throw new ParquetDecodingException("Failed to read a byte", e);
+ }
+ }
+
+ boolean v = (currentByte & (1 << bitOffset)) != 0;
+ bitOffset += 1;
+ if (bitOffset == 8) {
+ bitOffset = 0;
+ }
+ return v;
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/ByteColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/ByteColumnReader.java
new file mode 100644
index 0000000..ada3405
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/ByteColumnReader.java
@@ -0,0 +1,99 @@
+/*
+ * 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.dataformat.vector.writable.WritableByteVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableIntVector;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Byte {@link ColumnReader}. Using INT32 to store byte, so just cast int to byte.
+ */
+public class ByteColumnReader extends AbstractColumnReader<WritableByteVector> {
+
+ public ByteColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ super(descriptor, pageReader);
+ checkTypeName(PrimitiveType.PrimitiveTypeName.INT32);
+ }
+
+ @Override
+ protected void readBatch(int rowId, int num, WritableByteVector column) {
+ int left = num;
+ while (left > 0) {
+ if (runLenDecoder.currentCount == 0) {
+ runLenDecoder.readNextGroup();
+ }
+ int n = Math.min(left, runLenDecoder.currentCount);
+ switch (runLenDecoder.mode) {
+ case RLE:
+ if (runLenDecoder.currentValue == maxDefLevel) {
+ readBytes(n, column, rowId);
+ } else {
+ column.setNulls(rowId, n);
+ }
+ break;
+ case PACKED:
+ for (int i = 0; i < n; ++i) {
+ if (runLenDecoder.currentBuffer[runLenDecoder.currentBufferIdx++] == maxDefLevel) {
+ column.setByte(rowId + i, readByte());
+ } else {
+ column.setNullAt(rowId + i);
+ }
+ }
+ break;
+ }
+ rowId += n;
+ left -= n;
+ runLenDecoder.currentCount -= n;
+ }
+ }
+
+ @Override
+ protected void readBatchFromDictionaryIds(int rowId, int num, WritableByteVector column,
+ WritableIntVector dictionaryIds) {
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!column.isNullAt(i)) {
+ column.setByte(i, (byte) dictionary.decodeToInt(dictionaryIds.getInt(i)));
+ }
+ }
+ }
+
+ private byte readByte() {
+ return (byte) readDataBuffer(4).getInt();
+ }
+
+ private void readBytes(int total, WritableByteVector c, int rowId) {
+ // Bytes are stored as a 4-byte little endian int. Just read the first byte.
+ int requiredBytes = total * 4;
+ ByteBuffer buffer = readDataBuffer(requiredBytes);
+
+ for (int i = 0; i < total; i += 1) {
+ c.setByte(rowId + i, buffer.get());
+ // skip the next 3 bytes
+ buffer.position(buffer.position() + 3);
+ }
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/BytesColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/BytesColumnReader.java
new file mode 100644
index 0000000..3f6a135
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/BytesColumnReader.java
@@ -0,0 +1,98 @@
+/*
+ * 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.dataformat.vector.writable.WritableBytesVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableIntVector;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Bytes {@link ColumnReader}. A int length and bytes data.
+ */
+public class BytesColumnReader extends AbstractColumnReader<WritableBytesVector> {
+
+ public BytesColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ super(descriptor, pageReader);
+ checkTypeName(PrimitiveType.PrimitiveTypeName.BINARY);
+ }
+
+ @Override
+ protected void readBatch(int rowId, int num, WritableBytesVector column) {
+ int left = num;
+ while (left > 0) {
+ if (runLenDecoder.currentCount == 0) {
+ runLenDecoder.readNextGroup();
+ }
+ int n = Math.min(left, runLenDecoder.currentCount);
+ switch (runLenDecoder.mode) {
+ case RLE:
+ if (runLenDecoder.currentValue == maxDefLevel) {
+ readBinary(n, column, rowId);
+ } else {
+ column.setNulls(rowId, n);
+ }
+ break;
+ case PACKED:
+ for (int i = 0; i < n; ++i) {
+ if (runLenDecoder.currentBuffer[runLenDecoder.currentBufferIdx++] == maxDefLevel) {
+ readBinary(1, column, rowId + i);
+ } else {
+ column.setNullAt(rowId + i);
+ }
+ }
+ break;
+ }
+ rowId += n;
+ left -= n;
+ runLenDecoder.currentCount -= n;
+ }
+ }
+
+ @Override
+ protected void readBatchFromDictionaryIds(int rowId, int num, WritableBytesVector column,
+ WritableIntVector dictionaryIds) {
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!column.isNullAt(i)) {
+ byte[] bytes = dictionary.decodeToBinary(dictionaryIds.getInt(i)).getBytes();
+ column.appendBytes(i, bytes, 0, bytes.length);
+ }
+ }
+ }
+
+ private void readBinary(int total, WritableBytesVector v, int rowId) {
+ for (int i = 0; i < total; i++) {
+ int len = readDataBuffer(4).getInt();
+ ByteBuffer buffer = readDataBuffer(len);
+ if (buffer.hasArray()) {
+ v.appendBytes(rowId + i, buffer.array(), buffer.arrayOffset() + buffer.position(), len);
+ } else {
+ byte[] bytes = new byte[len];
+ buffer.get(bytes);
+ v.appendBytes(rowId + i, bytes, 0, bytes.length);
+ }
+ }
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/ColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/ColumnReader.java
new file mode 100644
index 0000000..f05dc6c
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/ColumnReader.java
@@ -0,0 +1,34 @@
+/*
+ * 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.dataformat.vector.writable.WritableColumnVector;
+
+import java.io.IOException;
+
+/**
+ * Read a batch of records for a column to {@link WritableColumnVector} from parquet data file.
+ */
+public interface ColumnReader<VECTOR extends WritableColumnVector> {
+
+ /**
+ * @param readNumber number to read.
+ * @param vector vector to write.
+ */
+ void readToVector(int readNumber, VECTOR vector) throws IOException;
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/DoubleColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/DoubleColumnReader.java
new file mode 100644
index 0000000..4999c72
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/DoubleColumnReader.java
@@ -0,0 +1,101 @@
+/*
+ * 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.dataformat.vector.writable.WritableDoubleVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableIntVector;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Double {@link ColumnReader}.
+ */
+public class DoubleColumnReader extends AbstractColumnReader<WritableDoubleVector> {
+
+ public DoubleColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ super(descriptor, pageReader);
+ checkTypeName(PrimitiveType.PrimitiveTypeName.DOUBLE);
+ }
+
+ @Override
+ protected void readBatch(int rowId, int num, WritableDoubleVector column) {
+ int left = num;
+ while (left > 0) {
+ if (runLenDecoder.currentCount == 0) {
+ runLenDecoder.readNextGroup();
+ }
+ int n = Math.min(left, runLenDecoder.currentCount);
+ switch (runLenDecoder.mode) {
+ case RLE:
+ if (runLenDecoder.currentValue == maxDefLevel) {
+ readDoubles(n, column, rowId);
+ } else {
+ column.setNulls(rowId, n);
+ }
+ break;
+ case PACKED:
+ for (int i = 0; i < n; ++i) {
+ if (runLenDecoder.currentBuffer[runLenDecoder.currentBufferIdx++] == maxDefLevel) {
+ column.setDouble(rowId + i, readDouble());
+ } else {
+ column.setNullAt(rowId + i);
+ }
+ }
+ break;
+ }
+ rowId += n;
+ left -= n;
+ runLenDecoder.currentCount -= n;
+ }
+ }
+
+ @Override
+ protected void readBatchFromDictionaryIds(int rowId, int num, WritableDoubleVector column,
+ WritableIntVector dictionaryIds) {
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!column.isNullAt(i)) {
+ column.setDouble(i, dictionary.decodeToDouble(dictionaryIds.getInt(i)));
+ }
+ }
+ }
+
+ private double readDouble() {
+ return readDataBuffer(8).getDouble();
+ }
+
+ private void readDoubles(int total, WritableDoubleVector c, int rowId) {
+ int requiredBytes = total * 8;
+ ByteBuffer buffer = readDataBuffer(requiredBytes);
+
+ if (buffer.hasArray()) {
+ int offset = buffer.arrayOffset() + buffer.position();
+ c.setDoublesFromBinary(rowId, total, buffer.array(), offset);
+ } else {
+ for (int i = 0; i < total; i += 1) {
+ c.setDouble(rowId + i, buffer.getDouble());
+ }
+ }
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/FixedLenBytesColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/FixedLenBytesColumnReader.java
new file mode 100644
index 0000000..835562e
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/FixedLenBytesColumnReader.java
@@ -0,0 +1,144 @@
+/*
+ * 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.dataformat.Decimal;
+import org.apache.flink.table.dataformat.vector.writable.WritableBytesVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableColumnVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableIntVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableLongVector;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Fixed length bytes {@link ColumnReader}, just for decimal.
+ */
+public class FixedLenBytesColumnReader<VECTOR extends WritableColumnVector> extends AbstractColumnReader<VECTOR> {
+
+ private final int precision;
+
+ public FixedLenBytesColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader,
+ int precision) throws IOException {
+ super(descriptor, pageReader);
+ checkTypeName(PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY);
+ this.precision = precision;
+ }
+
+ @Override
+ protected void readBatch(int rowId, int num, VECTOR column) {
+ int bytesLen = descriptor.getPrimitiveType().getTypeLength();
+ if (Decimal.is32BitDecimal(precision)) {
+ WritableIntVector intVector = (WritableIntVector) column;
+ for (int i = 0; i < num; i++) {
+ if (runLenDecoder.readInteger() == maxDefLevel) {
+ intVector.setInt(rowId + i, (int) heapBinaryToLong(readDataBinary(bytesLen)));
+ } else {
+ intVector.setNullAt(rowId + i);
+ }
+ }
+ } else if (Decimal.is64BitDecimal(precision)) {
+ WritableLongVector longVector = (WritableLongVector) column;
+ for (int i = 0; i < num; i++) {
+ if (runLenDecoder.readInteger() == maxDefLevel) {
+ longVector.setLong(rowId + i, heapBinaryToLong(readDataBinary(bytesLen)));
+ } else {
+ longVector.setNullAt(rowId + i);
+ }
+ }
+ } else {
+ WritableBytesVector bytesVector = (WritableBytesVector) column;
+ for (int i = 0; i < num; i++) {
+ if (runLenDecoder.readInteger() == maxDefLevel) {
+ byte[] bytes = readDataBinary(bytesLen).getBytes();
+ bytesVector.appendBytes(rowId + i, bytes, 0, bytes.length);
+ } else {
+ bytesVector.setNullAt(rowId + i);
+ }
+ }
+ }
+ }
+
+ @Override
+ protected void readBatchFromDictionaryIds(
+ int rowId,
+ int num,
+ VECTOR column,
+ WritableIntVector dictionaryIds) {
+ if (Decimal.is32BitDecimal(precision)) {
+ WritableIntVector intVector = (WritableIntVector) column;
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!intVector.isNullAt(i)) {
+ Binary v = dictionary.decodeToBinary(dictionaryIds.getInt(i));
+ intVector.setInt(i, (int) heapBinaryToLong(v));
+ }
+ }
+ } else if (Decimal.is64BitDecimal(precision)) {
+ WritableLongVector longVector = (WritableLongVector) column;
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!longVector.isNullAt(i)) {
+ Binary v = dictionary.decodeToBinary(dictionaryIds.getInt(i));
+ longVector.setLong(i, heapBinaryToLong(v));
+ }
+ }
+ } else {
+ WritableBytesVector bytesVector = (WritableBytesVector) column;
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!bytesVector.isNullAt(i)) {
+ byte[] v = dictionary.decodeToBinary(dictionaryIds.getInt(i)).getBytes();
+ bytesVector.appendBytes(i, v, 0, v.length);
+ }
+ }
+ }
+ }
+
+ private long heapBinaryToLong(Binary binary) {
+ ByteBuffer buffer = binary.toByteBuffer();
+ byte[] bytes = buffer.array();
+ int start = buffer.arrayOffset() + buffer.position();
+ int end = buffer.arrayOffset() + buffer.limit();
+
+ long unscaled = 0L;
+
+ for (int i = start; i < end; i++) {
+ unscaled = (unscaled << 8) | (bytes[i] & 0xff);
+ }
+
+ int bits = 8 * (end - start);
+ return (unscaled << (64 - bits)) >> (64 - bits);
+ }
+
+ private Binary readDataBinary(int len) {
+ ByteBuffer buffer = readDataBuffer(len);
+ if (buffer.hasArray()) {
+ return Binary.fromConstantByteArray(
+ buffer.array(), buffer.arrayOffset() + buffer.position(), len);
+ } else {
+ byte[] bytes = new byte[len];
+ buffer.get(bytes);
+ return Binary.fromConstantByteArray(bytes);
+ }
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/FloatColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/FloatColumnReader.java
new file mode 100644
index 0000000..71b794a
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/FloatColumnReader.java
@@ -0,0 +1,101 @@
+/*
+ * 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.dataformat.vector.writable.WritableFloatVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableIntVector;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Float {@link ColumnReader}.
+ */
+public class FloatColumnReader extends AbstractColumnReader<WritableFloatVector> {
+
+ public FloatColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ super(descriptor, pageReader);
+ checkTypeName(PrimitiveType.PrimitiveTypeName.FLOAT);
+ }
+
+ @Override
+ protected void readBatch(int rowId, int num, WritableFloatVector column) {
+ int left = num;
+ while (left > 0) {
+ if (runLenDecoder.currentCount == 0) {
+ runLenDecoder.readNextGroup();
+ }
+ int n = Math.min(left, runLenDecoder.currentCount);
+ switch (runLenDecoder.mode) {
+ case RLE:
+ if (runLenDecoder.currentValue == maxDefLevel) {
+ readFloats(n, column, rowId);
+ } else {
+ column.setNulls(rowId, n);
+ }
+ break;
+ case PACKED:
+ for (int i = 0; i < n; ++i) {
+ if (runLenDecoder.currentBuffer[runLenDecoder.currentBufferIdx++] == maxDefLevel) {
+ column.setFloat(rowId + i, readFloat());
+ } else {
+ column.setNullAt(rowId + i);
+ }
+ }
+ break;
+ }
+ rowId += n;
+ left -= n;
+ runLenDecoder.currentCount -= n;
+ }
+ }
+
+ @Override
+ protected void readBatchFromDictionaryIds(int rowId, int num, WritableFloatVector column,
+ WritableIntVector dictionaryIds) {
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!column.isNullAt(i)) {
+ column.setFloat(i, dictionary.decodeToFloat(dictionaryIds.getInt(i)));
+ }
+ }
+ }
+
+ private float readFloat() {
+ return readDataBuffer(4).getFloat();
+ }
+
+ private void readFloats(int total, WritableFloatVector c, int rowId) {
+ int requiredBytes = total * 4;
+ ByteBuffer buffer = readDataBuffer(requiredBytes);
+
+ if (buffer.hasArray()) {
+ int offset = buffer.arrayOffset() + buffer.position();
+ c.setFloatsFromBinary(rowId, total, buffer.array(), offset);
+ } else {
+ for (int i = 0; i < total; i += 1) {
+ c.setFloat(rowId + i, buffer.getFloat());
+ }
+ }
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/IntColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/IntColumnReader.java
new file mode 100644
index 0000000..9b07367
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/IntColumnReader.java
@@ -0,0 +1,100 @@
+/*
+ * 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.dataformat.vector.writable.WritableIntVector;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Int {@link ColumnReader}.
+ */
+public class IntColumnReader extends AbstractColumnReader<WritableIntVector> {
+
+ public IntColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ super(descriptor, pageReader);
+ checkTypeName(PrimitiveType.PrimitiveTypeName.INT32);
+ }
+
+ @Override
+ protected void readBatch(int rowId, int num, WritableIntVector column) {
+ int left = num;
+ while (left > 0) {
+ if (runLenDecoder.currentCount == 0) {
+ runLenDecoder.readNextGroup();
+ }
+ int n = Math.min(left, runLenDecoder.currentCount);
+ switch (runLenDecoder.mode) {
+ case RLE:
+ if (runLenDecoder.currentValue == maxDefLevel) {
+ readIntegers(n, column, rowId);
+ } else {
+ column.setNulls(rowId, n);
+ }
+ break;
+ case PACKED:
+ for (int i = 0; i < n; ++i) {
+ if (runLenDecoder.currentBuffer[runLenDecoder.currentBufferIdx++] == maxDefLevel) {
+ column.setInt(rowId + i, readInteger());
+ } else {
+ column.setNullAt(rowId + i);
+ }
+ }
+ break;
+ }
+ rowId += n;
+ left -= n;
+ runLenDecoder.currentCount -= n;
+ }
+ }
+
+ @Override
+ protected void readBatchFromDictionaryIds(int rowId, int num, WritableIntVector column,
+ WritableIntVector dictionaryIds) {
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!column.isNullAt(i)) {
+ column.setInt(i, dictionary.decodeToInt(dictionaryIds.getInt(i)));
+ }
+ }
+ }
+
+ private int readInteger() {
+ return readDataBuffer(4).getInt();
+ }
+
+ private void readIntegers(int total, WritableIntVector c, int rowId) {
+ int requiredBytes = total * 4;
+ ByteBuffer buffer = readDataBuffer(requiredBytes);
+
+ if (buffer.hasArray()) {
+ int offset = buffer.arrayOffset() + buffer.position();
+ c.setIntsFromBinary(rowId, total, buffer.array(), offset);
+ } else {
+ for (int i = 0; i < total; i += 1) {
+ c.setInt(rowId + i, buffer.getInt());
+ }
+ }
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/LongColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/LongColumnReader.java
new file mode 100644
index 0000000..b68a732
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/LongColumnReader.java
@@ -0,0 +1,101 @@
+/*
+ * 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.dataformat.vector.writable.WritableIntVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableLongVector;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Long {@link ColumnReader}.
+ */
+public class LongColumnReader extends AbstractColumnReader<WritableLongVector> {
+
+ public LongColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ super(descriptor, pageReader);
+ checkTypeName(PrimitiveType.PrimitiveTypeName.INT64);
+ }
+
+ @Override
+ protected void readBatch(int rowId, int num, WritableLongVector column) {
+ int left = num;
+ while (left > 0) {
+ if (runLenDecoder.currentCount == 0) {
+ runLenDecoder.readNextGroup();
+ }
+ int n = Math.min(left, runLenDecoder.currentCount);
+ switch (runLenDecoder.mode) {
+ case RLE:
+ if (runLenDecoder.currentValue == maxDefLevel) {
+ readLongs(n, column, rowId);
+ } else {
+ column.setNulls(rowId, n);
+ }
+ break;
+ case PACKED:
+ for (int i = 0; i < n; ++i) {
+ if (runLenDecoder.currentBuffer[runLenDecoder.currentBufferIdx++] == maxDefLevel) {
+ column.setLong(rowId + i, readLong());
+ } else {
+ column.setNullAt(rowId + i);
+ }
+ }
+ break;
+ }
+ rowId += n;
+ left -= n;
+ runLenDecoder.currentCount -= n;
+ }
+ }
+
+ @Override
+ protected void readBatchFromDictionaryIds(int rowId, int num, WritableLongVector column,
+ WritableIntVector dictionaryIds) {
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!column.isNullAt(i)) {
+ column.setLong(i, dictionary.decodeToLong(dictionaryIds.getInt(i)));
+ }
+ }
+ }
+
+ private long readLong() {
+ return readDataBuffer(8).getLong();
+ }
+
+ private void readLongs(int total, WritableLongVector c, int rowId) {
+ int requiredBytes = total * 8;
+ ByteBuffer buffer = readDataBuffer(requiredBytes);
+
+ if (buffer.hasArray()) {
+ int offset = buffer.arrayOffset() + buffer.position();
+ c.setLongsFromBinary(rowId, total, buffer.array(), offset);
+ } else {
+ for (int i = 0; i < total; i += 1) {
+ c.setLong(rowId + i, buffer.getLong());
+ }
+ }
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/RunLengthDecoder.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/RunLengthDecoder.java
new file mode 100644
index 0000000..3337f6d
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/RunLengthDecoder.java
@@ -0,0 +1,293 @@
+/*
+ * 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.dataformat.vector.writable.WritableColumnVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableIntVector;
+
+import org.apache.parquet.Preconditions;
+import org.apache.parquet.bytes.ByteBufferInputStream;
+import org.apache.parquet.bytes.BytesUtils;
+import org.apache.parquet.column.values.bitpacking.BytePacker;
+import org.apache.parquet.column.values.bitpacking.Packer;
+import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridDecoder;
+import org.apache.parquet.io.ParquetDecodingException;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Run length decoder for data and dictionary ids.
+ * See https://github.com/apache/parquet-format/blob/master/Encodings.md
+ * See {@link RunLengthBitPackingHybridDecoder}.
+ */
+final class RunLengthDecoder {
+
+ /**
+ * If true, the bit width is fixed. This decoder is used in different places and this also
+ * controls if we need to read the bitwidth from the beginning of the data stream.
+ */
+ private final boolean fixedWidth;
+ private final boolean readLength;
+
+ // Encoded data.
+ private ByteBufferInputStream in;
+
+ // bit/byte width of decoded data and utility to batch unpack them.
+ private int bitWidth;
+ private int bytesWidth;
+ private BytePacker packer;
+
+ // Current decoding mode and values
+ MODE mode;
+ int currentCount;
+ int currentValue;
+
+ // Buffer of decoded values if the values are PACKED.
+ int[] currentBuffer = new int[16];
+ int currentBufferIdx = 0;
+
+ RunLengthDecoder() {
+ this.fixedWidth = false;
+ this.readLength = false;
+ }
+
+ RunLengthDecoder(int bitWidth) {
+ this.fixedWidth = true;
+ this.readLength = bitWidth != 0;
+ initWidthAndPacker(bitWidth);
+ }
+
+ RunLengthDecoder(int bitWidth, boolean readLength) {
+ this.fixedWidth = true;
+ this.readLength = readLength;
+ initWidthAndPacker(bitWidth);
+ }
+
+ /**
+ * Init from input stream.
+ */
+ void initFromStream(int valueCount, ByteBufferInputStream in) throws IOException {
+ this.in = in;
+ if (fixedWidth) {
+ // initialize for repetition and definition levels
+ if (readLength) {
+ int length = readIntLittleEndian();
+ this.in = in.sliceStream(length);
+ }
+ } else {
+ // initialize for values
+ if (in.available() > 0) {
+ initWidthAndPacker(in.read());
+ }
+ }
+ if (bitWidth == 0) {
+ // 0 bit width, treat this as an RLE run of valueCount number of 0's.
+ this.mode = MODE.RLE;
+ this.currentCount = valueCount;
+ this.currentValue = 0;
+ } else {
+ this.currentCount = 0;
+ }
+ }
+
+ /**
+ * Initializes the internal state for decoding ints of `bitWidth`.
+ */
+ private void initWidthAndPacker(int bitWidth) {
+ Preconditions.checkArgument(bitWidth >= 0 && bitWidth <= 32, "bitWidth must be >= 0 and <= 32");
+ this.bitWidth = bitWidth;
+ this.bytesWidth = BytesUtils.paddedByteCountFromBits(bitWidth);
+ this.packer = Packer.LITTLE_ENDIAN.newBytePacker(bitWidth);
+ }
+
+ int readInteger() {
+ if (this.currentCount == 0) {
+ this.readNextGroup();
+ }
+
+ this.currentCount--;
+ switch (mode) {
+ case RLE:
+ return this.currentValue;
+ case PACKED:
+ return this.currentBuffer[currentBufferIdx++];
+ }
+ throw new RuntimeException("Unreachable");
+ }
+
+ /**
+ * Decoding for dictionary ids. The IDs are populated into `values` and the nullability is
+ * populated into `nulls`.
+ */
+ void readDictionaryIds(
+ int total,
+ WritableIntVector values,
+ WritableColumnVector nulls,
+ int rowId,
+ int level,
+ RunLengthDecoder data) {
+ int left = total;
+ while (left > 0) {
+ if (this.currentCount == 0) {
+ this.readNextGroup();
+ }
+ int n = Math.min(left, this.currentCount);
+ switch (mode) {
+ case RLE:
+ if (currentValue == level) {
+ data.readDictionaryIdData(n, values, rowId);
+ } else {
+ nulls.setNulls(rowId, n);
+ }
+ break;
+ case PACKED:
+ for (int i = 0; i < n; ++i) {
+ if (currentBuffer[currentBufferIdx++] == level) {
+ values.setInt(rowId + i, data.readInteger());
+ } else {
+ nulls.setNullAt(rowId + i);
+ }
+ }
+ break;
+ }
+ rowId += n;
+ left -= n;
+ currentCount -= n;
+ }
+ }
+
+ /**
+ * It is used to decode dictionary IDs.
+ */
+ private void readDictionaryIdData(int total, WritableIntVector c, int rowId) {
+ int left = total;
+ while (left > 0) {
+ if (this.currentCount == 0) {
+ this.readNextGroup();
+ }
+ int n = Math.min(left, this.currentCount);
+ switch (mode) {
+ case RLE:
+ c.setInts(rowId, n, currentValue);
+ break;
+ case PACKED:
+ c.setInts(rowId, n, currentBuffer, currentBufferIdx);
+ currentBufferIdx += n;
+ break;
+ }
+ rowId += n;
+ left -= n;
+ currentCount -= n;
+ }
+ }
+
+ /**
+ * Reads the next varint encoded int.
+ */
+ private int readUnsignedVarInt() throws IOException {
+ int value = 0;
+ int shift = 0;
+ int b;
+ do {
+ b = in.read();
+ value |= (b & 0x7F) << shift;
+ shift += 7;
+ } while ((b & 0x80) != 0);
+ return value;
+ }
+
+ /**
+ * Reads the next 4 byte little endian int.
+ */
+ private int readIntLittleEndian() throws IOException {
+ int ch4 = in.read();
+ int ch3 = in.read();
+ int ch2 = in.read();
+ int ch1 = in.read();
+ return ((ch1 << 24) + (ch2 << 16) + (ch3 << 8) + ch4);
+ }
+
+ /**
+ * Reads the next byteWidth little endian int.
+ */
+ private int readIntLittleEndianPaddedOnBitWidth() throws IOException {
+ switch (bytesWidth) {
+ case 0:
+ return 0;
+ case 1:
+ return in.read();
+ case 2: {
+ int ch2 = in.read();
+ int ch1 = in.read();
+ return (ch1 << 8) + ch2;
+ }
+ case 3: {
+ int ch3 = in.read();
+ int ch2 = in.read();
+ int ch1 = in.read();
+ return (ch1 << 16) + (ch2 << 8) + ch3;
+ }
+ case 4: {
+ return readIntLittleEndian();
+ }
+ }
+ throw new RuntimeException("Unreachable");
+ }
+
+ /**
+ * Reads the next group.
+ */
+ void readNextGroup() {
+ try {
+ int header = readUnsignedVarInt();
+ this.mode = (header & 1) == 0 ? MODE.RLE : MODE.PACKED;
+ switch (mode) {
+ case RLE:
+ this.currentCount = header >>> 1;
+ this.currentValue = readIntLittleEndianPaddedOnBitWidth();
+ return;
+ case PACKED:
+ int numGroups = header >>> 1;
+ this.currentCount = numGroups * 8;
+
+ if (this.currentBuffer.length < this.currentCount) {
+ this.currentBuffer = new int[this.currentCount];
+ }
+ currentBufferIdx = 0;
+ int valueIndex = 0;
+ while (valueIndex < this.currentCount) {
+ // values are bit packed 8 at a time, so reading bitWidth will always work
+ ByteBuffer buffer = in.slice(bitWidth);
+ this.packer.unpack8Values(buffer, buffer.position(), this.currentBuffer, valueIndex);
+ valueIndex += 8;
+ }
+ return;
+ default:
+ throw new ParquetDecodingException("not a valid mode " + this.mode);
+ }
+ } catch (IOException e) {
+ throw new ParquetDecodingException("Failed to read from input stream", e);
+ }
+ }
+
+ enum MODE {
+ RLE,
+ PACKED
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/ShortColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/ShortColumnReader.java
new file mode 100644
index 0000000..a98051d
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/ShortColumnReader.java
@@ -0,0 +1,84 @@
+/*
+ * 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.dataformat.vector.writable.WritableIntVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableShortVector;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.IOException;
+
+/**
+ * Short {@link ColumnReader}. Using INT32 to store short, so just cast int to short.
+ */
+public class ShortColumnReader extends AbstractColumnReader<WritableShortVector> {
+
+ public ShortColumnReader(
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ super(descriptor, pageReader);
+ checkTypeName(PrimitiveType.PrimitiveTypeName.INT32);
+ }
+
+ @Override
+ protected void readBatch(int rowId, int num, WritableShortVector column) {
+ int left = num;
+ while (left > 0) {
+ if (runLenDecoder.currentCount == 0) {
+ runLenDecoder.readNextGroup();
+ }
+ int n = Math.min(left, runLenDecoder.currentCount);
+ switch (runLenDecoder.mode) {
+ case RLE:
+ if (runLenDecoder.currentValue == maxDefLevel) {
+ for (int i = 0; i < n; i++) {
+ column.setShort(rowId + i, (short) readDataBuffer(4).getInt());
+ }
+ } else {
+ column.setNulls(rowId, n);
+ }
+ break;
+ case PACKED:
+ for (int i = 0; i < n; ++i) {
+ if (runLenDecoder.currentBuffer[runLenDecoder.currentBufferIdx++] == maxDefLevel) {
+ column.setShort(rowId + i, (short) readDataBuffer(4).getInt());
+ } else {
+ column.setNullAt(rowId + i);
+ }
+ }
+ break;
+ }
+ rowId += n;
+ left -= n;
+ runLenDecoder.currentCount -= n;
+ }
+ }
+
+ @Override
+ protected void readBatchFromDictionaryIds(int rowId, int num, WritableShortVector column,
+ WritableIntVector dictionaryIds) {
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!column.isNullAt(i)) {
+ column.setShort(i, (short) dictionary.decodeToInt(dictionaryIds.getInt(i)));
+ }
+ }
+ }
+}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/TimestampColumnReader.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/TimestampColumnReader.java
new file mode 100644
index 0000000..256ae72
--- /dev/null
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/vector/reader/TimestampColumnReader.java
@@ -0,0 +1,120 @@
+/*
+ * 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.dataformat.SqlTimestamp;
+import org.apache.flink.table.dataformat.vector.writable.WritableIntVector;
+import org.apache.flink.table.dataformat.vector.writable.WritableTimestampVector;
+
+import org.apache.parquet.Preconditions;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReader;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.sql.Timestamp;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Timestamp {@link ColumnReader}. We only support INT96 bytes now, julianDay(4) + nanosOfDay(8).
+ * See https://github.com/apache/parquet-format/blob/master/LogicalTypes.md#timestamp
+ * TIMESTAMP_MILLIS and TIMESTAMP_MICROS are the deprecated ConvertedType.
+ */
+public class TimestampColumnReader extends AbstractColumnReader<WritableTimestampVector> {
+
+ public static final int JULIAN_EPOCH_OFFSET_DAYS = 2_440_588;
+ public static final long MILLIS_IN_DAY = TimeUnit.DAYS.toMillis(1);
+ public static final long NANOS_PER_MILLISECOND = TimeUnit.MILLISECONDS.toNanos(1);
+ public static final long NANOS_PER_SECOND = TimeUnit.SECONDS.toNanos(1);
+
+ private final boolean utcTimestamp;
+
+ public TimestampColumnReader(
+ boolean utcTimestamp,
+ ColumnDescriptor descriptor,
+ PageReader pageReader) throws IOException {
+ super(descriptor, pageReader);
+ this.utcTimestamp = utcTimestamp;
+ checkTypeName(PrimitiveType.PrimitiveTypeName.INT96);
+ }
+
+ @Override
+ protected boolean supportLazyDecode() {
+ return utcTimestamp;
+ }
+
+ @Override
+ protected void readBatch(int rowId, int num, WritableTimestampVector column) {
+ for (int i = 0; i < num; i++) {
+ if (runLenDecoder.readInteger() == maxDefLevel) {
+ ByteBuffer buffer = readDataBuffer(12);
+ column.setTimestamp(rowId + i, int96ToTimestamp(
+ utcTimestamp, buffer.getLong(), buffer.getInt()));
+ } else {
+ column.setNullAt(rowId + i);
+ }
+ }
+ }
+
+ @Override
+ protected void readBatchFromDictionaryIds(
+ int rowId,
+ int num,
+ WritableTimestampVector column,
+ WritableIntVector dictionaryIds) {
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!column.isNullAt(i)) {
+ column.setTimestamp(i, decodeInt96ToTimestamp(
+ utcTimestamp, dictionary, dictionaryIds.getInt(i)));
+ }
+ }
+ }
+
+ public static SqlTimestamp decodeInt96ToTimestamp(
+ boolean utcTimestamp,
+ org.apache.parquet.column.Dictionary dictionary,
+ int id) {
+ Binary binary = dictionary.decodeToBinary(id);
+ Preconditions.checkArgument(
+ binary.length() == 12,
+ "Timestamp with int96 should be 12 bytes.");
+ ByteBuffer buffer = binary.toByteBuffer().order(ByteOrder.LITTLE_ENDIAN);
+ return int96ToTimestamp(utcTimestamp, buffer.getLong(), buffer.getInt());
+ }
+
+ private static SqlTimestamp int96ToTimestamp(
+ boolean utcTimestamp, long nanosOfDay, int julianDay) {
+ long millisecond = julianDayToMillis(julianDay) + (nanosOfDay / NANOS_PER_MILLISECOND);
+
+ if (utcTimestamp) {
+ int nanoOfMillisecond = (int) (nanosOfDay % NANOS_PER_MILLISECOND);
+ return SqlTimestamp.fromEpochMillis(millisecond, nanoOfMillisecond);
+ } else {
+ Timestamp timestamp = new Timestamp(millisecond);
+ timestamp.setNanos((int) (nanosOfDay % NANOS_PER_SECOND));
+ return SqlTimestamp.fromTimestamp(timestamp);
+ }
+ }
+
+ private static long julianDayToMillis(int julianDay) {
+ return (julianDay - JULIAN_EPOCH_OFFSET_DAYS) * MILLIS_IN_DAY;
+ }
+}
diff --git a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/utils/ParquetWriterUtil.java b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/utils/ParquetWriterUtil.java
new file mode 100644
index 0000000..f14c295
--- /dev/null
+++ b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/utils/ParquetWriterUtil.java
@@ -0,0 +1,166 @@
+/*
+ * 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.core.fs.Path;
+import org.apache.flink.types.Row;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.hadoop.ParquetWriter;
+import org.apache.parquet.hadoop.api.WriteSupport;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.io.api.RecordConsumer;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.io.File;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.charset.StandardCharsets;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.UUID;
+
+import static org.apache.flink.formats.parquet.vector.reader.TimestampColumnReader.JULIAN_EPOCH_OFFSET_DAYS;
+import static org.apache.flink.formats.parquet.vector.reader.TimestampColumnReader.MILLIS_IN_DAY;
+import static org.apache.flink.formats.parquet.vector.reader.TimestampColumnReader.NANOS_PER_SECOND;
+
+/**
+ * Parquet writer util to write Row to file.
+ */
+public class ParquetWriterUtil {
+
+ public static Path createTempParquetFile(File folder, MessageType schema, List<Row> records, int rowGroupSize) throws IOException {
+ Path path = new Path(folder.getPath(), UUID.randomUUID().toString());
+ WriteSupport<Row> support = new WriteSupport<Row>() {
+ private RecordConsumer consumer;
+
+ @Override
+ public WriteContext init(Configuration configuration) {
+ return new WriteContext(schema, new HashMap<>());
+ }
+
+ @Override
+ public void prepareForWrite(RecordConsumer consumer) {
+ this.consumer = consumer;
+ }
+
+ @Override
+ public void write(Row row) {
+ consumer.startMessage();
+ for (int i = 0; i < row.getArity(); i++) {
+ PrimitiveType type = schema.getColumns().get(i).getPrimitiveType();
+ Object field = row.getField(i);
+ if (field != null) {
+ consumer.startField("f" + i, i);
+ switch (type.getPrimitiveTypeName()) {
+ case INT64:
+ consumer.addLong(((Number) field).longValue());
+ break;
+ case INT32:
+ consumer.addInteger(((Number) field).intValue());
+ break;
+ case BOOLEAN:
+ consumer.addBoolean((Boolean) field);
+ break;
+ case BINARY:
+ if (field instanceof String) {
+ field = ((String) field).getBytes(StandardCharsets.UTF_8);
+ } else if (field instanceof BigDecimal) {
+ field = ((BigDecimal) field).unscaledValue().toByteArray();
+ }
+ consumer.addBinary(Binary.fromConstantByteArray((byte[]) field));
+ break;
+ case FLOAT:
+ consumer.addFloat(((Number) field).floatValue());
+ break;
+ case DOUBLE:
+ consumer.addDouble(((Number) field).doubleValue());
+ break;
+ case INT96:
+ consumer.addBinary(timestampToInt96((LocalDateTime) field));
+ break;
+ case FIXED_LEN_BYTE_ARRAY:
+ byte[] bytes = ((BigDecimal) field).unscaledValue().toByteArray();
+ byte signByte = (byte) (bytes[0] < 0 ? -1 : 0);
+ int numBytes = 16;
+ byte[] newBytes = new byte[numBytes];
+ Arrays.fill(newBytes, 0, numBytes - bytes.length, signByte);
+ System.arraycopy(bytes, 0, newBytes, numBytes - bytes.length, bytes.length);
+ consumer.addBinary(Binary.fromConstantByteArray(newBytes));
+ break;
+ }
+ consumer.endField("f" + i, i);
+ }
+ }
+ consumer.endMessage();
+ }
+ };
+ ParquetWriter<Row> writer = new ParquetWriterBuilder(
+ new org.apache.hadoop.fs.Path(path.getPath()), support)
+ .withRowGroupSize(rowGroupSize)
+ .build();
+
+ for (Row record : records) {
+ writer.write(record);
+ }
+
+ writer.close();
+ return path;
+ }
+
+ private static class ParquetWriterBuilder extends ParquetWriter.Builder<Row, ParquetWriterBuilder> {
+
+ private final WriteSupport<Row> support;
+
+ private ParquetWriterBuilder(org.apache.hadoop.fs.Path path, WriteSupport<Row> support) {
+ super(path);
+ this.support = support;
+ }
+
+ @Override
+ protected ParquetWriterBuilder self() {
+ return this;
+ }
+
+ @Override
+ protected WriteSupport<Row> getWriteSupport(Configuration conf) {
+ return support;
+ }
+ }
+
+ private static Binary timestampToInt96(LocalDateTime time) {
+ Timestamp timestamp = Timestamp.valueOf(time);
+ long mills = timestamp.getTime();
+ int julianDay = (int) ((mills / MILLIS_IN_DAY) + JULIAN_EPOCH_OFFSET_DAYS);
+ long nanosOfDay = ((mills % MILLIS_IN_DAY) / 1000) * NANOS_PER_SECOND + timestamp.getNanos();
+
+ ByteBuffer buf = ByteBuffer.allocate(12);
+ buf.order(ByteOrder.LITTLE_ENDIAN);
+ buf.putLong(nanosOfDay);
+ buf.putInt(julianDay);
+ buf.flip();
+ return Binary.fromConstantByteBuffer(buf);
+ }
+}
diff --git a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/utils/TestUtil.java b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/utils/TestUtil.java
index 6a4f176..9220aaa 100644
--- a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/utils/TestUtil.java
+++ b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/utils/TestUtil.java
@@ -230,7 +230,7 @@ public class TestUtil {
return mockContext;
}
- private static Schema getTestSchema(String schemaName) {
+ public static Schema getTestSchema(String schemaName) {
try {
InputStream inputStream = TestUtil.class.getClassLoader()
.getResourceAsStream("avro/" + schemaName);
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
new file mode 100644
index 0000000..758c267
--- /dev/null
+++ b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/vector/ParquetColumnarRowSplitReaderTest.java
@@ -0,0 +1,545 @@
+/*
+ * 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;
+
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.table.dataformat.ColumnarRow;
+import org.apache.flink.table.dataformat.Decimal;
+import org.apache.flink.table.dataformat.vector.VectorizedColumnBatch;
+import org.apache.flink.table.runtime.functions.SqlDateTimeUtils;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.BigIntType;
+import org.apache.flink.table.types.logical.BooleanType;
+import org.apache.flink.table.types.logical.DateType;
+import org.apache.flink.table.types.logical.DecimalType;
+import org.apache.flink.table.types.logical.DoubleType;
+import org.apache.flink.table.types.logical.FloatType;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.SmallIntType;
+import org.apache.flink.table.types.logical.TimestampType;
+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.types.Row;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
+import org.apache.parquet.schema.Types;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.stream.IntStream;
+
+import static org.apache.flink.formats.parquet.utils.ParquetWriterUtil.createTempParquetFile;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Test for {@link ParquetColumnarRowSplitReader}.
+ */
+@RunWith(Parameterized.class)
+public class ParquetColumnarRowSplitReaderTest {
+
+ private static final int FIELD_NUMBER = 15;
+ private static final LocalDateTime BASE_TIME = LocalDateTime.now();
+
+ private static final MessageType PARQUET_SCHEMA = new MessageType(
+ "TOP",
+ Types.primitive(PrimitiveType.PrimitiveTypeName.BINARY, Type.Repetition.OPTIONAL)
+ .named("f0"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.BOOLEAN, Type.Repetition.OPTIONAL)
+ .named("f1"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT32, Type.Repetition.OPTIONAL)
+ .named("f2"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT32, Type.Repetition.OPTIONAL)
+ .named("f3"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT32, Type.Repetition.OPTIONAL)
+ .named("f4"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT64, Type.Repetition.OPTIONAL)
+ .named("f5"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.FLOAT, Type.Repetition.OPTIONAL)
+ .named("f6"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.DOUBLE, Type.Repetition.OPTIONAL)
+ .named("f7"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT96, Type.Repetition.OPTIONAL)
+ .named("f8"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT32, Type.Repetition.OPTIONAL)
+ .precision(5)
+ .as(OriginalType.DECIMAL)
+ .named("f9"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.INT64, Type.Repetition.OPTIONAL)
+ .precision(15)
+ .as(OriginalType.DECIMAL)
+ .named("f10"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.BINARY, Type.Repetition.OPTIONAL)
+ .precision(20)
+ .as(OriginalType.DECIMAL)
+ .named("f11"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY, Type.Repetition.OPTIONAL)
+ .length(16)
+ .precision(5)
+ .as(OriginalType.DECIMAL)
+ .named("f12"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY, Type.Repetition.OPTIONAL)
+ .length(16)
+ .precision(15)
+ .as(OriginalType.DECIMAL)
+ .named("f13"),
+ Types.primitive(PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY, Type.Repetition.OPTIONAL)
+ .length(16)
+ .precision(20)
+ .as(OriginalType.DECIMAL)
+ .named("f14")
+ );
+
+ @ClassRule
+ public static final TemporaryFolder TEMPORARY_FOLDER = new TemporaryFolder();
+
+ private final int rowGroupSize;
+
+ @Parameterized.Parameters(name = "rowGroupSize-{0}")
+ public static Collection<Integer> parameters() {
+ return Arrays.asList(10, 1000);
+ }
+
+ public ParquetColumnarRowSplitReaderTest(int rowGroupSize) {
+ this.rowGroupSize = rowGroupSize;
+ }
+
+ @Test
+ public void testNormalTypesReadWithSplits() throws IOException {
+ // prepare parquet file
+ int number = 10000;
+ List<Row> records = new ArrayList<>(number);
+ List<Integer> values = new ArrayList<>(number);
+ Random random = new Random();
+ for (int i = 0; i < number; i++) {
+ Integer v = random.nextInt(number / 2);
+ if (v % 10 == 0) {
+ values.add(null);
+ records.add(new Row(FIELD_NUMBER));
+ } else {
+ values.add(v);
+ records.add(newRow(v));
+ }
+ }
+
+ testNormalTypes(number, records, values);
+ }
+
+ private void testNormalTypes(int number, List<Row> records,
+ List<Integer> values) throws IOException {
+ Path testPath = createTempParquetFile(
+ TEMPORARY_FOLDER.newFolder(), PARQUET_SCHEMA, records, rowGroupSize);
+
+ // test reading and splitting
+ long fileLen = testPath.getFileSystem().getFileStatus(testPath).getLen();
+ int len1 = readSplitAndCheck(0, testPath, 0, fileLen / 3, values);
+ int len2 = readSplitAndCheck(len1, testPath, fileLen / 3, fileLen * 2 / 3, values);
+ int len3 = readSplitAndCheck(len1 + len2, testPath, fileLen * 2 / 3, Long.MAX_VALUE, values);
+ assertEquals(number, len1 + len2 + len3);
+ }
+
+ private int readSplitAndCheck(
+ int start,
+ Path testPath,
+ long splitStart,
+ long splitLength,
+ List<Integer> values) throws IOException {
+ LogicalType[] fieldTypes = new LogicalType[]{
+ new VarCharType(VarCharType.MAX_LENGTH),
+ new BooleanType(),
+ new TinyIntType(),
+ new SmallIntType(),
+ new IntType(),
+ new BigIntType(),
+ new FloatType(),
+ new DoubleType(),
+ new TimestampType(9),
+ new DecimalType(5, 0),
+ new DecimalType(15, 0),
+ new DecimalType(20, 0),
+ new DecimalType(5, 0),
+ new DecimalType(15, 0),
+ new DecimalType(20, 0)};
+
+ ParquetColumnarRowSplitReader reader = new ParquetColumnarRowSplitReader(
+ false,
+ new Configuration(),
+ fieldTypes,
+ new String[] {
+ "f0", "f1", "f2", "f3", "f4", "f5", "f6", "f7",
+ "f8", "f9", "f10", "f11", "f12", "f13", "f14"},
+ VectorizedColumnBatch::new,
+ 500,
+ new org.apache.hadoop.fs.Path(testPath.getPath()),
+ splitStart,
+ splitLength);
+
+ int i = start;
+ while (!reader.reachedEnd()) {
+ ColumnarRow row = reader.nextRecord();
+ Integer v = values.get(i);
+ if (v == null) {
+ assertTrue(row.isNullAt(0));
+ assertTrue(row.isNullAt(1));
+ assertTrue(row.isNullAt(2));
+ assertTrue(row.isNullAt(3));
+ assertTrue(row.isNullAt(4));
+ assertTrue(row.isNullAt(5));
+ assertTrue(row.isNullAt(6));
+ assertTrue(row.isNullAt(7));
+ assertTrue(row.isNullAt(8));
+ assertTrue(row.isNullAt(9));
+ assertTrue(row.isNullAt(10));
+ assertTrue(row.isNullAt(11));
+ assertTrue(row.isNullAt(12));
+ assertTrue(row.isNullAt(13));
+ assertTrue(row.isNullAt(14));
+ } else {
+ assertEquals("" + v, row.getString(0).toString());
+ assertEquals(v % 2 == 0, row.getBoolean(1));
+ assertEquals(v.byteValue(), row.getByte(2));
+ assertEquals(v.shortValue(), row.getShort(3));
+ assertEquals(v.intValue(), row.getInt(4));
+ assertEquals(v.longValue(), row.getLong(5));
+ assertEquals(v.floatValue(), row.getFloat(6), 0);
+ assertEquals(v.doubleValue(), row.getDouble(7), 0);
+ assertEquals(
+ toDateTime(v),
+ row.getTimestamp(8, 9).toLocalDateTime());
+ assertEquals(BigDecimal.valueOf(v), row.getDecimal(9, 5, 0).toBigDecimal());
+ assertEquals(BigDecimal.valueOf(v), row.getDecimal(10, 15, 0).toBigDecimal());
+ assertEquals(BigDecimal.valueOf(v), row.getDecimal(11, 20, 0).toBigDecimal());
+ assertEquals(BigDecimal.valueOf(v), row.getDecimal(12, 5, 0).toBigDecimal());
+ assertEquals(BigDecimal.valueOf(v), row.getDecimal(13, 15, 0).toBigDecimal());
+ assertEquals(BigDecimal.valueOf(v), row.getDecimal(14, 20, 0).toBigDecimal());
+ }
+ i++;
+ }
+ reader.close();
+ return i - start;
+ }
+
+ private Row newRow(Integer v) {
+ return Row.of(
+ "" + v,
+ v % 2 == 0,
+ v,
+ v,
+ v,
+ v.longValue(),
+ v.floatValue(),
+ v.doubleValue(),
+ toDateTime(v),
+ BigDecimal.valueOf(v),
+ BigDecimal.valueOf(v),
+ BigDecimal.valueOf(v),
+ BigDecimal.valueOf(v),
+ BigDecimal.valueOf(v),
+ BigDecimal.valueOf(v));
+ }
+
+ private LocalDateTime toDateTime(Integer v) {
+ v = (v > 0 ? v : -v) % 10000;
+ return BASE_TIME.plusNanos(v).plusSeconds(v);
+ }
+
+ @Test
+ public void testDictionary() throws IOException {
+ // prepare parquet file
+ int number = 10000;
+ List<Row> records = new ArrayList<>(number);
+ List<Integer> values = new ArrayList<>(number);
+ Random random = new Random();
+ int[] intValues = new int[10];
+ // test large values in dictionary
+ for (int i = 0; i < intValues.length; i++) {
+ intValues[i] = random.nextInt();
+ }
+ for (int i = 0; i < number; i++) {
+ Integer v = intValues[random.nextInt(10)];
+ if (v == 0) {
+ values.add(null);
+ records.add(new Row(FIELD_NUMBER));
+ } else {
+ values.add(v);
+ records.add(newRow(v));
+ }
+ }
+
+ testNormalTypes(number, records, values);
+ }
+
+ @Test
+ public void testPartialDictionary() throws IOException {
+ // prepare parquet file
+ int number = 10000;
+ List<Row> records = new ArrayList<>(number);
+ List<Integer> values = new ArrayList<>(number);
+ Random random = new Random();
+ int[] intValues = new int[10];
+ // test large values in dictionary
+ for (int i = 0; i < intValues.length; i++) {
+ intValues[i] = random.nextInt();
+ }
+ for (int i = 0; i < number; i++) {
+ Integer v = i < 5000 ? intValues[random.nextInt(10)] : i;
+ if (v == 0) {
+ values.add(null);
+ records.add(new Row(FIELD_NUMBER));
+ } else {
+ values.add(v);
+ records.add(newRow(v));
+ }
+ }
+
+ testNormalTypes(number, records, values);
+ }
+
+ @Test
+ public void testContinuousRepetition() throws IOException {
+ // prepare parquet file
+ int number = 10000;
+ List<Row> records = new ArrayList<>(number);
+ List<Integer> values = new ArrayList<>(number);
+ Random random = new Random();
+ for (int i = 0; i < 100; i++) {
+ Integer v = random.nextInt(10);
+ for (int j = 0; j < 100; j++) {
+ if (v == 0) {
+ values.add(null);
+ records.add(new Row(FIELD_NUMBER));
+ } else {
+ values.add(v);
+ records.add(newRow(v));
+ }
+ }
+ }
+
+ testNormalTypes(number, records, values);
+ }
+
+ @Test
+ public void testLargeValue() throws IOException {
+ // prepare parquet file
+ int number = 10000;
+ List<Row> records = new ArrayList<>(number);
+ List<Integer> values = new ArrayList<>(number);
+ Random random = new Random();
+ for (int i = 0; i < number; i++) {
+ Integer v = random.nextInt();
+ if (v % 10 == 0) {
+ values.add(null);
+ records.add(new Row(FIELD_NUMBER));
+ } else {
+ values.add(v);
+ records.add(newRow(v));
+ }
+ }
+
+ testNormalTypes(number, records, values);
+ }
+
+ @Test
+ public void testProject() throws IOException {
+ // prepare parquet file
+ int number = 1000;
+ List<Row> records = new ArrayList<>(number);
+ for (int i = 0; i < number; i++) {
+ Integer v = i;
+ records.add(newRow(v));
+ }
+
+ Path testPath = createTempParquetFile(
+ TEMPORARY_FOLDER.newFolder(), PARQUET_SCHEMA, records, rowGroupSize);
+
+ // test reader
+ LogicalType[] fieldTypes = new LogicalType[]{
+ new DoubleType(),
+ new TinyIntType(),
+ new IntType()};
+ ParquetColumnarRowSplitReader reader = new ParquetColumnarRowSplitReader(
+ false,
+ new Configuration(),
+ fieldTypes,
+ new String[] {"f7", "f2", "f4"},
+ VectorizedColumnBatch::new,
+ 500,
+ new org.apache.hadoop.fs.Path(testPath.getPath()),
+ 0,
+ Long.MAX_VALUE);
+ int i = 0;
+ while (!reader.reachedEnd()) {
+ ColumnarRow row = reader.nextRecord();
+ assertEquals(i, row.getDouble(0), 0);
+ assertEquals((byte) i, row.getByte(1));
+ assertEquals(i, row.getInt(2));
+ i++;
+ }
+ reader.close();
+ }
+
+ @Test
+ public void testPartitionValues() throws IOException {
+ // prepare parquet file
+ int number = 1000;
+ List<Row> records = new ArrayList<>(number);
+ for (int i = 0; i < number; i++) {
+ Integer v = i;
+ records.add(newRow(v));
+ }
+
+ Path testPath = createTempParquetFile(
+ TEMPORARY_FOLDER.newFolder(), PARQUET_SCHEMA, records, rowGroupSize);
+
+ // test reader
+ Map<String, Object> partSpec = new HashMap<>();
+ partSpec.put("f15", true);
+ partSpec.put("f16", Date.valueOf("2020-11-23"));
+ partSpec.put("f17", LocalDateTime.of(1999, 1, 1, 1, 1));
+ partSpec.put("f18", 6.6);
+ partSpec.put("f19", (byte) 9);
+ partSpec.put("f20", (short) 10);
+ partSpec.put("f21", 11);
+ partSpec.put("f22", 12L);
+ partSpec.put("f23", 13f);
+ partSpec.put("f24", new BigDecimal(24));
+ partSpec.put("f25", new BigDecimal(25));
+ partSpec.put("f26", new BigDecimal(26));
+ partSpec.put("f27", "f27");
+
+ innerTestPartitionValues(testPath, partSpec, false);
+
+ for (String k : new ArrayList<>(partSpec.keySet())) {
+ partSpec.put(k, null);
+ }
+
+ innerTestPartitionValues(testPath, partSpec, true);
+ }
+
+ private void innerTestPartitionValues(
+ Path testPath,
+ Map<String, Object> partSpec,
+ boolean nullPartValue) throws IOException {
+ LogicalType[] fieldTypes = new LogicalType[]{
+ new VarCharType(VarCharType.MAX_LENGTH),
+ new BooleanType(),
+ new TinyIntType(),
+ new SmallIntType(),
+ new IntType(),
+ new BigIntType(),
+ new FloatType(),
+ new DoubleType(),
+ new TimestampType(9),
+ new DecimalType(5, 0),
+ new DecimalType(15, 0),
+ new DecimalType(20, 0),
+ new DecimalType(5, 0),
+ new DecimalType(15, 0),
+ new DecimalType(20, 0),
+ new BooleanType(),
+ new DateType(),
+ new TimestampType(9),
+ new DoubleType(),
+ new TinyIntType(),
+ new SmallIntType(),
+ new IntType(),
+ new BigIntType(),
+ new FloatType(),
+ new DecimalType(5, 0),
+ new DecimalType(15, 0),
+ new DecimalType(20, 0),
+ new VarCharType(VarCharType.MAX_LENGTH)};
+ ParquetColumnarRowSplitReader reader = ParquetSplitReaderUtil.genPartColumnarRowReader(
+ false,
+ new Configuration(),
+ IntStream.range(0, 28).mapToObj(i -> "f" + i).toArray(String[]::new),
+ Arrays.stream(fieldTypes)
+ .map(TypeConversions::fromLogicalToDataType)
+ .toArray(DataType[]::new),
+ partSpec,
+ new int[]{7, 2, 4, 15, 19, 20, 21, 22, 23, 18, 16, 17, 24, 25, 26, 27},
+ rowGroupSize,
+ new Path(testPath.getPath()),
+ 0,
+ Long.MAX_VALUE);
+ int i = 0;
+ while (!reader.reachedEnd()) {
+ ColumnarRow row = reader.nextRecord();
+
+ // common values
+ assertEquals(i, row.getDouble(0), 0);
+ assertEquals((byte) i, row.getByte(1));
+ assertEquals(i, row.getInt(2));
+
+ // partition values
+ if (nullPartValue) {
+ for (int j = 3; j < 16; j++) {
+ assertTrue(row.isNullAt(j));
+ }
+ } else {
+ assertTrue(row.getBoolean(3));
+ assertEquals(9, row.getByte(4));
+ assertEquals(10, row.getShort(5));
+ assertEquals(11, row.getInt(6));
+ assertEquals(12, row.getLong(7));
+ assertEquals(13, row.getFloat(8), 0);
+ assertEquals(6.6, row.getDouble(9), 0);
+ assertEquals(
+ SqlDateTimeUtils.dateToInternal(Date.valueOf("2020-11-23")),
+ row.getInt(10));
+ assertEquals(
+ LocalDateTime.of(1999, 1, 1, 1, 1),
+ row.getTimestamp(11, 9).toLocalDateTime());
+ assertEquals(
+ Decimal.fromBigDecimal(new BigDecimal(24), 5, 0),
+ row.getDecimal(12, 5, 0));
+ assertEquals(
+ Decimal.fromBigDecimal(new BigDecimal(25), 15, 0),
+ row.getDecimal(13, 15, 0));
+ assertEquals(
+ Decimal.fromBigDecimal(new BigDecimal(26), 20, 0),
+ row.getDecimal(14, 20, 0));
+ assertEquals("f27", row.getString(15).toString());
+ }
+
+ i++;
+ }
+ reader.close();
+ }
+}