You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2020/09/25 09:17:40 UTC
[GitHub] [flink] JingsongLi opened a new pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
JingsongLi opened a new pull request #13479:
URL: https://github.com/apache/flink/pull/13479
## What is the purpose of the change
Introduce `ParquetColumnarRowInputFormat` for filesystem connector.
(Migrate Old interface to new/FLIP-27 source API)
## Brief change log
- `ParquetColumnarRowSplitReader` -> `ParquetColumnarRowInputFormat extends ParquetVectorizedInputFormat`.
- `ParquetColumnarRowSplitReaderTest` -> `ParquetColumnarRowInputFormatTest`.
Some other extraction:
- Introduce `BulkFormatUtils` to consume a `BulkFormat.Reader`.
- Introduce `ColumnarRowIterator` to be a columnar row `RecyclableIterator`.
- Introduce `PartitionValueConverter` to abstract Hive and Flink Filesystem partition value conversions.
## Verifying this change
`ParquetColumnarRowInputFormatTest`
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): no
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
- The serializers: no
- The runtime per-record code paths (performance sensitive): no
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: no
- The S3 file system connector: no
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (JavaDocs)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-707569225
@JingsongLi I think we merged something here that has a problem: We should not have a `test-jar` for the `flink-connector-files` project, because the module it is not scala-versioned. Thus we cannot release an artifact with a dependency on a specific scala version (like the test jar).
We need to have the connector test utils in a different project, or add a scala version to the module (I would like to avoid a scala version if possible).
Can you send all PRs that do anything on the file connector for my review before merging them, until we have enough common understanding on such issues?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lirui-apache commented on a change in pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
lirui-apache commented on a change in pull request #13479:
URL: https://github.com/apache/flink/pull/13479#discussion_r502767584
##########
File path: flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
##########
@@ -0,0 +1,471 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.connector.file.src.util.CheckpointedPosition;
+import org.apache.flink.connector.file.src.util.Pool;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
+import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
+import org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import 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 javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+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;
+
+/**
+ * Parquet {@link BulkFormat} that reads data from the file to {@link VectorizedColumnBatch} in
+ * vectorized mode.
+ */
+public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final SerializableConfiguration hadoopConfig;
+ private final String[] projectedFields;
+ private final LogicalType[] projectedTypes;
+ private final ColumnBatchFactory batchFactory;
+ private final int batchSize;
+ private final boolean isUtcTimestamp;
+ private final boolean isCaseSensitive;
+
+ public ParquetVectorizedInputFormat(
+ SerializableConfiguration hadoopConfig,
+ String[] projectedFields,
+ LogicalType[] projectedTypes,
+ ColumnBatchFactory batchFactory,
+ int batchSize,
+ boolean isUtcTimestamp,
+ boolean isCaseSensitive) {
+ Preconditions.checkArgument(
+ projectedFields.length == projectedTypes.length,
+ "The length(%s) of projectedFields should equal to the length(%s) projectedTypes",
+ projectedFields.length,
+ projectedTypes.length);
+
+ this.hadoopConfig = hadoopConfig;
+ this.projectedFields = projectedFields;
+ this.projectedTypes = projectedTypes;
+ this.batchFactory = batchFactory;
+ this.batchSize = batchSize;
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.isCaseSensitive = isCaseSensitive;
+ }
+
+ @Override
+ public ParquetReader createReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength) throws IOException {
+ org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path(filePath.toUri());
+ ParquetMetadata footer = readFooter(
+ hadoopConfig.conf(), hadoopPath, range(splitOffset, splitOffset + splitLength));
+ MessageType fileSchema = footer.getFileMetaData().getSchema();
+ FilterCompat.Filter filter = getFilter(hadoopConfig.conf());
+ List<BlockMetaData> blocks = filterRowGroups(filter, footer.getBlocks(), fileSchema);
+
+ MessageType requestedSchema = clipParquetSchema(fileSchema);
+ ParquetFileReader reader = new ParquetFileReader(
+ hadoopConfig.conf(),
+ footer.getFileMetaData(),
+ hadoopPath,
+ blocks,
+ requestedSchema.getColumns());
+
+ long totalRowCount = 0;
+ for (BlockMetaData block : blocks) {
+ totalRowCount += block.getRowCount();
+ }
+
+ checkSchema(fileSchema, requestedSchema);
+
+ final int numBatchesToCirculate = config.getInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY) + 1;
+ final Pool<ParquetReaderBatch<T>> poolOfBatches =
+ createPoolOfBatches(filePath, requestedSchema, numBatchesToCirculate);
+
+ return new ParquetReader(reader, requestedSchema, totalRowCount, poolOfBatches);
+ }
+
+ @Override
+ public ParquetReader restoreReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength,
+ CheckpointedPosition checkpointedPosition) throws IOException {
+ ParquetReader reader = createReader(config, filePath, splitOffset, splitLength);
+ // Offset is record count too.
+ reader.seek(checkpointedPosition.getOffset() + checkpointedPosition.getRecordsAfterOffset());
Review comment:
-1 can be used to represent `NO_OFFSET`, will that cause a problem here?
##########
File path: flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
##########
@@ -0,0 +1,471 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.connector.file.src.util.CheckpointedPosition;
+import org.apache.flink.connector.file.src.util.Pool;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
+import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
+import org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import 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 javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+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;
+
+/**
+ * Parquet {@link BulkFormat} that reads data from the file to {@link VectorizedColumnBatch} in
+ * vectorized mode.
+ */
+public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final SerializableConfiguration hadoopConfig;
+ private final String[] projectedFields;
+ private final LogicalType[] projectedTypes;
+ private final ColumnBatchFactory batchFactory;
+ private final int batchSize;
+ private final boolean isUtcTimestamp;
+ private final boolean isCaseSensitive;
+
+ public ParquetVectorizedInputFormat(
+ SerializableConfiguration hadoopConfig,
+ String[] projectedFields,
+ LogicalType[] projectedTypes,
+ ColumnBatchFactory batchFactory,
+ int batchSize,
+ boolean isUtcTimestamp,
+ boolean isCaseSensitive) {
+ Preconditions.checkArgument(
+ projectedFields.length == projectedTypes.length,
+ "The length(%s) of projectedFields should equal to the length(%s) projectedTypes",
+ projectedFields.length,
+ projectedTypes.length);
+
+ this.hadoopConfig = hadoopConfig;
+ this.projectedFields = projectedFields;
+ this.projectedTypes = projectedTypes;
+ this.batchFactory = batchFactory;
+ this.batchSize = batchSize;
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.isCaseSensitive = isCaseSensitive;
+ }
+
+ @Override
+ public ParquetReader createReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength) throws IOException {
+ org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path(filePath.toUri());
+ ParquetMetadata footer = readFooter(
+ hadoopConfig.conf(), hadoopPath, range(splitOffset, splitOffset + splitLength));
+ MessageType fileSchema = footer.getFileMetaData().getSchema();
+ FilterCompat.Filter filter = getFilter(hadoopConfig.conf());
+ List<BlockMetaData> blocks = filterRowGroups(filter, footer.getBlocks(), fileSchema);
+
+ MessageType requestedSchema = clipParquetSchema(fileSchema);
+ ParquetFileReader reader = new ParquetFileReader(
+ hadoopConfig.conf(),
+ footer.getFileMetaData(),
+ hadoopPath,
+ blocks,
+ requestedSchema.getColumns());
+
+ long totalRowCount = 0;
+ for (BlockMetaData block : blocks) {
+ totalRowCount += block.getRowCount();
+ }
+
+ checkSchema(fileSchema, requestedSchema);
+
+ final int numBatchesToCirculate = config.getInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY) + 1;
+ final Pool<ParquetReaderBatch<T>> poolOfBatches =
+ createPoolOfBatches(filePath, requestedSchema, numBatchesToCirculate);
+
+ return new ParquetReader(reader, requestedSchema, totalRowCount, poolOfBatches);
+ }
+
+ @Override
+ public ParquetReader restoreReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength,
+ CheckpointedPosition checkpointedPosition) throws IOException {
+ ParquetReader reader = createReader(config, filePath, splitOffset, splitLength);
+ // Offset is record count too.
+ reader.seek(checkpointedPosition.getOffset() + checkpointedPosition.getRecordsAfterOffset());
+ return reader;
+ }
+
+ @Override
+ public boolean isSplittable() {
+ return true;
+ }
+
+ /**
+ * Clips `parquetSchema` according to `fieldNames`.
+ */
+ private MessageType clipParquetSchema(GroupType parquetSchema) {
+ Type[] types = new Type[projectedFields.length];
+ if (isCaseSensitive) {
+ for (int i = 0; i < projectedFields.length; ++i) {
+ String fieldName = projectedFields[i];
+ if (parquetSchema.getFieldIndex(fieldName) < 0) {
+ throw new IllegalArgumentException(fieldName + " does not exist");
+ }
+ types[i] = parquetSchema.getType(fieldName);
+ }
+ } else {
+ Map<String, Type> caseInsensitiveFieldMap = new HashMap<>();
+ for (Type type : parquetSchema.getFields()) {
+ caseInsensitiveFieldMap.compute(type.getName().toLowerCase(Locale.ROOT),
+ (key, previousType) -> {
+ if (previousType != null) {
+ throw new FlinkRuntimeException(
+ "Parquet with case insensitive mode should have no duplicate key: " + key);
+ }
+ return type;
+ });
+ }
+ for (int i = 0; i < projectedFields.length; ++i) {
+ Type type = caseInsensitiveFieldMap.get(projectedFields[i].toLowerCase(Locale.ROOT));
+ if (type == null) {
+ throw new IllegalArgumentException(projectedFields[i] + " does not exist");
+ }
+ // TODO clip for array,map,row types.
+ types[i] = type;
+ }
+ }
+
+ return Types.buildMessage().addFields(types).named("flink-parquet");
+ }
+
+ private void checkSchema(
+ MessageType fileSchema,
+ MessageType requestedSchema) throws IOException, UnsupportedOperationException {
+ if (projectedFields.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));
+ }
+ }
+ }
+ }
+
+ private Pool<ParquetReaderBatch<T>> createPoolOfBatches(
+ Path filePath, MessageType requestedSchema, int numBatches) {
+ final Pool<ParquetReaderBatch<T>> pool = new Pool<>(numBatches);
+
+ for (int i = 0; i < numBatches; i++) {
+ pool.add(createReaderBatch(filePath, requestedSchema, pool.recycler()));
+ }
+
+ return pool;
+ }
+
+ private ParquetReaderBatch<T> createReaderBatch(
+ Path filePath,
+ MessageType requestedSchema,
+ Pool.Recycler<ParquetReaderBatch<T>> recycler) {
+ WritableColumnVector[] writableVectors = createWritableVectors(requestedSchema);
+ VectorizedColumnBatch columnarBatch =
+ batchFactory.create(filePath, createReadableVectors(writableVectors));
+ return createReaderBatch(writableVectors, columnarBatch, recycler);
+ }
+
+ private WritableColumnVector[] createWritableVectors(MessageType requestedSchema) {
+ WritableColumnVector[] columns = new WritableColumnVector[projectedTypes.length];
+ for (int i = 0; i < projectedTypes.length; i++) {
+ columns[i] = createWritableColumnVector(
+ batchSize,
+ projectedTypes[i],
+ requestedSchema.getColumns().get(i).getPrimitiveType());
+ }
+ return columns;
+ }
+
+ /**
+ * Create readable vectors from writable vectors.
+ * Especially for decimal, see {@link ParquetDecimalVector}.
+ */
+ private ColumnVector[] createReadableVectors(WritableColumnVector[] writableVectors) {
+ ColumnVector[] vectors = new ColumnVector[writableVectors.length];
+ for (int i = 0; i < writableVectors.length; i++) {
+ vectors[i] = projectedTypes[i].getTypeRoot() == LogicalTypeRoot.DECIMAL ?
+ new ParquetDecimalVector(writableVectors[i]) :
+ writableVectors[i];
+ }
+ return vectors;
+ }
+
+ private class ParquetReader implements BulkFormat.Reader<T> {
+
+ private ParquetFileReader reader;
+
+ 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 Pool<ParquetReaderBatch<T>> pool;
+
+ /**
+ * 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;
+
+ /**
+ * 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.
+ */
+ @SuppressWarnings("rawtypes")
+ private ColumnReader[] columnReaders;
+
+ private long recordsToSkip;
+
+ private ParquetReader(
+ ParquetFileReader reader,
+ MessageType requestedSchema,
+ long totalRowCount,
+ Pool<ParquetReaderBatch<T>> pool) {
+ this.reader = reader;
+ this.requestedSchema = requestedSchema;
+ this.totalRowCount = totalRowCount;
+ this.pool = pool;
+ this.rowsReturned = 0;
+ this.totalCountLoadedSoFar = 0;
+ this.recordsToSkip = 0;
+ }
+
+ @Nullable
+ @Override
+ public RecordIterator<T> readBatch() throws IOException {
+ final ParquetReaderBatch<T> batch = getCachedEntry();
+
+ final long startingRowNum = rowsReturned;
+ if (!nextBatch(batch)) {
+ batch.recycle();
+ return null;
+ }
+
+ final RecordIterator<T> records = batch.convertAndGetIterator(startingRowNum);
+ if (recordsToSkip > 0) {
Review comment:
Seems this check is unnecessary.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
}, {
"hash" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411",
"triggerID" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8b7772756eb215c09ef7a77bec62a2e258661454 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #13479:
URL: https://github.com/apache/flink/pull/13479#discussion_r503066114
##########
File path: flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
##########
@@ -0,0 +1,471 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.connector.file.src.util.CheckpointedPosition;
+import org.apache.flink.connector.file.src.util.Pool;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
+import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
+import org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import 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 javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+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;
+
+/**
+ * Parquet {@link BulkFormat} that reads data from the file to {@link VectorizedColumnBatch} in
+ * vectorized mode.
+ */
+public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final SerializableConfiguration hadoopConfig;
+ private final String[] projectedFields;
+ private final LogicalType[] projectedTypes;
+ private final ColumnBatchFactory batchFactory;
+ private final int batchSize;
+ private final boolean isUtcTimestamp;
+ private final boolean isCaseSensitive;
+
+ public ParquetVectorizedInputFormat(
+ SerializableConfiguration hadoopConfig,
+ String[] projectedFields,
+ LogicalType[] projectedTypes,
+ ColumnBatchFactory batchFactory,
+ int batchSize,
+ boolean isUtcTimestamp,
+ boolean isCaseSensitive) {
+ Preconditions.checkArgument(
+ projectedFields.length == projectedTypes.length,
+ "The length(%s) of projectedFields should equal to the length(%s) projectedTypes",
+ projectedFields.length,
+ projectedTypes.length);
+
+ this.hadoopConfig = hadoopConfig;
+ this.projectedFields = projectedFields;
+ this.projectedTypes = projectedTypes;
+ this.batchFactory = batchFactory;
+ this.batchSize = batchSize;
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.isCaseSensitive = isCaseSensitive;
+ }
+
+ @Override
+ public ParquetReader createReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength) throws IOException {
+ org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path(filePath.toUri());
+ ParquetMetadata footer = readFooter(
+ hadoopConfig.conf(), hadoopPath, range(splitOffset, splitOffset + splitLength));
+ MessageType fileSchema = footer.getFileMetaData().getSchema();
+ FilterCompat.Filter filter = getFilter(hadoopConfig.conf());
+ List<BlockMetaData> blocks = filterRowGroups(filter, footer.getBlocks(), fileSchema);
+
+ MessageType requestedSchema = clipParquetSchema(fileSchema);
+ ParquetFileReader reader = new ParquetFileReader(
+ hadoopConfig.conf(),
+ footer.getFileMetaData(),
+ hadoopPath,
+ blocks,
+ requestedSchema.getColumns());
+
+ long totalRowCount = 0;
+ for (BlockMetaData block : blocks) {
+ totalRowCount += block.getRowCount();
+ }
+
+ checkSchema(fileSchema, requestedSchema);
+
+ final int numBatchesToCirculate = config.getInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY) + 1;
+ final Pool<ParquetReaderBatch<T>> poolOfBatches =
+ createPoolOfBatches(filePath, requestedSchema, numBatchesToCirculate);
+
+ return new ParquetReader(reader, requestedSchema, totalRowCount, poolOfBatches);
+ }
+
+ @Override
+ public ParquetReader restoreReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength,
+ CheckpointedPosition checkpointedPosition) throws IOException {
+ ParquetReader reader = createReader(config, filePath, splitOffset, splitLength);
+ // Offset is record count too.
+ reader.seek(checkpointedPosition.getOffset() + checkpointedPosition.getRecordsAfterOffset());
Review comment:
I think make sense to use `offset` as a binary offset. Otherwise, it looks strange.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
}, {
"hash" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411",
"triggerID" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"triggerType" : "PUSH"
}, {
"hash" : "acf0eed867c7b85fb44381802649d2ade720e942",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7417",
"triggerID" : "acf0eed867c7b85fb44381802649d2ade720e942",
"triggerType" : "PUSH"
}, {
"hash" : "0513fe1d2ce9768deaba3b7da20b9dff11746a0c",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7439",
"triggerID" : "0513fe1d2ce9768deaba3b7da20b9dff11746a0c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 0513fe1d2ce9768deaba3b7da20b9dff11746a0c Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7439)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #13479:
URL: https://github.com/apache/flink/pull/13479#discussion_r503081381
##########
File path: flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormat.java
##########
@@ -0,0 +1,189 @@
+/*
+ * 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;
+
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.connector.file.src.util.Pool;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
+import org.apache.flink.table.data.ColumnarRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.filesystem.ColumnarRowIterator;
+import org.apache.flink.table.filesystem.PartitionValueConverter;
+import org.apache.flink.table.runtime.typeutils.InternalTypeInfo;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.utils.PartitionPathUtils;
+
+import org.apache.hadoop.conf.Configuration;
+
+import java.util.Arrays;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.stream.Collectors;
+
+import static org.apache.flink.formats.parquet.vector.ParquetSplitReaderUtil.createVectorFromConstant;
+
+/**
+ * A {@link ParquetVectorizedInputFormat} to provide {@link RowData} iterator.
+ * Using {@link ColumnarRowData} to provide a row view of column batch.
+ */
+public class ParquetColumnarRowInputFormat extends ParquetVectorizedInputFormat<RowData> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final RowType producedType;
+
+ /**
+ * Constructor to create parquet format without other fields.
+ */
+ public ParquetColumnarRowInputFormat(
+ Configuration hadoopConfig,
+ String[] projectedFields,
+ LogicalType[] projectedTypes,
+ int batchSize,
+ boolean isUtcTimestamp,
+ boolean isCaseSensitive) {
+ this(
+ hadoopConfig,
+ projectedFields,
+ projectedTypes,
+ RowType.of(projectedTypes, projectedFields),
+ ColumnBatchFactory.DEFAULT,
+ batchSize,
+ isUtcTimestamp,
+ isCaseSensitive);
+ }
+
+ /**
+ * Constructor to create parquet format with other fields created by {@link ColumnBatchFactory}.
+ */
+ public ParquetColumnarRowInputFormat(
+ Configuration hadoopConfig,
+ String[] projectedFields,
+ LogicalType[] projectedTypes,
+ RowType producedType,
+ ColumnBatchFactory batchFactory,
+ int batchSize,
+ boolean isUtcTimestamp,
+ boolean isCaseSensitive) {
+ super(
+ new SerializableConfiguration(hadoopConfig),
+ projectedFields,
+ projectedTypes,
+ batchFactory,
+ batchSize,
+ isUtcTimestamp,
+ isCaseSensitive);
+ this.producedType = producedType;
+ }
+
+ @Override
+ protected ParquetReaderBatch<RowData> createReaderBatch(
+ WritableColumnVector[] writableVectors,
+ VectorizedColumnBatch columnarBatch,
+ Pool.Recycler<ParquetReaderBatch<RowData>> recycler) {
+ return new ColumnarRowReaderBatch(writableVectors, columnarBatch, recycler);
+ }
+
+ @Override
+ public TypeInformation<RowData> getProducedType() {
+ return InternalTypeInfo.of(producedType);
+ }
+
+ private static class ColumnarRowReaderBatch extends ParquetReaderBatch<RowData> {
+
+ private final ColumnarRowIterator result;
+
+ private ColumnarRowReaderBatch(
+ WritableColumnVector[] writableVectors,
+ VectorizedColumnBatch columnarBatch,
+ Pool.Recycler<ParquetReaderBatch<RowData>> recycler) {
+ super(writableVectors, columnarBatch, recycler);
+ this.result = new ColumnarRowIterator(new ColumnarRowData(columnarBatch), this::recycle);
+ }
+
+ @Override
+ public RecordIterator<RowData> convertAndGetIterator(long rowsReturned) {
+ result.set(columnarBatch.getNumRows(), rowsReturned);
+ return result;
+ }
+ }
+
+ /**
+ * Create a partitioned {@link ParquetColumnarRowInputFormat}, the partition columns can be
+ * generated by {@link Path}.
+ */
+ public static ParquetColumnarRowInputFormat createPartitionedFormat(
+ Configuration hadoopConfig,
+ RowType tableType,
+ List<String> partitionKeys,
+ String defaultPartValue,
+ int[] selectedFields,
+ PartitionValueConverter valueConverter,
Review comment:
This can be improved when it takes a Split in the create/restore method, not just the path and position. Then we can get partition directly.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
}, {
"hash" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411",
"triggerID" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"triggerType" : "PUSH"
}, {
"hash" : "acf0eed867c7b85fb44381802649d2ade720e942",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7417",
"triggerID" : "acf0eed867c7b85fb44381802649d2ade720e942",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8b7772756eb215c09ef7a77bec62a2e258661454 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411)
* acf0eed867c7b85fb44381802649d2ade720e942 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7417)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
}, {
"hash" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411",
"triggerID" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"triggerType" : "PUSH"
}, {
"hash" : "acf0eed867c7b85fb44381802649d2ade720e942",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7417",
"triggerID" : "acf0eed867c7b85fb44381802649d2ade720e942",
"triggerType" : "PUSH"
}, {
"hash" : "0513fe1d2ce9768deaba3b7da20b9dff11746a0c",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7439",
"triggerID" : "0513fe1d2ce9768deaba3b7da20b9dff11746a0c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8b7772756eb215c09ef7a77bec62a2e258661454 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411)
* acf0eed867c7b85fb44381802649d2ade720e942 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7417)
* 0513fe1d2ce9768deaba3b7da20b9dff11746a0c Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7439)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 31e024e6f968a5d88f251f910cce133c60b19bfa Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016)
* be67383a2c22ee4fb6a5330510b4549dd029af51 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362)
* 6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
}, {
"hash" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411",
"triggerID" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"triggerType" : "PUSH"
}, {
"hash" : "acf0eed867c7b85fb44381802649d2ade720e942",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7417",
"triggerID" : "acf0eed867c7b85fb44381802649d2ade720e942",
"triggerType" : "PUSH"
}, {
"hash" : "0513fe1d2ce9768deaba3b7da20b9dff11746a0c",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "0513fe1d2ce9768deaba3b7da20b9dff11746a0c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8b7772756eb215c09ef7a77bec62a2e258661454 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411)
* acf0eed867c7b85fb44381802649d2ade720e942 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7417)
* 0513fe1d2ce9768deaba3b7da20b9dff11746a0c UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #13479:
URL: https://github.com/apache/flink/pull/13479#discussion_r503048887
##########
File path: flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
##########
@@ -0,0 +1,471 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.connector.file.src.util.CheckpointedPosition;
+import org.apache.flink.connector.file.src.util.Pool;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
+import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
+import org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import 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 javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+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;
+
+/**
+ * Parquet {@link BulkFormat} that reads data from the file to {@link VectorizedColumnBatch} in
+ * vectorized mode.
+ */
+public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final SerializableConfiguration hadoopConfig;
+ private final String[] projectedFields;
+ private final LogicalType[] projectedTypes;
+ private final ColumnBatchFactory batchFactory;
+ private final int batchSize;
+ private final boolean isUtcTimestamp;
+ private final boolean isCaseSensitive;
+
+ public ParquetVectorizedInputFormat(
+ SerializableConfiguration hadoopConfig,
+ String[] projectedFields,
+ LogicalType[] projectedTypes,
+ ColumnBatchFactory batchFactory,
+ int batchSize,
+ boolean isUtcTimestamp,
+ boolean isCaseSensitive) {
+ Preconditions.checkArgument(
+ projectedFields.length == projectedTypes.length,
+ "The length(%s) of projectedFields should equal to the length(%s) projectedTypes",
+ projectedFields.length,
+ projectedTypes.length);
+
+ this.hadoopConfig = hadoopConfig;
+ this.projectedFields = projectedFields;
+ this.projectedTypes = projectedTypes;
+ this.batchFactory = batchFactory;
+ this.batchSize = batchSize;
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.isCaseSensitive = isCaseSensitive;
+ }
+
+ @Override
+ public ParquetReader createReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength) throws IOException {
+ org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path(filePath.toUri());
+ ParquetMetadata footer = readFooter(
+ hadoopConfig.conf(), hadoopPath, range(splitOffset, splitOffset + splitLength));
+ MessageType fileSchema = footer.getFileMetaData().getSchema();
+ FilterCompat.Filter filter = getFilter(hadoopConfig.conf());
+ List<BlockMetaData> blocks = filterRowGroups(filter, footer.getBlocks(), fileSchema);
+
+ MessageType requestedSchema = clipParquetSchema(fileSchema);
+ ParquetFileReader reader = new ParquetFileReader(
+ hadoopConfig.conf(),
+ footer.getFileMetaData(),
+ hadoopPath,
+ blocks,
+ requestedSchema.getColumns());
+
+ long totalRowCount = 0;
+ for (BlockMetaData block : blocks) {
+ totalRowCount += block.getRowCount();
+ }
+
+ checkSchema(fileSchema, requestedSchema);
+
+ final int numBatchesToCirculate = config.getInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY) + 1;
+ final Pool<ParquetReaderBatch<T>> poolOfBatches =
+ createPoolOfBatches(filePath, requestedSchema, numBatchesToCirculate);
+
+ return new ParquetReader(reader, requestedSchema, totalRowCount, poolOfBatches);
+ }
+
+ @Override
+ public ParquetReader restoreReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength,
+ CheckpointedPosition checkpointedPosition) throws IOException {
+ ParquetReader reader = createReader(config, filePath, splitOffset, splitLength);
+ // Offset is record count too.
+ reader.seek(checkpointedPosition.getOffset() + checkpointedPosition.getRecordsAfterOffset());
+ return reader;
+ }
+
+ @Override
+ public boolean isSplittable() {
+ return true;
+ }
+
+ /**
+ * Clips `parquetSchema` according to `fieldNames`.
+ */
+ private MessageType clipParquetSchema(GroupType parquetSchema) {
+ Type[] types = new Type[projectedFields.length];
+ if (isCaseSensitive) {
+ for (int i = 0; i < projectedFields.length; ++i) {
+ String fieldName = projectedFields[i];
+ if (parquetSchema.getFieldIndex(fieldName) < 0) {
+ throw new IllegalArgumentException(fieldName + " does not exist");
+ }
+ types[i] = parquetSchema.getType(fieldName);
+ }
+ } else {
+ Map<String, Type> caseInsensitiveFieldMap = new HashMap<>();
+ for (Type type : parquetSchema.getFields()) {
+ caseInsensitiveFieldMap.compute(type.getName().toLowerCase(Locale.ROOT),
+ (key, previousType) -> {
+ if (previousType != null) {
+ throw new FlinkRuntimeException(
+ "Parquet with case insensitive mode should have no duplicate key: " + key);
+ }
+ return type;
+ });
+ }
+ for (int i = 0; i < projectedFields.length; ++i) {
+ Type type = caseInsensitiveFieldMap.get(projectedFields[i].toLowerCase(Locale.ROOT));
+ if (type == null) {
+ throw new IllegalArgumentException(projectedFields[i] + " does not exist");
+ }
+ // TODO clip for array,map,row types.
+ types[i] = type;
+ }
+ }
+
+ return Types.buildMessage().addFields(types).named("flink-parquet");
+ }
+
+ private void checkSchema(
+ MessageType fileSchema,
+ MessageType requestedSchema) throws IOException, UnsupportedOperationException {
+ if (projectedFields.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));
+ }
+ }
+ }
+ }
+
+ private Pool<ParquetReaderBatch<T>> createPoolOfBatches(
+ Path filePath, MessageType requestedSchema, int numBatches) {
+ final Pool<ParquetReaderBatch<T>> pool = new Pool<>(numBatches);
+
+ for (int i = 0; i < numBatches; i++) {
+ pool.add(createReaderBatch(filePath, requestedSchema, pool.recycler()));
+ }
+
+ return pool;
+ }
+
+ private ParquetReaderBatch<T> createReaderBatch(
+ Path filePath,
+ MessageType requestedSchema,
+ Pool.Recycler<ParquetReaderBatch<T>> recycler) {
+ WritableColumnVector[] writableVectors = createWritableVectors(requestedSchema);
+ VectorizedColumnBatch columnarBatch =
+ batchFactory.create(filePath, createReadableVectors(writableVectors));
+ return createReaderBatch(writableVectors, columnarBatch, recycler);
+ }
+
+ private WritableColumnVector[] createWritableVectors(MessageType requestedSchema) {
+ WritableColumnVector[] columns = new WritableColumnVector[projectedTypes.length];
+ for (int i = 0; i < projectedTypes.length; i++) {
+ columns[i] = createWritableColumnVector(
+ batchSize,
+ projectedTypes[i],
+ requestedSchema.getColumns().get(i).getPrimitiveType());
+ }
+ return columns;
+ }
+
+ /**
+ * Create readable vectors from writable vectors.
+ * Especially for decimal, see {@link ParquetDecimalVector}.
+ */
+ private ColumnVector[] createReadableVectors(WritableColumnVector[] writableVectors) {
+ ColumnVector[] vectors = new ColumnVector[writableVectors.length];
+ for (int i = 0; i < writableVectors.length; i++) {
+ vectors[i] = projectedTypes[i].getTypeRoot() == LogicalTypeRoot.DECIMAL ?
+ new ParquetDecimalVector(writableVectors[i]) :
+ writableVectors[i];
+ }
+ return vectors;
+ }
+
+ private class ParquetReader implements BulkFormat.Reader<T> {
+
+ private ParquetFileReader reader;
+
+ 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 Pool<ParquetReaderBatch<T>> pool;
+
+ /**
+ * 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;
+
+ /**
+ * 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.
+ */
+ @SuppressWarnings("rawtypes")
+ private ColumnReader[] columnReaders;
+
+ private long recordsToSkip;
+
+ private ParquetReader(
+ ParquetFileReader reader,
+ MessageType requestedSchema,
+ long totalRowCount,
+ Pool<ParquetReaderBatch<T>> pool) {
+ this.reader = reader;
+ this.requestedSchema = requestedSchema;
+ this.totalRowCount = totalRowCount;
+ this.pool = pool;
+ this.rowsReturned = 0;
+ this.totalCountLoadedSoFar = 0;
+ this.recordsToSkip = 0;
+ }
+
+ @Nullable
+ @Override
+ public RecordIterator<T> readBatch() throws IOException {
+ final ParquetReaderBatch<T> batch = getCachedEntry();
+
+ final long startingRowNum = rowsReturned;
+ if (!nextBatch(batch)) {
+ batch.recycle();
+ return null;
+ }
+
+ final RecordIterator<T> records = batch.convertAndGetIterator(startingRowNum);
+ if (recordsToSkip > 0) {
Review comment:
It is necessary, It is possible to read only half of the batch data.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-699805304
CC: @lirui-apache
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] StephanEwen commented on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
StephanEwen commented on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-707573526
One more comment: I am not sure that it is good to reuse the `FileSourceTextLinesITCase` for other tests. That is an ITCase that is rather heavy and not very targeted. It is only there so we have one end-2-end execution of a file source.
Most of what this test represent is not relevant or helpful to execute for other formats. Instead, what is much more relevant is splitting/checkpointing/restoring, which is not tested here at all.
I would suggest to not use this test and instead we should create a testing base for format readers that test the above mentioned aspects (splitting/checkpointing/restoring). The https://github.com/apache/flink/blob/master/flink-connectors/flink-connector-files/src/test/java/org/apache/flink/connector/file/src/impl/AdapterTestBase.java goes in that direction.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lirui-apache commented on a change in pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
lirui-apache commented on a change in pull request #13479:
URL: https://github.com/apache/flink/pull/13479#discussion_r503061879
##########
File path: flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
##########
@@ -0,0 +1,471 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.connector.file.src.util.CheckpointedPosition;
+import org.apache.flink.connector.file.src.util.Pool;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
+import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
+import org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import 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 javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+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;
+
+/**
+ * Parquet {@link BulkFormat} that reads data from the file to {@link VectorizedColumnBatch} in
+ * vectorized mode.
+ */
+public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final SerializableConfiguration hadoopConfig;
+ private final String[] projectedFields;
+ private final LogicalType[] projectedTypes;
+ private final ColumnBatchFactory batchFactory;
+ private final int batchSize;
+ private final boolean isUtcTimestamp;
+ private final boolean isCaseSensitive;
+
+ public ParquetVectorizedInputFormat(
+ SerializableConfiguration hadoopConfig,
+ String[] projectedFields,
+ LogicalType[] projectedTypes,
+ ColumnBatchFactory batchFactory,
+ int batchSize,
+ boolean isUtcTimestamp,
+ boolean isCaseSensitive) {
+ Preconditions.checkArgument(
+ projectedFields.length == projectedTypes.length,
+ "The length(%s) of projectedFields should equal to the length(%s) projectedTypes",
+ projectedFields.length,
+ projectedTypes.length);
+
+ this.hadoopConfig = hadoopConfig;
+ this.projectedFields = projectedFields;
+ this.projectedTypes = projectedTypes;
+ this.batchFactory = batchFactory;
+ this.batchSize = batchSize;
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.isCaseSensitive = isCaseSensitive;
+ }
+
+ @Override
+ public ParquetReader createReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength) throws IOException {
+ org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path(filePath.toUri());
+ ParquetMetadata footer = readFooter(
+ hadoopConfig.conf(), hadoopPath, range(splitOffset, splitOffset + splitLength));
+ MessageType fileSchema = footer.getFileMetaData().getSchema();
+ FilterCompat.Filter filter = getFilter(hadoopConfig.conf());
+ List<BlockMetaData> blocks = filterRowGroups(filter, footer.getBlocks(), fileSchema);
+
+ MessageType requestedSchema = clipParquetSchema(fileSchema);
+ ParquetFileReader reader = new ParquetFileReader(
+ hadoopConfig.conf(),
+ footer.getFileMetaData(),
+ hadoopPath,
+ blocks,
+ requestedSchema.getColumns());
+
+ long totalRowCount = 0;
+ for (BlockMetaData block : blocks) {
+ totalRowCount += block.getRowCount();
+ }
+
+ checkSchema(fileSchema, requestedSchema);
+
+ final int numBatchesToCirculate = config.getInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY) + 1;
+ final Pool<ParquetReaderBatch<T>> poolOfBatches =
+ createPoolOfBatches(filePath, requestedSchema, numBatchesToCirculate);
+
+ return new ParquetReader(reader, requestedSchema, totalRowCount, poolOfBatches);
+ }
+
+ @Override
+ public ParquetReader restoreReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength,
+ CheckpointedPosition checkpointedPosition) throws IOException {
+ ParquetReader reader = createReader(config, filePath, splitOffset, splitLength);
+ // Offset is record count too.
+ reader.seek(checkpointedPosition.getOffset() + checkpointedPosition.getRecordsAfterOffset());
+ return reader;
+ }
+
+ @Override
+ public boolean isSplittable() {
+ return true;
+ }
+
+ /**
+ * Clips `parquetSchema` according to `fieldNames`.
+ */
+ private MessageType clipParquetSchema(GroupType parquetSchema) {
+ Type[] types = new Type[projectedFields.length];
+ if (isCaseSensitive) {
+ for (int i = 0; i < projectedFields.length; ++i) {
+ String fieldName = projectedFields[i];
+ if (parquetSchema.getFieldIndex(fieldName) < 0) {
+ throw new IllegalArgumentException(fieldName + " does not exist");
+ }
+ types[i] = parquetSchema.getType(fieldName);
+ }
+ } else {
+ Map<String, Type> caseInsensitiveFieldMap = new HashMap<>();
+ for (Type type : parquetSchema.getFields()) {
+ caseInsensitiveFieldMap.compute(type.getName().toLowerCase(Locale.ROOT),
+ (key, previousType) -> {
+ if (previousType != null) {
+ throw new FlinkRuntimeException(
+ "Parquet with case insensitive mode should have no duplicate key: " + key);
+ }
+ return type;
+ });
+ }
+ for (int i = 0; i < projectedFields.length; ++i) {
+ Type type = caseInsensitiveFieldMap.get(projectedFields[i].toLowerCase(Locale.ROOT));
+ if (type == null) {
+ throw new IllegalArgumentException(projectedFields[i] + " does not exist");
+ }
+ // TODO clip for array,map,row types.
+ types[i] = type;
+ }
+ }
+
+ return Types.buildMessage().addFields(types).named("flink-parquet");
+ }
+
+ private void checkSchema(
+ MessageType fileSchema,
+ MessageType requestedSchema) throws IOException, UnsupportedOperationException {
+ if (projectedFields.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));
+ }
+ }
+ }
+ }
+
+ private Pool<ParquetReaderBatch<T>> createPoolOfBatches(
+ Path filePath, MessageType requestedSchema, int numBatches) {
+ final Pool<ParquetReaderBatch<T>> pool = new Pool<>(numBatches);
+
+ for (int i = 0; i < numBatches; i++) {
+ pool.add(createReaderBatch(filePath, requestedSchema, pool.recycler()));
+ }
+
+ return pool;
+ }
+
+ private ParquetReaderBatch<T> createReaderBatch(
+ Path filePath,
+ MessageType requestedSchema,
+ Pool.Recycler<ParquetReaderBatch<T>> recycler) {
+ WritableColumnVector[] writableVectors = createWritableVectors(requestedSchema);
+ VectorizedColumnBatch columnarBatch =
+ batchFactory.create(filePath, createReadableVectors(writableVectors));
+ return createReaderBatch(writableVectors, columnarBatch, recycler);
+ }
+
+ private WritableColumnVector[] createWritableVectors(MessageType requestedSchema) {
+ WritableColumnVector[] columns = new WritableColumnVector[projectedTypes.length];
+ for (int i = 0; i < projectedTypes.length; i++) {
+ columns[i] = createWritableColumnVector(
+ batchSize,
+ projectedTypes[i],
+ requestedSchema.getColumns().get(i).getPrimitiveType());
+ }
+ return columns;
+ }
+
+ /**
+ * Create readable vectors from writable vectors.
+ * Especially for decimal, see {@link ParquetDecimalVector}.
+ */
+ private ColumnVector[] createReadableVectors(WritableColumnVector[] writableVectors) {
+ ColumnVector[] vectors = new ColumnVector[writableVectors.length];
+ for (int i = 0; i < writableVectors.length; i++) {
+ vectors[i] = projectedTypes[i].getTypeRoot() == LogicalTypeRoot.DECIMAL ?
+ new ParquetDecimalVector(writableVectors[i]) :
+ writableVectors[i];
+ }
+ return vectors;
+ }
+
+ private class ParquetReader implements BulkFormat.Reader<T> {
+
+ private ParquetFileReader reader;
+
+ 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 Pool<ParquetReaderBatch<T>> pool;
+
+ /**
+ * 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;
+
+ /**
+ * 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.
+ */
+ @SuppressWarnings("rawtypes")
+ private ColumnReader[] columnReaders;
+
+ private long recordsToSkip;
+
+ private ParquetReader(
+ ParquetFileReader reader,
+ MessageType requestedSchema,
+ long totalRowCount,
+ Pool<ParquetReaderBatch<T>> pool) {
+ this.reader = reader;
+ this.requestedSchema = requestedSchema;
+ this.totalRowCount = totalRowCount;
+ this.pool = pool;
+ this.rowsReturned = 0;
+ this.totalCountLoadedSoFar = 0;
+ this.recordsToSkip = 0;
+ }
+
+ @Nullable
+ @Override
+ public RecordIterator<T> readBatch() throws IOException {
+ final ParquetReaderBatch<T> batch = getCachedEntry();
+
+ final long startingRowNum = rowsReturned;
+ if (!nextBatch(batch)) {
+ batch.recycle();
+ return null;
+ }
+
+ final RecordIterator<T> records = batch.convertAndGetIterator(startingRowNum);
+ if (recordsToSkip > 0) {
Review comment:
I meant we can simply call `skipRecord(records)` here. It checks `recordsToSkip > 0` by itself, no?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
}, {
"hash" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369)
* 8b7772756eb215c09ef7a77bec62a2e258661454 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953)
* 31e024e6f968a5d88f251f910cce133c60b19bfa Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
}, {
"hash" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411",
"triggerID" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"triggerType" : "PUSH"
}, {
"hash" : "acf0eed867c7b85fb44381802649d2ade720e942",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "acf0eed867c7b85fb44381802649d2ade720e942",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 8b7772756eb215c09ef7a77bec62a2e258661454 Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411)
* acf0eed867c7b85fb44381802649d2ade720e942 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot commented on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot commented on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698822482
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 31e024e6f968a5d88f251f910cce133c60b19bfa Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016)
* be67383a2c22ee4fb6a5330510b4549dd029af51 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-707590479
Thanks @StephanEwen for pointing out this problem, I agree with you,
- Created https://github.com/apache/flink/pull/13610 for fixing it.
- Reopened FLINK-19414
- Created https://issues.apache.org/jira/browse/FLINK-19613 for testing that we should do.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #13479:
URL: https://github.com/apache/flink/pull/13479#discussion_r503034628
##########
File path: flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
##########
@@ -0,0 +1,471 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.connector.file.src.util.CheckpointedPosition;
+import org.apache.flink.connector.file.src.util.Pool;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
+import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
+import org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import 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 javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+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;
+
+/**
+ * Parquet {@link BulkFormat} that reads data from the file to {@link VectorizedColumnBatch} in
+ * vectorized mode.
+ */
+public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final SerializableConfiguration hadoopConfig;
+ private final String[] projectedFields;
+ private final LogicalType[] projectedTypes;
+ private final ColumnBatchFactory batchFactory;
+ private final int batchSize;
+ private final boolean isUtcTimestamp;
+ private final boolean isCaseSensitive;
+
+ public ParquetVectorizedInputFormat(
+ SerializableConfiguration hadoopConfig,
+ String[] projectedFields,
+ LogicalType[] projectedTypes,
+ ColumnBatchFactory batchFactory,
+ int batchSize,
+ boolean isUtcTimestamp,
+ boolean isCaseSensitive) {
+ Preconditions.checkArgument(
+ projectedFields.length == projectedTypes.length,
+ "The length(%s) of projectedFields should equal to the length(%s) projectedTypes",
+ projectedFields.length,
+ projectedTypes.length);
+
+ this.hadoopConfig = hadoopConfig;
+ this.projectedFields = projectedFields;
+ this.projectedTypes = projectedTypes;
+ this.batchFactory = batchFactory;
+ this.batchSize = batchSize;
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.isCaseSensitive = isCaseSensitive;
+ }
+
+ @Override
+ public ParquetReader createReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength) throws IOException {
+ org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path(filePath.toUri());
+ ParquetMetadata footer = readFooter(
+ hadoopConfig.conf(), hadoopPath, range(splitOffset, splitOffset + splitLength));
+ MessageType fileSchema = footer.getFileMetaData().getSchema();
+ FilterCompat.Filter filter = getFilter(hadoopConfig.conf());
+ List<BlockMetaData> blocks = filterRowGroups(filter, footer.getBlocks(), fileSchema);
+
+ MessageType requestedSchema = clipParquetSchema(fileSchema);
+ ParquetFileReader reader = new ParquetFileReader(
+ hadoopConfig.conf(),
+ footer.getFileMetaData(),
+ hadoopPath,
+ blocks,
+ requestedSchema.getColumns());
+
+ long totalRowCount = 0;
+ for (BlockMetaData block : blocks) {
+ totalRowCount += block.getRowCount();
+ }
+
+ checkSchema(fileSchema, requestedSchema);
+
+ final int numBatchesToCirculate = config.getInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY) + 1;
+ final Pool<ParquetReaderBatch<T>> poolOfBatches =
+ createPoolOfBatches(filePath, requestedSchema, numBatchesToCirculate);
+
+ return new ParquetReader(reader, requestedSchema, totalRowCount, poolOfBatches);
+ }
+
+ @Override
+ public ParquetReader restoreReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength,
+ CheckpointedPosition checkpointedPosition) throws IOException {
+ ParquetReader reader = createReader(config, filePath, splitOffset, splitLength);
+ // Offset is record count too.
+ reader.seek(checkpointedPosition.getOffset() + checkpointedPosition.getRecordsAfterOffset());
Review comment:
Good point
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
}, {
"hash" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411",
"triggerID" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369)
* 8b7772756eb215c09ef7a77bec62a2e258661454 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi merged pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
JingsongLi merged pull request #13479:
URL: https://github.com/apache/flink/pull/13479
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #13479:
URL: https://github.com/apache/flink/pull/13479#discussion_r503054664
##########
File path: flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
##########
@@ -0,0 +1,471 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.connector.file.src.util.CheckpointedPosition;
+import org.apache.flink.connector.file.src.util.Pool;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
+import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
+import org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import 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 javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+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;
+
+/**
+ * Parquet {@link BulkFormat} that reads data from the file to {@link VectorizedColumnBatch} in
+ * vectorized mode.
+ */
+public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final SerializableConfiguration hadoopConfig;
+ private final String[] projectedFields;
+ private final LogicalType[] projectedTypes;
+ private final ColumnBatchFactory batchFactory;
+ private final int batchSize;
+ private final boolean isUtcTimestamp;
+ private final boolean isCaseSensitive;
+
+ public ParquetVectorizedInputFormat(
+ SerializableConfiguration hadoopConfig,
+ String[] projectedFields,
+ LogicalType[] projectedTypes,
+ ColumnBatchFactory batchFactory,
+ int batchSize,
+ boolean isUtcTimestamp,
+ boolean isCaseSensitive) {
+ Preconditions.checkArgument(
+ projectedFields.length == projectedTypes.length,
+ "The length(%s) of projectedFields should equal to the length(%s) projectedTypes",
+ projectedFields.length,
+ projectedTypes.length);
+
+ this.hadoopConfig = hadoopConfig;
+ this.projectedFields = projectedFields;
+ this.projectedTypes = projectedTypes;
+ this.batchFactory = batchFactory;
+ this.batchSize = batchSize;
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.isCaseSensitive = isCaseSensitive;
+ }
+
+ @Override
+ public ParquetReader createReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength) throws IOException {
+ org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path(filePath.toUri());
+ ParquetMetadata footer = readFooter(
+ hadoopConfig.conf(), hadoopPath, range(splitOffset, splitOffset + splitLength));
+ MessageType fileSchema = footer.getFileMetaData().getSchema();
+ FilterCompat.Filter filter = getFilter(hadoopConfig.conf());
+ List<BlockMetaData> blocks = filterRowGroups(filter, footer.getBlocks(), fileSchema);
+
+ MessageType requestedSchema = clipParquetSchema(fileSchema);
+ ParquetFileReader reader = new ParquetFileReader(
+ hadoopConfig.conf(),
+ footer.getFileMetaData(),
+ hadoopPath,
+ blocks,
+ requestedSchema.getColumns());
+
+ long totalRowCount = 0;
+ for (BlockMetaData block : blocks) {
+ totalRowCount += block.getRowCount();
+ }
+
+ checkSchema(fileSchema, requestedSchema);
+
+ final int numBatchesToCirculate = config.getInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY) + 1;
+ final Pool<ParquetReaderBatch<T>> poolOfBatches =
+ createPoolOfBatches(filePath, requestedSchema, numBatchesToCirculate);
+
+ return new ParquetReader(reader, requestedSchema, totalRowCount, poolOfBatches);
+ }
+
+ @Override
+ public ParquetReader restoreReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength,
+ CheckpointedPosition checkpointedPosition) throws IOException {
+ ParquetReader reader = createReader(config, filePath, splitOffset, splitLength);
+ // Offset is record count too.
+ reader.seek(checkpointedPosition.getOffset() + checkpointedPosition.getRecordsAfterOffset());
Review comment:
I think we should only use `recordsAfterOffset`. Offset is useless. See comments in `CheckpointedPosition`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 31e024e6f968a5d88f251f910cce133c60b19bfa Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on a change in pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #13479:
URL: https://github.com/apache/flink/pull/13479#discussion_r503071625
##########
File path: flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
##########
@@ -0,0 +1,471 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.connector.file.src.util.CheckpointedPosition;
+import org.apache.flink.connector.file.src.util.Pool;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
+import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
+import org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import 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 javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+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;
+
+/**
+ * Parquet {@link BulkFormat} that reads data from the file to {@link VectorizedColumnBatch} in
+ * vectorized mode.
+ */
+public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final SerializableConfiguration hadoopConfig;
+ private final String[] projectedFields;
+ private final LogicalType[] projectedTypes;
+ private final ColumnBatchFactory batchFactory;
+ private final int batchSize;
+ private final boolean isUtcTimestamp;
+ private final boolean isCaseSensitive;
+
+ public ParquetVectorizedInputFormat(
+ SerializableConfiguration hadoopConfig,
+ String[] projectedFields,
+ LogicalType[] projectedTypes,
+ ColumnBatchFactory batchFactory,
+ int batchSize,
+ boolean isUtcTimestamp,
+ boolean isCaseSensitive) {
+ Preconditions.checkArgument(
+ projectedFields.length == projectedTypes.length,
+ "The length(%s) of projectedFields should equal to the length(%s) projectedTypes",
+ projectedFields.length,
+ projectedTypes.length);
+
+ this.hadoopConfig = hadoopConfig;
+ this.projectedFields = projectedFields;
+ this.projectedTypes = projectedTypes;
+ this.batchFactory = batchFactory;
+ this.batchSize = batchSize;
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.isCaseSensitive = isCaseSensitive;
+ }
+
+ @Override
+ public ParquetReader createReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength) throws IOException {
+ org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path(filePath.toUri());
+ ParquetMetadata footer = readFooter(
+ hadoopConfig.conf(), hadoopPath, range(splitOffset, splitOffset + splitLength));
+ MessageType fileSchema = footer.getFileMetaData().getSchema();
+ FilterCompat.Filter filter = getFilter(hadoopConfig.conf());
+ List<BlockMetaData> blocks = filterRowGroups(filter, footer.getBlocks(), fileSchema);
+
+ MessageType requestedSchema = clipParquetSchema(fileSchema);
+ ParquetFileReader reader = new ParquetFileReader(
+ hadoopConfig.conf(),
+ footer.getFileMetaData(),
+ hadoopPath,
+ blocks,
+ requestedSchema.getColumns());
+
+ long totalRowCount = 0;
+ for (BlockMetaData block : blocks) {
+ totalRowCount += block.getRowCount();
+ }
+
+ checkSchema(fileSchema, requestedSchema);
+
+ final int numBatchesToCirculate = config.getInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY) + 1;
+ final Pool<ParquetReaderBatch<T>> poolOfBatches =
+ createPoolOfBatches(filePath, requestedSchema, numBatchesToCirculate);
+
+ return new ParquetReader(reader, requestedSchema, totalRowCount, poolOfBatches);
+ }
+
+ @Override
+ public ParquetReader restoreReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength,
+ CheckpointedPosition checkpointedPosition) throws IOException {
+ ParquetReader reader = createReader(config, filePath, splitOffset, splitLength);
+ // Offset is record count too.
+ reader.seek(checkpointedPosition.getOffset() + checkpointedPosition.getRecordsAfterOffset());
+ return reader;
+ }
+
+ @Override
+ public boolean isSplittable() {
+ return true;
+ }
+
+ /**
+ * Clips `parquetSchema` according to `fieldNames`.
+ */
+ private MessageType clipParquetSchema(GroupType parquetSchema) {
+ Type[] types = new Type[projectedFields.length];
+ if (isCaseSensitive) {
+ for (int i = 0; i < projectedFields.length; ++i) {
+ String fieldName = projectedFields[i];
+ if (parquetSchema.getFieldIndex(fieldName) < 0) {
+ throw new IllegalArgumentException(fieldName + " does not exist");
+ }
+ types[i] = parquetSchema.getType(fieldName);
+ }
+ } else {
+ Map<String, Type> caseInsensitiveFieldMap = new HashMap<>();
+ for (Type type : parquetSchema.getFields()) {
+ caseInsensitiveFieldMap.compute(type.getName().toLowerCase(Locale.ROOT),
+ (key, previousType) -> {
+ if (previousType != null) {
+ throw new FlinkRuntimeException(
+ "Parquet with case insensitive mode should have no duplicate key: " + key);
+ }
+ return type;
+ });
+ }
+ for (int i = 0; i < projectedFields.length; ++i) {
+ Type type = caseInsensitiveFieldMap.get(projectedFields[i].toLowerCase(Locale.ROOT));
+ if (type == null) {
+ throw new IllegalArgumentException(projectedFields[i] + " does not exist");
+ }
+ // TODO clip for array,map,row types.
+ types[i] = type;
+ }
+ }
+
+ return Types.buildMessage().addFields(types).named("flink-parquet");
+ }
+
+ private void checkSchema(
+ MessageType fileSchema,
+ MessageType requestedSchema) throws IOException, UnsupportedOperationException {
+ if (projectedFields.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));
+ }
+ }
+ }
+ }
+
+ private Pool<ParquetReaderBatch<T>> createPoolOfBatches(
+ Path filePath, MessageType requestedSchema, int numBatches) {
+ final Pool<ParquetReaderBatch<T>> pool = new Pool<>(numBatches);
+
+ for (int i = 0; i < numBatches; i++) {
+ pool.add(createReaderBatch(filePath, requestedSchema, pool.recycler()));
+ }
+
+ return pool;
+ }
+
+ private ParquetReaderBatch<T> createReaderBatch(
+ Path filePath,
+ MessageType requestedSchema,
+ Pool.Recycler<ParquetReaderBatch<T>> recycler) {
+ WritableColumnVector[] writableVectors = createWritableVectors(requestedSchema);
+ VectorizedColumnBatch columnarBatch =
+ batchFactory.create(filePath, createReadableVectors(writableVectors));
+ return createReaderBatch(writableVectors, columnarBatch, recycler);
+ }
+
+ private WritableColumnVector[] createWritableVectors(MessageType requestedSchema) {
+ WritableColumnVector[] columns = new WritableColumnVector[projectedTypes.length];
+ for (int i = 0; i < projectedTypes.length; i++) {
+ columns[i] = createWritableColumnVector(
+ batchSize,
+ projectedTypes[i],
+ requestedSchema.getColumns().get(i).getPrimitiveType());
+ }
+ return columns;
+ }
+
+ /**
+ * Create readable vectors from writable vectors.
+ * Especially for decimal, see {@link ParquetDecimalVector}.
+ */
+ private ColumnVector[] createReadableVectors(WritableColumnVector[] writableVectors) {
+ ColumnVector[] vectors = new ColumnVector[writableVectors.length];
+ for (int i = 0; i < writableVectors.length; i++) {
+ vectors[i] = projectedTypes[i].getTypeRoot() == LogicalTypeRoot.DECIMAL ?
+ new ParquetDecimalVector(writableVectors[i]) :
+ writableVectors[i];
+ }
+ return vectors;
+ }
+
+ private class ParquetReader implements BulkFormat.Reader<T> {
+
+ private ParquetFileReader reader;
+
+ 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 Pool<ParquetReaderBatch<T>> pool;
+
+ /**
+ * 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;
+
+ /**
+ * 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.
+ */
+ @SuppressWarnings("rawtypes")
+ private ColumnReader[] columnReaders;
+
+ private long recordsToSkip;
+
+ private ParquetReader(
+ ParquetFileReader reader,
+ MessageType requestedSchema,
+ long totalRowCount,
+ Pool<ParquetReaderBatch<T>> pool) {
+ this.reader = reader;
+ this.requestedSchema = requestedSchema;
+ this.totalRowCount = totalRowCount;
+ this.pool = pool;
+ this.rowsReturned = 0;
+ this.totalCountLoadedSoFar = 0;
+ this.recordsToSkip = 0;
+ }
+
+ @Nullable
+ @Override
+ public RecordIterator<T> readBatch() throws IOException {
+ final ParquetReaderBatch<T> batch = getCachedEntry();
+
+ final long startingRowNum = rowsReturned;
+ if (!nextBatch(batch)) {
+ batch.recycle();
+ return null;
+ }
+
+ final RecordIterator<T> records = batch.convertAndGetIterator(startingRowNum);
+ if (recordsToSkip > 0) {
Review comment:
I see
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 31e024e6f968a5d88f251f910cce133c60b19bfa Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016)
* be67383a2c22ee4fb6a5330510b4549dd029af51 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] lirui-apache commented on a change in pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
lirui-apache commented on a change in pull request #13479:
URL: https://github.com/apache/flink/pull/13479#discussion_r503060323
##########
File path: flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
##########
@@ -0,0 +1,471 @@
+/*
+ * 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;
+
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.reader.SourceReaderOptions;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
+import org.apache.flink.connector.file.src.util.CheckpointedPosition;
+import org.apache.flink.connector.file.src.util.Pool;
+import org.apache.flink.core.fs.Path;
+import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
+import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
+import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
+import org.apache.flink.formats.parquet.vector.reader.AbstractColumnReader;
+import org.apache.flink.formats.parquet.vector.reader.ColumnReader;
+import org.apache.flink.table.data.vector.ColumnVector;
+import org.apache.flink.table.data.vector.VectorizedColumnBatch;
+import org.apache.flink.table.data.vector.writable.WritableColumnVector;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.flink.util.FlinkRuntimeException;
+import org.apache.flink.util.Preconditions;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import 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 javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+
+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;
+
+/**
+ * Parquet {@link BulkFormat} that reads data from the file to {@link VectorizedColumnBatch} in
+ * vectorized mode.
+ */
+public abstract class ParquetVectorizedInputFormat<T> implements BulkFormat<T> {
+
+ private static final long serialVersionUID = 1L;
+
+ private final SerializableConfiguration hadoopConfig;
+ private final String[] projectedFields;
+ private final LogicalType[] projectedTypes;
+ private final ColumnBatchFactory batchFactory;
+ private final int batchSize;
+ private final boolean isUtcTimestamp;
+ private final boolean isCaseSensitive;
+
+ public ParquetVectorizedInputFormat(
+ SerializableConfiguration hadoopConfig,
+ String[] projectedFields,
+ LogicalType[] projectedTypes,
+ ColumnBatchFactory batchFactory,
+ int batchSize,
+ boolean isUtcTimestamp,
+ boolean isCaseSensitive) {
+ Preconditions.checkArgument(
+ projectedFields.length == projectedTypes.length,
+ "The length(%s) of projectedFields should equal to the length(%s) projectedTypes",
+ projectedFields.length,
+ projectedTypes.length);
+
+ this.hadoopConfig = hadoopConfig;
+ this.projectedFields = projectedFields;
+ this.projectedTypes = projectedTypes;
+ this.batchFactory = batchFactory;
+ this.batchSize = batchSize;
+ this.isUtcTimestamp = isUtcTimestamp;
+ this.isCaseSensitive = isCaseSensitive;
+ }
+
+ @Override
+ public ParquetReader createReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength) throws IOException {
+ org.apache.hadoop.fs.Path hadoopPath = new org.apache.hadoop.fs.Path(filePath.toUri());
+ ParquetMetadata footer = readFooter(
+ hadoopConfig.conf(), hadoopPath, range(splitOffset, splitOffset + splitLength));
+ MessageType fileSchema = footer.getFileMetaData().getSchema();
+ FilterCompat.Filter filter = getFilter(hadoopConfig.conf());
+ List<BlockMetaData> blocks = filterRowGroups(filter, footer.getBlocks(), fileSchema);
+
+ MessageType requestedSchema = clipParquetSchema(fileSchema);
+ ParquetFileReader reader = new ParquetFileReader(
+ hadoopConfig.conf(),
+ footer.getFileMetaData(),
+ hadoopPath,
+ blocks,
+ requestedSchema.getColumns());
+
+ long totalRowCount = 0;
+ for (BlockMetaData block : blocks) {
+ totalRowCount += block.getRowCount();
+ }
+
+ checkSchema(fileSchema, requestedSchema);
+
+ final int numBatchesToCirculate = config.getInteger(SourceReaderOptions.ELEMENT_QUEUE_CAPACITY) + 1;
+ final Pool<ParquetReaderBatch<T>> poolOfBatches =
+ createPoolOfBatches(filePath, requestedSchema, numBatchesToCirculate);
+
+ return new ParquetReader(reader, requestedSchema, totalRowCount, poolOfBatches);
+ }
+
+ @Override
+ public ParquetReader restoreReader(
+ Configuration config,
+ Path filePath,
+ long splitOffset,
+ long splitLength,
+ CheckpointedPosition checkpointedPosition) throws IOException {
+ ParquetReader reader = createReader(config, filePath, splitOffset, splitLength);
+ // Offset is record count too.
+ reader.seek(checkpointedPosition.getOffset() + checkpointedPosition.getRecordsAfterOffset());
Review comment:
I wasn't using `offset` for my Hive source PoC. Honestly, I don't quite understand the difference between `offset` and `recordsAfterOffset`. Could you elaborate on these and how should they be used?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* be67383a2c22ee4fb6a5330510b4549dd029af51 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362)
* 6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "SUCCESS",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 31e024e6f968a5d88f251f910cce133c60b19bfa Azure: [SUCCESS](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016)
* be67383a2c22ee4fb6a5330510b4549dd029af51 Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362)
* 6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] JingsongLi commented on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-706923294
@lirui-apache Please take a look~
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "FAILURE",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d Azure: [FAILURE](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953)
* 31e024e6f968a5d88f251f910cce133c60b19bfa UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink] flinkbot edited a comment on pull request #13479: [FLINK-19414][parquet] Introduce ParquetColumnarRowInputFormat
Posted by GitBox <gi...@apache.org>.
flinkbot edited a comment on pull request #13479:
URL: https://github.com/apache/flink/pull/13479#issuecomment-698831350
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=6953",
"triggerID" : "9df0a7e3cff230a37dbd41f1edfb57ba9c060d0d",
"triggerType" : "PUSH"
}, {
"hash" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7016",
"triggerID" : "31e024e6f968a5d88f251f910cce133c60b19bfa",
"triggerType" : "PUSH"
}, {
"hash" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7362",
"triggerID" : "be67383a2c22ee4fb6a5330510b4549dd029af51",
"triggerType" : "PUSH"
}, {
"hash" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7369",
"triggerID" : "6fc43dcfa07a2f1488f33d54d26ed8c4bdeff2ef",
"triggerType" : "PUSH"
}, {
"hash" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"status" : "DELETED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7411",
"triggerID" : "8b7772756eb215c09ef7a77bec62a2e258661454",
"triggerType" : "PUSH"
}, {
"hash" : "acf0eed867c7b85fb44381802649d2ade720e942",
"status" : "CANCELED",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7417",
"triggerID" : "acf0eed867c7b85fb44381802649d2ade720e942",
"triggerType" : "PUSH"
}, {
"hash" : "0513fe1d2ce9768deaba3b7da20b9dff11746a0c",
"status" : "PENDING",
"url" : "https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7439",
"triggerID" : "0513fe1d2ce9768deaba3b7da20b9dff11746a0c",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* acf0eed867c7b85fb44381802649d2ade720e942 Azure: [CANCELED](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7417)
* 0513fe1d2ce9768deaba3b7da20b9dff11746a0c Azure: [PENDING](https://dev.azure.com/apache-flink/98463496-1af2-4620-8eab-a2ecc1a2e6fe/_build/results?buildId=7439)
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run travis` re-run the last Travis build
- `@flinkbot run azure` re-run the last Azure build
</details>
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org