You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by lz...@apache.org on 2023/02/27 10:20:59 UTC
[flink-table-store] branch master updated: [FLINK-31236] Limit pushdown will open useless RecordReader
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new ed454b8f [FLINK-31236] Limit pushdown will open useless RecordReader
ed454b8f is described below
commit ed454b8f909a85cdd43fac70c4280b24d0c0e34f
Author: Jingsong Lee <ji...@gmail.com>
AuthorDate: Mon Feb 27 18:20:53 2023 +0800
[FLINK-31236] Limit pushdown will open useless RecordReader
This closes #561
---
.../source/FileStoreSourceSplitReader.java | 41 ++++++++++++++++++----
.../source/FileStoreSourceSplitReaderTest.java | 8 +++++
2 files changed, 42 insertions(+), 7 deletions(-)
diff --git a/flink-table-store-flink/flink-table-store-flink-common/src/main/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitReader.java b/flink-table-store-flink/flink-table-store-flink-common/src/main/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitReader.java
index 7bfaa962..7b32b0c4 100644
--- a/flink-table-store-flink/flink-table-store-flink-common/src/main/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitReader.java
+++ b/flink-table-store-flink/flink-table-store-flink-common/src/main/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitReader.java
@@ -31,6 +31,7 @@ import org.apache.flink.table.store.connector.FlinkRowData;
import org.apache.flink.table.store.data.InternalRow;
import org.apache.flink.table.store.reader.RecordReader;
import org.apache.flink.table.store.reader.RecordReader.RecordIterator;
+import org.apache.flink.table.store.table.source.Split;
import org.apache.flink.table.store.table.source.TableRead;
import javax.annotation.Nullable;
@@ -52,8 +53,9 @@ public class FileStoreSourceSplitReader<T> implements SplitReader<T, FileStoreSo
private final Pool<FileStoreRecordIterator> pool;
- @Nullable private RecordReader<InternalRow> currentReader;
+ @Nullable private LazyRecordReader currentReader;
@Nullable private String currentSplitId;
+ private long totalNumRead;
private long currentNumRead;
private RecordIterator<InternalRow> currentFirstBatch;
@@ -80,7 +82,7 @@ public class FileStoreSourceSplitReader<T> implements SplitReader<T, FileStoreSo
nextBatch = currentFirstBatch;
currentFirstBatch = null;
} else {
- nextBatch = reachLimit() ? null : currentReader.readBatch();
+ nextBatch = reachLimit() ? null : currentReader.recordReader().readBatch();
}
if (nextBatch == null) {
pool.recycler().recycle(iterator);
@@ -90,7 +92,7 @@ public class FileStoreSourceSplitReader<T> implements SplitReader<T, FileStoreSo
}
private boolean reachLimit() {
- return limit != null && currentNumRead >= limit;
+ return limit != null && totalNumRead >= limit;
}
private FileStoreRecordIterator pool() throws IOException {
@@ -120,7 +122,9 @@ public class FileStoreSourceSplitReader<T> implements SplitReader<T, FileStoreSo
@Override
public void close() throws Exception {
if (currentReader != null) {
- currentReader.close();
+ if (currentReader.lazyRecordReader != null) {
+ currentReader.lazyRecordReader.close();
+ }
}
}
@@ -135,8 +139,9 @@ public class FileStoreSourceSplitReader<T> implements SplitReader<T, FileStoreSo
}
currentSplitId = nextSplit.splitId();
- currentReader = tableRead.createReader(nextSplit.split());
+ currentReader = new LazyRecordReader(nextSplit.split());
currentNumRead = nextSplit.recordsToSkip();
+ totalNumRead += currentNumRead;
if (currentNumRead > 0) {
seek(currentNumRead);
}
@@ -144,7 +149,7 @@ public class FileStoreSourceSplitReader<T> implements SplitReader<T, FileStoreSo
private void seek(long toSkip) throws IOException {
while (true) {
- RecordIterator<InternalRow> nextBatch = currentReader.readBatch();
+ RecordIterator<InternalRow> nextBatch = currentReader.recordReader().readBatch();
if (nextBatch == null) {
throw new RuntimeException(
String.format(
@@ -163,7 +168,9 @@ public class FileStoreSourceSplitReader<T> implements SplitReader<T, FileStoreSo
private RecordsWithSplitIds<T> finishSplit() throws IOException {
if (currentReader != null) {
- currentReader.close();
+ if (currentReader.lazyRecordReader != null) {
+ currentReader.lazyRecordReader.close();
+ }
currentReader = null;
}
@@ -204,6 +211,7 @@ public class FileStoreSourceSplitReader<T> implements SplitReader<T, FileStoreSo
recordAndPosition.setNext(new FlinkRowData(row));
currentNumRead++;
+ totalNumRead++;
return recordAndPosition;
}
@@ -213,4 +221,23 @@ public class FileStoreSourceSplitReader<T> implements SplitReader<T, FileStoreSo
pool.recycler().recycle(this);
}
}
+
+ /** Lazy to create {@link RecordReader} to improve performance for limit. */
+ private class LazyRecordReader {
+
+ private final Split split;
+
+ private RecordReader<InternalRow> lazyRecordReader;
+
+ private LazyRecordReader(Split split) {
+ this.split = split;
+ }
+
+ public RecordReader<InternalRow> recordReader() throws IOException {
+ if (lazyRecordReader == null) {
+ lazyRecordReader = tableRead.createReader(split);
+ }
+ return lazyRecordReader;
+ }
+ }
}
diff --git a/flink-table-store-flink/flink-table-store-flink-common/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitReaderTest.java b/flink-table-store-flink/flink-table-store-flink-common/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitReaderTest.java
index f0b8f0ff..9ab21e06 100644
--- a/flink-table-store-flink/flink-table-store-flink-common/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitReaderTest.java
+++ b/flink-table-store-flink/flink-table-store-flink-common/src/test/java/org/apache/flink/table/store/connector/source/FileStoreSourceSplitReaderTest.java
@@ -56,6 +56,7 @@ import java.util.stream.Collectors;
import java.util.stream.Stream;
import static org.apache.flink.table.store.connector.LogicalTypeConversion.toDataType;
+import static org.apache.flink.table.store.connector.source.FileStoreSourceSplitSerializerTest.newFile;
import static org.apache.flink.table.store.connector.source.FileStoreSourceSplitSerializerTest.newSourceSplit;
import static org.apache.flink.table.store.file.mergetree.compact.MergeTreeCompactManagerTest.row;
import static org.assertj.core.api.Assertions.assertThat;
@@ -378,6 +379,13 @@ public class FileStoreSourceSplitReaderTest {
records = reader.fetch();
assertRecords(records, "id1", null, 0, Collections.emptyList());
+ // test limit without opening reader
+ // create a new fake new file, throw exception if open it
+ assignSplit(
+ reader, newSourceSplit("id2", row(1), 0, Collections.singletonList(newFile(0)), 0));
+ records = reader.fetch();
+ assertRecords(records, "id2", null, 0, null);
+
reader.close();
}