You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by GitBox <gi...@apache.org> on 2022/03/02 12:07:11 UTC
[GitHub] [flink-table-store] SteNicholas opened a new pull request #31: [FLINK-26293] Store log offsets in snapshot
SteNicholas opened a new pull request #31:
URL: https://github.com/apache/flink-table-store/pull/31
Now `ManifestCommittable` has the `logOffsets` inside, but it is not saved when generating snapshot in `FileStoreCommitImpl`.`logOffsets` needs to be saved in the snapshot. When `logOffsets` doesn't contain all buckets, it means that the current generated snapshot does not write data to all buckets, and it should be read from the last snapshot to complete the bucket's offsets.
**The brief change log**
- `Snapshot` introduces the `logOffsets` inside.
- `FileStoreCommitImpl` saves the `logOffsets` in the `Snapshot`.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi merged pull request #31: [FLINK-26293] Store log offsets in snapshot
Posted by GitBox <gi...@apache.org>.
JingsongLi merged pull request #31:
URL: https://github.com/apache/flink-table-store/pull/31
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a change in pull request #31: [FLINK-26293] Store log offsets in snapshot
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #31:
URL: https://github.com/apache/flink-table-store/pull/31#discussion_r823271958
##########
File path: flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
##########
@@ -188,6 +191,9 @@ public FileStoreExpireImpl newExpire(int numRetained, long millisRetained) {
Increment increment = entryWithBucket.getValue().prepareCommit();
committable.addFileCommittable(
entryWithPartition.getKey(), entryWithBucket.getKey(), increment);
+ if (!committable.logOffsets().containsKey(entryWithBucket.getKey())) {
+ committable.addLogOffset(entryWithBucket.getKey(), ID.getAndIncrement());
Review comment:
Can we just test expected data instead of generation data?
##########
File path: flink-table-store-core/src/test/java/org/apache/flink/table/store/file/TestFileStore.java
##########
@@ -188,6 +191,9 @@ public FileStoreExpireImpl newExpire(int numRetained, long millisRetained) {
Increment increment = entryWithBucket.getValue().prepareCommit();
committable.addFileCommittable(
entryWithPartition.getKey(), entryWithBucket.getKey(), increment);
+ if (!committable.logOffsets().containsKey(entryWithBucket.getKey())) {
+ committable.addLogOffset(entryWithBucket.getKey(), ID.getAndIncrement());
Review comment:
Can we just test expected data instead of generation data?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a change in pull request #31: [FLINK-26293] Store log offsets in snapshot
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #31:
URL: https://github.com/apache/flink-table-store/pull/31#discussion_r818260224
##########
File path: flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
##########
@@ -304,10 +301,17 @@ private boolean tryCommitOnce(
String manifestListName = null;
List<ManifestFileMeta> oldMetas = new ArrayList<>();
List<ManifestFileMeta> newMetas = new ArrayList<>();
+ Map<Integer, Long> logOffsets = committable.logOffsets();
try {
if (latestSnapshot != null) {
// read all previous manifest files
oldMetas.addAll(manifestList.read(latestSnapshot.manifestList()));
+ // read the last snapshot to complete the bucket's offsets when logOffsets does not
+ // contain all buckets
+ logOffsets.putAll(
Review comment:
`latestSnapshot.getLogOffsets().forEach(logOffsets::putIfAbsent)`
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a change in pull request #31: [FLINK-26293] Store log offsets in snapshot
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #31:
URL: https://github.com/apache/flink-table-store/pull/31#discussion_r818260657
##########
File path: flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
##########
@@ -276,7 +273,7 @@ private void tryOverwrite(
private boolean tryCommitOnce(
List<ManifestEntry> changes,
- String identifier,
+ ManifestCommittable committable,
Review comment:
I prefer pass `identifier` and `logOffsets` here
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [flink-table-store] JingsongLi commented on a change in pull request #31: [FLINK-26293] Store log offsets in snapshot
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #31:
URL: https://github.com/apache/flink-table-store/pull/31#discussion_r818263016
##########
File path: flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreCommitTest.java
##########
@@ -150,9 +150,12 @@ public void testOverwritePartialCommit() throws Exception {
TestFileStore store = createStore(false);
store.commitData(
- data1.values().stream().flatMap(Collection::stream).collect(Collectors.toList()),
- gen::getPartition,
- kv -> 0);
+ data1.values().stream()
+ .flatMap(Collection::stream)
+ .collect(Collectors.toList()),
+ gen::getPartition,
+ kv -> 0)
+ .forEach(snapshot -> assertThat(snapshot.getLogOffsets().get(0)).isEqualTo(1));
Review comment:
I didn't quite understand your test. As I understand it, the test logic should be:
- commit with full log offsets, assert.
- commit with partial log offsets (Complemented by the previous snapshot), assert.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org