JingsongLi commented on code in PR #99:
URL: https://github.com/apache/flink-table-store/pull/99#discussion_r865755555
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreReadImpl.java:
##########
@@ -81,14 +87,42 @@ public FileStoreRead withValueProjection(int[][]
projectedFields) {
}
@Override
- public RecordReader createReader(BinaryRowData partition, int bucket,
List<SstFileMeta> files)
+ public RecordReader createReader(BinaryRowData partition, int bucket,
List<DataFileMeta> files)
throws IOException {
+ switch (writeMode) {
+ case APPEND_ONLY:
+ return createAppendOnlyReader(partition, bucket, files);
+
+ case CHANGE_LOG:
+ return createLSMReader(partition, bucket, files);
+
+ default:
+ throw new UnsupportedOperationException("Unknown write mode: "
+ writeMode);
+ }
+ }
+
+ private RecordReader createAppendOnlyReader(
+ BinaryRowData partition, int bucket, List<DataFileMeta> files)
throws IOException {
+ Preconditions.checkArgument(
+ !dropDelete, "Cannot drop delete message for append-only
table.");
+
+ SstFileReader fileReader = sstFileReaderFactory.create(partition,
bucket);
Review Comment:
I think we can start renaming in a separate JIRA/PR.
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreReadImpl.java:
##########
@@ -81,14 +87,42 @@ public FileStoreRead withValueProjection(int[][]
projectedFields) {
}
@Override
- public RecordReader createReader(BinaryRowData partition, int bucket,
List<SstFileMeta> files)
+ public RecordReader createReader(BinaryRowData partition, int bucket,
List<DataFileMeta> files)
throws IOException {
+ switch (writeMode) {
+ case APPEND_ONLY:
+ return createAppendOnlyReader(partition, bucket, files);
+
+ case CHANGE_LOG:
+ return createLSMReader(partition, bucket, files);
+
+ default:
+ throw new UnsupportedOperationException("Unknown write mode: "
+ writeMode);
+ }
+ }
+
+ private RecordReader createAppendOnlyReader(
+ BinaryRowData partition, int bucket, List<DataFileMeta> files)
throws IOException {
+ Preconditions.checkArgument(
+ !dropDelete, "Cannot drop delete message for append-only
table.");
+
+ SstFileReader fileReader = sstFileReaderFactory.create(partition,
bucket);
Review Comment:
I think we can start renaming in a separate JIRA/PR now.
--
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.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
