zhuxiangyi commented on code in PR #966: URL: https://github.com/apache/paimon-rust/pull/966#discussion_r4213547483
########## crates/paimon/src/table/expire_snapshots.rs: ########## @@ -0,0 +1,468 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Snapshot expiration. +//! +//! Reference: Java `ExpireSnapshotsImpl`, configured like the Flink and Spark +//! `expire_snapshots` procedures (`ProcedureUtils.fillInSnapshotOptions`). +//! +//! Expiring snapshots `[earliest, end)` removes, in order: +//! 1. data files deleted by the delta manifests of `(earliest, end]` that no +//! tag still reads; +//! 2. changelog files added by `[earliest, end)`; +//! 3. manifest lists, manifests, index manifests, index files, statistics and +//! reassign plans of `[earliest, end)` that neither `end` nor a tag in the +//! range still references; +//! 4. the snapshot files themselves, then the EARLIEST hint moves to `end`. +//! +//! Snapshot files go last, so an interrupted run leaves every remaining +//! snapshot readable and a later run finishes the job. + +use crate::spec::{CoreOptions, Snapshot}; +use crate::table::snapshot_deletion::{read_long_lived_changelogs, DataFileKey, SnapshotDeletion}; +use crate::table::{BranchManager, Table}; +use crate::{Error, Result}; +use futures::{stream, StreamExt, TryStreamExt}; +use std::collections::hash_map::Entry; +use std::collections::{HashMap, HashSet}; +use std::time::{SystemTime, UNIX_EPOCH}; + +const SNAPSHOT_READ_CONCURRENCY: usize = 16; + +/// Retention rules for one expiration run. Java `ExpireConfig` (snapshot part). +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ExpireConfig { + retain_max: i32, + retain_min: i32, + /// Snapshots whose successor was committed at or after this time are kept. + older_than_millis: i64, + max_deletes: i32, +} + +/// Expire old snapshots of a table and delete the files only they reference. +/// +/// Defaults come from the table options `snapshot.num-retained.min`, +/// `snapshot.num-retained.max`, `snapshot.time-retained` and +/// `snapshot.expire.limit`; each can be overridden per run, like the arguments +/// of Java's `expire_snapshots` procedure. +pub struct ExpireSnapshots<'a> { + table: &'a Table, + retain_max: Option<i32>, + retain_min: Option<i32>, + older_than_millis: Option<i64>, + max_deletes: Option<i32>, + current_time_millis: Option<i64>, +} + +impl<'a> ExpireSnapshots<'a> { + pub(crate) fn new(table: &'a Table) -> Self { + Self { + table, + retain_max: None, + retain_min: None, + older_than_millis: None, + max_deletes: None, + current_time_millis: None, + } + } + + /// Maximum number of completed snapshots to retain. + pub fn with_retain_max(&mut self, retain_max: i32) -> &mut Self { + self.retain_max = Some(retain_max); + self + } + + /// Minimum number of completed snapshots to retain. + pub fn with_retain_min(&mut self, retain_min: i32) -> &mut Self { + self.retain_min = Some(retain_min); + self + } + + /// Expire snapshots older than this epoch-millisecond timestamp instead of + /// `snapshot.time-retained`. + pub fn with_older_than_millis(&mut self, older_than_millis: i64) -> &mut Self { + self.older_than_millis = Some(older_than_millis); + self + } + + /// Maximum number of snapshots to expire in this run. + pub fn with_max_deletes(&mut self, max_deletes: i32) -> &mut Self { + self.max_deletes = Some(max_deletes); + self + } + + #[cfg(test)] + pub(crate) fn with_current_time_millis(&mut self, current_time_millis: i64) -> &mut Self { + self.current_time_millis = Some(current_time_millis); + self + } + + fn config(&self) -> Result<ExpireConfig> { + let core = CoreOptions::new(self.table.schema().options()); + let retain_max = match self.retain_max { + Some(value) => positive("retain_max", value)?, + None => core.snapshot_num_retained_max()?, + }; + let retain_min = match self.retain_min { + Some(value) => positive("retain_min", value)?, + None => core.snapshot_num_retained_min()?, + }; + if retain_max < retain_min { + return Err(Error::DataInvalid { + message: format!( + "retainMax ({retain_max}) must not be less than retainMin ({retain_min})." + ), + source: None, + }); + } + let max_deletes = match self.max_deletes { + Some(value) => positive("max_deletes", value)?, + None => core.snapshot_expire_limit()?, + }; + let older_than_millis = match self.older_than_millis { + Some(value) => value, + None => { + let now = self.current_time_millis.unwrap_or_else(current_time_millis); + let retained = i64::try_from(core.snapshot_time_retained_ms()?).unwrap_or(i64::MAX); + now.saturating_sub(retained) + } + }; + Ok(ExpireConfig { + retain_max, + retain_min, + older_than_millis, + max_deletes, + }) + } + + /// Expire snapshots and return how many snapshot files were removed. + pub async fn execute(&self) -> Result<usize> { + self.table.ensure_not_branch_reference_for_write()?; + let config = self.config()?; + + let snapshot_manager = self.table.snapshot_manager(); + let Some(latest) = snapshot_manager.get_latest_snapshot().await? else { + return Ok(0); + }; + let latest = latest.id(); + let Some(earliest) = snapshot_manager.earliest_snapshot_id().await? else { + return Ok(0); + }; + + // The oldest snapshot `snapshot.num-retained.max` lets us keep. + let min = (latest - i64::from(config.retain_max) + 1).max(earliest); + // `snapshot.num-retained.min` protects the newest snapshots. + let mut max_exclusive = latest - i64::from(config.retain_min) + 1; + // A snapshot a consumer still reads from cannot be deleted. + let consumers = self.table.consumer_manager().list_all().await?; + if let Some(min_next) = consumers.iter().map(|(_, next)| *next).min() { + max_exclusive = max_exclusive.min(min_next); + } + // `snapshot.expire.limit` bounds one run. + max_exclusive = max_exclusive.min(earliest.saturating_add(i64::from(config.max_deletes))); + + for id in min..max_exclusive { + // A snapshot expires only once its successor has also outlived the + // retention time. + if let Some(next) = snapshot_manager.try_get_snapshot(id + 1).await? { + if config.older_than_millis <= next.time_millis() as i64 { + return self.expire_until(earliest, id).await; + } + } + } + self.expire_until(earliest, max_exclusive).await + } + + async fn expire_until(&self, earliest: i64, end_exclusive: i64) -> Result<usize> { + let snapshot_manager = self.table.snapshot_manager(); + if end_exclusive <= earliest { + // Nothing expires; record the earliest snapshot so later runs find + // it without listing the directory. + if !snapshot_manager.earliest_hint_exists().await? { + snapshot_manager.write_earliest_hint(earliest).await?; + } + return Ok(0); + } + + // Futures are built eagerly: a borrowing closure inside the stream + // would make callers' futures lose `Send`. + let reads = (earliest..=end_exclusive) + .map(|id| snapshot_manager.try_get_snapshot(id)) + .collect::<Vec<_>>(); + let snapshots_including_end = stream::iter(reads) + .buffered(SNAPSHOT_READ_CONCURRENCY) + .try_collect::<Vec<_>>() + .await? + .into_iter() + .flatten() + .collect::<Vec<_>>(); + let Some(first) = snapshots_including_end.first() else { + return Ok(0); + }; + let begin_inclusive = first.id(); + let snapshots_excluding_end = snapshots_including_end + .iter() + .filter(|snapshot| snapshot.id() != end_exclusive) + .collect::<Vec<_>>(); + + let tagged = self.tagged_snapshots().await?; + let deletion = SnapshotDeletion::new(self.table)?; + // Branches and long-lived changelogs share files with this history. + // Read everything they reference before deleting anything; if that is + // impossible, the run stops with nothing changed. + let external = self.external_owners(&deletion).await?; + // Their index and changelog manifests too: nothing may be deleted + // before every owner outside main's history has been read. + let external_snapshots = external.snapshots.iter().collect::<Vec<_>>(); + let external_skipping = deletion + .manifest_skipping_set(&external_snapshots, true) + .await?; + + // Data files deleted by a snapshot are unused from that snapshot on, + // so the range is (begin, end]. + let data_snapshots = snapshots_including_end + .iter() + .filter(|snapshot| snapshot.id() != begin_inclusive) + .collect::<Vec<_>>(); + self.clean_data_files(&deletion, &data_snapshots, &tagged, &external.data_files) + .await; + + let mut changelog_files = Vec::new(); + for snapshot in &snapshots_excluding_end { + let still_read = snapshot + .changelog_manifest_list() + .is_some_and(|list| external.changelog_lists.contains(list)); + if !still_read { + changelog_files.extend(deletion.changelog_files(snapshot).await); + } + } + deletion.delete_quietly(changelog_files).await; + + let last = snapshots_including_end + .last() + .expect("checked non-empty above"); + if last.id() != end_exclusive { + // The end snapshot is gone, so nothing protects the manifests it + // would share; stop before deleting any of them. + return Ok(0); + } + let mut skipping_snapshots = find_skipping_tags(&tagged, begin_inclusive, end_exclusive); + skipping_snapshots.push(last); + match deletion + .manifest_skipping_set(&skipping_snapshots, false) + .await + { + Ok(mut skipping) => { + skipping.extend(external_skipping); + let mut manifest_files = Vec::new(); + for snapshot in &snapshots_excluding_end { + manifest_files.extend( + deletion + .unused_manifest_paths(snapshot, &mut skipping) + .await?, + ); + } + deletion.delete_quietly(manifest_files).await; + } + Err(error) => log::info!( + "Skip cleaning manifest files because the skipping set cannot be built: {error}" + ), + } + + for snapshot in &snapshots_excluding_end { + if let Err(error) = snapshot_manager.delete_snapshot(snapshot.id()).await { + log::warn!("Failed to delete snapshot {}: {error}", snapshot.id()); + } + } + snapshot_manager.write_earliest_hint(end_exclusive).await?; + Ok(snapshots_excluding_end.len()) + } + + /// Delete the data files each snapshot's delta removes, keeping the ones + /// the closest earlier tag still reads. A file removed after a tag was taken + /// is live in that tag and in no later snapshot, so the closest earlier tag + /// is the only one that can still read it. + async fn clean_data_files( + &self, + deletion: &SnapshotDeletion, + snapshots: &[&Snapshot], + tagged: &[Snapshot], + external: &HashSet<DataFileKey>, + ) { + let plans = snapshots + .iter() + .map(|snapshot| deletion.plan_deleted_in_delta_manifest(snapshot)) + .collect::<Vec<_>>(); + let plans = stream::iter(plans) + .buffered(SNAPSHOT_READ_CONCURRENCY) + .collect::<Vec<_>>() + .await; + + let mut tag_files: HashMap<i64, Option<HashSet<DataFileKey>>> = HashMap::new(); + let mut paths = Vec::new(); + for (snapshot, plan) in snapshots.iter().zip(&plans) { + let Some(tag) = previous_tag(tagged, snapshot.id()) else { + paths.extend(plan.paths_to_delete(&[external])); + continue; + }; + if let Entry::Vacant(slot) = tag_files.entry(tag.id()) { + let files = match deletion.tag_data_files(tag).await { + Ok(files) => Some(files), + Err(error) => { + log::info!( + "Failed to read the data files of tag snapshot {}: {error}", + tag.id() + ); + None + } + }; + slot.insert(files); + } + match &tag_files[&tag.id()] { + Some(files) => paths.extend(plan.paths_to_delete(&[files, external])), + None => log::info!( + "Skip cleaning data files of snapshot {} because tag snapshot {} cannot be read", + snapshot.id(), + tag.id() + ), + } + } + deletion.delete_quietly(paths).await; + } + + /// Snapshots and tags of every branch, and long-lived changelogs of main + /// and every branch. Unlike main's own history, their files cannot be + /// reasoned about from the expired range, so everything they reference is + /// kept. + async fn external_owners(&self, deletion: &SnapshotDeletion) -> Result<ExternalOwners> { + let file_io = self.table.file_io(); + let branch_manager = BranchManager::new(file_io.clone(), self.table.location().to_string()); + let mut snapshots = read_long_lived_changelogs(file_io, self.table.location()).await?; + for branch in branch_manager.list_all().await? { + let snapshot_manager = self.table.snapshot_manager().with_branch(&branch); + for id in snapshot_manager.list_all_ids().await? { Review Comment: Fixed by the shared #965 commit 6e3242c, which this branch now includes: each branch's latest catalog snapshot is a protected owner during the preflight, and a failed lookup aborts before anything is deleted. `test_empty_commit_keeps_catalog_only_branch` reproduces your case: main at snapshot 2, branch `b1` reading snapshot 1 only through the REST catalog, no tags, `min = max = 1`. After a successful empty `TableCommit`, main still reads row 2 and the branch still reads row 1. It fails without the fix. ########## crates/paimon/src/table/table_commit.rs: ########## @@ -384,7 +387,9 @@ impl TableCommit { commit_identifier, filter_committed, ) - .await + .await?; + self.maintain().await; Review Comment: Fixed in 8cca3eb. `filter_and_commit` now commits every pending group without maintenance, then runs `maintain()` once if at least one group was committed, as Java `commitMultiple` does. Single commits still run maintenance after each commit. `test_recovery_batch_runs_maintenance_after_all_groups` follows your native reproduction: a seed row (1, 10) at snapshot 1 committed by another user, an append (2, 20) and an `update_by_arrow_with_row_id` setting row 0 to 100, recovered together as identifiers 10 and 11 with `min = max = 1`. Both groups commit, the table reads `[(1, 100), (2, 20)]`, and only the latest snapshot remains. Without the fix it fails with `SnapshotNotExist(1)`. ########## crates/paimon/src/spec/core_options.rs: ########## @@ -1352,6 +1366,96 @@ impl<'a> CoreOptions<'a> { .unwrap_or(DEFAULT_COMMIT_MAX_RETRY_WAIT_MS) } + /// Minimum number of completed snapshots to retain + /// (`snapshot.num-retained.min`, default 10). Must be at least 1. + pub fn snapshot_num_retained_min(&self) -> crate::Result<i32> { + self.positive_i32_option( + SNAPSHOT_NUM_RETAINED_MIN_OPTION, + DEFAULT_SNAPSHOT_NUM_RETAINED_MIN, + ) + } + + /// Maximum number of completed snapshots to retain + /// (`snapshot.num-retained.max`, default unbounded). Must be at least 1. + pub fn snapshot_num_retained_max(&self) -> crate::Result<i32> { + self.positive_i32_option( + SNAPSHOT_NUM_RETAINED_MAX_OPTION, + DEFAULT_SNAPSHOT_NUM_RETAINED_MAX, + ) + } + + /// How long a completed snapshot is retained (`snapshot.time-retained`, + /// default 1 h), in milliseconds. + pub fn snapshot_time_retained_ms(&self) -> crate::Result<u64> { + match self.options.get(SNAPSHOT_TIME_RETAINED_OPTION) { + None => Ok(DEFAULT_SNAPSHOT_TIME_RETAINED_MS), + Some(value) => parse_duration_millis(value).ok_or_else(|| crate::Error::DataInvalid { + message: format!("Invalid value for {SNAPSHOT_TIME_RETAINED_OPTION}: '{value}'"), + source: None, + }), + } + } + + /// Maximum number of snapshots expired in one run (`snapshot.expire.limit`, + /// default 50). Must be at least 1. + pub fn snapshot_expire_limit(&self) -> crate::Result<i32> { + self.positive_i32_option(SNAPSHOT_EXPIRE_LIMIT_OPTION, DEFAULT_SNAPSHOT_EXPIRE_LIMIT) + } + + /// Whether changelogs outlive snapshots: `changelog.num-retained.min/max` or + /// `changelog.time-retained` exceed the snapshot counterparts they default + /// to. Java `CoreOptions#changelogLifecycleDecoupled`. + pub fn changelog_lifecycle_decoupled(&self) -> crate::Result<bool> { + let snapshot_max = self.snapshot_num_retained_max()?; + let snapshot_min = self.snapshot_num_retained_min()?; + let snapshot_time = self.snapshot_time_retained_ms()?; + let changelog_max = + self.positive_i32_option(CHANGELOG_NUM_RETAINED_MAX_OPTION, snapshot_max)?; + let changelog_min = + self.positive_i32_option(CHANGELOG_NUM_RETAINED_MIN_OPTION, snapshot_min)?; + let changelog_time = match self.options.get(CHANGELOG_TIME_RETAINED_OPTION) { + None => snapshot_time, + Some(value) => { + parse_duration_millis(value).ok_or_else(|| crate::Error::DataInvalid { + message: format!( + "Invalid value for {CHANGELOG_TIME_RETAINED_OPTION}: '{value}'" + ), + source: None, + })? + } + }; + Ok(changelog_max > snapshot_max + || changelog_min > snapshot_min + || changelog_time > snapshot_time) + } + + /// Whether writers skip compaction and snapshot expiration (`write-only`, + /// fallback `write.compaction-skip`, default false). + pub fn write_only(&self) -> bool { + self.options + .get(WRITE_ONLY_OPTION) + .or_else(|| self.options.get(WRITE_COMPACTION_SKIP_OPTION)) + .map(|v| v.eq_ignore_ascii_case("true")) Review Comment: Fixed in 8cca3eb. `CoreOptions::write_only()` now returns `Result<bool>`: it reads `write-only` first and falls back to `write.compaction-skip`, accepting only `true` / `false` (case-insensitive, trimmed). Any other value is `DataInvalid`. `maintain()` then skips automatic expiration and logs a warning, like it already does for malformed retention options. `test_malformed_no_expiration_switch_skips_maintenance` seeds three snapshots with each key set to `true` and `min = max = 1`, changes only that key to `tru`, and makes an empty commit. All three snapshots remain. The `CoreOptions` unit tests cover `tru` for both keys and an explicit `write-only = FALSE` taking precedence over the fallback. -- 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]
