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 08:06:05 UTC
[GitHub] [flink-table-store] tsreaper opened a new pull request #30: [FLINK-26247] Optimize expire by only reading new changes in a snapshot
tsreaper opened a new pull request #30:
URL: https://github.com/apache/flink-table-store/pull/30
Current implementation of `FileStoreExpire` will read the whole snapshot for each snapshot to be expired. This is unnecessary because later snapshots also contain changes from previous snapshots.
This PR optimizes this by only reading new changes in a 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 #30: [FLINK-26247] Optimize expire by only reading new changes in a snapshot
Posted by GitBox <gi...@apache.org>.
JingsongLi merged pull request #30:
URL: https://github.com/apache/flink-table-store/pull/30
--
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 #30: [FLINK-26247] Optimize expire by only reading new changes in a snapshot
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #30:
URL: https://github.com/apache/flink-table-store/pull/30#discussion_r817471690
##########
File path: flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
##########
@@ -316,24 +318,35 @@ private boolean tryCommitOnce(
manifestFile,
manifestTargetSize.getBytes(),
manifestMergeMinCount));
+ previousChangesListName = manifestList.write(newMetas);
+
// write new changes into manifest files
+ List<ManifestFileMeta> newChangesManifests = new ArrayList<>();
if (!changes.isEmpty()) {
- newMetas.addAll(manifestFile.write(changes));
+ newChangesManifests.addAll(manifestFile.write(changes));
Review comment:
`manifestFile.write` supports empty entries?
##########
File path: flink-table-store-core/src/main/java/org/apache/flink/table/store/file/Snapshot.java
##########
@@ -44,8 +45,14 @@
@JsonProperty(FIELD_ID)
private final long id;
- @JsonProperty(FIELD_MANIFEST_LIST)
- private final String manifestList;
+ // a manifest list recording all changes from the previous snapshots
+ @JsonProperty(FIELD_PREVIOUS_CHANGES)
+ private final String previousChanges;
Review comment:
`baseManifestList` and `deltaManifestList`?
--
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 #30: [FLINK-26247] Optimize expire by only reading new changes in a snapshot
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #30:
URL: https://github.com/apache/flink-table-store/pull/30#discussion_r817506286
##########
File path: flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreExpireImpl.java
##########
@@ -129,38 +124,64 @@ private void expireUntil(long exclusiveId) {
// assume that all snapshots preceding it have been removed
break;
}
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to determine if snapshot #" + id + " still exists", e);
+ }
- Snapshot toExpire = Snapshot.fromPath(pathFactory.toSnapshotPath(id));
+ Snapshot toExpire = Snapshot.fromPath(pathFactory.toSnapshotPath(id));
+ List<ManifestFileMeta> previousChanges = manifestList.read(toExpire.previousChanges());
+ List<ManifestFileMeta> newChanges = manifestList.read(toExpire.newChanges());
- for (ManifestEntry entry : scan.withSnapshot(toExpire.id()).plan().files()) {
+ // we cannot delete an sst file directly when we meet a DELETE entry, because that
+ // file might be upgraded
+ Set<Path> sstToDelete = new HashSet<>();
+ for (ManifestFileMeta meta : newChanges) {
+ for (ManifestEntry entry : manifestFile.read(meta.fileName())) {
SstPathFactory sstPathFactory =
sstPathFactoryCache.getSstPathFactory(
entry.partition(), entry.bucket());
Path sstPath = sstPathFactory.toPath(entry.file().fileName());
- if (!sstInUse.contains(sstPath)) {
- sstToDelete.add(sstPath);
+ switch (entry.kind()) {
+ case ADD:
+ sstToDelete.remove(sstPath);
+ break;
+ case DELETE:
+ sstToDelete.add(sstPath);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Unknown value kind " + entry.kind().name());
}
}
+ }
+ for (Path sst : sstToDelete) {
+ FileUtils.deleteOrWarn(sst);
+ }
- for (ManifestFileMeta manifest : manifestList.read(toExpire.manifestList())) {
- if (!manifestsInUse.contains(manifest)) {
- manifestsToDelete.add(manifest.fileName());
- }
+ // collect useless manifests
+ for (ManifestFileMeta manifest : previousChanges) {
+ if (!manifestsInUse.contains(manifest)) {
+ manifestsToDelete.add(manifest);
+ }
+ }
+ for (ManifestFileMeta manifest : newChanges) {
+ if (!manifestsInUse.contains(manifest)) {
+ manifestsToDelete.add(manifest);
}
-
- manifestList.delete(toExpire.manifestList());
- FileUtils.deleteOrWarn(pathFactory.toSnapshotPath(id));
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to determine if snapshot #" + id + " still exists", e);
}
- }
- for (Path sst : sstToDelete) {
- FileUtils.deleteOrWarn(sst);
+ // delete manifest lists
+ manifestList.delete(toExpire.previousChanges());
+ manifestList.delete(toExpire.newChanges());
+
+ // delete snapshot
+ FileUtils.deleteOrWarn(pathFactory.toSnapshotPath(id));
}
- for (String manifestName : manifestsToDelete) {
- FileUtils.deleteOrWarn(pathFactory.toManifestFilePath(manifestName));
+
+ // actual delete of manifest files
+ for (ManifestFileMeta manifest : manifestsToDelete) {
Review comment:
delete manifests before deleting manifest list?
This ensures that the final files are deleted.
##########
File path: flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreExpireImpl.java
##########
@@ -129,38 +124,64 @@ private void expireUntil(long exclusiveId) {
// assume that all snapshots preceding it have been removed
break;
}
+ } catch (IOException e) {
+ throw new RuntimeException(
+ "Failed to determine if snapshot #" + id + " still exists", e);
+ }
- Snapshot toExpire = Snapshot.fromPath(pathFactory.toSnapshotPath(id));
+ Snapshot toExpire = Snapshot.fromPath(pathFactory.toSnapshotPath(id));
+ List<ManifestFileMeta> previousChanges = manifestList.read(toExpire.previousChanges());
+ List<ManifestFileMeta> newChanges = manifestList.read(toExpire.newChanges());
- for (ManifestEntry entry : scan.withSnapshot(toExpire.id()).plan().files()) {
+ // we cannot delete an sst file directly when we meet a DELETE entry, because that
+ // file might be upgraded
+ Set<Path> sstToDelete = new HashSet<>();
+ for (ManifestFileMeta meta : newChanges) {
+ for (ManifestEntry entry : manifestFile.read(meta.fileName())) {
SstPathFactory sstPathFactory =
sstPathFactoryCache.getSstPathFactory(
entry.partition(), entry.bucket());
Path sstPath = sstPathFactory.toPath(entry.file().fileName());
- if (!sstInUse.contains(sstPath)) {
- sstToDelete.add(sstPath);
+ switch (entry.kind()) {
+ case ADD:
+ sstToDelete.remove(sstPath);
+ break;
+ case DELETE:
+ sstToDelete.add(sstPath);
+ break;
+ default:
+ throw new UnsupportedOperationException(
+ "Unknown value kind " + entry.kind().name());
}
}
+ }
+ for (Path sst : sstToDelete) {
+ FileUtils.deleteOrWarn(sst);
+ }
- for (ManifestFileMeta manifest : manifestList.read(toExpire.manifestList())) {
- if (!manifestsInUse.contains(manifest)) {
- manifestsToDelete.add(manifest.fileName());
- }
+ // collect useless manifests
+ for (ManifestFileMeta manifest : previousChanges) {
+ if (!manifestsInUse.contains(manifest)) {
+ manifestsToDelete.add(manifest);
+ }
+ }
+ for (ManifestFileMeta manifest : newChanges) {
+ if (!manifestsInUse.contains(manifest)) {
+ manifestsToDelete.add(manifest);
}
-
- manifestList.delete(toExpire.manifestList());
- FileUtils.deleteOrWarn(pathFactory.toSnapshotPath(id));
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to determine if snapshot #" + id + " still exists", e);
}
- }
- for (Path sst : sstToDelete) {
- FileUtils.deleteOrWarn(sst);
+ // delete manifest lists
+ manifestList.delete(toExpire.previousChanges());
+ manifestList.delete(toExpire.newChanges());
+
+ // delete snapshot
+ FileUtils.deleteOrWarn(pathFactory.toSnapshotPath(id));
}
- for (String manifestName : manifestsToDelete) {
- FileUtils.deleteOrWarn(pathFactory.toManifestFilePath(manifestName));
+
+ // actual delete of manifest files
+ for (ManifestFileMeta manifest : manifestsToDelete) {
Review comment:
delete manifests before deleting manifest list?
This ensures that the final all files are deleted.
--
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 #30: [FLINK-26247] Optimize expire by only reading new changes in a snapshot
Posted by GitBox <gi...@apache.org>.
JingsongLi commented on a change in pull request #30:
URL: https://github.com/apache/flink-table-store/pull/30#discussion_r817509022
##########
File path: flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScanImpl.java
##########
@@ -155,7 +155,9 @@ public Plan plan() {
manifests = Collections.emptyList();
} else {
Snapshot snapshot = Snapshot.fromPath(pathFactory.toSnapshotPath(snapshotId));
- manifests = manifestList.read(snapshot.manifestList());
+ manifests = new ArrayList<>();
+ manifests.addAll(manifestList.read(snapshot.previousChanges()));
+ manifests.addAll(manifestList.read(snapshot.newChanges()));
Review comment:
Can we provide a method to snapshot: `List<String> allManifestLists()`
--
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