You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2021/08/31 07:38:33 UTC
[flink] branch master updated: [FLINK-23715][parquet] Support for
reading fields that do not exist in Parquet files
This is an automated email from the ASF dual-hosted git repository.
lirui pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 8abf8c6 [FLINK-23715][parquet] Support for reading fields that do not exist in Parquet files
8abf8c6 is described below
commit 8abf8c627ca5ceab7a7e4f5b40f89b7ccceb9ceb
Author: Tartarus0zm <zh...@163.com>
AuthorDate: Wed Aug 11 16:11:36 2021 +0800
[FLINK-23715][parquet] Support for reading fields that do not exist in Parquet files
This closes #16778
---
.../parquet/ParquetVectorizedInputFormat.java | 52 +++++++++++++++++-----
.../parquet/utils/ParquetSchemaConverter.java | 2 +-
.../parquet/ParquetColumnarRowInputFormatTest.java | 41 +++++++++++++++++
3 files changed, 82 insertions(+), 13 deletions(-)
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
index 341cf59..5246334 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/ParquetVectorizedInputFormat.java
@@ -25,6 +25,7 @@ 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.ParquetSchemaConverter;
import org.apache.flink.formats.parquet.utils.SerializableConfiguration;
import org.apache.flink.formats.parquet.vector.ColumnBatchFactory;
import org.apache.flink.formats.parquet.vector.ParquetDecimalVector;
@@ -49,15 +50,19 @@ 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 org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
import java.io.IOException;
import java.util.Arrays;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Locale;
import java.util.Map;
+import java.util.Set;
import static org.apache.flink.formats.parquet.vector.ParquetSplitReaderUtil.createColumnReader;
import static org.apache.flink.formats.parquet.vector.ParquetSplitReaderUtil.createWritableColumnVector;
@@ -73,6 +78,7 @@ import static org.apache.parquet.hadoop.ParquetInputFormat.getFilter;
public abstract class ParquetVectorizedInputFormat<T, SplitT extends FileSourceSplit>
implements BulkFormat<T, SplitT> {
+ private static final Logger LOG = LoggerFactory.getLogger(ParquetVectorizedInputFormat.class);
private static final long serialVersionUID = 1L;
private final SerializableConfiguration hadoopConfig;
@@ -82,6 +88,7 @@ public abstract class ParquetVectorizedInputFormat<T, SplitT extends FileSourceS
private final int batchSize;
private final boolean isUtcTimestamp;
private final boolean isCaseSensitive;
+ private final Set<Integer> unknownFieldsIndices = new HashSet<>();
public ParquetVectorizedInputFormat(
SerializableConfiguration hadoopConfig,
@@ -169,10 +176,18 @@ public abstract class ParquetVectorizedInputFormat<T, SplitT extends FileSourceS
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");
+ if (!parquetSchema.containsField(fieldName)) {
+ LOG.warn(
+ "{} does not exist in {}, will fill the field with null.",
+ fieldName,
+ parquetSchema);
+ types[i] =
+ ParquetSchemaConverter.convertToParquetType(
+ fieldName, projectedTypes[i]);
+ unknownFieldsIndices.add(i);
+ } else {
+ types[i] = parquetSchema.getType(fieldName);
}
- types[i] = parquetSchema.getType(fieldName);
}
} else {
Map<String, Type> caseInsensitiveFieldMap = new HashMap<>();
@@ -192,7 +207,14 @@ public abstract class ParquetVectorizedInputFormat<T, SplitT extends FileSourceS
Type type =
caseInsensitiveFieldMap.get(projectedFields[i].toLowerCase(Locale.ROOT));
if (type == null) {
- throw new IllegalArgumentException(projectedFields[i] + " does not exist");
+ LOG.warn(
+ "{} does not exist in {}, will fill the field with null.",
+ projectedFields[i],
+ parquetSchema);
+ type =
+ ParquetSchemaConverter.convertToParquetType(
+ projectedFields[i].toLowerCase(Locale.ROOT), projectedTypes[i]);
+ unknownFieldsIndices.add(i);
}
// TODO clip for array,map,row types.
types[i] = type;
@@ -361,8 +383,12 @@ public abstract class ParquetVectorizedInputFormat<T, SplitT extends FileSourceS
int num = (int) Math.min(batchSize, totalCountLoadedSoFar - rowsReturned);
for (int i = 0; i < columnReaders.length; ++i) {
- //noinspection unchecked
- columnReaders[i].readToVector(num, batch.writableVectors[i]);
+ if (columnReaders[i] == null) {
+ batch.writableVectors[i].fillWithNulls();
+ } else {
+ //noinspection unchecked
+ columnReaders[i].readToVector(num, batch.writableVectors[i]);
+ }
}
rowsReturned += num;
batch.columnarBatch.setNumRows(num);
@@ -381,12 +407,14 @@ public abstract class ParquetVectorizedInputFormat<T, SplitT extends FileSourceS
List<ColumnDescriptor> columns = requestedSchema.getColumns();
columnReaders = new AbstractColumnReader[columns.size()];
for (int i = 0; i < columns.size(); ++i) {
- columnReaders[i] =
- createColumnReader(
- isUtcTimestamp,
- projectedTypes[i],
- columns.get(i),
- pages.getPageReader(columns.get(i)));
+ if (!unknownFieldsIndices.contains(i)) {
+ columnReaders[i] =
+ createColumnReader(
+ isUtcTimestamp,
+ projectedTypes[i],
+ columns.get(i),
+ pages.getPageReader(columns.get(i)));
+ }
}
totalCountLoadedSoFar += pages.getRowCount();
}
diff --git a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/ParquetSchemaConverter.java b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/ParquetSchemaConverter.java
index 37aa43f..b3a0296 100644
--- a/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/ParquetSchemaConverter.java
+++ b/flink-formats/flink-parquet/src/main/java/org/apache/flink/formats/parquet/utils/ParquetSchemaConverter.java
@@ -39,7 +39,7 @@ public class ParquetSchemaConverter {
return new MessageType(name, types);
}
- private static Type convertToParquetType(String name, LogicalType type) {
+ public static Type convertToParquetType(String name, LogicalType type) {
return convertToParquetType(name, type, Type.Repetition.OPTIONAL);
}
diff --git a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormatTest.java b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormatTest.java
index 63f2116..8cb2af3 100644
--- a/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormatTest.java
+++ b/flink-formats/flink-parquet/src/test/java/org/apache/flink/formats/parquet/ParquetColumnarRowInputFormatTest.java
@@ -254,6 +254,47 @@ public class ParquetColumnarRowInputFormatTest {
}
@Test
+ public void testProjectionReadUnknownField() throws IOException {
+ int number = 1000;
+ List<Row> records = new ArrayList<>(number);
+ for (int i = 0; i < number; i++) {
+ Integer v = i;
+ records.add(newRow(v));
+ }
+
+ Path testPath =
+ createTempParquetFile(
+ TEMPORARY_FOLDER.newFolder(), PARQUET_SCHEMA, records, rowGroupSize);
+
+ // test reader
+ LogicalType[] fieldTypes =
+ new LogicalType[] {
+ new DoubleType(), new TinyIntType(), new IntType(), new VarCharType()
+ };
+ ParquetColumnarRowInputFormat<FileSourceSplit> format =
+ new ParquetColumnarRowInputFormat<>(
+ new Configuration(),
+ // f99 not exist in parquet file.
+ RowType.of(fieldTypes, new String[] {"f7", "f2", "f4", "f99"}),
+ 500,
+ false,
+ true);
+
+ AtomicInteger cnt = new AtomicInteger(0);
+ forEachRemaining(
+ format.createReader(
+ EMPTY_CONF, new FileSourceSplit("id", testPath, 0, Long.MAX_VALUE)),
+ row -> {
+ int i = cnt.get();
+ assertEquals(i, row.getDouble(0), 0);
+ assertEquals((byte) i, row.getByte(1));
+ assertEquals(i, row.getInt(2));
+ assertTrue(row.isNullAt(3));
+ cnt.incrementAndGet();
+ });
+ }
+
+ @Test
public void testPartitionValues() throws IOException {
// prepare parquet file
int number = 1000;