From 0c08d1696894448d74c7ae42589295e6c632ade9 Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Mon, 13 Jun 2022 19:39:52 -0700 Subject: [PATCH 1/3] Retain unreferenced snapshots using table settings. --- .../org/apache/iceberg/RemoveSnapshots.java | 98 +++++++------------ 1 file changed, 34 insertions(+), 64 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java b/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java index a7cdcaaf0cc6..e80370fc3f99 100644 --- a/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java +++ b/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java @@ -20,13 +20,13 @@ package org.apache.iceberg; import java.io.IOException; +import java.util.Collection; import java.util.List; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ExecutorService; import java.util.function.Consumer; -import java.util.stream.Collectors; import org.apache.iceberg.avro.Avro; import org.apache.iceberg.exceptions.CommitFailedException; import org.apache.iceberg.exceptions.NotFoundException; @@ -35,7 +35,6 @@ import org.apache.iceberg.io.CloseableIterable; import org.apache.iceberg.relocated.com.google.common.base.Joiner; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; -import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.relocated.com.google.common.collect.Sets; @@ -180,16 +179,8 @@ private TableMetadata internalApply() { } Set idsToRetain = Sets.newHashSet(); - - // Compute the snapshots for each reference - Map> refSnapshots = computeRefSnapshots(base.refs()); - - // Identify unreferenced snapshots which should be retained - Set unreferencedSnapshotsToRetain = computeUnreferencedSnapshotsToRetain(refSnapshots); - idsToRetain.addAll(unreferencedSnapshotsToRetain); - // Identify refs that should be removed - Map retainedRefs = computeRetainedRefs(base.refs()); + Map retainedRefs = computeRetainedRefs(base.refs()); Map> retainedIdToRefs = Maps.newHashMap(); for (Map.Entry retainedRefEntry : retainedRefs.entrySet()) { long snapshotId = retainedRefEntry.getValue().snapshotId(); @@ -204,45 +195,24 @@ private TableMetadata internalApply() { "Cannot expire %s. Still referenced by refs: %s", idToRemove, refsForId); } - Set branchSnapshotsToRetain = computeAllBranchSnapshotsToRetain(retainedRefs, refSnapshots); - idsToRetain.addAll(branchSnapshotsToRetain); + idsToRetain.addAll(computeAllBranchSnapshotsToRetain(retainedRefs.values())); + idsToRetain.addAll(unreferencedSnapshotsToRetain(retainedRefs.values())); + TableMetadata.Builder updatedMetaBuilder = TableMetadata.buildFrom(base); + base.refs().keySet().stream() + .filter(ref -> !retainedRefs.containsKey(ref)) + .forEach(updatedMetaBuilder::removeRef); + base.snapshots().stream() .map(Snapshot::snapshotId) .filter(snapshot -> !idsToRetain.contains(snapshot)) .forEach(idsToRemove::add); updatedMetaBuilder.removeSnapshots(idsToRemove); - base.refs().keySet().stream() - .filter(ref -> !retainedRefs.containsKey(ref)) - .forEach(updatedMetaBuilder::removeRef); - return updatedMetaBuilder.build(); } - /** - * Helper to compute the mapping of a ref to its snapshots. If it's a branch, the snapshots is an ordered set - * of all the snapshots on the branch. If it's a tag, the snapshot is a set of the single snapshot the tag refers to - */ - private Map> computeRefSnapshots(Map refs) { - Map> refSnapshots = Maps.newHashMap(); - for (Map.Entry refEntry : refs.entrySet()) { - String name = refEntry.getKey(); - SnapshotRef ref = refEntry.getValue(); - if (ref.isBranch()) { - Set branchAncestors = Sets.newLinkedHashSet(); - Iterable snapshots = SnapshotUtil.ancestorsOf(ref.snapshotId(), base::snapshot); - snapshots.forEach(snapshot -> branchAncestors.add(snapshot.snapshotId())); - refSnapshots.put(name, branchAncestors); - } else { - refSnapshots.put(name, ImmutableSet.of(ref.snapshotId())); - } - } - - return refSnapshots; - } - private Map computeRetainedRefs(Map refs) { Map retainedRefs = Maps.newHashMap(); for (Map.Entry refEntry : refs.entrySet()) { @@ -268,34 +238,16 @@ private Map computeRetainedRefs(Map re return retainedRefs; } - private Set computeUnreferencedSnapshotsToRetain(Map> refSnapshots) { - Set referencedSnapshots = Sets.newHashSet(); - refSnapshots.values().forEach(referencedSnapshots::addAll); - - return base.snapshots().stream() - .filter(snapshot -> !referencedSnapshots.contains(snapshot.snapshotId())) - .filter(snapshot -> snapshot.timestampMillis() >= defaultExpireOlderThan) - .map(Snapshot::snapshotId) - .collect(Collectors.toSet()); - } - - private Set computeAllBranchSnapshotsToRetain( - Map refs, - Map> refSnapshots) { - + private Set computeAllBranchSnapshotsToRetain(Collection refs) { Set branchSnapshotsToRetain = Sets.newHashSet(); - for (Map.Entry refEntry : refs.entrySet()) { - final String name = refEntry.getKey(); - final SnapshotRef ref = refEntry.getValue(); - + for (SnapshotRef ref : refs) { if (ref.isBranch()) { long expireSnapshotsOlderThan = ref.maxSnapshotAgeMs() != null ? now - ref.maxSnapshotAgeMs() : defaultExpireOlderThan; int minSnapshotsToKeep = ref.minSnapshotsToKeep() != null ? ref.minSnapshotsToKeep() : defaultMinNumSnapshots; - Set branchAncestors = refSnapshots.get(name); branchSnapshotsToRetain.addAll( - computeBranchSnapshotsToRetain(branchAncestors, expireSnapshotsOlderThan, minSnapshotsToKeep)); + computeBranchSnapshotsToRetain(ref.snapshotId(), expireSnapshotsOlderThan, minSnapshotsToKeep)); } } @@ -303,12 +255,11 @@ private Set computeAllBranchSnapshotsToRetain( } private Set computeBranchSnapshotsToRetain( - Set branchSnapshots, + long snapshot, long expireSnapshotsOlderThan, int minSnapshotsToKeep) { Set idsToRetain = Sets.newHashSet(); - for (long snapshot : branchSnapshots) { - Snapshot ancestor = base.snapshot(snapshot); + for (Snapshot ancestor : SnapshotUtil.ancestorsOf(snapshot, base::snapshot)) { if (idsToRetain.size() < minSnapshotsToKeep || ancestor.timestampMillis() >= expireSnapshotsOlderThan) { idsToRetain.add(ancestor.snapshotId()); } else { @@ -319,6 +270,25 @@ private Set computeBranchSnapshotsToRetain( return idsToRetain; } + private Set unreferencedSnapshotsToRetain(Collection refs) { + Set referencedSnapshots = Sets.newHashSet(); + for (SnapshotRef ref : refs) { + for (Snapshot snapshot : SnapshotUtil.ancestorsOf(ref.snapshotId(), base::snapshot)) { + referencedSnapshots.add(snapshot.snapshotId()); + } + } + + Set snapshotsToRetain = Sets.newHashSet(); + for (Snapshot snapshot : base.snapshots()) { + if (!referencedSnapshots.contains(snapshot.snapshotId()) && // unreferenced + snapshot.timestampMillis() >= defaultExpireOlderThan) { // not old enough to expire + snapshotsToRetain.add(snapshot.snapshotId()); + } + } + + return snapshotsToRetain; + } + @Override public void commit() { Tasks.foreach(ops) @@ -601,7 +571,7 @@ private CloseableIterable readManifestFiles(Snapshot snapshot) { .build(); } else { - return CloseableIterable.withNoopClose(snapshot.allManifests(ops.io())); + return CloseableIterable.withNoopClose(snapshot.allManifests()); } } } From c3ae8b03a71ca79741fe64a3e3dc7757293da62f Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Mon, 13 Jun 2022 20:10:02 -0700 Subject: [PATCH 2/3] Improve tests and default the main branch. --- .../apache/iceberg/TableMetadataParser.java | 7 +- .../apache/iceberg/TestRemoveSnapshots.java | 176 ++++++++++++++---- 2 files changed, 140 insertions(+), 43 deletions(-) diff --git a/core/src/main/java/org/apache/iceberg/TableMetadataParser.java b/core/src/main/java/org/apache/iceberg/TableMetadataParser.java index a709fe5ee352..c167b08706a1 100644 --- a/core/src/main/java/org/apache/iceberg/TableMetadataParser.java +++ b/core/src/main/java/org/apache/iceberg/TableMetadataParser.java @@ -415,12 +415,15 @@ static TableMetadata fromJson(FileIO io, String metadataLocation, JsonNode node) // parse properties map Map properties = JsonUtil.getStringMap(PROPERTIES, node); - long currentVersionId = JsonUtil.getLong(CURRENT_SNAPSHOT_ID, node); + long currentSnapshotId = JsonUtil.getLong(CURRENT_SNAPSHOT_ID, node); long lastUpdatedMillis = JsonUtil.getLong(LAST_UPDATED_MILLIS, node); Map refs; if (node.has(REFS)) { refs = refsFromJson(node.get(REFS)); + } else if (currentSnapshotId != -1) { + // initialize the main branch if there are no refs + refs = ImmutableMap.of(SnapshotRef.MAIN_BRANCH, SnapshotRef.branchBuilder(currentSnapshotId).build()); } else { refs = ImmutableMap.of(); } @@ -457,7 +460,7 @@ static TableMetadata fromJson(FileIO io, String metadataLocation, JsonNode node) return new TableMetadata(metadataLocation, formatVersion, uuid, location, lastSequenceNumber, lastUpdatedMillis, lastAssignedColumnId, currentSchemaId, schemas, defaultSpecId, specs, - lastAssignedPartitionId, defaultSortOrderId, sortOrders, properties, currentVersionId, + lastAssignedPartitionId, defaultSortOrderId, sortOrders, properties, currentSnapshotId, snapshots, entries.build(), metadataEntries.build(), refs, ImmutableList.of() /* no changes from the file */); } diff --git a/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java b/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java index fd3bf4d110b4..99cba44d3842 100644 --- a/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java +++ b/core/src/test/java/org/apache/iceberg/TestRemoveSnapshots.java @@ -1206,30 +1206,66 @@ public void testExpireWithDeleteFiles() { } @Test - public void testRefExpiration() { + public void testTagExpiration() { table.newAppend() .appendFile(FILE_A) .commit(); + long now = System.currentTimeMillis(); - long tagARefAgeMs = 100; + long maxAgeMs = 100; + long expirationTime = now + maxAgeMs; + table.manageSnapshots() - .createTag("TagA", table.currentSnapshot().snapshotId()) - .setMaxRefAgeMs("TagA", tagARefAgeMs) + .createTag("tag", table.currentSnapshot().snapshotId()) + .setMaxRefAgeMs("tag", maxAgeMs) .commit(); - long expirationTime = now + tagARefAgeMs; + table.newAppend() .appendFile(FILE_B) .commit(); + table.manageSnapshots() - .createBranch("BranchB", table.currentSnapshot().snapshotId()) + .createBranch("branch", table.currentSnapshot().snapshotId()) .commit(); waitUntilAfter(expirationTime); table.expireSnapshots().cleanExpiredFiles(false).commit(); - Assert.assertNull(table.ops().current().ref("TagA")); - Assert.assertNotNull(table.ops().current().ref("BranchB")); + Assert.assertNull(table.ops().current().ref("tag")); + Assert.assertNotNull(table.ops().current().ref("branch")); + Assert.assertNotNull(table.ops().current().ref(SnapshotRef.MAIN_BRANCH)); + } + + @Test + public void testBranchExpiration() { + table.newAppend() + .appendFile(FILE_A) + .commit(); + + long now = System.currentTimeMillis(); + long maxAgeMs = 100; + long expirationTime = now + maxAgeMs; + + table.manageSnapshots() + .createBranch("branch", table.currentSnapshot().snapshotId()) + .setMaxRefAgeMs("branch", maxAgeMs) + .commit(); + + table.newAppend() + .appendFile(FILE_B) + .commit(); + + table.manageSnapshots() + .createTag("tag", table.currentSnapshot().snapshotId()) + .commit(); + + waitUntilAfter(expirationTime); + + table.expireSnapshots().cleanExpiredFiles(false).commit(); + + Assert.assertNull(table.ops().current().ref("branch")); + Assert.assertNotNull(table.ops().current().ref("tag")); Assert.assertNotNull(table.ops().current().ref(SnapshotRef.MAIN_BRANCH)); } @@ -1238,6 +1274,7 @@ public void testMultipleRefsAndCleanExpiredFilesFails() { table.newAppend() .appendFile(FILE_A) .commit(); + table.manageSnapshots() .createTag("TagA", table.currentSnapshot().snapshotId()) .commit(); @@ -1250,29 +1287,63 @@ public void testMultipleRefsAndCleanExpiredFilesFails() { } @Test - public void testFailRemovingSnapshotWhenStillReferenced() { + public void testFailRemovingSnapshotWhenStillReferencedByBranch() { + table.newAppend() + .appendFile(FILE_A) + .commit(); + + AppendFiles append = table.newAppend() + .appendFile(FILE_B) + .stageOnly(); + + long snapshotId = append.apply().snapshotId(); + + append.commit(); + + table.manageSnapshots() + .createBranch("branch", snapshotId) + .commit(); + + AssertHelpers.assertThrows( + "Should fail removing snapshot when it is still referenced", + IllegalArgumentException.class, + "Cannot expire 2. Still referenced by refs: [branch]", + () -> table.expireSnapshots().expireSnapshotId(snapshotId).commit()); + } + + @Test + public void testFailRemovingSnapshotWhenStillReferencedByTag() { table.newAppend() .appendFile(FILE_A) .commit(); + + long snapshotId = table.currentSnapshot().snapshotId(); + + table.manageSnapshots() + .createTag("tag", snapshotId) + .commit(); + + // commit another snapshot so the first one isn't referenced by main table.newAppend() .appendFile(FILE_B) - .stageOnly() .commit(); - table.manageSnapshots().createTag("stagedB", 2).commit(); AssertHelpers.assertThrows( "Should fail removing snapshot when it is still referenced", IllegalArgumentException.class, - "Cannot expire 2. Still referenced by refs: [stagedB]", - () -> table.expireSnapshots().expireSnapshotId(2).commit()); + "Cannot expire 1. Still referenced by refs: [tag]", + () -> table.expireSnapshots().expireSnapshotId(snapshotId).commit()); } @Test - public void testRetainStagedSnapshotsWithinExpirationAge() { + public void testRetainUnreferencedSnapshotsWithinExpirationAge() { table.newAppend() .appendFile(FILE_A) .commit(); + long expireTimestampSnapshotA = waitUntilAfter(table.currentSnapshot().timestampMillis()); + waitUntilAfter(expireTimestampSnapshotA); + table.newAppend() .appendFile(FILE_B) .stageOnly() @@ -1282,7 +1353,9 @@ public void testRetainStagedSnapshotsWithinExpirationAge() { .appendFile(FILE_C) .commit(); - table.expireSnapshots().expireOlderThan(expireTimestampSnapshotA).commit(); + table.expireSnapshots() + .expireOlderThan(expireTimestampSnapshotA) + .commit(); Assert.assertEquals(2, table.ops().current().snapshots().size()); } @@ -1292,71 +1365,92 @@ public void testRetainStagedSnapshotsWithinExpirationAge() { @Test public void testMinSnapshotsToKeepMultipleBranches() { table.newAppend().appendFile(FILE_A).commit(); + long initialSnapshotId = table.currentSnapshot().snapshotId(); table.newAppend().appendFile(FILE_B).commit(); - table.newAppend().appendFile(FILE_C).stageOnly().commit(); - int branchSnapshot = 3; - String branchName = "branchC"; + + // stage a snapshot and get its id + AppendFiles append = table.newAppend().appendFile(FILE_C).stageOnly(); + long branchSnapshotId = append.apply().snapshotId(); + append.commit(); + + Assert.assertEquals("Should have 3 snapshots", 3, Iterables.size(table.snapshots())); + long maxSnapshotAgeMs = 1; long expirationTime = System.currentTimeMillis() + maxSnapshotAgeMs; - // Retain 1 snapshot on main + // configure main so that the initial snapshot will expire table.manageSnapshots() .setMinSnapshotsToKeep(SnapshotRef.MAIN_BRANCH, 1) .setMaxSnapshotAgeMs(SnapshotRef.MAIN_BRANCH, 1) .commit(); - // Retain all snapshots on branchC + // retain 3 snapshots on branch (including the initial snapshot) table.manageSnapshots() - .createBranch(branchName, branchSnapshot) - .setMinSnapshotsToKeep(branchName, 3) - .setMaxSnapshotAgeMs(branchName, 1) + .createBranch("branch", branchSnapshotId) + .setMinSnapshotsToKeep("branch", 3) + .setMaxSnapshotAgeMs("branch", maxSnapshotAgeMs) .commit(); waitUntilAfter(expirationTime); table.expireSnapshots().cleanExpiredFiles(false).commit(); - Assert.assertEquals(3, table.ops().current().snapshots().size()); + Assert.assertEquals("Should have 3 snapshots (none removed)", 3, Iterables.size(table.snapshots())); + + // stop retaining snapshots from the branch + table.manageSnapshots() + .setMinSnapshotsToKeep("branch", 1) + .commit(); - table.manageSnapshots().setMinSnapshotsToKeep(branchName, 1).commit(); table.expireSnapshots().cleanExpiredFiles(false).commit(); - Assert.assertEquals(2, table.ops().current().snapshots().size()); - Assert.assertNull(table.ops().current().snapshot(1)); + Assert.assertEquals("Should have 2 snapshots (initial removed)", 2, Iterables.size(table.snapshots())); + Assert.assertNull(table.ops().current().snapshot(initialSnapshotId)); } @Test public void testMaxSnapshotAgeMultipleBranches() { table.newAppend().appendFile(FILE_A).commit(); - long ageFirstSnapshot = 10; - long expirationTime = System.currentTimeMillis() + ageFirstSnapshot; + long initialSnapshotId = table.currentSnapshot().snapshotId(); + + long ageMs = 10; + long expirationTime = System.currentTimeMillis() + ageMs; + + waitUntilAfter(expirationTime); + + table.newAppend().appendFile(FILE_B).commit(); + + // configure main so that the initial snapshot will expire table.manageSnapshots() - .setMaxSnapshotAgeMs(SnapshotRef.MAIN_BRANCH, ageFirstSnapshot) + .setMaxSnapshotAgeMs(SnapshotRef.MAIN_BRANCH, ageMs) .setMinSnapshotsToKeep(SnapshotRef.MAIN_BRANCH, 1) .commit(); - String branchName = "branchC"; - waitUntilAfter(expirationTime); + // stage a snapshot and get its id + AppendFiles append = table.newAppend().appendFile(FILE_C).stageOnly(); + long branchSnapshotId = append.apply().snapshotId(); + append.commit(); - table.newAppend().appendFile(FILE_B).commit(); - table.newAppend().appendFile(FILE_C).stageOnly().commit(); + Assert.assertEquals("Should have 3 snapshots", 3, Iterables.size(table.snapshots())); + // retain all snapshots on branch (including the initial snapshot) table.manageSnapshots() - .createBranch(branchName, 3) - .setMinSnapshotsToKeep(branchName, 1) - .setMaxSnapshotAgeMs(branchName, Long.MAX_VALUE) + .createBranch("branch", branchSnapshotId) + .setMinSnapshotsToKeep("branch", 1) + .setMaxSnapshotAgeMs("branch", Long.MAX_VALUE) .commit(); table.expireSnapshots().cleanExpiredFiles(false).commit(); - Assert.assertEquals(3, table.ops().current().snapshots().size()); + Assert.assertEquals("Should have 3 snapshots (none removed)", 3, Iterables.size(table.snapshots())); + // allow the initial snapshot to age off from branch table.manageSnapshots() - .setMaxSnapshotAgeMs(branchName, ageFirstSnapshot) + .setMaxSnapshotAgeMs("branch", ageMs) .commit(); table.expireSnapshots().cleanExpiredFiles(false).commit(); - Assert.assertEquals(2, table.ops().current().snapshots().size()); - Assert.assertNull(table.ops().current().snapshot(1)); + Assert.assertEquals("Should have 2 snapshots (initial removed)", 2, Iterables.size(table.snapshots())); + Assert.assertNull(table.ops().current().snapshot(initialSnapshotId)); } } From 24548e3f5efd4f5f40fd6374dd4fb41aaaad030f Mon Sep 17 00:00:00 2001 From: Ryan Blue Date: Mon, 13 Jun 2022 20:14:18 -0700 Subject: [PATCH 3/3] Fix accidental change. --- core/src/main/java/org/apache/iceberg/RemoveSnapshots.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java b/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java index e80370fc3f99..2f9fa023d69d 100644 --- a/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java +++ b/core/src/main/java/org/apache/iceberg/RemoveSnapshots.java @@ -571,7 +571,7 @@ private CloseableIterable readManifestFiles(Snapshot snapshot) { .build(); } else { - return CloseableIterable.withNoopClose(snapshot.allManifests()); + return CloseableIterable.withNoopClose(snapshot.allManifests(ops.io())); } } }