Add custom iceberg metadata cleaner #621
Conversation
| .expireOlderThan( | ||
| Instant.now().minus(snapshotRetentionInHours, ChronoUnit.HOURS).toEpochMilli()) | ||
| .cleanExpiredFiles(!useInternalIcebergCleaner); // is internal cleaner is enabled, disable iceberg cleaner | ||
| List<Snapshot> removedSnapshots = expireSnapshots.apply(); |
There was a problem hiding this comment.
expiredSnapshots.apply() will apply the changes and returns list of snapshots that will be deleted based on provided parameters without committing them.
| // only remove files that were deleted in an ancestor of the current table state to avoid | ||
| // physically deleting files that were logically deleted in a commit that was rolled back. | ||
| Set<Long> ancestorIds = | ||
| Sets.newHashSet(SnapshotUtil.ancestorIds(latest, snapshotLookup(removedSnapshots, table))); |
There was a problem hiding this comment.
To fetch the ancestors of table before expiring snapshots, we assume total table snapshots before committing = total table snapshots after committing + removed snapshots
| transaction.commitTransaction(); | ||
| // after commit is complete, clean up the manifest files | ||
| if (useInternalIcebergCleaner) { | ||
| cleanExpiredSnapshots(removedSnapshots); |
There was a problem hiding this comment.
To clean up expired snapshots, we're relying on table's current state and list of removed snapshots.
Another approach would be compare metadata before expire snapshots is committed and after. This would give us the exact changes expire snapshots commit has made (Iceberg uses this approach).
But the problem with this approach is how to get the intermediate TableMetadata before commit is made, since Transaction class doesn't expose intermediate state before transaction is committed. One way is casting Transaction to BaseTransaction which exposes TransactionTableOperations which provides intermediate TableMetadata
TableMetadata beforeExpiration = ((BaseTransaction) transaction).underlyingOps().refresh();
transaction
.expireSnapshots()
.expireOlderThan(
Instant.now().minus(snapshotRetentionInHours, ChronoUnit.HOURS).toEpochMilli())
.deleteWith(this::safeDelete) // ensures that only metadata files are deleted
.cleanExpiredFiles(true)
.commit();
TableMetadata afterExpiration = ((BaseTransaction) transaction).underlyingOps().refresh();
// do metadata clean based on before and after expiration metadata
cleanUp(beforeExpiration, afterExpiration)
| private void cleanExpiredSnapshots(List<Snapshot> removedSnapshots) { | ||
| IcebergMetadataCleanupStrategy cleanupStrategy = new IcebergMetadataFileCleaner(transaction.table().io(), deleteExecutorService, planExecutorService, this::safeDelete); | ||
| cleanupStrategy.cleanFiles(table, removedSnapshots); | ||
| } |
There was a problem hiding this comment.
This can be run in async to further improve the performance
Important Read
What is the purpose of the pull request
(For example: This pull request implements the sync for delta format.)
Brief change log
(for example:)
Verify this pull request
(Please pick either of the following options)
This pull request is a trivial rework / code cleanup without any test coverage.
(or)
This pull request is already covered by existing tests, such as (please describe tests).
(or)
This change added tests and can be verified as follows:
(example:)