[core] Add TimeTravelUtils to resolve snapshot from scan params #4307
Conversation
|
Expansion ：We can support more scan startup mode, like Add parameters ： |
Thanks @LinMingQiang ,good reminder, had added timestamp config, but mode not add due to user not need add two config if they use snapshot or timestamp. |
xuzifu666
changed the title
|
Maybe you want |
Yes, it is a direct way, I had test for it, can query right snapshot and reset consumer. Thanks @JingsongLi |
xuzifu666
changed the title
| import static org.junit.jupiter.api.Assertions.assertThrows; | ||
|
|
||
| /** Tests for {@link TimeTravelUtilsTest}. */ | ||
| public class TimeTravelUtilsTest extends ScannerTestBase { |
xuzifu666
changed the title
| CoreOptions.SCAN_TAG_NAME.key(), | ||
| CoreOptions.SCAN_WATERMARK.key(), | ||
| CoreOptions.SCAN_TIMESTAMP_MILLIS.key())); | ||
| } |
There was a problem hiding this comment.
Use checkArgument will be better.
| scanKeys.add(CoreOptions.SCAN_TAG_NAME.key()); | ||
| scanKeys.add(CoreOptions.SCAN_WATERMARK.key()); | ||
| scanKeys.add(CoreOptions.SCAN_TIMESTAMP_MILLIS.key()); | ||
| } |
There was a problem hiding this comment.
Use String[]{} is better.
| if (scanHandleKey.size() > 1) { | ||
| throw new IllegalArgumentException( | ||
| String.format( | ||
| "%s %s %s and %s can contains only one", |
There was a problem hiding this comment.
Only one of the following parameters may be set : [%s, %s, %s]
There was a problem hiding this comment.
Thanks @LinMingQiang had addressed
|
|
||
| public static Snapshot resolveSnapshotFromOption( | ||
| CoreOptions options, SnapshotManager snapshotManager) { | ||
| List<String> scanHandleKey = new ArrayList<>(); |
There was a problem hiding this comment.
List scanHandleKey = new ArrayList<>(1);
| /** Contains time travel functions. */ | ||
| public class TimeTravelUtil { | ||
|
|
||
| public static String[] scanKeys = { |
There was a problem hiding this comment.
private static final String[] SCAN_KEYS = {
| CoreOptions.SCAN_TIMESTAMP_MILLIS.key() | ||
| }; | ||
|
|
||
| public static Snapshot resolveSnapshotFromOption( |
There was a problem hiding this comment.
resolveSnapshotFromOptions
| } | ||
| } | ||
|
|
||
| if (scanHandleKey.size() == 0) { |
| return snapshot; | ||
| } | ||
|
|
||
| private static Snapshot handleSnapshotId(SnapshotManager snapshotManager, CoreOptions options) { |
There was a problem hiding this comment.
resolveSnapshotBySnapshotId
| optMap.put("scan.timestamp-millis", ts + ""); | ||
| options = CoreOptions.fromMap(optMap); | ||
| snapshot = TimeTravelUtil.resolveSnapshotFromOption(options, snapshotManager); | ||
| assertThat(snapshot.id() == 1); |
| optMap.put("scan.snapshot-id", "2"); | ||
| CoreOptions options = CoreOptions.fromMap(optMap); | ||
| Snapshot snapshot = TimeTravelUtil.resolveSnapshotFromOption(options, snapshotManager); | ||
| assertThat(snapshot.id() == 2); |
There was a problem hiding this comment.
add assert snapshot.id() is not null
| public class TimeTravelUtilsTest extends ScannerTestBase { | ||
|
|
||
| @Test | ||
| public void testScan() throws Exception { |
There was a problem hiding this comment.
testResolveSnapshotFromOptions
|
|
||
| write.write(rowData(1, 10, 100L)); | ||
| commit.commit(0, write.prepareCommit(true, 0)); | ||
| long ts = System.currentTimeMillis(); |
| write.write(rowData(3, 50, 500L)); | ||
| commit.commit(2, write.prepareCommit(true, 2)); | ||
|
|
||
| HashMap<String, String> optMap = new HashMap<>(); |
There was a problem hiding this comment.
Map<String, String> optMap = new HashMap<>(4);
|
@wwj6591812 had addressd, Thanks. |
|
+1 |
Purpose
Add TimeTravelUtils to resolve snapshot from scan params
Linked issue: close #xxx
Tests
API and Format
Documentation