Flink: Use Tag or Branch to scan data. #5029
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -42,11 +42,15 @@ public class ScanContext implements Serializable { | |
| private final boolean caseSensitive; | ||
| private final boolean exposeLocality; | ||
| private final Long snapshotId; | ||
| private final String branch; | ||
| private final String tag; | ||
| private final StreamingStartingStrategy startingStrategy; | ||
| private final Long startSnapshotId; | ||
| private final Long startSnapshotTimestamp; | ||
| private final Long endSnapshotId; | ||
| private final Long asOfTimestamp; | ||
| private final String startTag; | ||
| private final String endTag; | ||
| private final Long splitSize; | ||
| private final Integer splitLookback; | ||
| private final Long splitOpenFileCost; | ||
|
|
@@ -81,14 +85,22 @@ private ScanContext( | |
| boolean includeColumnStats, | ||
| boolean exposeLocality, | ||
| Integer planParallelism, | ||
| int maxPlanningSnapshotCount, | ||
| String branch, | ||
| String tag, | ||
| String startTag, | ||
| String endTag) { | ||
| this.caseSensitive = caseSensitive; | ||
| this.snapshotId = snapshotId; | ||
| this.tag = tag; | ||
| this.branch = branch; | ||
| this.startingStrategy = startingStrategy; | ||
| this.startSnapshotTimestamp = startSnapshotTimestamp; | ||
| this.startSnapshotId = startSnapshotId; | ||
| this.endSnapshotId = endSnapshotId; | ||
| this.asOfTimestamp = asOfTimestamp; | ||
| this.startTag = startTag; | ||
| this.endTag = endTag; | ||
| this.splitSize = splitSize; | ||
| this.splitLookback = splitLookback; | ||
| this.splitOpenFileCost = splitOpenFileCost; | ||
|
|
@@ -125,7 +137,24 @@ private void validate() { | |
| startSnapshotId == null, | ||
| "Invalid starting snapshot id for SPECIFIC_START_SNAPSHOT_ID strategy: not null"); | ||
| } | ||
|
|
||
| Preconditions.checkArgument( | ||
| branch == null, | ||
| String.format( | ||
| "Cannot scan table using ref %s configured for streaming reader yet", branch)); | ||
|
Contributor
Author
There was a problem hiding this comment. #5984 seems to be the prerequisite for flink to implement stream incremental read with branch. |
||
|
|
||
| Preconditions.checkArgument( | ||
| tag == null, | ||
| String.format("Cannot scan table using ref %s configured for streaming reader", tag)); | ||
| } | ||
|
|
||
| Preconditions.checkArgument( | ||
| !(startTag != null && startSnapshotId() != null), | ||
| "START_SNAPSHOT_ID and START_TAG cannot both be set."); | ||
|
|
||
| Preconditions.checkArgument( | ||
| !(endTag != null && endSnapshotId() != null), | ||
| "END_SNAPSHOT_ID and END_TAG cannot both be set."); | ||
| } | ||
|
|
||
| public boolean caseSensitive() { | ||
|
|
@@ -136,6 +165,22 @@ public Long snapshotId() { | |
| return snapshotId; | ||
| } | ||
|
|
||
| public String branch() { | ||
| return branch; | ||
| } | ||
|
|
||
| public String tag() { | ||
| return tag; | ||
| } | ||
|
|
||
| public String startTag() { | ||
| return startTag; | ||
| } | ||
|
|
||
| public String endTag() { | ||
| return endTag; | ||
| } | ||
|
|
||
| public StreamingStartingStrategy streamingStartingStrategy() { | ||
| return startingStrategy; | ||
| } | ||
|
|
@@ -212,8 +257,12 @@ public ScanContext copyWithAppendsBetween(Long newStartSnapshotId, long newEndSn | |
| return ScanContext.builder() | ||
| .caseSensitive(caseSensitive) | ||
| .useSnapshotId(null) | ||
| .useBranch(branch) | ||
| .useTag(null) | ||
| .startSnapshotId(newStartSnapshotId) | ||
| .endSnapshotId(newEndSnapshotId) | ||
| .startTag(null) | ||
| .endTag(null) | ||
| .asOfTimestamp(null) | ||
| .splitSize(splitSize) | ||
| .splitLookback(splitLookback) | ||
|
|
@@ -235,8 +284,12 @@ public ScanContext copyWithSnapshotId(long newSnapshotId) { | |
| return ScanContext.builder() | ||
| .caseSensitive(caseSensitive) | ||
| .useSnapshotId(newSnapshotId) | ||
| .useBranch(branch) | ||
| .useTag(tag) | ||
| .startSnapshotId(null) | ||
| .endSnapshotId(null) | ||
| .startTag(null) | ||
| .endTag(null) | ||
| .asOfTimestamp(null) | ||
| .splitSize(splitSize) | ||
| .splitLookback(splitLookback) | ||
|
|
@@ -261,6 +314,10 @@ public static Builder builder() { | |
| public static class Builder { | ||
| private boolean caseSensitive = FlinkReadOptions.CASE_SENSITIVE_OPTION.defaultValue(); | ||
| private Long snapshotId = FlinkReadOptions.SNAPSHOT_ID.defaultValue(); | ||
| private String branch = FlinkReadOptions.BRANCH.defaultValue(); | ||
| private String tag = FlinkReadOptions.TAG.defaultValue(); | ||
| private String startTag = FlinkReadOptions.START_TAG.defaultValue(); | ||
| private String endTag = FlinkReadOptions.END_TAG.defaultValue(); | ||
| private StreamingStartingStrategy startingStrategy = | ||
| FlinkReadOptions.STARTING_STRATEGY_OPTION.defaultValue(); | ||
| private Long startSnapshotTimestamp = FlinkReadOptions.START_SNAPSHOT_TIMESTAMP.defaultValue(); | ||
|
|
@@ -297,6 +354,16 @@ public Builder useSnapshotId(Long newSnapshotId) { | |
| return this; | ||
| } | ||
|
|
||
| public Builder useTag(String newTag) { | ||
| this.tag = newTag; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder useBranch(String newBranch) { | ||
| this.branch = newBranch; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder startingStrategy(StreamingStartingStrategy newStartingStrategy) { | ||
| this.startingStrategy = newStartingStrategy; | ||
| return this; | ||
|
|
@@ -317,6 +384,16 @@ public Builder endSnapshotId(Long newEndSnapshotId) { | |
| return this; | ||
| } | ||
|
|
||
| public Builder startTag(String newStartTag) { | ||
| this.startTag = newStartTag; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder endTag(String newEndTag) { | ||
| this.endTag = newEndTag; | ||
| return this; | ||
| } | ||
|
|
||
| public Builder asOfTimestamp(Long newAsOfTimestamp) { | ||
| this.asOfTimestamp = newAsOfTimestamp; | ||
| return this; | ||
|
|
@@ -392,6 +469,10 @@ public Builder resolveConfig( | |
| FlinkReadConf flinkReadConf = new FlinkReadConf(table, readOptions, readableConfig); | ||
|
|
||
| return this.useSnapshotId(flinkReadConf.snapshotId()) | ||
| .useTag(flinkReadConf.tag()) | ||
| .useBranch(flinkReadConf.branch()) | ||
| .startTag(flinkReadConf.startTag()) | ||
| .endTag(flinkReadConf.endTag()) | ||
| .caseSensitive(flinkReadConf.caseSensitive()) | ||
| .asOfTimestamp(flinkReadConf.asOfTimestamp()) | ||
| .startingStrategy(flinkReadConf.startingStrategy()) | ||
|
|
@@ -431,7 +512,11 @@ public ScanContext build() { | |
| includeColumnStats, | ||
| exposeLocality, | ||
| planParallelism, | ||
| maxPlanningSnapshotCount, | ||
| branch, | ||
| tag, | ||
| startTag, | ||
| endTag); | ||
| } | ||
| } | ||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
nit: this can be
appendToTable(null, records), right?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Currently it won't work because we have two methods:
appendToTable(String branch, List<Record> records)and
appendToTable(StructLike partition, List<Record> records)Unless I rearrange the order of
branchand place it afterrecords, but I prefer to leave it as it is currently, at the front.