delta_kernel 0.30.0

//! Builder for creating [`Snapshot`] instances.

use std::sync::Arc;

use delta_kernel_derive::internal_api;
use tracing::{info, instrument};

use crate::cancellation::CancellationTokenRef;
use crate::log_path::LogPath;
use crate::log_segment::{validate_catalog_managed_log_tail, LogSegment};
use crate::log_segment_files::CheckpointHandling;
use crate::metrics::events::SNAPSHOT_COMPLETED_SPAN;
use crate::metrics::{MetricId, SnapshotLoadMetricContext, SnapshotLoadType};
use crate::path::DELTA_LOG_DIR_WITH_SLASH;
use crate::snapshot::{SnapshotHint, SnapshotRef};
use crate::utils::{require, try_parse_uri, PhantomType};
use crate::{Engine, KernelError, KernelResult, Snapshot, Version};

/// Marker for builders that load a snapshot from a table root.
#[doc(hidden)]
pub struct FromTableRoot;

/// Marker for builders that incrementally update an existing snapshot.
#[doc(hidden)]
pub struct FromSnapshot;

/// Builder for creating [`Snapshot`] instances.
///
/// # Example
///
/// ```no_run
/// # use delta_kernel::{Snapshot, Engine};
/// # use url::Url;
/// # fn example(engine: &dyn Engine) -> delta_kernel::KernelResult<()> {
/// let table_root = Url::parse("file:///path/to/table")?;
///
/// // Build a snapshot
/// let snapshot = Snapshot::builder_for(table_root.clone())
///     .at_version(5) // Optional: specify a time-travel version (default is latest version)
///     .build(engine)?;
///
/// # Ok(())
/// # }
/// ```
pub struct SnapshotBuilder<Mode = FromTableRoot> {
    table_root: Option<String>,
    existing_snapshot: Option<SnapshotRef>,
    version: Option<Version>,
    log_tail: Vec<LogPath>,
    max_catalog_version: Option<Version>,
    incremental_replay: IncrementalReplay,
    checkpoint_handling: CheckpointHandling,
    snapshot_hint: Option<Box<SnapshotHint>>,
    /// Kernel-minted id correlating this build's metric events with its child events.
    operation_id: MetricId,
    /// Opaque, caller-supplied id recorded on this build's metric events. Not interpreted by
    /// kernel; set via [`with_correlation_id`](Self::with_correlation_id).
    correlation_id: Option<Arc<str>>,
    /// Optional cooperative cancellation token supplied via
    /// [`with_cancellation_token`](Self::with_cancellation_token). `None` means the build is not
    /// cancellable.
    cancellation_token: Option<CancellationTokenRef>,
    // Carries the zero-sized typestate that limits mode-specific methods at compile time.
    mode: PhantomType<Mode>,
}

/// Builder for incrementally updating an existing [`Snapshot`].
pub type IncrementalSnapshotBuilder = SnapshotBuilder<FromSnapshot>;

// Hand-written because `CancellationToken` is not `Debug`: the token is projected to a bool. Ends
// with `finish_non_exhaustive` so a future field is not silently dropped from the output.
impl<Mode> std::fmt::Debug for SnapshotBuilder<Mode> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SnapshotBuilder")
            .field("table_root", &self.table_root)
            .field("existing_snapshot", &self.existing_snapshot)
            .field("version", &self.version)
            .field("log_tail", &self.log_tail)
            .field("max_catalog_version", &self.max_catalog_version)
            .field("incremental_replay", &self.incremental_replay)
            .field("checkpoint_handling", &self.checkpoint_handling)
            .field("snapshot_hint", &self.snapshot_hint)
            .field("operation_id", &self.operation_id)
            .field("correlation_id", &self.correlation_id)
            .field("cancellable", &self.cancellation_token.is_some())
            .finish_non_exhaustive()
    }
}

/// Controls whether kernel replays commits to advance a stale base CRC (the existing snapshot's
/// in-memory CRC, or an on-disk CRC) to the target snapshot version on load. A CRC already at the
/// target version is always used regardless of this setting; this only bounds the cost of
/// advancing a *stale* CRC.
///
/// A resolved CRC gives the snapshot precomputed file statistics (file count and sizes, useful
/// for query optimization and for writers producing a post-commit CRC) along with domain metadata
/// and set transactions (useful for writers), all without extra log replay.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub enum IncrementalReplay {
    /// Never advance a stale CRC; fall back to normal log replay. `UpToCommits(0)` is equivalent.
    #[default]
    Disabled,
    /// Advance only when the CRC is within `n` commits of the target version, i.e.
    /// `target_version - crc_version <= n`.
    UpToCommits(u64),
    /// Advance regardless of how stale the CRC is.
    Unlimited,
}

impl IncrementalReplay {
    pub(super) fn is_disabled(self) -> bool {
        self == Self::Disabled || self == Self::UpToCommits(0)
    }

    /// Whether the configured budget permits advancing a CRC at `crc_version` to `target_version`.
    /// Errors if `crc_version` is ahead of `target_version`, which violates a caller invariant.
    ///
    /// Example: 95.crc with commits 96.json through 100.json is 5 commits, so `UpToCommits(5)`
    /// advances and `UpToCommits(4)` does not; `Unlimited` always advances.
    pub(crate) fn should_advance(
        self,
        crc_version: Version,
        target_version: Version,
    ) -> KernelResult<bool> {
        let distance = target_version.checked_sub(crc_version).ok_or_else(|| {
            KernelError::internal_error(format!(
                "CRC version {crc_version} is ahead of target version {target_version}"
            ))
        })?;
        Ok(match self {
            IncrementalReplay::Disabled => false,
            IncrementalReplay::UpToCommits(n) => distance <= n,
            IncrementalReplay::Unlimited => true,
        })
    }
}

impl SnapshotBuilder<FromTableRoot> {
    // ============================================================================
    // Constructors
    // ============================================================================

    pub(crate) fn new_for(table_root: impl AsRef<str>) -> Self {
        Self {
            table_root: Some(table_root.as_ref().to_string()),
            existing_snapshot: None,
            version: None,
            log_tail: Vec::new(),
            max_catalog_version: None,
            incremental_replay: IncrementalReplay::default(),
            checkpoint_handling: CheckpointHandling::Adopt,
            snapshot_hint: None,
            operation_id: MetricId::new(),
            correlation_id: None,
            cancellation_token: None,
            mode: PhantomType::default(),
        }
    }

    /// Supply complete snapshot state that Kernel can validate and construct without engine I/O.
    ///
    /// The hint conflicts with [`with_log_tail`](Self::with_log_tail) and non-disabled incremental
    /// CRC replay. An explicit [`at_version`](Self::at_version) must equal the hint version. When
    /// no explicit version is set, a supplied maximum catalog version must also equal the hint
    /// version. The caller must canonicalize supplied log paths into the same URL form as the table
    /// root. Kernel preserves the paths, requires them to be beneath this builder's table log root,
    /// and validates structural consistency without reading the supplied files. The caller must
    /// ensure the protocol and metadata came from those files.
    ///
    /// # Errors
    ///
    /// [`build`](Self::build) returns [`KernelError::SnapshotHint`] for hint conflicts and hinted
    /// log-segment validation failures. Catalog-version, table-root URI, protocol, and metadata
    /// failures retain their normal error variants.
    #[allow(dead_code)]
    #[internal_api]
    pub(crate) fn with_snapshot_hint(mut self, hint: impl Into<Box<SnapshotHint>>) -> Self {
        self.snapshot_hint = Some(hint.into());
        self
    }
}

impl SnapshotBuilder<FromSnapshot> {
    pub(crate) fn new_from(existing_snapshot: SnapshotRef) -> Self {
        Self {
            table_root: None,
            existing_snapshot: Some(existing_snapshot),
            version: None,
            log_tail: Vec::new(),
            max_catalog_version: None,
            incremental_replay: IncrementalReplay::default(),
            checkpoint_handling: CheckpointHandling::Adopt,
            snapshot_hint: None,
            operation_id: MetricId::new(),
            correlation_id: None,
            cancellation_token: None,
            mode: PhantomType::default(),
        }
    }

    /// Skip adopting new checkpoints while updating the input snapshot.
    ///
    /// The resulting snapshot retains newly listed commits instead of adopting a newer checkpoint
    /// as its replay base. For example, when updating snapshot N to N+5, a checkpoint at N+3 is
    /// recognized but not adopted, and commits N+1 through N+5 remain in the log segment.
    ///
    /// The update fails if the new commits are not contiguous.
    ///
    /// This option increases memory use and may increase full-scan log replay. It is available
    /// only for builders created by [`Snapshot::builder_from`].
    pub fn skip_new_checkpoints(mut self) -> Self {
        self.checkpoint_handling = CheckpointHandling::Ignore;
        self
    }
}

impl<Mode> SnapshotBuilder<Mode> {
    // ============================================================================
    // Chainable configuration
    // ============================================================================

    /// Sets the target version of the [`Snapshot`].
    ///
    /// Without a snapshot hint, omitting this targets the latest table version. With a hint,
    /// omitting this uses the hint version and its supplied freshness.
    pub fn at_version(mut self, version: Version) -> Self {
        self.version = Some(version);
        self
    }

    /// Set the log tail to use when building the snapshot. This allows catalogs or external
    /// systems to provide an up-to-date log tail when used to build a snapshot.
    ///
    /// Note that the log tail must be a contiguous sequence of commits from M..=N where N is the
    /// target version of the snapshot and 0 <= M <= N.
    ///
    /// See [`with_max_catalog_version`] for additional constraints when loading catalog-managed
    /// tables.
    ///
    /// [`with_max_catalog_version`]: Self::with_max_catalog_version
    pub fn with_log_tail(mut self, log_tail: Vec<LogPath>) -> Self {
        self.log_tail = log_tail;
        self
    }

    /// Set the maximum catalog-ratified version. When set, the snapshot will not load versions
    /// beyond this limit, even if later commits exist on the filesystem. This ensures the catalog
    /// remains the source of truth for catalog-managed tables.
    ///
    /// When no explicit time-travel version is set via [`at_version`], `max_catalog_version` is
    /// used as the effective target version. When time-travelling to an explicit version,
    /// `max_catalog_version` must still be set for catalog-managed tables -- the requested version
    /// must not exceed it.
    ///
    /// # Log tail requirements
    ///
    /// When `max_catalog_version` is set and no time-travel version is specified, the last entry in
    /// the log tail must match `max_catalog_version` exactly. When time-travelling, the last log
    /// tail entry must be >= the requested version.
    ///
    /// [`at_version`]: Self::at_version
    pub fn with_max_catalog_version(mut self, max_catalog_version: Version) -> Self {
        self.max_catalog_version = Some(max_catalog_version);
        self
    }

    /// Supply a [`CancellationToken`] for snapshot builds that list or read the log.
    ///
    /// Kernel forwards the token (if any) to cancellation-aware [`Engine`] listing and read
    /// operations, and [`build`](Self::build) fails with [`KernelError::Cancelled`] if cancellation
    /// is observed before they complete. Snapshot-hint builds perform no listing or reads, so
    /// the token has no effect on them. By default (or when passing `None`), the build is not
    /// cancellable.
    ///
    /// [`CancellationToken`]: crate::CancellationToken
    /// [`KernelError::Cancelled`]: crate::KernelError::Cancelled
    pub fn with_cancellation_token(
        mut self,
        token: impl Into<Option<CancellationTokenRef>>,
    ) -> Self {
        self.cancellation_token = token.into();
        self
    }

    /// Bound how many commits kernel will replay to advance a stale CRC to the target version.
    /// See [`IncrementalReplay`]. Defaults to [`IncrementalReplay::Disabled`].
    ///
    /// Writers should set this to [`IncrementalReplay::Unlimited`] for faster writes, as should
    /// readers that always want table-level file statistics for query optimization.
    ///
    /// Applies to both fresh and incremental builds.
    pub fn with_incremental_crc_replay(mut self, mode: IncrementalReplay) -> Self {
        self.incremental_replay = mode;
        self
    }

    /// Attach an opaque, caller-supplied correlation id for joining this build's metric events to
    /// the caller's own request or operation id. An empty id is treated as unset. When unset,
    /// behavior is unchanged.
    pub fn with_correlation_id(mut self, correlation_id: impl Into<Arc<str>>) -> Self {
        self.correlation_id = Some(correlation_id.into()).filter(|id| !id.is_empty());
        self
    }

    // ============================================================================
    // Terminal: build the Snapshot
    // ============================================================================

    /// Create a new [`Snapshot`]. This returns a [`SnapshotRef`] (`Arc<Snapshot>`), perhaps
    /// returning a reference to an existing snapshot if the request to build a new snapshot
    /// matches the version of an existing snapshot.
    ///
    /// Reports metrics: [`MetricEvent::SnapshotBuildSuccess`] or
    /// [`MetricEvent::SnapshotBuildFailure`]. Events include `load_type`; hinted builds use
    /// `snapshot_hint` and emit no child log-load events because they perform no engine I/O.
    ///
    /// # Parameters
    ///
    /// - `engine`: Implementation of [`Engine`] apis.
    ///
    /// [`MetricEvent::SnapshotBuildSuccess`]: crate::metrics::MetricEvent::SnapshotBuildSuccess
    /// [`MetricEvent::SnapshotBuildFailure`]: crate::metrics::MetricEvent::SnapshotBuildFailure
    // `is_catalog_managed` is the requested load mode, not the confirmed protocol
    // (see `IS_CATALOG_MANAGED_FIELD`).
    #[instrument(
        name = SNAPSHOT_COMPLETED_SPAN,
        skip_all,
        fields(path = %self.table_path(), report, enable_call_frame, version = tracing::field::Empty, operation_id = %self.operation_id, is_catalog_managed = self.max_catalog_version.is_some(), correlation_id = self.correlation_id.as_deref().unwrap_or(""), load_type = self.load_type().as_ref()),
        err
    )]
    pub fn build(self, engine: &dyn Engine) -> KernelResult<SnapshotRef> {
        // Fold the context into the message string rather than passing structured fields: this
        // `info!` fires inside the `snap.build` metrics span, where any field the
        // `SnapshotBuildSuccess` event doesn't recognize would trip a spurious "Invalid field"
        // warning from the metrics layer.
        info!(
            "building snapshot: target={}, from_version={:?}, log_tail_len={}, \
             max_catalog_version={:?}",
            self.target_version_str(),
            self.existing_snapshot.as_ref().map(|s| s.version()),
            self.log_tail.len(),
            self.max_catalog_version
        );

        let load_type = self.load_type();

        // Destructure self so fields can be moved independently
        let Self {
            table_root,
            existing_snapshot,
            version,
            log_tail,
            max_catalog_version,
            incremental_replay,
            checkpoint_handling,
            snapshot_hint,
            operation_id,
            correlation_id,
            cancellation_token,
            mode: _,
        } = self;

        let metric_context = SnapshotLoadMetricContext {
            operation_id,
            is_catalog_managed: max_catalog_version.is_some(),
            correlation_id,
            load_type,
        };

        let snapshot = if let Some(snapshot_hint) = snapshot_hint {
            snapshot_hint.build(
                table_root,
                version,
                log_tail,
                max_catalog_version,
                incremental_replay,
            )?
        } else {
            let log_tail: Vec<_> = log_tail.into_iter().map(Into::into).collect();

            // Pre-build validations for catalog-managed tables
            Self::validate_catalog_managed_build_inputs(version, max_catalog_version, &log_tail)?;

            // Use time-travel version if set, otherwise fall back to max_catalog_version. Passing
            // this as the version to LogSegment::for_snapshot does NOT skip the _last_checkpoint
            // hint -- the hint is still used when its version <= effective_version.
            let effective_version = version.or(max_catalog_version);

            // A snapshot is latest when no explicit time-travel version is requested, or when the
            // requested version is exactly the max_catalog_version.
            let built_as_latest = version.is_none() || version == max_catalog_version;

            if let Some(table_root) = table_root {
                let table_url = try_parse_uri(table_root)?;
                let log_segment = LogSegment::for_snapshot(
                    engine.storage_handler().as_ref(),
                    table_url.join(DELTA_LOG_DIR_WITH_SLASH)?,
                    log_tail,
                    effective_version,
                    metric_context.clone(),
                    cancellation_token.as_ref(),
                )?;
                Snapshot::try_new_from_log_segment(
                    table_url,
                    log_segment,
                    engine,
                    metric_context,
                    incremental_replay,
                    built_as_latest,
                )
                .map(Into::into)?
            } else {
                let Some(existing_snapshot) = existing_snapshot else {
                    return Err(KernelError::internal_error(
                        "SnapshotBuilder should have either table_root or existing_snapshot",
                    ));
                };
                Snapshot::try_new_from(
                    existing_snapshot,
                    log_tail,
                    engine,
                    effective_version,
                    metric_context,
                    incremental_replay,
                    checkpoint_handling,
                    built_as_latest,
                    cancellation_token.as_ref(),
                )?
            }
        };

        // Post-build validations for catalog-managed tables
        Self::validate_catalog_managed_build_result(&snapshot, max_catalog_version)?;
        tracing::Span::current().record("version", snapshot.version());
        Ok(snapshot)
    }

    // ============================================================================
    // Helpers
    // ============================================================================

    // ===== Catalog-managed Validations =====

    /// Pre-build validations for catalog-managed table invariants.
    fn validate_catalog_managed_build_inputs(
        version: Option<Version>,
        max_catalog_version: Option<Version>,
        log_tail: &[crate::path::ParsedLogPath],
    ) -> KernelResult<()> {
        validate_catalog_managed_log_tail(version, max_catalog_version, log_tail)
    }

    /// Post-build validation: catalog-managed tables must have max_catalog_version, and
    /// non-catalog-managed tables must not.
    fn validate_catalog_managed_build_result(
        snapshot: &SnapshotRef,
        max_catalog_version: Option<Version>,
    ) -> KernelResult<()> {
        let is_catalog_managed = snapshot.table_configuration().is_catalog_managed();

        require!(
            !is_catalog_managed || max_catalog_version.is_some(),
            KernelError::MaxCatalogVersion(
                "Max catalog version is required when loading a catalog-managed table. \
                 Use with_max_catalog_version()."
                    .to_string()
            )
        );
        if let Some(max_catalog_version) = max_catalog_version {
            require!(
                is_catalog_managed,
                KernelError::MaxCatalogVersion(format!(
                    "Max catalog version {max_catalog_version} must not be set for a \
                     non-catalog-managed table"
                ))
            );
        }

        Ok(())
    }

    // ===== Instrumentation Helpers =====

    fn table_path(&self) -> &str {
        self.table_root
            .as_deref()
            .or_else(|| {
                self.existing_snapshot
                    .as_ref()
                    .map(|s| s.table_root().as_str())
            })
            .unwrap_or("unknown")
    }

    /// A build from a table root is a fresh, full log listing; a build from an existing snapshot
    /// reuses that snapshot's log root and lists only the commits above it (`table_root` is None).
    fn load_type(&self) -> SnapshotLoadType {
        if self.snapshot_hint.is_some() {
            SnapshotLoadType::SnapshotHint
        } else if self.table_root.is_some() {
            SnapshotLoadType::Full
        } else {
            SnapshotLoadType::Incremental
        }
    }

    fn target_version_str(&self) -> String {
        if let Some(mcv) = self.max_catalog_version {
            return match self.version {
                Some(v) => format!("{v} (max_catalog_version={mcv})"),
                None => format!("{mcv} (max_catalog_version)"),
            };
        }

        self.version
            .map(|v| v.to_string())
            .unwrap_or_else(|| "LATEST".into())
    }
}

#[cfg(test)]
pub(super) mod tests {
    use std::sync::Arc;
    use std::time::Duration;

    use itertools::Itertools;
    use serde_json::json;
    use test_utils::{actions_to_string, add_commit, TestAction};

    use super::*;
    use crate::engine::sync::SyncEngine;
    use crate::metrics::{LogSegmentLoadType, MetricEvent};
    use crate::object_store::memory::InMemory;
    use crate::object_store::path::Path;
    use crate::object_store::{DynObjectStore, ObjectStoreExt as _};
    use crate::unit_test_utils::{install_thread_local_metrics_reporter, CapturingReporter};
    use crate::utils::FoldWithOption as _;

    pub(in crate::snapshot) fn setup_test() -> (Arc<SyncEngine>, Arc<DynObjectStore>, String) {
        let table_root = String::from("memory:///");
        let store = Arc::new(InMemory::new());
        let engine = Arc::new(SyncEngine::new_with_store(store.clone()));
        (engine, store, table_root)
    }

    pub(in crate::snapshot) async fn create_table(
        store: &Arc<DynObjectStore>,
        table_root: &str,
    ) -> KernelResult<(), Box<dyn std::error::Error>> {
        add_commit(
            table_root,
            store.as_ref(),
            0,
            actions_to_string(vec![TestAction::Metadata]),
        )
        .await?;
        add_commit(
            table_root,
            store.as_ref(),
            1,
            format!(
                "{}\n{}",
                json!({"commitInfo": {"operation": "WRITE"}}),
                actions_to_string(vec![TestAction::Add("part-00000-test.parquet".into())])
            ),
        )
        .await?;
        Ok(())
    }

    #[test_log::test(tokio::test)]
    async fn test_snapshot_builder() -> KernelResult<(), Box<dyn std::error::Error>> {
        let (engine, store, table_root) = setup_test();
        let engine = engine.as_ref();
        create_table(&store, &table_root).await?;

        let snapshot = SnapshotBuilder::new_for(table_root.clone()).build(engine)?;
        assert_eq!(snapshot.version(), 1);

        let snapshot = SnapshotBuilder::new_for(table_root.clone())
            .at_version(0)
            .build(engine)?;
        assert_eq!(snapshot.version(), 0);

        Ok(())
    }

    #[test_log::test(tokio::test)]
    async fn test_snapshot_with_unsupported_type() -> KernelResult<()> {
        let (engine, store, table_root) = setup_test();
        let engine = engine.as_ref();

        // Create a table with an unsupported type in the schema
        let protocol = json!({
            "minReaderVersion": 1,
            "minWriterVersion": 2,
        });

        let metadata = json!({
            "id": "test-table-id",
            "format": {
                "provider": "parquet",
                "options": {}
            },
            "schemaString": "{\"type\":\"struct\",\"fields\":[{\"name\":\"id\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"interval_col\",\"type\":\"interval year to second\",\"nullable\":true,\"metadata\":{}}]}",
            "partitionColumns": [],
            "configuration": {},
            "createdTime": 1587968585495i64
        });

        let commit0 = [
            json!({
                "protocol": protocol
            }),
            json!({
                "metaData": metadata
            }),
        ];

        let commit0_data = commit0
            .iter()
            .map(ToString::to_string)
            .collect_vec()
            .join("\n");

        let path = Path::from("_delta_log/00000000000000000000.json");
        store.put(&path, commit0_data.into()).await?;

        // Try to build a snapshot and expect a clear error message
        let result = SnapshotBuilder::new_for(table_root.clone()).build(engine);
        assert!(result.is_err());

        let err = result.unwrap_err();
        let err_msg = err.to_string();
        assert!(
            err_msg.contains("Unsupported Delta table type: 'interval year to second'"),
            "Expected clear error message about unsupported type, got: {err_msg}"
        );

        Ok(())
    }

    pub(in crate::snapshot) fn measuring_reporter(
    ) -> (Arc<CapturingReporter>, tracing::subscriber::DefaultGuard) {
        let reporter = Arc::new(CapturingReporter::default());
        let guard = install_thread_local_metrics_reporter(reporter.clone());
        (reporter, guard)
    }

    #[test_log::test(tokio::test)]
    async fn snapshot_failed_emits_metric_on_error() -> KernelResult<()> {
        let (engine, store, table_root) = setup_test();

        // Unsupported schema type forces a build failure
        let protocol = json!({"minReaderVersion": 1, "minWriterVersion": 2});
        let metadata = json!({
            "id": "test-table-id",
            "format": {"provider": "parquet", "options": {}},
            "schemaString": r#"{"type":"struct","fields":[{"name":"id","type":"interval year to second","nullable":true,"metadata":{}}]}"#,
            "partitionColumns": [],
            "configuration": {},
            "createdTime": 1587968585495i64
        });
        let commit0_data = [json!({"protocol": protocol}), json!({"metaData": metadata})]
            .iter()
            .map(ToString::to_string)
            .collect_vec()
            .join("\n");
        store
            .put(
                &Path::from("_delta_log/00000000000000000000.json"),
                commit0_data.into(),
            )
            .await?;

        let (reporter, _guard) = measuring_reporter();
        let result = SnapshotBuilder::new_for(table_root).build(engine.as_ref());
        assert!(result.is_err());

        let events = reporter.events();
        assert!(
            events
                .iter()
                .any(|e| matches!(e, MetricEvent::SnapshotBuildFailure(_))),
            "expected SnapshotBuildFailure event on build failure"
        );
        assert!(
            !events
                .iter()
                .any(|e| matches!(e, MetricEvent::SnapshotBuildSuccess(_))),
            "should not emit SnapshotBuildSuccess on failure"
        );
        Ok(())
    }

    #[test_log::test(tokio::test)]
    async fn log_segment_load_failure_emits_metric_on_empty_log() -> KernelResult<()> {
        let (engine, _store, table_root) = setup_test();
        let (reporter, _guard) = measuring_reporter();

        assert!(SnapshotBuilder::new_for(table_root)
            .build(engine.as_ref())
            .is_err());

        let events = reporter.events();
        let failure = events
            .iter()
            .find_map(|e| match e {
                MetricEvent::LogSegmentLoadFailure(f) => Some(f),
                _ => None,
            })
            .expect("expected LogSegmentLoadFailure when the log has no commits");
        assert_eq!(failure.load_type, LogSegmentLoadType::Full);
        Ok(())
    }

    #[test_log::test(tokio::test)]
    async fn protocol_metadata_load_failure_emits_metric_when_actions_absent(
    ) -> KernelResult<(), Box<dyn std::error::Error>> {
        let (engine, store, table_root) = setup_test();
        // A commit with no protocol/metadata: the segment lists fine, then the read fails.
        add_commit(
            &table_root,
            store.as_ref(),
            0,
            actions_to_string(vec![TestAction::Add("part-00000-test.parquet".into())]),
        )
        .await?;
        let (reporter, _guard) = measuring_reporter();

        assert!(SnapshotBuilder::new_for(table_root)
            .build(engine.as_ref())
            .is_err());

        let events = reporter.events();
        assert!(
            events
                .iter()
                .any(|e| matches!(e, MetricEvent::ProtocolMetadataLoadFailure(_))),
            "expected ProtocolMetadataLoadFailure when protocol/metadata are absent"
        );
        assert!(
            !events
                .iter()
                .any(|e| matches!(e, MetricEvent::ProtocolMetadataLoadSuccess(_))),
            "must not emit ProtocolMetadataLoadSuccess when the load fails"
        );
        Ok(())
    }

    #[test_log::test(tokio::test)]
    async fn snapshot_update_from_existing_emits_metric(
    ) -> KernelResult<(), Box<dyn std::error::Error>> {
        let (engine, store, table_root) = setup_test();
        create_table(&store, &table_root).await?;

        // Build v0 snapshot before installing the reporter so only the update is measured
        let snap_v0 = SnapshotBuilder::new_for(table_root)
            .at_version(0)
            .build(engine.as_ref())?;
        assert_eq!(snap_v0.version(), 0);

        let (reporter, _guard) = measuring_reporter();

        let snap_v1 = SnapshotBuilder::new_from(snap_v0).build(engine.as_ref())?;
        assert_eq!(snap_v1.version(), 1);

        let events = reporter.events();
        let (version, duration) = events
            .iter()
            .find_map(|e| match e {
                MetricEvent::SnapshotBuildSuccess(s) => Some((s.version, s.duration)),
                _ => None,
            })
            .expect("expected SnapshotBuildSuccess event");
        assert_eq!(version, 1, "version should match the updated snapshot");
        assert!(duration > Duration::ZERO, "duration should be non-zero");
        Ok(())
    }

    #[test_log::test(tokio::test)]
    async fn snapshot_update_to_earlier_version_emits_failed_metric(
    ) -> KernelResult<(), Box<dyn std::error::Error>> {
        let (engine, store, table_root) = setup_test();
        create_table(&store, &table_root).await?;

        // Build v1 snapshot before installing the reporter
        let snap_v1 = SnapshotBuilder::new_for(table_root).build(engine.as_ref())?;
        assert_eq!(snap_v1.version(), 1);

        let (reporter, _guard) = measuring_reporter();

        let result = SnapshotBuilder::new_from(snap_v1)
            .at_version(0)
            .build(engine.as_ref());
        assert!(
            result.is_err(),
            "updating to an earlier version should fail"
        );

        let events = reporter.events();
        assert!(
            events
                .iter()
                .any(|e| matches!(e, MetricEvent::SnapshotBuildFailure(_))),
            "expected SnapshotBuildFailure when version update goes backwards"
        );
        assert!(
            !events
                .iter()
                .any(|e| matches!(e, MetricEvent::SnapshotBuildSuccess(_))),
            "should not emit SnapshotBuildSuccess when version update fails"
        );
        Ok(())
    }

    #[test_log::test(tokio::test)]
    async fn snapshot_completed_duration_exceeds_log_segment_load_duration(
    ) -> KernelResult<(), Box<dyn std::error::Error>> {
        let (engine, store, table_root) = setup_test();
        create_table(&store, &table_root).await?;

        let (reporter, _guard) = measuring_reporter();
        let _snap = SnapshotBuilder::new_for(table_root).build(engine.as_ref())?;

        let events = reporter.events();
        let snap_duration = events
            .iter()
            .find_map(|e| match e {
                MetricEvent::SnapshotBuildSuccess(s) => Some(s.duration),
                _ => None,
            })
            .expect("expected SnapshotBuildSuccess event");
        let segment_duration = events
            .iter()
            .find_map(|e| match e {
                MetricEvent::LogSegmentLoadSuccess(s) => Some(s.duration),
                _ => None,
            })
            .expect("expected LogSegmentLoadSuccess event");

        assert!(
            snap_duration > Duration::ZERO,
            "duration should be non-zero"
        );
        assert!(
            snap_duration >= segment_duration,
            "SnapshotBuildSuccess.duration ({snap_duration:?}) should be >= LogSegmentLoadSuccess.duration ({segment_duration:?})"
        );
        Ok(())
    }

    #[rstest::rstest]
    #[case::with_id(Some("req-abc-123"))]
    #[case::without_id(None)]
    #[test_log::test(tokio::test)]
    async fn snapshot_build_and_child_events_carry_correlation_id(
        #[case] correlation_id: Option<&str>,
    ) -> KernelResult<(), Box<dyn std::error::Error>> {
        let (engine, store, table_root) = setup_test();
        create_table(&store, &table_root).await?;

        let (reporter, _guard) = measuring_reporter();
        let _ = SnapshotBuilder::new_for(table_root)
            .fold_with(correlation_id, SnapshotBuilder::with_correlation_id)
            .build(engine.as_ref())?;

        // The build event and its snapshot-load child events must all carry the id, since they
        // ride the same SnapshotLoadMetricContext.
        let events = reporter.events();
        let id_of = |pick: fn(&MetricEvent) -> Option<&Option<Arc<str>>>| {
            events
                .iter()
                .find_map(pick)
                .expect("expected event")
                .as_deref()
                .map(str::to_string)
        };
        let build_id = id_of(|e| match e {
            MetricEvent::SnapshotBuildSuccess(s) => Some(&s.correlation_id),
            _ => None,
        });
        let segment_id = id_of(|e| match e {
            MetricEvent::LogSegmentLoadSuccess(s) => Some(&s.correlation_id),
            _ => None,
        });
        let metadata_id = id_of(|e| match e {
            MetricEvent::ProtocolMetadataLoadSuccess(s) => Some(&s.correlation_id),
            _ => None,
        });
        let expected = correlation_id.map(str::to_string);
        assert_eq!(build_id, expected);
        assert_eq!(segment_id, expected, "log-segment child must carry the id");
        assert_eq!(
            metadata_id, expected,
            "protocol/metadata child must carry the id"
        );
        Ok(())
    }

    pub(in crate::snapshot) mod catalog_managed_tests {
        use test_utils::{
            actions_to_string, actions_to_string_catalog_managed, add_commit, add_staged_commit,
            TestAction,
        };

        use super::*;
        use crate::log_path::LogPath;
        use crate::utils::try_parse_uri;
        use crate::FileMeta;

        pub(in crate::snapshot) fn create_log_path(table_root: &str, commit_path: Path) -> LogPath {
            let table_url = try_parse_uri(table_root).expect("Failed to parse table root");
            let commit_url = table_url.join(commit_path.as_ref()).unwrap();
            let file_meta = FileMeta {
                location: commit_url,
                last_modified: 123,
                size: 100,
            };
            LogPath::try_new(file_meta).expect("Failed to create LogPath")
        }

        /// Creates an in-memory engine, store, and table root with an initial catalog-managed
        /// commit at version 0 (protocol + metadata).
        pub(in crate::snapshot) async fn setup_catalog_managed_test(
        ) -> (Arc<SyncEngine>, Arc<DynObjectStore>, String) {
            let (engine, store, table_root) = setup_test();
            let actions = vec![TestAction::Metadata];
            add_commit(
                &table_root,
                store.as_ref(),
                0,
                actions_to_string_catalog_managed(actions),
            )
            .await
            .expect("Failed to write initial catalog-managed commit");
            (engine, store, table_root)
        }

        #[test_log::test(tokio::test)]
        async fn test_staged_commits_without_max_catalog_version_errors(
        ) -> KernelResult<(), Box<dyn std::error::Error>> {
            let (engine, store, table_root) = setup_catalog_managed_test().await;
            let path1 =
                add_staged_commit(&table_root, store.as_ref(), 1, String::from("{}")).await?;

            let log_tail = vec![create_log_path(&table_root, path1)];

            let result = SnapshotBuilder::new_for(table_root)
                .with_log_tail(log_tail)
                .build(engine.as_ref());

            assert!(matches!(result, Err(KernelError::MaxCatalogVersion(_))));

            Ok(())
        }

        #[test_log::test(tokio::test)]
        async fn test_version_exceeds_max_catalog_version_errors() -> KernelResult<()> {
            let (engine, _store, table_root) = setup_catalog_managed_test().await;

            let result = SnapshotBuilder::new_for(table_root)
                .at_version(5)
                .with_max_catalog_version(3)
                .build(engine.as_ref());

            assert!(matches!(result, Err(KernelError::MaxCatalogVersion(_))));

            Ok(())
        }

        #[test_log::test(tokio::test)]
        async fn test_log_tail_last_version_mismatch_errors(
        ) -> KernelResult<(), Box<dyn std::error::Error>> {
            let (engine, store, table_root) = setup_catalog_managed_test().await;
            let actions = vec![TestAction::Add("file_1.parquet".to_string())];
            add_commit(&table_root, store.as_ref(), 1, actions_to_string(actions)).await?;
            let actions = vec![TestAction::Add("file_2.parquet".to_string())];
            add_commit(&table_root, store.as_ref(), 2, actions_to_string(actions)).await?;

            let log_tail = vec![
                create_log_path(&table_root, test_utils::delta_path_for_version(1, "json")),
                create_log_path(&table_root, test_utils::delta_path_for_version(2, "json")),
            ];

            // log_tail ends at v2, max_catalog_version=3, no time-travel -> error
            let result = SnapshotBuilder::new_for(table_root)
                .with_log_tail(log_tail)
                .with_max_catalog_version(3)
                .build(engine.as_ref());

            assert!(matches!(result, Err(KernelError::MaxCatalogVersion(_))));

            Ok(())
        }

        #[test_log::test(tokio::test)]
        async fn test_catalog_managed_table_without_max_catalog_version_errors() -> KernelResult<()>
        {
            let (engine, _store, table_root) = setup_catalog_managed_test().await;

            let result = SnapshotBuilder::new_for(table_root).build(engine.as_ref());

            assert!(matches!(result, Err(KernelError::MaxCatalogVersion(_))));

            Ok(())
        }

        #[test_log::test(tokio::test)]
        async fn test_non_catalog_managed_table_with_max_catalog_version_errors(
        ) -> KernelResult<(), Box<dyn std::error::Error>> {
            let (engine, store, table_root) = setup_test();

            let actions = vec![TestAction::Metadata];
            add_commit(&table_root, store.as_ref(), 0, actions_to_string(actions)).await?;

            let result = SnapshotBuilder::new_for(table_root)
                .with_max_catalog_version(0)
                .build(engine.as_ref());

            assert!(matches!(result, Err(KernelError::MaxCatalogVersion(_))));

            Ok(())
        }

        #[test_log::test(tokio::test)]
        async fn test_log_tail_last_version_less_than_time_travel_version_errors(
        ) -> KernelResult<(), Box<dyn std::error::Error>> {
            let (engine, store, table_root) = setup_catalog_managed_test().await;
            let actions = vec![TestAction::Add("file_1.parquet".to_string())];
            add_commit(&table_root, store.as_ref(), 1, actions_to_string(actions)).await?;

            let log_tail = vec![create_log_path(
                &table_root,
                test_utils::delta_path_for_version(1, "json"),
            )];

            // Time travel to v2, but log tail only goes up to v1
            let result = SnapshotBuilder::new_for(table_root)
                .at_version(2)
                .with_log_tail(log_tail)
                .with_max_catalog_version(3)
                .build(engine.as_ref());

            assert!(matches!(result, Err(KernelError::MaxCatalogVersion(_))));

            Ok(())
        }

        #[test_log::test(tokio::test)]
        async fn test_max_catalog_version_as_effective_version(
        ) -> KernelResult<(), Box<dyn std::error::Error>> {
            let (engine, store, table_root) = setup_catalog_managed_test().await;
            let actions = vec![TestAction::Add("file_1.parquet".to_string())];
            add_commit(&table_root, store.as_ref(), 1, actions_to_string(actions)).await?;
            let actions = vec![TestAction::Add("file_2.parquet".to_string())];
            add_commit(&table_root, store.as_ref(), 2, actions_to_string(actions)).await?;

            // max_catalog_version=1, no time-travel -> snapshot at v1
            let snapshot = SnapshotBuilder::new_for(table_root)
                .with_max_catalog_version(1)
                .build(engine.as_ref())?;
            assert_eq!(snapshot.version(), 1);

            Ok(())
        }

        #[test_log::test(tokio::test)]
        async fn test_time_travel_with_max_catalog_version(
        ) -> KernelResult<(), Box<dyn std::error::Error>> {
            let (engine, store, table_root) = setup_catalog_managed_test().await;
            let actions = vec![TestAction::Add("file_1.parquet".to_string())];
            add_commit(&table_root, store.as_ref(), 1, actions_to_string(actions)).await?;

            // at_version(0) + max_catalog_version=1 -> snapshot at v0
            let snapshot = SnapshotBuilder::new_for(table_root)
                .at_version(0)
                .with_max_catalog_version(1)
                .build(engine.as_ref())?;
            assert_eq!(snapshot.version(), 0);

            Ok(())
        }

        #[test_log::test(tokio::test)]
        async fn test_builder_from_catalog_managed_without_mcv_errors(
        ) -> KernelResult<(), Box<dyn std::error::Error>> {
            let (engine, store, table_root) = setup_catalog_managed_test().await;
            let actions = vec![TestAction::Add("file_1.parquet".to_string())];
            add_commit(&table_root, store.as_ref(), 1, actions_to_string(actions)).await?;

            let initial = SnapshotBuilder::new_for(table_root)
                .with_max_catalog_version(1)
                .build(engine.as_ref())?;

            // Incremental update without mcv should fail
            let result = SnapshotBuilder::new_from(initial).build(engine.as_ref());

            assert!(matches!(result, Err(KernelError::MaxCatalogVersion(_))));

            Ok(())
        }

        #[rstest::rstest]
        #[case::gap(vec![1, 3], vec![1, 3], 3)]
        #[case::duplicates(vec![1], vec![1, 1], 1)]
        #[case::unsorted(vec![1, 2], vec![2, 1], 2)]
        #[case::overflow(vec![], vec![Version::MAX, 0], 0)]
        #[test_log::test(tokio::test)]
        async fn test_non_contiguous_log_tail_errors(
            #[case] commit_versions: Vec<u64>,
            #[case] log_tail_versions: Vec<u64>,
            #[case] mcv: u64,
        ) -> KernelResult<(), Box<dyn std::error::Error>> {
            let (engine, store, table_root) = setup_catalog_managed_test().await;
            for v in &commit_versions {
                let actions = vec![TestAction::Add(format!("file_{v}.parquet"))];
                add_commit(&table_root, store.as_ref(), *v, actions_to_string(actions)).await?;
            }

            let log_tail: Vec<_> = log_tail_versions
                .iter()
                .map(|v| {
                    create_log_path(&table_root, test_utils::delta_path_for_version(*v, "json"))
                })
                .collect();

            let result = SnapshotBuilder::new_for(table_root)
                .with_log_tail(log_tail)
                .with_max_catalog_version(mcv)
                .build(engine.as_ref());

            assert!(matches!(
                result,
                Err(KernelError::LogTailVersionsNotContiguous { .. })
            ));

            Ok(())
        }
    }
}