You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "vamossagar12 (via GitHub)" <gi...@apache.org> on 2026/03/16 10:21:00 UTC
[PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
vamossagar12 opened a new pull request, #2881:
URL: https://github.com/apache/fluss/pull/2881
<!--
*Thank you very much for contributing to Fluss - we are happy that you want to help us improve Fluss. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [GitHub issue](https://github.com/apache/fluss/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no issue.
- Name the pull request in the format "[component] Title of the pull request", where *[component]* should be replaced by the name of the component being changed. Typically, this corresponds to the component label assigned to the issue (e.g., [kv], [log], [client], [flink]). Skip *[component]* if you are unsure about which is the best component.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes.
- Each pull request should address only one issue, not mix up code from multiple issues.
**(The sections below can be removed for hotfixes or typos)**
-->
### Purpose
<!-- Linking this pull request to the issue -->
Linked issue: close #2213
<!-- What is the purpose of the change -->
### Brief change log
Similar to https://github.com/apache/fluss/issues/2182, this PR adds support for snapshot expiration for Iceberg tables.
### Tests
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java
### API and Format
N/A
### Documentation
No. The configs `table.datalake.auto-expire-snapshot` and `tiering service config lake.tiering.auto-expire-snapshot` already exist. This PR makes Iceberg implement it.
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on code in PR #2881:
URL: https://github.com/apache/fluss/pull/2881#discussion_r3143190578
##########
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeCommitter.java:
##########
@@ -262,6 +279,22 @@ public void close() throws Exception {
}
}
+ /**
+ * Expires old snapshots from the Iceberg table. Uses Iceberg's built-in snapshot expiration
+ * which respects table properties like {@code history.expire.max-snapshot-age-ms} and {@code
+ * history.expire.min-snapshots-to-keep}.
+ */
+ private void expireSnapshots() {
+ try {
+ ExpireSnapshots expireSnapshots =
+ icebergTable.expireSnapshots().cleanExpiredFiles(true);
Review Comment:
You're right — with aggressive expiration settings (e.g., `min-snapshots-to-keep`=1 and `max-snapshot-age-ms`=1ms), the expired snapshot could cause `getMissingLakeSnapshot()` to fail during recovery because `icebergTable.snapshot(expiredId)` returns null.
I've fixed this by making `getMissingLakeSnapshot()` handle expired snapshots gracefully, similar to how Paimon's `SnapshotManager.pickOrLatest()` works. Instead of throwing IllegalStateException, the code now logs a warning and falls through to return the latest Fluss-committed live snapshot for recovery. Since Iceberg never expires the current snapshot, the latest Fluss-committed snapshot is guaranteed to be newer than the expired one.
Added test `testSnapshotExpirationDoesNotBreakRecovery` to validate this scenario. Let me know what you think!
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #2881:
URL: https://github.com/apache/fluss/pull/2881#issuecomment-4087579058
@luoyuxia gentle ping!
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "leekeiabstraction (via GitHub)" <gi...@apache.org>.
leekeiabstraction commented on code in PR #2881:
URL: https://github.com/apache/fluss/pull/2881#discussion_r2991326712
##########
fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/tiering/IcebergLakeCommitter.java:
##########
@@ -262,6 +279,22 @@ public void close() throws Exception {
}
}
+ /**
+ * Expires old snapshots from the Iceberg table. Uses Iceberg's built-in snapshot expiration
+ * which respects table properties like {@code history.expire.max-snapshot-age-ms} and {@code
+ * history.expire.min-snapshots-to-keep}.
+ */
+ private void expireSnapshots() {
+ try {
+ ExpireSnapshots expireSnapshots =
+ icebergTable.expireSnapshots().cleanExpiredFiles(true);
Review Comment:
Does this expire all non latest snapshots? If so, would this cause recovery failure in the case when tiering service failover and restarts from latest Flink checkpoint and couldn't find a deleted iceberg 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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #2881:
URL: https://github.com/apache/fluss/pull/2881#discussion_r2939861902
##########
fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/tiering/IcebergTieringTest.java:
##########
@@ -274,6 +285,108 @@ public Configuration flussClientConfig() {
});
}
+ @ParameterizedTest
+ @MethodSource("snapshotExpireArgs")
+ void testSnapshotExpiration(
+ boolean isTableAutoExpireSnapshot, boolean isLakeTieringExpireSnapshot)
+ throws Exception {
+ int bucketNum = 3;
+ TablePath tablePath =
+ TablePath.of(
+ "iceberg",
+ String.format(
+ "test_snapshot_expire_%s_%s",
+ isTableAutoExpireSnapshot, isLakeTieringExpireSnapshot));
+
+ // Create Iceberg table with snapshot retention properties
+ Map<String, String> tableProperties = new HashMap<>();
+ tableProperties.put(TableProperties.MIN_SNAPSHOTS_TO_KEEP, "1");
+ tableProperties.put(TableProperties.MAX_SNAPSHOT_AGE_MS, "1");
+ createTable(tablePath, false, false, tableProperties);
+
+ TableDescriptor descriptor =
+ TableDescriptor.builder()
+ .schema(
+ org.apache.fluss.metadata.Schema.newBuilder()
+ .column("c1", DataTypes.INT())
+ .column("c2", DataTypes.STRING())
+ .column("c3", DataTypes.STRING())
+ .build())
+ .distributedBy(bucketNum)
+ .property(ConfigOptions.TABLE_DATALAKE_ENABLED, true)
+ .property(
+ ConfigOptions.TABLE_DATALAKE_AUTO_EXPIRE_SNAPSHOT,
+ isTableAutoExpireSnapshot)
+ .build();
+ TableInfo tableInfo = TableInfo.of(tablePath, 0, 1, descriptor, 1L, 1L);
+
+ Configuration lakeTieringConfig = new Configuration();
+ lakeTieringConfig.set(
+ ConfigOptions.LAKE_TIERING_AUTO_EXPIRE_SNAPSHOT, isLakeTieringExpireSnapshot);
+
+ // Write data multiple times to generate snapshots
+ for (int round = 0; round < 5; round++) {
+ writeData(tablePath, tableInfo, lakeTieringConfig, bucketNum);
+ // Small delay to ensure snapshots are older than MAX_SNAPSHOT_AGE_MS=1ms
+ Thread.sleep(10);
+ }
Review Comment:
This test relies on `Thread.sleep` + a 1ms snapshot age to make snapshots eligible for expiration, which can be timing-sensitive and may cause flakes on slow/loaded CI environments. Consider making the expiration condition deterministic (e.g., use a larger max age with a clear margin, or drive expiration based on snapshot timestamps rather than wall-clock sleeps) to avoid time-based dependencies in the test.
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #2881:
URL: https://github.com/apache/fluss/pull/2881#issuecomment-4581500435
`org.apache.fluss.client.security.acl.FlussAuthorizationITCase.testDeleteProducerOffsetsAuthorization` is a test which failed but it's known to be flake. I see https://github.com/apache/fluss/pull/2866 to fix it (not merged yet).
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #2881:
URL: https://github.com/apache/fluss/pull/2881#issuecomment-4321638354
One of the failing tests `CoordinatorEventManagerTest.testMetricsUpdatedImmediatelyOnStartup:95 ` seems to be flaky. Related bug: #3188 . Can raise a separate PR to fix this.
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #2881:
URL: https://github.com/apache/fluss/pull/2881#issuecomment-4692021746
@polyzos , gentle ping, can you please take a look at this one again when you have some time.. Thanks..
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #2881:
URL: https://github.com/apache/fluss/pull/2881#issuecomment-4581392875
> @vamossagar12 Thanks for this.. I have one concen that i would like your thoughts on as well us from @luoyuxia and @MehulBatra ... Unlike Paimon, where setting `WRITE_ONLY=false` lets expiration run in-band as part of the normal commit, Iceberg's `ExpireSnapshots.commit()` is a **separate metadata commit** that:
>
> * Plans across manifests
> * Writes a new `TableMetadata` JSON
> * Synchronously deletes files (because `cleanExpiredFiles=true`)
>
> In a streaming tiering scenario, this happens on **every commit** with no throttling. This can materially increase per-commit latency and produce a metadata/delete storm on object stores. Could we:
>
> * Add an opt-in interval (e.g. `lake.iceberg.expire-snapshot.interval` defaulting to every N commits or M minutes), OR
> * At minimum, document this characteristic on the config option and recommend conservative `history.expire.max-snapshot-age-ms` / `history.expire.min-snapshots-to-keep` settings?
>
> I don't know whether it's a blocker, but I thought to mention.
Thanks @polyzos , it makes sense. Actually I went ahead with option 1 to allow ease of use for the users. I introduced 2 Iceberg-specific config options:
```
lake.iceberg.expire-snapshot.min-interval-commits (default: 10) — minimum commits since last expiration
lake.iceberg.expire-snapshot.min-interval-ms (default: 10min) — minimum time elapsed since last expiration
```
Expiration only fires when both thresholds are satisfied simultaneously. This means:
- In a fast-streaming scenario (e.g. 1 commit/sec), the 10-minute time gate acts as the primary rate-limiter — expiration runs at most once every 10 minutes regardless of commit frequency
- In a slow scenario (e.g. 1 commit/hour), the count gate ensures we don't expire too frequently either
This way the snapshotting can be controlled by the users based on commit counts or time based. Let me now what you think!
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "polyzos (via GitHub)" <gi...@apache.org>.
polyzos commented on PR #2881:
URL: https://github.com/apache/fluss/pull/2881#issuecomment-4368932801
@vamossagar12 can you rebase? I think this PR is good, but @MehulBatra @luoyuxia can you also PTAL?
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #2881:
URL: https://github.com/apache/fluss/pull/2881#issuecomment-4067010939
@luoyuxia please review this PR. Thanks!
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "polyzos (via GitHub)" <gi...@apache.org>.
polyzos commented on PR #2881:
URL: https://github.com/apache/fluss/pull/2881#issuecomment-4532627129
@vamossagar12 Thanks for this.. I have one concen that i would like your thoughts on as well us from @luoyuxia and @MehulBatra ... Unlike Paimon, where setting `WRITE_ONLY=false` lets expiration run in-band as part of the normal commit, Iceberg's `ExpireSnapshots.commit()` is a **separate metadata commit** that:
- Plans across manifests
- Writes a new `TableMetadata` JSON
- Synchronously deletes files (because `cleanExpiredFiles=true`)
In a streaming tiering scenario, this happens on **every commit** with no throttling. This can materially increase per-commit latency and produce a metadata/delete storm on object stores. Could we:
- Add an opt-in interval (e.g. `lake.iceberg.expire-snapshot.interval` defaulting to every N commits or M minutes), OR
- At minimum, document this characteristic on the config option and recommend conservative `history.expire.max-snapshot-age-ms` / `history.expire.min-snapshots-to-keep` settings?
I don't know whether it's a blocker, but I thought to mention.
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #2881:
URL: https://github.com/apache/fluss/pull/2881#issuecomment-4357616952
@leekeiabstraction please take a look when you get a chance.
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [lake/iceberg] Support snapshot expiration for lake tables [fluss]
Posted by "vamossagar12 (via GitHub)" <gi...@apache.org>.
vamossagar12 commented on PR #2881:
URL: https://github.com/apache/fluss/pull/2881#issuecomment-4396909138
@polyzos thanks, I rebased and pushed. @MehulBatra , @luoyuxia can you also take a look. Thanks!
--
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@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org