risingwave_meta/manager/iceberg_compaction/

gc.rs

// Copyright 2026 RisingWave Labs
//
// Licensed 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.

use std::collections::{HashMap, HashSet};

use iceberg::actions::RemoveOrphanFilesAction;
use iceberg::spec::{FormatVersion, ManifestContentType, ManifestFile};
use iceberg::table::Table;
use iceberg::transaction::{ApplyTransactionAction, Transaction};
use itertools::Itertools;
use risingwave_connector::sink::SinkError;
use risingwave_connector::sink::catalog::SinkId;
use risingwave_connector::sink::iceberg::commit_branch;
use thiserror_ext::AsReport;
use tokio::sync::oneshot::Sender;
use tokio::task::JoinHandle;

use super::*;

const MAX_SNAPSHOT_AGE_MS_DEFAULT: i64 = 24 * 60 * 60 * 1000;
const ORPHAN_FILE_LOG_SAMPLE_SIZE: usize = 10;

async fn remove_orphan_files_from_table(
    table: Table,
    min_age: std::time::Duration,
) -> risingwave_connector::sink::Result<Vec<String>> {
    // The pinned iceberg-rust action does not enforce this table-level protection.
    // Check before discovering or deleting files, which may be shared with other tables.
    if !table.metadata().table_properties()?.gc_enabled {
        return Err(SinkError::Config(anyhow!(
            "Cannot remove orphan files when gc.enabled=false: files may be shared with other tables"
        )));
    }
    // The action keeps the current metadata file only when the table carries its
    // location; without it, cleanup would delete the file the catalog points to.
    if table.metadata_location().is_none() {
        return Err(SinkError::Config(anyhow!(
            "Cannot remove orphan files: current metadata location of table {} is unknown",
            table.identifier()
        )));
    }

    Ok(RemoveOrphanFilesAction::new(table)
        .older_than(min_age)
        .execute()
        .await?)
}

#[derive(Debug, PartialEq, Eq)]
struct ManifestRewritePlan {
    rewrite_paths: HashSet<String>,
    data_manifest_count: usize,
    selected_manifest_count: usize,
    estimated_output_manifest_count: usize,
}

fn plan_manifest_rewrite(
    manifests: &[ManifestFile],
    target_size_bytes: u64,
    min_count_to_merge: usize,
) -> ManifestRewritePlan {
    debug_assert!(target_size_bytes > 0);
    debug_assert!(min_count_to_merge > 0);

    let mut candidates_by_spec = HashMap::<i32, Vec<&ManifestFile>>::new();
    let mut data_manifest_count = 0;
    for manifest in manifests {
        if manifest.content != ManifestContentType::Data {
            continue;
        }
        data_manifest_count += 1;
        if manifest.manifest_length >= 0 && (manifest.manifest_length as u64) < target_size_bytes {
            candidates_by_spec
                .entry(manifest.partition_spec_id)
                .or_default()
                .push(manifest);
        }
    }

    let mut rewrite_paths = HashSet::new();
    let mut estimated_output_manifest_count = 0;
    for mut candidates in candidates_by_spec.into_values() {
        // The iceberg-rust version currently pinned by RisingWave creates one
        // output manifest per cluster key and partition spec. Limit each spec
        // to one target-sized batch so the rewrite cannot replace many small
        // manifests with a single oversized manifest. Later maintenance runs
        // will consume the remaining candidates.
        //
        // Pack from the oldest end so completed bins are rewritten first and
        // the newest under-filled bin can accumulate across maintenance runs.
        // A stable sort preserves manifest-list order for V1 manifests and
        // manifests written by the same snapshot.
        candidates.sort_by_key(|manifest| manifest.sequence_number);
        let mut current_bin = Vec::new();
        let mut current_bin_size_bytes = 0_u64;
        let mut completed_bin = None;
        for candidate in candidates {
            let candidate_size_bytes = candidate.manifest_length as u64;
            let next_size_bytes = current_bin_size_bytes.saturating_add(candidate_size_bytes);
            if !current_bin.is_empty() && next_size_bytes > target_size_bytes {
                if current_bin.len() >= 2 {
                    completed_bin = Some(std::mem::take(&mut current_bin));
                    break;
                }
                current_bin = Vec::new();
                current_bin_size_bytes = 0;
            }

            current_bin.push(candidate);
            current_bin_size_bytes = current_bin_size_bytes.saturating_add(candidate_size_bytes);
        }

        // A completed target-sized bin is always worth merging. The newest
        // under-filled bin is merged only after it reaches the configured
        // count threshold. Either path must reduce at least two manifests to
        // one; otherwise the rewrite would only create snapshot churn.
        let selected = if let Some(completed_bin) = completed_bin {
            completed_bin
        } else if current_bin.len() >= 2
            && (current_bin_size_bytes >= target_size_bytes
                || current_bin.len() >= min_count_to_merge)
        {
            current_bin
        } else {
            continue;
        };

        estimated_output_manifest_count += 1;
        rewrite_paths.extend(
            selected
                .into_iter()
                .map(|manifest| manifest.manifest_path.clone()),
        );
    }

    ManifestRewritePlan {
        selected_manifest_count: rewrite_paths.len(),
        rewrite_paths,
        data_manifest_count,
        estimated_output_manifest_count,
    }
}

fn snapshot_expiration_cutoff_ms(iceberg_config: &IcebergConfig, now: i64) -> i64 {
    iceberg_config
        .snapshot_expiration_timestamp_ms(now)
        .unwrap_or(now - MAX_SNAPSHOT_AGE_MS_DEFAULT)
}

impl IcebergCompactionManager {
    pub fn gc_loop(manager: Arc<Self>, interval_sec: u64) -> (JoinHandle<()>, Sender<()>) {
        assert!(
            interval_sec > 0,
            "Iceberg GC interval must be greater than 0"
        );
        let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
        let join_handle = tokio::spawn(async move {
            tracing::info!(
                interval_sec = interval_sec,
                "Starting Iceberg GC loop with configurable interval"
            );
            let mut interval = tokio::time::interval(std::time::Duration::from_secs(interval_sec));

            loop {
                tokio::select! {
                    _ = interval.tick() => {
                        if let Err(e) = manager.perform_gc_operations().await {
                            tracing::error!(error = ?e.as_report(), "GC operations failed");
                        }
                    },
                    _ = &mut shutdown_rx => {
                        tracing::info!("Iceberg GC loop is stopped");
                        return;
                    }
                }
            }
        });

        (join_handle, shutdown_tx)
    }

    pub fn orphan_file_cleanup_loop(
        manager: Arc<Self>,
        interval_sec: u64,
    ) -> (JoinHandle<()>, Sender<()>) {
        assert!(
            interval_sec > 0,
            "Iceberg orphan file cleanup interval must be greater than 0"
        );
        let (shutdown_tx, mut shutdown_rx) = tokio::sync::oneshot::channel();
        let join_handle = tokio::spawn(async move {
            tracing::info!(interval_sec, "Starting Iceberg orphan file cleanup loop");
            let period = std::time::Duration::from_secs(interval_sec);
            let mut interval =
                tokio::time::interval_at(tokio::time::Instant::now() + period, period);
            interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

            loop {
                tokio::select! {
                    _ = interval.tick() => {
                        if let Err(e) = manager.perform_orphan_file_cleanup().await {
                            tracing::error!(
                                error = ?e.as_report(),
                                "Iceberg orphan file cleanup failed",
                            );
                        }
                    },
                    _ = &mut shutdown_rx => {
                        tracing::info!("Iceberg orphan file cleanup loop is stopped");
                        return;
                    }
                }
            }
        });

        (join_handle, shutdown_tx)
    }

    async fn perform_gc_operations(&self) -> MetaResult<()> {
        let (snapshot_expiration_sink_ids, manifest_rewrite_sink_ids) = {
            let guard = self.inner.read();
            (
                guard
                    .snapshot_expiration_sink_ids
                    .iter()
                    .copied()
                    .collect::<Vec<_>>(),
                guard
                    .manifest_rewrite_sink_ids
                    .iter()
                    .copied()
                    .collect::<Vec<_>>(),
            )
        };

        tracing::info!(
            snapshot_expiration_sink_count = snapshot_expiration_sink_ids.len(),
            manifest_rewrite_sink_count = manifest_rewrite_sink_ids.len(),
            "Starting Iceberg metadata maintenance operations",
        );

        // Rewrite first so snapshot expiration in the same tick can clean up
        // manifests replaced by the new snapshot.
        for sink_id in manifest_rewrite_sink_ids {
            if let Err(e) = self.check_and_rewrite_manifests(sink_id).await {
                tracing::error!(
                    error = ?e.as_report(),
                    %sink_id,
                    "Failed to rewrite Iceberg manifests",
                );
            }
        }

        for sink_id in snapshot_expiration_sink_ids {
            if let Err(e) = self.check_and_expire_snapshots(sink_id).await {
                tracing::error!(error = ?e.as_report(), "Failed to perform GC for sink {}", sink_id);
            }
        }

        tracing::info!("Iceberg metadata maintenance operations completed");
        Ok(())
    }

    async fn perform_orphan_file_cleanup(&self) -> MetaResult<()> {
        let orphan_file_cleanup_sink_ids = {
            let guard = self.inner.read();
            guard
                .orphan_file_cleanup_sink_ids
                .iter()
                .copied()
                .collect::<Vec<_>>()
        };

        tracing::info!(
            orphan_file_cleanup_sink_count = orphan_file_cleanup_sink_ids.len(),
            "Starting Iceberg orphan file cleanup operations",
        );

        for sink_id in orphan_file_cleanup_sink_ids {
            if let Err(e) = self.check_and_remove_orphan_files(sink_id).await {
                tracing::error!(
                    error = ?e.as_report(),
                    %sink_id,
                    "Failed to remove Iceberg orphan files",
                );
            }
        }

        tracing::info!("Iceberg orphan file cleanup operations completed");
        Ok(())
    }

    async fn load_maintenance_config(&self, sink_id: SinkId) -> MetaResult<Option<IcebergConfig>> {
        {
            let guard = self.inner.read();
            if guard.is_recovering(sink_id)
                || guard
                    .sink_schedules
                    .get(&sink_id)
                    .is_some_and(|track| !track.source_is_ready())
            {
                return Ok(None);
            }
        }
        let iceberg_config = self.load_iceberg_config(sink_id).await?;
        if iceberg_config.enable_pk_index
            && !self
                .inner
                .read()
                .sink_schedules
                .get(&sink_id)
                .is_some_and(|track| track.source_is_ready())
        {
            return Ok(None);
        }
        Ok(Some(iceberg_config))
    }

    pub async fn check_and_expire_snapshots(&self, sink_id: SinkId) -> MetaResult<()> {
        let now = chrono::Utc::now().timestamp_millis();
        let Some(iceberg_config) = self.load_maintenance_config(sink_id).await? else {
            return Ok(());
        };
        if !iceberg_config.enable_snapshot_expiration {
            let mut guard = self.inner.write();
            guard.snapshot_expiration_sink_ids.remove(&sink_id);
            return Ok(());
        }

        let active_attempt_gc_watermark_snapshot = {
            let guard = self.inner.read();
            // The old track is discarded during recovery, but durable pending rewrites still
            // need protection until the new coordinator has settled them.
            if guard.is_recovering(sink_id) {
                return Ok(());
            }
            guard
                .sink_schedules
                .get(&sink_id)
                .and_then(|track| track.active_attempt_gc_watermark_snapshot())
                .map(|snapshot| snapshot.cloned())
        };

        let mut snapshot_expiration_timestamp_ms =
            snapshot_expiration_cutoff_ms(&iceberg_config, now);

        // Outer `None` means there is no active compaction attempt. Inner `None`
        // means an active attempt has no safe snapshot watermark, so GC skips.
        match active_attempt_gc_watermark_snapshot {
            None => {}
            Some(None) => {
                tracing::info!(
                    catalog_name = iceberg_config.catalog_name(),
                    table_name = iceberg_config.full_table_name()?.to_string(),
                    %sink_id,
                    "Skip snapshot expiration because an active iceberg compaction attempt has no observed GC watermark",
                );
                return Ok(());
            }
            Some(Some(snapshot)) => {
                // A running compaction task may still need snapshots up to its
                // captured watermark, so GC must not expire newer snapshots.
                snapshot_expiration_timestamp_ms =
                    snapshot_expiration_timestamp_ms.min(snapshot.timestamp_ms);
                tracing::info!(
                    catalog_name = iceberg_config.catalog_name(),
                    table_name = iceberg_config.full_table_name()?.to_string(),
                    %sink_id,
                    gc_watermark_branch = %snapshot.branch,
                    gc_watermark_snapshot_id = snapshot.snapshot_id,
                    gc_watermark_timestamp_ms = snapshot.timestamp_ms,
                    protected_snapshot_expiration_timestamp_ms = snapshot_expiration_timestamp_ms,
                    "Protect snapshots expiration with iceberg compaction GC watermark",
                );
            }
        }

        let catalog = iceberg_config.create_catalog().await?;
        let mut table = catalog
            .load_table(&iceberg_config.full_table_name()?)
            .await
            .map_err(|e| SinkError::Iceberg(e.into()))?;

        let metadata = table.metadata();
        let mut snapshots = metadata.snapshots().collect_vec();
        snapshots.sort_by_key(|s| s.timestamp_ms());

        if snapshots.is_empty()
            || snapshots.first().unwrap().timestamp_ms() > snapshot_expiration_timestamp_ms
        {
            return Ok(());
        }

        tracing::info!(
            catalog_name = iceberg_config.catalog_name(),
            table_name = iceberg_config.full_table_name()?.to_string(),
            %sink_id,
            snapshots_len = snapshots.len(),
            snapshot_expiration_timestamp_ms = snapshot_expiration_timestamp_ms,
            snapshot_expiration_retain_last = ?iceberg_config.snapshot_expiration_retain_last,
            clear_expired_files = ?iceberg_config.snapshot_expiration_clear_expired_files,
            clear_expired_meta_data = ?iceberg_config.snapshot_expiration_clear_expired_meta_data,
            "try trigger snapshots expiration",
        );

        let txn = Transaction::new(&table);

        let mut expired_snapshots = txn
            .expire_snapshots()
            .expire_older_than_ms(snapshot_expiration_timestamp_ms)
            .clear_expired_meta_data(iceberg_config.snapshot_expiration_clear_expired_meta_data);

        if let Some(retain_last) = iceberg_config.snapshot_expiration_retain_last {
            expired_snapshots = expired_snapshots.retain_last(
                usize::try_from(retain_last).map_err(|e| SinkError::Config(e.into()))?,
            );
        }

        let before_metadata = table.metadata_ref();
        let tx = expired_snapshots
            .apply(txn)
            .map_err(|e| SinkError::Iceberg(e.into()))?;
        table = tx
            .commit(catalog.as_ref())
            .await
            .map_err(|e| SinkError::Iceberg(e.into()))?;

        if iceberg_config.snapshot_expiration_clear_expired_files {
            table
                .cleanup_expired_files(&before_metadata)
                .await
                .map_err(|e| SinkError::Iceberg(e.into()))?;
        }

        tracing::info!(
            catalog_name = iceberg_config.catalog_name(),
            table_name = iceberg_config.full_table_name()?.to_string(),
            %sink_id,
            "Expired snapshots for iceberg table",
        );

        Ok(())
    }

    pub async fn check_and_remove_orphan_files(&self, sink_id: SinkId) -> MetaResult<()> {
        let Some(iceberg_config) = self.load_maintenance_config(sink_id).await? else {
            return Ok(());
        };
        if !iceberg_config.enable_orphan_file_cleanup {
            self.inner
                .write()
                .orphan_file_cleanup_sink_ids
                .remove(&sink_id);
            return Ok(());
        }

        self.remove_orphan_files_with_config(sink_id, iceberg_config)
            .await?;
        Ok(())
    }

    /// Remove orphan files immediately, independently of the periodic cleanup switch.
    pub async fn remove_orphan_files(&self, sink_id: SinkId) -> MetaResult<u64> {
        let iceberg_config = self
            .load_maintenance_config(sink_id)
            .await?
            .ok_or_else(|| anyhow!("iceberg sink {} is not ready for orphan cleanup", sink_id))?;
        self.remove_orphan_files_with_config(sink_id, iceberg_config)
            .await
    }

    async fn remove_orphan_files_with_config(
        &self,
        sink_id: SinkId,
        iceberg_config: IcebergConfig,
    ) -> MetaResult<u64> {
        // Storage catalog relies on metadata/version-hint.text to load and
        // commit the table, but that catalog-owned file is not tracked by
        // Iceberg table metadata. Do not expose orphan cleanup for this catalog.
        iceberg_config.ensure_orphan_file_cleanup_supported()?;

        // Serialize scheduled and manual cleanup for the same sink. Deleting
        // the same candidate concurrently can otherwise turn an idempotent
        // maintenance request into a partial failure.
        let cleanup_lock = self.orphan_file_cleanup_lock(sink_id);
        let _cleanup_guard = cleanup_lock.lock().await;

        let catalog = iceberg_config.create_catalog().await?;
        let table_ident = iceberg_config.full_table_name()?;
        let table = catalog
            .load_table(&table_ident)
            .await
            .map_err(|e| SinkError::Iceberg(e.into()))?;
        let min_age_millis = iceberg_config.orphan_file_cleanup_min_age_millis();

        tracing::info!(
            iceberg_component = "orphan_file_maintenance",
            iceberg_operation = "remove_orphan_files",
            catalog_name = iceberg_config.catalog_name(),
            table = %table_ident,
            %sink_id,
            min_age_millis,
            "Starting Iceberg orphan file cleanup",
        );

        let orphan_files =
            remove_orphan_files_from_table(table, std::time::Duration::from_millis(min_age_millis))
                .await?;

        let orphan_file_sample = orphan_files
            .iter()
            .take(ORPHAN_FILE_LOG_SAMPLE_SIZE)
            .cloned()
            .collect_vec();
        let orphan_file_count = orphan_files.len() as u64;

        tracing::info!(
            iceberg_component = "orphan_file_maintenance",
            iceberg_operation = "remove_orphan_files",
            catalog_name = iceberg_config.catalog_name(),
            table = %table_ident,
            %sink_id,
            min_age_millis,
            orphan_file_count,
            deleted_file_count = orphan_file_count,
            orphan_file_sample = ?orphan_file_sample,
            "Iceberg orphan file cleanup completed",
        );

        Ok(orphan_file_count)
    }

    pub async fn check_and_rewrite_manifests(&self, sink_id: SinkId) -> MetaResult<()> {
        let Some(iceberg_config) = self.load_maintenance_config(sink_id).await? else {
            return Ok(());
        };
        if !iceberg_config.enable_manifest_rewrite {
            self.inner
                .write()
                .manifest_rewrite_sink_ids
                .remove(&sink_id);
            return Ok(());
        }

        let catalog = iceberg_config.create_catalog().await?;
        let table_ident = iceberg_config.full_table_name()?;
        let table = catalog
            .load_table(&table_ident)
            .await
            .map_err(|e| SinkError::Iceberg(e.into()))?;

        if table.metadata().format_version() >= FormatVersion::V3 {
            // Iceberg format upgrades cannot be downgraded, so stop retrying
            // periodic rewrites for this sink.
            self.inner
                .write()
                .manifest_rewrite_sink_ids
                .remove(&sink_id);
            tracing::warn!(
                iceberg_component = "manifest_maintenance",
                iceberg_operation = "rewrite_manifests",
                catalog_name = iceberg_config.catalog_name(),
                table = %table_ident,
                %sink_id,
                format_version = %table.metadata().format_version(),
                "Skipping manifest rewrite because Iceberg row lineage is enabled",
            );
            return Ok(());
        }

        let branch = commit_branch(iceberg_config.r#type.as_str(), iceberg_config.write_mode);
        let Some(current_snapshot) = table.metadata().snapshot_for_ref(&branch) else {
            tracing::debug!(
                iceberg_component = "manifest_maintenance",
                iceberg_operation = "rewrite_manifests",
                table = %table_ident,
                %sink_id,
                %branch,
                "Skipping manifest rewrite because the target branch has no snapshot",
            );
            return Ok(());
        };
        let current_snapshot_id = current_snapshot.snapshot_id();
        let manifest_list = table
            .object_cache()
            .get_manifest_list(current_snapshot, &table.metadata_ref())
            .await
            .map_err(|e| SinkError::Iceberg(e.into()))?;

        let target_size_bytes = iceberg_config.manifest_rewrite_target_size_bytes();
        let min_count_to_merge = iceberg_config.manifest_rewrite_min_count_to_merge();
        let plan = plan_manifest_rewrite(
            manifest_list.entries(),
            target_size_bytes,
            min_count_to_merge,
        );
        if plan.rewrite_paths.is_empty() {
            tracing::debug!(
                iceberg_component = "manifest_maintenance",
                iceberg_operation = "rewrite_manifests",
                table = %table_ident,
                %sink_id,
                %branch,
                data_manifest_count = plan.data_manifest_count,
                target_size_bytes,
                min_count_to_merge,
                "Skipping manifest rewrite because manifests are not fragmented enough",
            );
            return Ok(());
        }

        tracing::info!(
            iceberg_component = "manifest_maintenance",
            iceberg_operation = "rewrite_manifests",
            catalog_name = iceberg_config.catalog_name(),
            table = %table_ident,
            %sink_id,
            %branch,
            current_snapshot_id,
            data_manifest_count = plan.data_manifest_count,
            selected_manifest_count = plan.selected_manifest_count,
            estimated_output_manifest_count = plan.estimated_output_manifest_count,
            target_size_bytes,
            min_count_to_merge,
            "Starting Iceberg manifest rewrite",
        );

        let selected_manifest_count = plan.selected_manifest_count;
        let estimated_output_manifest_count = plan.estimated_output_manifest_count;
        let rewrite_paths = plan.rewrite_paths;
        let txn = Transaction::new(&table);
        let tx = txn
            .rewrite_manifests()
            .rewrite_if(Box::new(move |manifest| {
                rewrite_paths.contains(&manifest.manifest_path)
            }))
            .cluster_by(Box::new(|_| "risingwave-maintenance".to_owned()))
            .set_target_branch(branch.clone())
            .apply(txn)
            .map_err(|e| SinkError::Iceberg(e.into()))?;
        let table = tx
            .commit(catalog.as_ref())
            .await
            .map_err(|e| SinkError::Iceberg(e.into()))?;

        let Some(rewritten_snapshot) = table.metadata().snapshot_for_ref(&branch) else {
            return Err(anyhow!(
                "Iceberg branch {} disappeared after manifest rewrite for sink {}",
                branch,
                sink_id
            )
            .into());
        };
        if rewritten_snapshot.snapshot_id() == current_snapshot_id {
            tracing::warn!(
                iceberg_component = "manifest_maintenance",
                iceberg_operation = "rewrite_manifests",
                table = %table_ident,
                %sink_id,
                %branch,
                current_snapshot_id,
                selected_manifest_count,
                "Manifest rewrite completed without creating a new snapshot",
            );
            return Ok(());
        }

        let summary = &rewritten_snapshot.summary().additional_properties;
        tracing::info!(
            iceberg_component = "manifest_maintenance",
            iceberg_operation = "rewrite_manifests",
            catalog_name = iceberg_config.catalog_name(),
            table = %table_ident,
            %sink_id,
            %branch,
            previous_snapshot_id = current_snapshot_id,
            snapshot_id = rewritten_snapshot.snapshot_id(),
            selected_manifest_count,
            estimated_output_manifest_count,
            manifests_created = summary.get("manifests-created").map(String::as_str),
            manifests_replaced = summary.get("manifests-replaced").map(String::as_str),
            manifests_kept = summary.get("manifests-kept").map(String::as_str),
            entries_processed = summary.get("entries-processed").map(String::as_str),
            "Iceberg manifest rewrite succeeded",
        );

        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn manifest(
        path: &str,
        manifest_length: i64,
        partition_spec_id: i32,
        content: ManifestContentType,
    ) -> ManifestFile {
        ManifestFile {
            manifest_path: path.to_owned(),
            manifest_length,
            partition_spec_id,
            content,
            sequence_number: 1,
            min_sequence_number: 1,
            added_snapshot_id: 1,
            added_files_count: Some(1),
            existing_files_count: Some(0),
            deleted_files_count: Some(0),
            added_rows_count: Some(1),
            existing_rows_count: Some(0),
            deleted_rows_count: Some(0),
            partitions: None,
            key_metadata: None,
            first_row_id: None,
        }
    }

    #[test]
    fn test_plan_manifest_rewrite_selects_fragmented_data_manifests_per_spec() {
        let manifests = vec![
            manifest("small-a", 40, 0, ManifestContentType::Data),
            manifest("small-b", 40, 0, ManifestContentType::Data),
            manifest("single-other-spec", 40, 1, ManifestContentType::Data),
            manifest("already-large", 100, 0, ManifestContentType::Data),
            manifest("delete", 10, 0, ManifestContentType::Deletes),
        ];

        let plan = plan_manifest_rewrite(&manifests, 100, 2);

        assert_eq!(plan.data_manifest_count, 4);
        assert_eq!(plan.selected_manifest_count, 2);
        assert_eq!(plan.estimated_output_manifest_count, 1);
        assert_eq!(
            plan.rewrite_paths,
            ["small-a".to_owned(), "small-b".to_owned()]
                .into_iter()
                .collect()
        );
    }

    #[test]
    fn test_plan_manifest_rewrite_skips_when_packing_cannot_reduce_count() {
        let manifests = vec![
            manifest("almost-full-a", 60, 0, ManifestContentType::Data),
            manifest("almost-full-b", 60, 0, ManifestContentType::Data),
        ];

        let plan = plan_manifest_rewrite(&manifests, 100, 2);

        assert!(plan.rewrite_paths.is_empty());
        assert_eq!(plan.selected_manifest_count, 0);
        assert_eq!(plan.estimated_output_manifest_count, 0);
    }

    #[test]
    fn test_plan_manifest_rewrite_limits_each_spec_to_one_target_sized_batch() {
        let manifests = vec![
            manifest("small-a", 40, 0, ManifestContentType::Data),
            manifest("small-b", 40, 0, ManifestContentType::Data),
            manifest("small-c", 40, 0, ManifestContentType::Data),
        ];

        let plan = plan_manifest_rewrite(&manifests, 100, 2);

        assert_eq!(plan.selected_manifest_count, 2);
        assert_eq!(plan.estimated_output_manifest_count, 1);
        assert_eq!(
            plan.rewrite_paths,
            ["small-a".to_owned(), "small-b".to_owned()]
                .into_iter()
                .collect()
        );
    }

    #[test]
    fn test_plan_manifest_rewrite_selects_oldest_sequence_first() {
        let mut newest = manifest("newest", 40, 0, ManifestContentType::Data);
        newest.sequence_number = 3;
        let mut oldest_a = manifest("oldest-a", 40, 0, ManifestContentType::Data);
        oldest_a.sequence_number = 1;
        let mut middle = manifest("middle", 40, 0, ManifestContentType::Data);
        middle.sequence_number = 2;
        let mut oldest_b = manifest("oldest-b", 40, 0, ManifestContentType::Data);
        oldest_b.sequence_number = 1;

        let plan = plan_manifest_rewrite(&[newest, oldest_a, middle, oldest_b], 100, 2);

        assert_eq!(plan.selected_manifest_count, 2);
        assert_eq!(
            plan.rewrite_paths,
            ["oldest-a".to_owned(), "oldest-b".to_owned()]
                .into_iter()
                .collect()
        );
    }

    #[test]
    fn test_plan_manifest_rewrite_selects_completed_bin_below_min_count() {
        let manifests = (0..17)
            .map(|index| {
                manifest(
                    &format!("small-{index:03}"),
                    512 * 1024,
                    0,
                    ManifestContentType::Data,
                )
            })
            .collect_vec();

        let plan = plan_manifest_rewrite(&manifests, 8 * 1024 * 1024, 100);

        assert_eq!(plan.data_manifest_count, 17);
        assert_eq!(plan.selected_manifest_count, 16);
        assert_eq!(plan.estimated_output_manifest_count, 1);
        assert!(!plan.rewrite_paths.contains("small-016"));
    }

    #[test]
    fn test_plan_manifest_rewrite_waits_for_underfilled_bin_min_count() {
        let manifests = (0..99)
            .map(|index| {
                manifest(
                    &format!("small-{index:03}"),
                    1024,
                    0,
                    ManifestContentType::Data,
                )
            })
            .collect_vec();

        let plan = plan_manifest_rewrite(&manifests, 1024 * 1024, 100);

        assert_eq!(plan.data_manifest_count, 99);
        assert_eq!(plan.selected_manifest_count, 0);
        assert_eq!(plan.estimated_output_manifest_count, 0);
    }
}

// Iceberg's runtime and file IO need a real Tokio runtime, which madsim does not provide.
#[cfg(all(test, not(madsim)))]
mod orphan_file_tests {
    use std::collections::HashMap;
    use std::time::{Duration, SystemTime};

    use iceberg::io::FileIO;
    use iceberg::spec::{
        FormatVersion, NestedField, PrimitiveType, Schema, SortOrder, TableMetadataBuilder, Type,
        UnboundPartitionSpec,
    };
    use iceberg::table::Table;
    use iceberg::{Runtime, TableIdent};
    use thiserror_ext::AsReport;

    use super::remove_orphan_files_from_table;

    async fn check_orphan_file_cleanup(
        gc_enabled: Option<&str>,
        with_metadata_location: bool,
        expected_error: Option<&str>,
    ) {
        let directory = tempfile::tempdir().unwrap();
        let location = format!("file://{}", directory.path().display());
        let properties = gc_enabled
            .map(|value| HashMap::from([("gc.enabled".to_owned(), value.to_owned())]))
            .unwrap_or_default();
        let schema = Schema::builder()
            .with_fields([
                NestedField::required(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
            ])
            .build()
            .unwrap();
        let metadata = TableMetadataBuilder::new(
            schema,
            UnboundPartitionSpec::default(),
            SortOrder::unsorted_order(),
            location.clone(),
            FormatVersion::V2,
            properties,
        )
        .unwrap()
        .build()
        .unwrap()
        .metadata;
        let metadata_path = directory.path().join("metadata.json");
        let old_path = directory.path().join("old.parquet");
        let recent_path = directory.path().join("recent.parquet");
        std::fs::write(&metadata_path, serde_json::to_vec(&metadata).unwrap()).unwrap();
        std::fs::write(&old_path, b"old orphan").unwrap();
        std::fs::write(&recent_path, b"in-progress write").unwrap();
        for path in [&old_path, &metadata_path] {
            std::fs::File::open(path)
                .unwrap()
                .set_modified(SystemTime::now() - Duration::from_secs(8 * 24 * 3600))
                .unwrap();
        }
        let mut builder = Table::builder()
            .metadata(metadata)
            .identifier(TableIdent::from_strs(["ns", "table"]).unwrap())
            .file_io(FileIO::new_with_fs())
            .runtime(Runtime::current());
        if with_metadata_location {
            builder = builder.metadata_location(format!("{location}/metadata.json"));
        }
        let table = builder.build().unwrap();

        let result =
            remove_orphan_files_from_table(table, Duration::from_secs(7 * 24 * 3600)).await;
        match expected_error {
            None => assert_eq!(result.unwrap(), vec![format!("{location}/old.parquet")]),
            Some(error) => assert!(result.unwrap_err().to_report_string().contains(error)),
        }
        assert_eq!(old_path.exists(), expected_error.is_some());
        assert!(recent_path.exists());
        assert!(metadata_path.exists());
    }

    #[tokio::test]
    async fn test_orphan_file_cleanup_respects_disabled_gc() {
        check_orphan_file_cleanup(Some("false"), true, Some("gc.enabled")).await;
    }

    #[tokio::test]
    async fn test_orphan_file_cleanup_rejects_invalid_gc_property() {
        check_orphan_file_cleanup(Some("invalid"), true, Some("gc.enabled")).await;
    }

    #[tokio::test]
    async fn test_orphan_file_cleanup_requires_metadata_location() {
        check_orphan_file_cleanup(Some("true"), false, Some("metadata location")).await;
    }

    #[tokio::test]
    async fn test_orphan_file_cleanup_preserves_reachable_and_recent_files() {
        check_orphan_file_cleanup(Some("true"), true, None).await;
        check_orphan_file_cleanup(None, true, None).await;
    }
}