fresh-borzoni commented on code in PR #3519:
URL: https://github.com/apache/fluss/pull/3519#discussion_r3473536685
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/DvTableReadableSnapshotRetriever.java:
##########
@@ -426,6 +416,112 @@ public ReadableSnapshotResult
getReadableSnapshotAndOffsets(long tieredSnapshotI
earliestSnapshotIdToKeep);
}
+ /**
+ * Lowers {@code earliestSnapshotIdToKeep} to also cover the base anchors
of buckets that have
+ * no L0 files in the latest compacted snapshot.
+ *
+ * <p>A bucket without L0 has all of its data in base files, and its
readable offset is taken
+ * from the latest tiered snapshot. However, that base was established by
the bucket's most
+ * recent flush, and a later recomputation (once the bucket receives new
L0) will trace back to
+ * that flush's anchor snapshot (the previous APPEND of the latest
snapshot that exactly holds
+ * the flushed L0). That anchor snapshot must therefore stay retained
until the bucket is
+ * flushed again.
+ *
+ * <p>This is best-effort: if a bucket's flush history has expired and its
anchor cannot be
+ * determined, this conservatively returns {@link
LakeCommitResult#KEEP_ALL_PREVIOUS} so that no
+ * snapshot is deleted, rather than risk deleting one that is still needed.
+ *
+ * @param bucketsWithoutL0 buckets with no L0 in the latest compacted
snapshot
+ * @param flussTableBucketMapper mapper from Paimon partition-bucket to
Fluss table bucket
+ * @param latestCompactedSnapshotId the latest compacted snapshot id
(traversal start)
+ * @param earliestSnapshotId the earliest snapshot id still present in
Paimon (traversal end)
+ * @param earliestSnapshotIdToKeep the current value computed from buckets
with L0
+ * @return the tightened earliest snapshot id to keep
+ */
+ private long tightenEarliestSnapshotIdToKeepForBucketsWithoutL0(
+ Set<PaimonPartitionBucket> bucketsWithoutL0,
+ FlussTableBucketMapper flussTableBucketMapper,
+ long latestCompactedSnapshotId,
+ long earliestSnapshotId,
+ long earliestSnapshotIdToKeep)
+ throws IOException {
+ if (bucketsWithoutL0.isEmpty()) {
+ return earliestSnapshotIdToKeep;
+ }
+
+ // Only track buckets that map to a Fluss bucket; unmappable ones
(e.g. a partition not in
+ // Fluss) never need recomputation and must not pin retention.
+ Set<PaimonPartitionBucket> bucketsToAnchor = new HashSet<>();
+ for (PaimonPartitionBucket bucket : bucketsWithoutL0) {
+ if (flussTableBucketMapper.toTableBucket(bucket) != null) {
+ bucketsToAnchor.add(bucket);
+ }
+ }
+ if (bucketsToAnchor.isEmpty()) {
+ return earliestSnapshotIdToKeep;
+ }
+ boolean allAnchorsResolved = true;
+
+ for (long currentSnapshotId = latestCompactedSnapshotId;
+ currentSnapshotId >= earliestSnapshotId &&
!bucketsToAnchor.isEmpty();
+ currentSnapshotId--) {
+ Snapshot currentSnapshot =
snapshotManager.tryGetSnapshot(currentSnapshotId);
+ if (currentSnapshot == null
+ || currentSnapshot.commitKind() !=
Snapshot.CommitKind.COMPACT) {
+ continue;
+ }
+ // The first flush encountered going backwards is the bucket's
most recent flush.
+ for (PaimonPartitionBucket partitionBucket :
getBucketsWithFlushedL0(currentSnapshot)) {
+ if (!bucketsToAnchor.remove(partitionBucket)) {
+ continue;
+ }
+ Snapshot previousAppendSnapshot =
findBaseAnchorAppendSnapshot(currentSnapshot);
+ if (previousAppendSnapshot == null) {
+ // can't determine this bucket's base anchor; don't
tighten retention
+ allAnchorsResolved = false;
+ continue;
+ }
+ if (earliestSnapshotIdToKeep <= 0
+ || previousAppendSnapshot.id() <
earliestSnapshotIdToKeep) {
+ earliestSnapshotIdToKeep = previousAppendSnapshot.id();
+ }
+ }
+ }
+
+ if (!bucketsToAnchor.isEmpty()) {
+ // some bucket's most recent flush was not found within the
retained snapshots
+ allAnchorsResolved = false;
+ }
+
+ return allAnchorsResolved ? earliestSnapshotIdToKeep :
LakeCommitResult.KEEP_ALL_PREVIOUS;
Review Comment:
If one bucket's flush history has aged out, this returns KEEP_ALL_PREVIOUS
and throws away the bound the other buckets already gave us - so retention
never shrinks while that bucket stays cold.
The unresolvable anchor is by definition older than earliestSnapshotId
(already gone from Paimon), so keep-all can't recover it anyway, flooring at
earliestSnapshotId is just as safe and lets retention track Paimon's expiry.
WDYT?
##########
fluss-lake/fluss-lake-paimon/src/main/java/org/apache/fluss/lake/paimon/utils/DvTableReadableSnapshotRetriever.java:
##########
@@ -426,6 +416,112 @@ public ReadableSnapshotResult
getReadableSnapshotAndOffsets(long tieredSnapshotI
earliestSnapshotIdToKeep);
}
+ /**
+ * Lowers {@code earliestSnapshotIdToKeep} to also cover the base anchors
of buckets that have
+ * no L0 files in the latest compacted snapshot.
+ *
+ * <p>A bucket without L0 has all of its data in base files, and its
readable offset is taken
+ * from the latest tiered snapshot. However, that base was established by
the bucket's most
+ * recent flush, and a later recomputation (once the bucket receives new
L0) will trace back to
+ * that flush's anchor snapshot (the previous APPEND of the latest
snapshot that exactly holds
+ * the flushed L0). That anchor snapshot must therefore stay retained
until the bucket is
+ * flushed again.
+ *
+ * <p>This is best-effort: if a bucket's flush history has expired and its
anchor cannot be
+ * determined, this conservatively returns {@link
LakeCommitResult#KEEP_ALL_PREVIOUS} so that no
+ * snapshot is deleted, rather than risk deleting one that is still needed.
+ *
+ * @param bucketsWithoutL0 buckets with no L0 in the latest compacted
snapshot
+ * @param flussTableBucketMapper mapper from Paimon partition-bucket to
Fluss table bucket
+ * @param latestCompactedSnapshotId the latest compacted snapshot id
(traversal start)
+ * @param earliestSnapshotId the earliest snapshot id still present in
Paimon (traversal end)
+ * @param earliestSnapshotIdToKeep the current value computed from buckets
with L0
+ * @return the tightened earliest snapshot id to keep
+ */
+ private long tightenEarliestSnapshotIdToKeepForBucketsWithoutL0(
+ Set<PaimonPartitionBucket> bucketsWithoutL0,
+ FlussTableBucketMapper flussTableBucketMapper,
+ long latestCompactedSnapshotId,
+ long earliestSnapshotId,
+ long earliestSnapshotIdToKeep)
+ throws IOException {
+ if (bucketsWithoutL0.isEmpty()) {
+ return earliestSnapshotIdToKeep;
+ }
+
+ // Only track buckets that map to a Fluss bucket; unmappable ones
(e.g. a partition not in
+ // Fluss) never need recomputation and must not pin retention.
+ Set<PaimonPartitionBucket> bucketsToAnchor = new HashSet<>();
+ for (PaimonPartitionBucket bucket : bucketsWithoutL0) {
+ if (flussTableBucketMapper.toTableBucket(bucket) != null) {
+ bucketsToAnchor.add(bucket);
+ }
+ }
+ if (bucketsToAnchor.isEmpty()) {
+ return earliestSnapshotIdToKeep;
+ }
+ boolean allAnchorsResolved = true;
+
+ for (long currentSnapshotId = latestCompactedSnapshotId;
+ currentSnapshotId >= earliestSnapshotId &&
!bucketsToAnchor.isEmpty();
+ currentSnapshotId--) {
+ Snapshot currentSnapshot =
snapshotManager.tryGetSnapshot(currentSnapshotId);
+ if (currentSnapshot == null
+ || currentSnapshot.commitKind() !=
Snapshot.CommitKind.COMPACT) {
+ continue;
+ }
+ // The first flush encountered going backwards is the bucket's
most recent flush.
+ for (PaimonPartitionBucket partitionBucket :
getBucketsWithFlushedL0(currentSnapshot)) {
Review Comment:
This re-walks the same snapshots as the with-L0 loop above, so
getBucketsWithFlushedL0 scans manifests twice per compacted snapshot. Could it
fold into that loop: one pass, offsets for the with-L0 buckets, anchor-only for
the rest?
--
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]
