This is an automated email from the ASF dual-hosted git repository.
czweng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new 8e11702b [FLINK-29293] Set level to KeyValue when reading from data
files (#310)
8e11702b is described below
commit 8e11702be1fda0ad5692eaf78fce2ff914c36a62
Author: tsreaper <tsreape...@gmail.com>
AuthorDate: Mon Oct 10 10:06:25 2022 +0800
[FLINK-29293] Set level to KeyValue when reading from data files (#310)
This closes #310.
---
.../apache/flink/table/store/file/KeyValue.java | 29 ++++++++++++++--------
.../file/io/KeyValueDataFileRecordReader.java | 21 +++++++++-------
.../store/file/io/KeyValueFileReaderFactory.java | 5 ++--
.../store/file/mergetree/MergeTreeReader.java | 2 +-
.../file/operation/KeyValueFileStoreRead.java | 2 +-
.../store/file/io/KeyValueFileReadWriteTest.java | 6 +++--
6 files changed, 40 insertions(+), 25 deletions(-)
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValue.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValue.java
index 5f124fe4..fcea7467 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValue.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/KeyValue.java
@@ -39,11 +39,15 @@ import java.util.stream.IntStream;
public class KeyValue {
public static final long UNKNOWN_SEQUENCE = -1;
+ public static final int UNKNOWN_LEVEL = -1;
private RowData key;
+ // determined after written into memory table or read from file
private long sequenceNumber;
private RowKind valueKind;
private RowData value;
+ // determined after read from file
+ private int level;
public KeyValue replace(RowData key, RowKind valueKind, RowData value) {
return replace(key, UNKNOWN_SEQUENCE, valueKind, value);
@@ -54,6 +58,7 @@ public class KeyValue {
this.sequenceNumber = sequenceNumber;
this.valueKind = valueKind;
this.value = value;
+ this.level = UNKNOWN_LEVEL;
return this;
}
@@ -78,6 +83,15 @@ public class KeyValue {
return value;
}
+ public int level() {
+ return level;
+ }
+
+ public KeyValue setLevel(int level) {
+ this.level = level;
+ return this;
+ }
+
public static RowType schema(RowType keyType, RowType valueType) {
List<RowType.RowField> fields = new ArrayList<>(keyType.getFields());
fields.add(new RowType.RowField("_SEQUENCE_NUMBER", new
BigIntType(false)));
@@ -120,22 +134,17 @@ public class KeyValue {
keySerializer.copy(key),
sequenceNumber,
valueKind,
- valueSerializer.copy(value));
+ valueSerializer.copy(value))
+ .setLevel(level);
}
@VisibleForTesting
public String toString(RowType keyType, RowType valueType) {
String keyString = rowDataToString(key, keyType);
String valueString = rowDataToString(value, valueType);
- return "{kind: "
- + valueKind.name()
- + ", seq: "
- + sequenceNumber
- + ", key: ("
- + keyString
- + "), value: ("
- + valueString
- + ")}";
+ return String.format(
+ "{kind: %s, seq: %d, key: (%s), value: (%s), level: %d}",
+ valueKind.name(), sequenceNumber, keyString, valueString,
level);
}
public static String rowDataToString(RowData row, RowType type) {
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/KeyValueDataFileRecordReader.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/KeyValueDataFileRecordReader.java
index cd03c3de..5be4b733 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/KeyValueDataFileRecordReader.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/KeyValueDataFileRecordReader.java
@@ -38,22 +38,25 @@ public class KeyValueDataFileRecordReader implements
RecordReader<KeyValue> {
private final BulkFormat.Reader<RowData> reader;
private final KeyValueSerializer serializer;
+ private final int level;
public KeyValueDataFileRecordReader(
BulkFormat<RowData, FileSourceSplit> readerFactory,
Path path,
RowType keyType,
- RowType valueType)
+ RowType valueType,
+ int level)
throws IOException {
this.reader = FileUtils.createFormatReader(readerFactory, path);
this.serializer = new KeyValueSerializer(keyType, valueType);
+ this.level = level;
}
@Nullable
@Override
public RecordIterator<KeyValue> readBatch() throws IOException {
BulkFormat.RecordIterator<RowData> iterator = reader.readBatch();
- return iterator == null ? null : new
KeyValueDataFileRecordIterator(iterator, serializer);
+ return iterator == null ? null : new
KeyValueDataFileRecordIterator(iterator);
}
@Override
@@ -61,16 +64,12 @@ public class KeyValueDataFileRecordReader implements
RecordReader<KeyValue> {
reader.close();
}
- private static class KeyValueDataFileRecordIterator
- implements RecordReader.RecordIterator<KeyValue> {
+ private class KeyValueDataFileRecordIterator implements
RecordReader.RecordIterator<KeyValue> {
private final BulkFormat.RecordIterator<RowData> iterator;
- private final KeyValueSerializer serializer;
- private KeyValueDataFileRecordIterator(
- BulkFormat.RecordIterator<RowData> iterator,
KeyValueSerializer serializer) {
+ private
KeyValueDataFileRecordIterator(BulkFormat.RecordIterator<RowData> iterator) {
this.iterator = iterator;
- this.serializer = serializer;
}
@Override
@@ -78,7 +77,11 @@ public class KeyValueDataFileRecordReader implements
RecordReader<KeyValue> {
RecordAndPosition<RowData> result = iterator.next();
// TODO schema evolution
- return result == null ? null :
serializer.fromRow(result.getRecord());
+ if (result == null) {
+ return null;
+ } else {
+ return serializer.fromRow(result.getRecord()).setLevel(level);
+ }
}
@Override
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/KeyValueFileReaderFactory.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/KeyValueFileReaderFactory.java
index 5b414c23..a1064a64 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/KeyValueFileReaderFactory.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/io/KeyValueFileReaderFactory.java
@@ -64,9 +64,10 @@ public class KeyValueFileReaderFactory {
this.pathFactory = pathFactory;
}
- public RecordReader<KeyValue> createRecordReader(String fileName) throws
IOException {
+ public RecordReader<KeyValue> createRecordReader(String fileName, int
level)
+ throws IOException {
return new KeyValueDataFileRecordReader(
- readerFactory, pathFactory.toPath(fileName), keyType,
valueType);
+ readerFactory, pathFactory.toPath(fileName), keyType,
valueType, level);
}
public static Builder builder(
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReader.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReader.java
index 67e65b06..444d10ce 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReader.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/mergetree/MergeTreeReader.java
@@ -126,7 +126,7 @@ public class MergeTreeReader implements
RecordReader<KeyValue> {
SortedRun run, KeyValueFileReaderFactory readerFactory) throws
IOException {
List<ReaderSupplier<KeyValue>> readers = new ArrayList<>();
for (DataFileMeta file : run.files()) {
- readers.add(() ->
readerFactory.createRecordReader(file.fileName()));
+ readers.add(() ->
readerFactory.createRecordReader(file.fileName(), file.level()));
}
return ConcatRecordReader.create(readers);
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java
index d07d69c2..0565b1a5 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/KeyValueFileStoreRead.java
@@ -136,7 +136,7 @@ public class KeyValueFileStoreRead implements
FileStoreRead<KeyValue> {
suppliers.add(
() ->
readerFactory.createRecordReader(
-
changelogFile(file).orElse(file.fileName())));
+
changelogFile(file).orElse(file.fileName()), file.level()));
}
return ConcatRecordReader.create(suppliers);
} else {
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/io/KeyValueFileReadWriteTest.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/io/KeyValueFileReadWriteTest.java
index 6f87316e..5633c8b1 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/io/KeyValueFileReadWriteTest.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/io/KeyValueFileReadWriteTest.java
@@ -68,7 +68,7 @@ public class KeyValueFileReadWriteTest {
public void testReadNonExistentFile() {
KeyValueFileReaderFactory readerFactory =
createReaderFactory(tempDir.toString(), "avro", null, null);
- assertThatThrownBy(() ->
readerFactory.createRecordReader("dummy_file"))
+ assertThatThrownBy(() ->
readerFactory.createRecordReader("dummy_file", 0))
.hasMessageContaining(
"you can configure 'snapshot.time-retained' option
with a larger value.");
}
@@ -289,7 +289,8 @@ public class KeyValueFileReadWriteTest {
for (DataFileMeta meta : actualMetas) {
// check the contents of data file
CloseableIterator<KeyValue> actualKvsIterator =
- new
RecordReaderIterator<>(readerFactory.createRecordReader(meta.fileName()));
+ new RecordReaderIterator<>(
+ readerFactory.createRecordReader(meta.fileName(),
meta.level()));
while (actualKvsIterator.hasNext()) {
assertThat(expectedIterator.hasNext()).isTrue();
KeyValue actualKv = actualKvsIterator.next();
@@ -300,6 +301,7 @@ public class KeyValueFileReadWriteTest {
keySerializer,
projectedValueSerializer))
.isTrue();
+ assertThat(actualKv.level()).isEqualTo(meta.level());
}
actualKvsIterator.close();
