SteNicholas commented on code in PR #388:
URL: https://github.com/apache/flink-table-store/pull/388#discussion_r1027877730
##########
flink-table-store-core/src/main/java/org/apache/flink/table/store/table/source/SnapshotEnumerator.java:
##########
@@ -99,11 +103,48 @@ public static class EnumeratorResult {
public final long snapshotId;
- public final DataTableScan.DataFilePlan plan;
+ public final DataFilePlan plan;
- private EnumeratorResult(long snapshotId, DataTableScan.DataFilePlan
plan) {
+ private EnumeratorResult(long snapshotId, DataFilePlan plan) {
this.snapshotId = snapshotId;
this.plan = plan;
}
}
+
+ /** Startup snapshot enumerator, this is the first plan for continuous
reading. */
+ public static DataFilePlan startup(DataTableScan scan) {
+ CoreOptions options = scan.options();
+ SnapshotManager snapshotManager = scan.snapshotManager();
+ CoreOptions.LogStartupMode startupMode = options.logStartupMode();
+ switch (startupMode) {
+ case FULL:
+ DataFilePlan plan;
+ if (options.changelogProducer() == FULL_COMPACTION) {
+ // Read the results of the last full compaction.
+ // Only full compaction results will appear on the max
level.
+ plan = scan.withLevel(options.numLevels() - 1).plan();
+ } else {
+ plan = scan.plan();
+ }
+ return plan;
+ case LATEST:
+ return new DataFilePlan(
+ snapshotManager.latestSnapshotId(),
Collections.emptyList());
+ case FROM_TIMESTAMP:
+ Long timestampMills = options.logScanTimestampMills();
+ if (timestampMills == null) {
Review Comment:
Could the `LOG_SCAN_TIMESTAMP_MILLS ` validation move before creating source?
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
