JingsongLi commented on a change in pull request #16:
URL: https://github.com/apache/flink-table-store/pull/16#discussion_r805754480
##########
File path:
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreCommitImpl.java
##########
@@ -168,21 +157,67 @@ public void overwrite(
Map<String, String> partition,
ManifestCommittable committable,
Map<String, String> properties) {
- throw new UnsupportedOperationException();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Ready to overwrite partition "
+ + partition.toString()
+ + "\n"
+ + committable.toString());
+ }
+
+ BinaryRowData partitionRowData =
+ TypeUtils.partitionMapToBinaryRowData(partition,
partitionType);
+
+ List<ManifestEntry> appendChanges =
collectChanges(committable.newFiles(), ValueKind.ADD);
+ tryOverwrite(
+ partitionRowData, appendChanges, committable.uuid(),
Snapshot.CommitKind.APPEND);
+
+ List<ManifestEntry> compactChanges = new ArrayList<>();
+ compactChanges.addAll(collectChanges(committable.compactBefore(),
ValueKind.DELETE));
+ compactChanges.addAll(collectChanges(committable.compactAfter(),
ValueKind.ADD));
+ tryCommit(compactChanges, committable.uuid(),
Snapshot.CommitKind.COMPACT);
}
- private String digestManifestCommittable(ManifestCommittable committable) {
- try {
- return new String(
- Base64.getEncoder()
- .encode(
- MessageDigest.getInstance("MD5")
-
.digest(committableSerializer.serialize(committable))));
- } catch (NoSuchAlgorithmException e) {
- throw new RuntimeException("MD5 algorithm not found. This is
impossible.", e);
- } catch (IOException e) {
- throw new RuntimeException(
- "Failed to serialize ManifestCommittable. This is
unexpected.", e);
+ private void tryCommit(
+ List<ManifestEntry> changes, String hash, Snapshot.CommitKind
commitKind) {
+ while (true) {
+ Long latestSnapshotId = pathFactory.latestSnapshotId();
+ if (tryCommitOnce(changes, hash, commitKind, latestSnapshotId)) {
+ break;
+ }
+ }
+ }
+
+ private void tryOverwrite(
+ BinaryRowData partition,
+ List<ManifestEntry> changes,
+ String hash,
+ Snapshot.CommitKind commitKind) {
+ while (true) {
+ Long latestSnapshotId = pathFactory.latestSnapshotId();
+
+ List<ManifestEntry> changesWithOverwrite = new ArrayList<>();
+ if (latestSnapshotId != null) {
+ List<ManifestEntry> currentEntries =
+ scan.withSnapshot(latestSnapshotId)
+
.withPartitionFilter(Collections.singletonList(partition))
+ .plan()
+ .files();
+ for (ManifestEntry entry : currentEntries) {
+ changesWithOverwrite.add(
+ new ManifestEntry(
+ ValueKind.DELETE,
+ entry.partition(),
+ entry.bucket(),
+ entry.totalBuckets(),
+ entry.file()));
+ }
+ }
+ changesWithOverwrite.addAll(changes);
+
+ if (tryCommitOnce(changesWithOverwrite, hash, commitKind,
latestSnapshotId)) {
Review comment:
If I understand correctly, both checks are for the same snapshot, there
can't be a new snapshot 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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
