Flink: add _row_id and _last_updated_sequence_number readers #14148
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -34,7 +34,6 @@ | |
| import org.apache.flink.table.data.RowData; | ||
| import org.apache.flink.table.data.StringData; | ||
| import org.apache.flink.table.data.TimestampData; | ||
| import org.apache.iceberg.Schema; | ||
| import org.apache.iceberg.parquet.ParquetSchemaUtil; | ||
| import org.apache.iceberg.parquet.ParquetValueReader; | ||
|
|
@@ -90,10 +89,13 @@ public ParquetValueReader<RowData> message( | |
| @SuppressWarnings("checkstyle:CyclomaticComplexity") | ||
| public ParquetValueReader<RowData> struct( | ||
| Types.StructType expected, GroupType struct, List<ParquetValueReader<?>> fieldReaders) { | ||
|
|
||
| if (null == expected) { | ||
| return new RowDataReader(ImmutableList.of()); | ||
| } | ||
|
|
||
| // match the expected struct's order | ||
| Map<Integer, ParquetValueReader<?>> readersById = Maps.newHashMap(); | ||
| List<Type> fields = struct.getFields(); | ||
| for (int i = 0; i < fields.size(); i += 1) { | ||
| Type fieldType = fields.get(i); | ||
|
|
@@ -102,51 +104,39 @@ public ParquetValueReader<RowData> struct( | |
| if (fieldType.getId() != null) { | ||
| int id = fieldType.getId().intValue(); | ||
| readersById.put(id, ParquetValueReaders.option(fieldType, fieldD, fieldReaders.get(i))); | ||
| } | ||
| } | ||
| } | ||
|
|
||
| int constantDefinitionLevel = type.getMaxDefinitionLevel(currentPath()); | ||
| List<Types.NestedField> expectedFields = expected.fields(); | ||
| List<ParquetValueReader<?>> reorderedFields = | ||
| Lists.newArrayListWithExpectedSize(expectedFields.size()); | ||
| for (Types.NestedField field : expectedFields) { | ||
| int id = field.fieldId(); | ||
| ParquetValueReader<?> reader = | ||
| ParquetValueReaders.replaceWithMetadataReader( | ||
| id, readersById.get(id), idToConstant, constantDefinitionLevel); | ||
| reorderedFields.add(defaultReader(field, reader, constantDefinitionLevel)); | ||
|
Comment on lines
-121
to
+120
Contributor
There was a problem hiding this comment. Just curious, how did you decide to make this refactoring? You consolidate several branches of the if statement.
Contributor
Author
There was a problem hiding this comment. In #12836, the core layer integrated the same code between Spark and Core. I noticed that the code for Flink is also the same, so I reused it. |
||
| } | ||
|
|
||
| return new RowDataReader(reorderedFields); | ||
| } | ||
|
|
||
| private ParquetValueReader<?> defaultReader( | ||
| Types.NestedField field, ParquetValueReader<?> reader, int constantDL) { | ||
| if (reader != null) { | ||
| return reader; | ||
| } else if (field.initialDefault() != null) { | ||
| return ParquetValueReaders.constant( | ||
| RowDataUtil.convertConstant(field.type(), field.initialDefault()), constantDL); | ||
| } else if (field.isOptional()) { | ||
| return ParquetValueReaders.nulls(); | ||
| } | ||
|
|
||
| throw new IllegalArgumentException(String.format("Missing required field: %s", field.name())); | ||
| } | ||
|
|
||
| @Override | ||
| public ParquetValueReader<?> list( | ||
| Types.ListType expectedList, GroupType array, ParquetValueReader<?> elementReader) { | ||
|
|
||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -42,8 +42,10 @@ | |
| import org.apache.iceberg.data.parquet.GenericParquetWriter; | ||
| import org.apache.iceberg.flink.FlinkSchemaUtil; | ||
| import org.apache.iceberg.flink.TestHelpers; | ||
| import org.apache.iceberg.inmemory.InMemoryOutputFile; | ||
| import org.apache.iceberg.io.CloseableIterable; | ||
| import org.apache.iceberg.io.FileAppender; | ||
| import org.apache.iceberg.io.OutputFile; | ||
| import org.apache.iceberg.parquet.Parquet; | ||
| import org.apache.iceberg.parquet.ParquetValueReader; | ||
| import org.apache.iceberg.relocated.com.google.common.collect.Lists; | ||
|
|
@@ -74,6 +76,11 @@ protected boolean supportsTimestampNanos() { | |
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| protected boolean supportsRowLineage() { | ||
| return true; | ||
| } | ||
|
|
||
| @Test | ||
| public void testBuildReader() { | ||
| MessageType fileSchema = | ||
|
|
@@ -216,29 +223,31 @@ public void testTwoLevelList() throws IOException { | |
|
|
||
| private void writeAndValidate( | ||
| Iterable<Record> iterable, Schema writeSchema, Schema expectedSchema) throws IOException { | ||
|
|
||
| OutputFile output = new InMemoryOutputFile(); | ||
| try (FileAppender<Record> writer = | ||
|
Contributor
There was a problem hiding this comment. What's the reason for moving from local file to in memory?
Contributor
Author
There was a problem hiding this comment. In #12836, spark and Core have changed their behavior, so I aligned this part. |
||
| Parquet.write(output) | ||
| .schema(writeSchema) | ||
| .createWriterFunc(GenericParquetWriter::create) | ||
| .build()) { | ||
| writer.addAll(iterable); | ||
| } | ||
|
|
||
| try (CloseableIterable<RowData> reader = | ||
| Parquet.read(output.toInputFile()) | ||
| .project(expectedSchema) | ||
| .createReaderFunc( | ||
| type -> FlinkParquetReaders.buildReader(expectedSchema, type, ID_TO_CONSTANT)) | ||
| .build()) { | ||
| Iterator<RowData> rows = reader.iterator(); | ||
| LogicalType rowType = FlinkSchemaUtil.convert(writeSchema); | ||
| int pos = 0; | ||
| for (Record record : iterable) { | ||
| assertThat(rows).hasNext(); | ||
| TestHelpers.assertRowData( | ||
| writeSchema.asStruct(), rowType, record, rows.next(), ID_TO_CONSTANT, pos++); | ||
| } | ||
|
|
||
| assertThat(rows).isExhausted(); | ||
| } | ||
| } | ||
|
|
||
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The PR needs to go against the 2.1 target (currently: 2.0), as #13714 has been merged.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Can we focus on the 2.0.0 version for this PR first, and then backport it to 2.1.0 and 1.20 later?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I'm fine with both.