delta_kernel 0.30.0

//! Utilities for reading the `_last_checkpoint` file. Maybe this file should instead go under
//! log_segment module since it should only really be used there? as hint for listing?

use std::collections::HashMap;
use std::sync::Arc;

use delta_kernel_derive::internal_api;
use serde::{Deserialize, Serialize};
use tracing::{debug, info, instrument, warn};
use url::Url;

#[cfg(feature = "adaptive-metadata-in-dev")]
use crate::actions::CheckpointAction;
use crate::actions::{
    CheckpointMetadata, DomainMetadata, Metadata, Protocol, SetTransaction, Sidecar,
};
use crate::cancellation::CancellationTokenRef;
use crate::path::{CheckpointInstance, ParsedLogPath};
use crate::schema::SchemaRef;
use crate::{FileMeta, KernelError, KernelResult, StorageHandler, Version};

/// Name of the _last_checkpoint file that provides metadata about the last checkpoint
/// created for the table. This file is used as a hint for the engine to quickly locate
/// the latest checkpoint without a full directory listing.
const LAST_CHECKPOINT_FILE_NAME: &str = "_last_checkpoint";

/// Per-field count cap on a retained hint's `sidecarFiles` / `nonFileActions`. Matches the
/// Delta-Spark defaults for `lastCheckpoint.{sidecars,nonFileActions}.threshold` (both 30), which
/// drop the whole field by count when it exceeds the cap:
/// <https://github.com/delta-io/delta/blob/83002ef0bfdae90914edbcb0cae23dae5a9b9af5/spark/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSQLConf.scala#L1431-L1461>
const LAST_CHECKPOINT_SIDECARS_THRESHOLD: usize = 30;
const LAST_CHECKPOINT_NON_FILE_ACTIONS_THRESHOLD: usize = 30;

// Note: Schema can not be derived because the checkpoint schema is only known at runtime.
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
#[cfg_attr(test, derive(Default))]
#[serde(rename_all = "camelCase")]
#[internal_api]
pub(crate) struct LastCheckpointHint {
    /// The version of the table when the last checkpoint was made.
    #[allow(unreachable_pub)] // used by acceptance tests (TODO make an fn accessor?)
    pub version: Version,
    /// The number of actions that are stored in the checkpoint.
    pub(crate) size: i64,
    /// The number of fragments if the last checkpoint was written in multiple parts. `None` means
    /// a single-part or classic checkpoint (i.e. `numParts == 1`).
    pub(crate) parts: Option<usize>,
    /// The number of bytes of the checkpoint.
    pub(crate) size_in_bytes: Option<i64>,
    /// The number of AddFile actions in the checkpoint.
    pub(crate) num_of_add_files: Option<i64>,
    /// The schema of the checkpoint file.
    pub(crate) checkpoint_schema: Option<SchemaRef>,
    /// The checksum of the last checkpoint JSON.
    pub(crate) checksum: Option<String>,
    /// Additional metadata about the last checkpoint.
    pub(crate) tags: Option<HashMap<String, String>>,
    /// For a V2 checkpoint, the embedded V2 checkpoint info. Identifies the specific checkpoint
    /// file the hint describes. Absent for V1 / classic checkpoints.
    pub(crate) v2_checkpoint: Option<LastCheckpointV2>,

    /// The checkpoint format the writer tagged this hint with, which
    /// determines how the hint is consumed:
    ///
    /// - **absent** (`None`): a classic / multi-part / V2 checkpoint. Kernel knows this format, so
    ///   the hint is valid and used as-is (see [`Self::applies_to`]).
    /// - **`AdaptiveMetadataTree`**: an AMT checkpoint; the embedded
    ///   [`amt_checkpoint`](Self::amt_checkpoint) carries the prefetched checkpoint state.
    /// - **`Unknown`**: a `checkpointType` value kernel does not recognize (e.g. from a newer
    ///   writer). The whole hint is dropped at read time (see [`Self::try_read`]) and the reader
    ///   falls back to log replay.
    ///
    /// Absent and `Unknown` are thus distinct: absence is a known (legacy) checkpoint, whereas an
    /// unrecognized value invalidates the hint.
    ///
    /// Skipped on serialize when `None` so a classic hint's wire form is identical whether or not
    /// the `adaptive-metadata-in-dev` feature is compiled in.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) checkpoint_type: Option<CheckpointType>,

    /// For an adaptive-metadata (AMT) checkpoint, the embedded AMT checkpoint info: the manifest
    /// commit version plus optional prefetched checkpoint/leaves.
    ///
    /// A writer sets this exactly when `checkpoint_type == AdaptiveMetadataTree`, and never
    /// alongside `v2_checkpoint`. [`Self::try_read`] drops a hint that violates either rule.
    ///
    /// Skipped on serialize when `None` so a classic hint's wire form is identical whether or not
    /// the `adaptive-metadata-in-dev` feature is compiled in.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[serde(skip_serializing_if = "Option::is_none")]
    pub(crate) amt_checkpoint: Option<AmtCheckpoint>,
}

/// The checkpoint format recorded in a `_last_checkpoint` hint's `checkpointType` field.
/// An unrecognized wire value deserializes to [`CheckpointType::Unknown`]
/// rather than failing the parse, signaling `LastCheckpointHint::try_read` to drop the hint so
/// the reader falls back to log replay.
#[cfg(feature = "adaptive-metadata-in-dev")]
#[derive(Debug, Clone, Copy, Deserialize, PartialEq, Eq)]
#[internal_api]
pub(crate) enum CheckpointType {
    /// An adaptive-metadata (Iceberg V4) embedded-tree checkpoint.
    AdaptiveMetadataTree,
    /// Any value kernel does not recognize (e.g. from a newer writer). A hint carrying it is
    /// dropped entirely (the reader falls back to log replay), unlike an absent `checkpointType`,
    /// which is a usable legacy checkpoint. Read-only: it is produced only by deserializing an
    /// unrecognized wire value, never written by kernel (see the hand-written [`Serialize`], which
    /// refuses it).
    #[serde(other)]
    Unknown,
}

// `Serialize` is hand-written rather than derived so it can refuse `Unknown`: that variant is a
// read-only fallback sentinel with no wire representation a writer should produce, and the derived
// impl would emit the bogus string `"Unknown"`. Kernel never writes `Unknown` today, so this only
// fires if a read-in hint is ever serialized back -- in which case failing closed is correct.
#[cfg(feature = "adaptive-metadata-in-dev")]
impl Serialize for CheckpointType {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> KernelResult<S::Ok, S::Error> {
        match self {
            CheckpointType::AdaptiveMetadataTree => {
                serializer.serialize_str("AdaptiveMetadataTree")
            }
            CheckpointType::Unknown => Err(serde::ser::Error::custom(
                "refusing to serialize CheckpointType::Unknown, a read-only fallback sentinel",
            )),
        }
    }
}

/// The `amtCheckpoint` object embedded in a `_last_checkpoint` hint for an adaptive-metadata
/// checkpoint. `manifest_commit_version` lets a reader locate the checkpoint
/// action without full log replay; `checkpoint` and `leaves` are optional prefetch that writers may
/// omit. Absent for classic / V2 checkpoints.
#[cfg(feature = "adaptive-metadata-in-dev")]
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
#[cfg_attr(test, derive(Default))]
#[serde(rename_all = "camelCase")]
#[internal_api]
pub(crate) struct AmtCheckpoint {
    /// Version of the commit that emitted the latest checkpoint action. Distinct from the
    /// checkpoint's own `contentRoot.version`; lets a reader detect a stale hint and find the
    /// checkpoint action even when `checkpoint`/`leaves` are omitted.
    pub(crate) manifest_commit_version: Version,

    /// The embedded `checkpoint` action, prefetched alongside the hint. Serialized as an
    /// array of tagged action entries and folded into a typed [`CheckpointAction`] on parse (see
    /// its hand-written serde). `None` when the writer omitted it (e.g. to bound write
    /// latency); the reader then reads it from the manifest commit. A malformed array fails
    /// the whole-hint parse, which `try_read` drops so the reader falls back to log replay.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) checkpoint: Option<CheckpointAction>,

    /// The checkpoint's embedded content entries, prefetched alongside the hint. Retained as
    /// untyped [`serde_json::Value`] because a content entry's schema depends on the table's
    /// partition spec / schema, which is not known at hint-parse time; typed materialization is
    /// deferred to the read path. `None` when the writer omitted them.
    // TODO(#3438): parse into a typed content-entry struct mirroring `ContentTreeNodeEntry`
    // (keeping `partition`/`content_stats` raw until the read path has the table schema) in a
    // follow-up PR.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) leaves: Option<Vec<serde_json::Value>>,
}

/// The `v2Checkpoint` object embedded in a `_last_checkpoint` hint for a V2 checkpoint.
///
/// Carries the V2 checkpoint file's identity and metadata plus the actions a reader would otherwise
/// read from the checkpoint itself -- its sidecar references and its non-file actions. Absent for
/// V1 / classic checkpoints.
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
#[cfg_attr(test, derive(Default))]
#[serde(rename_all = "camelCase")]
#[internal_api]
pub(crate) struct LastCheckpointV2 {
    /// Bare file name of the V2 checkpoint this hint describes, matched against the selected
    /// checkpoint part's file name. Several V2 checkpoints can share a version, so this identifies
    /// which one the hint's fields describe.
    pub(crate) path: String,

    /// Size in bytes of the V2 checkpoint file named by `path`.
    pub(crate) size_in_bytes: Option<i64>,

    /// Modification time of the V2 checkpoint file named by `path`, in milliseconds since the Unix
    /// epoch.
    pub(crate) modification_time: Option<i64>,

    /// The sidecar files this checkpoint references, for a manifest (non-leaf) V2 checkpoint.
    /// Empty/absent for a leaf checkpoint that inlines its file actions. Also dropped to `None` by
    /// [`LastCheckpointHint::drop_oversized_fields`] when the count exceeds the threshold, so
    /// absence is a missing optimization, never a signal that the checkpoint is a leaf.
    pub(crate) sidecar_files: Option<Vec<Sidecar>>,

    /// The checkpoint's non-file actions (see [`HintAction`]), letting a reader obtain them
    /// without reading the checkpoint file. Dropped to `None` by
    /// [`LastCheckpointHint::drop_oversized_fields`] when the count exceeds the threshold.
    pub(crate) non_file_actions: Option<Vec<HintAction>>,
}

/// One element of [`LastCheckpointV2`]'s `non_file_actions`. A log action is exactly one action
/// type, so this is an externally-tagged enum keyed by the action name, reusing kernel's action
/// structs to yield the same types as log replay. An unrecognized action key fails the whole-hint
/// parse; `try_read` swallows that, so the reader falls back to reading the checkpoint.
#[derive(Debug, Clone, Deserialize, Serialize, PartialEq, Eq)]
#[serde(rename_all = "camelCase")]
#[internal_api]
pub(crate) enum HintAction {
    #[serde(rename = "metaData")]
    Metadata(Metadata),
    Protocol(Protocol),
    Txn(SetTransaction),
    DomainMetadata(DomainMetadata),
    CheckpointMetadata(CheckpointMetadata),
}

impl LastCheckpointHint {
    /// Returns the checkpoint version.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn version(&self) -> Version {
        self.version
    }

    /// Returns the checkpoint action count.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn size(&self) -> i64 {
        self.size
    }

    /// Returns the number of checkpoint parts, if captured.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn parts(&self) -> Option<usize> {
        self.parts
    }

    /// Returns the checkpoint byte size, if captured.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn size_in_bytes(&self) -> Option<i64> {
        self.size_in_bytes
    }

    /// Returns the Add action count, if captured.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn num_of_add_files(&self) -> Option<i64> {
        self.num_of_add_files
    }

    /// Returns the checkpoint schema, if captured.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn checkpoint_schema(&self) -> Option<&SchemaRef> {
        self.checkpoint_schema.as_ref()
    }

    /// Returns the checkpoint checksum, if captured.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn checksum(&self) -> Option<&str> {
        self.checksum.as_deref()
    }

    /// Returns checkpoint tags, preserving absent versus empty maps.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn tags(&self) -> Option<&HashMap<String, String>> {
        self.tags.as_ref()
    }

    /// Returns the embedded V2 checkpoint state, if captured.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn v2_checkpoint(&self) -> Option<&LastCheckpointV2> {
        self.v2_checkpoint.as_ref()
    }

    /// Reconstructs a checkpoint hint from its serialized fields, dropping oversized sidecar and
    /// non-file-action arrays so the retained hint is always bounded.
    ///
    /// # Errors
    ///
    /// Returns an error when the optional checkpoint schema string is not a valid Delta schema.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn from_parts(
        version: Version,
        size: i64,
        parts: Option<usize>,
        size_in_bytes: Option<i64>,
        num_of_add_files: Option<i64>,
        checkpoint_schema: Option<String>,
        checksum: Option<String>,
        tags: Option<HashMap<String, String>>,
        v2_checkpoint: Option<LastCheckpointV2>,
    ) -> KernelResult<Self> {
        let checkpoint_schema = checkpoint_schema
            .map(|schema| serde_json::from_str::<crate::schema::StructType>(&schema).map(Arc::new))
            .transpose()?;
        Ok(Self {
            version,
            size,
            parts,
            size_in_bytes,
            num_of_add_files,
            checkpoint_schema,
            checksum,
            tags,
            v2_checkpoint,
            #[cfg(feature = "adaptive-metadata-in-dev")]
            checkpoint_type: None,
            #[cfg(feature = "adaptive-metadata-in-dev")]
            amt_checkpoint: None,
        }
        .drop_oversized_fields())
    }

    /// The version of the manifest commit carrying an AMT hint's checkpoint action, or `None` if
    /// this is not an AMT hint with `amtCheckpoint`.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    pub(crate) fn amt_manifest_commit_version(&self) -> Option<Version> {
        if self.checkpoint_type != Some(CheckpointType::AdaptiveMetadataTree) {
            return None;
        }
        self.amt_checkpoint
            .as_ref()
            .map(|amt| amt.manifest_commit_version)
    }

    /// Whether this hint describes the checkpoint a log segment selected, given that segment's
    /// `checkpoint_parts`. Multiple checkpoints can share a version, so a matching version alone is
    /// not enough: the hint's own identity must equal the selected checkpoint's.
    ///
    /// On a mismatch, callers read the checkpoint file itself instead of trusting the hint's
    /// fields.
    pub(crate) fn applies_to(&self, checkpoint_parts: &[ParsedLogPath<FileMeta>]) -> bool {
        let Some(selected) = checkpoint_parts.first() else {
            return false;
        };
        self.version == selected.version
            && CheckpointInstance::of(selected) == self.implied_instance()
    }

    /// The checkpoint identity this hint's fields describe, mirroring Delta-Spark's
    /// `getFormatEnum`: an `amtCheckpoint` object means an AMT checkpoint, else a `v2Checkpoint`
    /// object means uuid-named, else `parts` means multi-part, else classic-named. `None` for an
    /// AMT checkpoint (it has no checkpoint file) or if `parts` overflows `u32`, so no checkpoint
    /// can match it.
    fn implied_instance(&self) -> Option<CheckpointInstance> {
        #[cfg(feature = "adaptive-metadata-in-dev")]
        if self.amt_checkpoint.is_some() {
            return None;
        }
        Some(match (&self.v2_checkpoint, self.parts) {
            (Some(v2), _) => CheckpointInstance::Uuid {
                filename: v2.path.clone(),
            },
            (None, Some(parts)) => CheckpointInstance::MultiPart {
                num_parts: parts.try_into().ok()?,
            },
            (None, None) => CheckpointInstance::Classic,
        })
    }

    /// Describes why this hint's `checkpointType`, `amtCheckpoint`, and `v2Checkpoint` fields
    /// cannot describe a single checkpoint, or `None` if they are consistent.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    fn amt_fields_problem(&self) -> Option<&'static str> {
        match (
            self.checkpoint_type,
            &self.amt_checkpoint,
            &self.v2_checkpoint,
        ) {
            (Some(CheckpointType::Unknown), _, _) => Some("has an unrecognized checkpointType"),
            (Some(CheckpointType::AdaptiveMetadataTree), None, _) => {
                Some("AMT hint has no amtCheckpoint")
            }
            (None, Some(_), _) => Some("has amtCheckpoint without an AdaptiveMetadataTree type"),
            (_, Some(_), Some(_)) => Some("has both amtCheckpoint and v2Checkpoint"),
            _ => None,
        }
    }

    /// Parses a hint from raw `_last_checkpoint` bytes, dropping oversized fields so the retained
    /// hint is always bounded. This is the only way to construct a hint from disk, so callers can
    /// never hold an untrimmed one.
    fn from_bytes_with_oversized_fields_dropped(bytes: &[u8]) -> serde_json::Result<Self> {
        let hint: Self = serde_json::from_slice(bytes)?;
        Ok(hint.drop_oversized_fields())
    }

    /// Drops `sidecarFiles` / `nonFileActions` over the threshold. Drops the whole field, never
    /// truncates. Absent means info missing, so this only loses an optimization.
    fn drop_oversized_fields(mut self) -> Self {
        if let Some(v2) = &mut self.v2_checkpoint {
            let version = self.version;
            if let Some(count) = v2
                .sidecar_files
                .as_ref()
                .map(Vec::len)
                .filter(|&n| n > LAST_CHECKPOINT_SIDECARS_THRESHOLD)
            {
                debug!(
                    version,
                    count, "dropping _last_checkpoint sidecarFiles above threshold"
                );
                v2.sidecar_files = None;
            }
            if let Some(count) = v2
                .non_file_actions
                .as_ref()
                .map(Vec::len)
                .filter(|&n| n > LAST_CHECKPOINT_NON_FILE_ACTIONS_THRESHOLD)
            {
                debug!(
                    version,
                    count, "dropping _last_checkpoint nonFileActions above threshold"
                );
                v2.non_file_actions = None;
            }
        }
        self
    }

    /// Returns the path of the `_last_checkpoint` file given the log root of a table.
    #[internal_api]
    pub(crate) fn path(log_root: &Url) -> KernelResult<Url> {
        Ok(log_root.join(LAST_CHECKPOINT_FILE_NAME)?)
    }

    /// Try reading the `_last_checkpoint` file.
    ///
    /// Note that we typically want to ignore a missing/invalid `_last_checkpoint` file without
    /// failing the read. Thus, the semantics of this function are to return `None` if the file is
    /// not found or is invalid JSON. Unexpected/unrecoverable errors are returned as `Err` case and
    /// are assumed to cause failure.
    // TODO(#1047): weird that we propagate FileNotFound as part of the iterator instead of top-
    // level result coming from storage.read_files
    #[instrument(
        name = "last_checkpoint.read",
        skip_all,
        fields(enable_call_frame),
        err
    )]
    pub(crate) fn try_read(
        storage: &dyn StorageHandler,
        log_root: &Url,
        cancellation_token: Option<&CancellationTokenRef>,
    ) -> KernelResult<Option<LastCheckpointHint>> {
        let file_path = Self::path(log_root)?;
        match storage
            .read_files_with_cancellation(vec![(file_path, None)], cancellation_token.cloned())?
            .next()
        {
            Some(Ok(data)) => {
                let result = Self::from_bytes_with_oversized_fields_dropped(&data)
                    .inspect_err(|e| warn!("invalid _last_checkpoint JSON: {e}"))
                    .ok()
                    // Kernel cannot tell which checkpoint a hint describes if its `checkpointType`
                    // is unrecognized or its AMT fields are inconsistent. Such a hint is dropped so
                    // the reader falls back to log replay.
                    .filter(|_hint| {
                        #[cfg(feature = "adaptive-metadata-in-dev")]
                        if let Some(problem) = _hint.amt_fields_problem() {
                            warn!(
                                version = _hint.version,
                                "_last_checkpoint {problem}; dropping"
                            );
                            return false;
                        }
                        true
                    });
                info!(hint = result.as_ref().map(|h| h.summary()));
                Ok(result)
            }
            Some(Err(KernelError::FileNotFound(_))) => {
                info!("_last_checkpoint file not found");
                Ok(None)
            }
            Some(Err(err)) => Err(err),
            None => {
                warn!("empty _last_checkpoint file");
                Ok(None)
            }
        }
    }

    /// Succinct summary string for logging purposes.
    fn summary(&self) -> String {
        format!(
            "{{v={}, size={}, parts={:?}}}",
            self.version, self.size, self.parts
        )
    }

    /// Convert the LastCheckpointHint to JSON bytes
    #[cfg(test)]
    pub(crate) fn to_json_bytes(&self) -> Vec<u8> {
        serde_json::to_vec(self).expect("Failed to convert LastCheckpointHint to JSON bytes")
    }
}

impl LastCheckpointV2 {
    /// Returns the checkpoint file name.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn path(&self) -> &str {
        &self.path
    }

    /// Returns the checkpoint byte size, if captured.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn size_in_bytes(&self) -> Option<i64> {
        self.size_in_bytes
    }

    /// Returns the checkpoint modification time, if captured.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn modification_time(&self) -> Option<i64> {
        self.modification_time
    }

    /// Returns sidecar actions, preserving absent versus empty arrays.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn sidecar_files(&self) -> Option<&[Sidecar]> {
        self.sidecar_files.as_deref()
    }

    /// Returns non-file actions, preserving absent versus empty arrays.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn non_file_actions(&self) -> Option<&[HintAction]> {
        self.non_file_actions.as_deref()
    }
}

impl LastCheckpointV2 {
    /// Reconstructs V2 checkpoint state from its serialized fields.
    #[internal_api]
    #[cfg_attr(not(feature = "internal-api"), allow(dead_code))]
    pub(crate) fn from_parts(
        path: String,
        size_in_bytes: Option<i64>,
        modification_time: Option<i64>,
        sidecar_files: Option<Vec<Sidecar>>,
        non_file_actions: Option<Vec<HintAction>>,
    ) -> Self {
        Self {
            path,
            size_in_bytes,
            modification_time,
            sidecar_files,
            non_file_actions,
        }
    }
}

#[cfg(test)]
mod tests {
    #[cfg(feature = "adaptive-metadata-in-dev")]
    use bytes::Bytes;
    use rstest::rstest;

    use super::*;
    use crate::schema::schema;
    use crate::table_features::TableFeature;
    use crate::unit_test_utils::create_log_path;
    use crate::KernelResultIteratorStatic;
    #[cfg(feature = "adaptive-metadata-in-dev")]
    use crate::{engine::sync::SyncEngine, object_store::memory::InMemory, Engine};

    /// A real `_last_checkpoint` for a V2 checkpoint carries a `v2Checkpoint` object; we parse its
    /// `path` and file metadata. An empty `sidecarFiles` (a leaf checkpoint) parses to `Some([])`,
    /// distinct from an absent field. Guards the `camelCase` wire keys -- a rename would otherwise
    /// silently parse to `None` (errors are swallowed in `try_read`) and disable the identity
    /// filter.
    #[test]
    fn parses_v2_checkpoint_path_from_wire_json() {
        let json = br#"{
            "version": 5,
            "size": 10,
            "v2Checkpoint": {
                "path": "00000000000000000005.checkpoint.0190e8f5-uuid.parquet",
                "sizeInBytes": 1234,
                "modificationTime": 1700000000000,
                "sidecarFiles": []
            }
        }"#;
        let hint: LastCheckpointHint = serde_json::from_slice(json).unwrap();
        let v2 = hint.v2_checkpoint.expect("v2Checkpoint present");
        assert_eq!(
            v2.path,
            "00000000000000000005.checkpoint.0190e8f5-uuid.parquet"
        );
        assert_eq!(v2.size_in_bytes, Some(1234));
        assert_eq!(v2.modification_time, Some(1700000000000));
        assert_eq!(v2.sidecar_files, Some(vec![]));
        assert_eq!(v2.non_file_actions, None);
    }

    /// A manifest V2 checkpoint hint carries its sidecar references and non-file actions. Each
    /// non-file action decodes to exactly one [`HintAction`] variant via its action key.
    #[test]
    fn parses_v2_checkpoint_sidecars_and_non_file_actions() {
        let json = br#"{
            "version": 5,
            "size": 10,
            "v2Checkpoint": {
                "path": "00000000000000000005.checkpoint.0190e8f5-uuid.parquet",
                "sidecarFiles": [
                    {"path": "sidecar-1.parquet", "sizeInBytes": 42, "modificationTime": 1700000000000}
                ],
                "nonFileActions": [
                    {"protocol": {"minReaderVersion": 3, "minWriterVersion": 7,
                        "readerFeatures": [], "writerFeatures": []}},
                    {"metaData": {"id": "table-id", "format": {"provider": "parquet", "options": {}},
                        "schemaString": "{\"type\":\"struct\",\"fields\":[]}",
                        "partitionColumns": [], "configuration": {}}},
                    {"txn": {"appId": "app", "version": 1}},
                    {"domainMetadata": {"domain": "d", "configuration": "c", "removed": false}},
                    {"checkpointMetadata": {"version": 5}}
                ]
            }
        }"#;
        let hint: LastCheckpointHint = serde_json::from_slice(json).unwrap();
        let v2 = hint.v2_checkpoint.expect("v2Checkpoint present");

        let sidecars = v2.sidecar_files.expect("sidecarFiles present");
        assert_eq!(sidecars.len(), 1);
        assert_eq!(sidecars[0].path, "sidecar-1.parquet");
        assert_eq!(sidecars[0].size_in_bytes, 42);

        let actions = v2.non_file_actions.expect("nonFileActions present");
        assert_eq!(actions.len(), 5);
        assert!(matches!(&actions[0], HintAction::Protocol(p) if p.min_reader_version() == 3));
        assert!(matches!(&actions[1], HintAction::Metadata(m) if m.id() == "table-id"));
        assert!(matches!(&actions[2], HintAction::Txn(t) if t.app_id == "app"));
        assert!(matches!(&actions[3], HintAction::DomainMetadata(_)));
        assert!(matches!(&actions[4], HintAction::CheckpointMetadata(c) if c.version == 5));
    }

    /// A `_last_checkpoint` without a `v2Checkpoint` object (V1 / classic) parses to `None`.
    #[test]
    fn v2_checkpoint_absent_parses_to_none() {
        let json = br#"{"version": 5, "size": 10}"#;
        let hint: LastCheckpointHint = serde_json::from_slice(json).unwrap();
        assert!(hint.v2_checkpoint.is_none());
    }

    /// A malformed `v2Checkpoint` -- missing the required `path`, or a type-mismatched field --
    /// fails the whole-hint parse. `try_read` swallows that to `None`, so the reader falls back to
    /// a footer read rather than trusting a partially-parsed hint.
    #[test]
    fn malformed_v2_checkpoint_fails_whole_hint_parse() {
        let missing_path = br#"{"version": 5, "size": 10, "v2Checkpoint": {"sizeInBytes": 1234}}"#;
        assert!(serde_json::from_slice::<LastCheckpointHint>(missing_path).is_err());

        let bad_type = br#"{"version": 5, "size": 10,
            "v2Checkpoint": {"path": "c.parquet", "sizeInBytes": "not-a-number"}}"#;
        assert!(serde_json::from_slice::<LastCheckpointHint>(bad_type).is_err());
    }

    /// The full JSON form of an AMT (`AdaptiveMetadataTree`) `_last_checkpoint` hint parses to its
    /// typed fields: `checkpointType`, the required `manifestCommitVersion`, the embedded
    /// `checkpoint` action (array of tagged entries), and the prefetched `leaves` (retained
    /// raw). Guards the `camelCase`/`PascalCase` wire keys -- a rename would silently parse to
    /// `None`/`Unknown` (errors are swallowed in `try_read`) and disable the AMT fast path.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn parses_amt_checkpoint_from_wire_json() {
        let json = br#"{
            "version": 7,
            "size": -1,
            "checkpointType": "AdaptiveMetadataTree",
            "amtCheckpoint": {
                "manifestCommitVersion": 6,
                "checkpoint": [
                    {"checkpointMetadata": {"version": 7}},
                    {"contentRoot": {"path": "metadata/root-v7.parquet", "sizeInBytes": 2048, "version": 7}},
                    {"protocol": {"minReaderVersion": 3, "minWriterVersion": 7,
                        "readerFeatures": ["adaptiveMetadata-preview"], "writerFeatures": ["adaptiveMetadata-preview"]}},
                    {"metaData": {"id": "tid", "format": {"provider": "parquet", "options": {}},
                        "schemaString": "{\"type\":\"struct\",\"fields\":[]}", "partitionColumns": [], "configuration": {}}}
                ],
                "leaves": [
                    {"contentType": 0, "location": "data/part-0.parquet", "recordCount": 3}
                ]
            }
        }"#;
        let hint: LastCheckpointHint = serde_json::from_slice(json).unwrap();
        assert_eq!(
            hint.checkpoint_type,
            Some(CheckpointType::AdaptiveMetadataTree)
        );
        let amt = hint.amt_checkpoint.expect("amtCheckpoint present");
        assert_eq!(amt.manifest_commit_version, 6);
        let checkpoint = amt.checkpoint.expect("checkpoint present");
        assert_eq!(checkpoint.version(), 7);
        assert_eq!(checkpoint.path(), "metadata/root-v7.parquet");
        assert_eq!(checkpoint.metadata().id(), "tid");
        assert_eq!(amt.leaves.expect("leaves present").len(), 1);
    }

    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[rstest]
    #[case::amt_hint(Some(CheckpointType::AdaptiveMetadataTree), Some(7), Some(7))]
    #[case::amt_type_without_amt_checkpoint(Some(CheckpointType::AdaptiveMetadataTree), None, None)]
    #[case::no_checkpoint_type(None, Some(7), None)]
    #[case::unknown_checkpoint_type(Some(CheckpointType::Unknown), Some(7), None)]
    fn amt_manifest_commit_version_requires_amt_type_and_amt_checkpoint(
        #[case] checkpoint_type: Option<CheckpointType>,
        #[case] manifest_commit_version: Option<Version>,
        #[case] expected: Option<Version>,
    ) {
        let hint = LastCheckpointHint {
            version: 5,
            checkpoint_type,
            amt_checkpoint: manifest_commit_version.map(|manifest_commit_version| AmtCheckpoint {
                manifest_commit_version,
                ..Default::default()
            }),
            ..Default::default()
        };
        assert_eq!(hint.amt_manifest_commit_version(), expected);
    }

    /// A `_last_checkpoint` without `checkpointType`/`amtCheckpoint` (classic / V2) leaves both
    /// `None`.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn amt_checkpoint_absent_parses_to_none() {
        let json = br#"{"version": 5, "size": 10}"#;
        let hint: LastCheckpointHint = serde_json::from_slice(json).unwrap();
        assert!(hint.checkpoint_type.is_none());
        assert!(hint.amt_checkpoint.is_none());
    }

    /// A classic hint (no `checkpoint_type` / `amt_checkpoint`) omits both AMT keys on serialize
    /// rather than emitting `"checkpointType": null` / `"amtCheckpoint": null`, so the wire form is
    /// identical whether or not the `adaptive-metadata-in-dev` feature is compiled in.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn classic_hint_omits_amt_keys_on_serialize() {
        let hint = LastCheckpointHint {
            version: 5,
            size: 10,
            ..Default::default()
        };
        let json = serde_json::to_value(&hint).unwrap();
        let obj = json.as_object().expect("hint serializes to an object");
        assert!(
            !obj.contains_key("checkpointType"),
            "classic hint must omit checkpointType, got: {json}"
        );
        assert!(
            !obj.contains_key("amtCheckpoint"),
            "classic hint must omit amtCheckpoint, got: {json}"
        );
    }

    /// A `checkpointType` value kernel does not recognize parses to `Unknown` rather than failing
    /// the whole-hint parse.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn unrecognized_checkpoint_type_parses_to_unknown() {
        let json = br#"{"version": 5, "size": 10, "checkpointType": "SomethingNewer"}"#;
        let hint: LastCheckpointHint = serde_json::from_slice(json).unwrap();
        assert_eq!(hint.checkpoint_type, Some(CheckpointType::Unknown));
    }

    /// `try_read` retains a hint whose `checkpointType` and AMT fields describe one checkpoint, and
    /// drops one it cannot interpret so the reader falls back to log replay.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[rstest]
    #[case::no_checkpoint_type(r#"{"version": 5, "size": 10}"#, true)]
    #[case::amt(
        r#"{"version": 6, "size": -1, "checkpointType": "AdaptiveMetadataTree",
            "amtCheckpoint": {"manifestCommitVersion": 6}}"#,
        true
    )]
    #[case::amt_type_without_amt_checkpoint(
        r#"{"version": 6, "size": -1, "checkpointType": "AdaptiveMetadataTree"}"#,
        false
    )]
    #[case::amt_checkpoint_without_amt_type(
        r#"{"version": 6, "size": -1, "amtCheckpoint": {"manifestCommitVersion": 6}}"#,
        false
    )]
    #[case::amt_checkpoint_with_v2_checkpoint(
        r#"{"version": 6, "size": -1, "checkpointType": "AdaptiveMetadataTree",
            "amtCheckpoint": {"manifestCommitVersion": 6},
            "v2Checkpoint": {"path": "00000000000000000006.checkpoint.abc.parquet"}}"#,
        false
    )]
    #[case::unknown_checkpoint_type(
        r#"{"version": 5, "size": 10, "checkpointType": "SomethingNewer"}"#,
        false
    )]
    fn try_read_drops_unrecognized_or_inconsistent_checkpoint_type(
        #[case] json: &str,
        #[case] retained: bool,
    ) {
        let log_root = Url::parse("memory:///_delta_log/").unwrap();
        let engine = SyncEngine::new_with_store(Arc::new(InMemory::new()));
        let storage = engine.storage_handler();
        storage
            .put(
                &LastCheckpointHint::path(&log_root).unwrap(),
                Bytes::copy_from_slice(json.as_bytes()),
                true,
            )
            .unwrap();
        let hint = LastCheckpointHint::try_read(storage.as_ref(), &log_root, None).unwrap();
        assert_eq!(hint.is_some(), retained, "{json}");
    }

    /// `AdaptiveMetadataTree` serializes to its wire string, but `Unknown` refuses to serialize --
    /// it is a read-only fallback sentinel with no value a writer should emit.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn checkpoint_type_unknown_is_not_serializable() {
        assert_eq!(
            serde_json::to_string(&CheckpointType::AdaptiveMetadataTree).unwrap(),
            r#""AdaptiveMetadataTree""#
        );
        assert!(serde_json::to_string(&CheckpointType::Unknown).is_err());
    }

    /// A malformed `amtCheckpoint` -- missing the required `manifestCommitVersion` -- fails the
    /// whole-hint parse. `try_read` swallows that to `None`, so the reader falls back to log replay
    /// rather than trusting a partially-parsed hint.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn malformed_amt_checkpoint_fails_whole_hint_parse() {
        let json = br#"{"version": 5, "size": 10, "checkpointType": "AdaptiveMetadataTree",
            "amtCheckpoint": {}}"#;
        assert!(serde_json::from_slice::<LastCheckpointHint>(json).is_err());
    }

    /// An AMT hint round-trips through serialization: the embedded `checkpoint` action re-emits its
    /// tagged-array form and the raw `leaves` re-emit verbatim, so the reparsed hint equals the
    /// original.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn amt_checkpoint_hint_json_round_trips() {
        let json = br#"{
            "version": 7,
            "size": -1,
            "checkpointType": "AdaptiveMetadataTree",
            "amtCheckpoint": {
                "manifestCommitVersion": 6,
                "checkpoint": [
                    {"checkpointMetadata": {"version": 7}},
                    {"contentRoot": {"path": "metadata/root-v7.parquet", "sizeInBytes": 2048, "version": 7}},
                    {"protocol": {"minReaderVersion": 3, "minWriterVersion": 7,
                        "readerFeatures": ["adaptiveMetadata-preview"], "writerFeatures": ["adaptiveMetadata-preview"]}},
                    {"metaData": {"id": "tid", "format": {"provider": "parquet", "options": {}},
                        "schemaString": "{\"type\":\"struct\",\"fields\":[]}", "partitionColumns": [], "configuration": {}}}
                ],
                "leaves": [{"contentType": 0, "location": "data/part-0.parquet", "recordCount": 3}]
            }
        }"#;
        let hint: LastCheckpointHint = serde_json::from_slice(json).unwrap();
        let reparsed: LastCheckpointHint = serde_json::from_slice(&hint.to_json_bytes()).unwrap();
        assert_eq!(hint, reparsed);
    }

    /// A checkpoint action array carrying every element kind -- including repeatable `txn` /
    /// `domainMetadata` entries and a `sidecar` with its `type` discriminator -- folds into a typed
    /// [`CheckpointAction`] with each entry in the right field, and round-trips unchanged.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn amt_checkpoint_full_action_array_parses_and_round_trips() {
        let json = br#"{
            "version": 7,
            "size": -1,
            "checkpointType": "AdaptiveMetadataTree",
            "amtCheckpoint": {
                "manifestCommitVersion": 6,
                "checkpoint": [
                    {"checkpointMetadata": {"version": 7}},
                    {"contentRoot": {"path": "metadata/root-v7.parquet", "sizeInBytes": 2048, "version": 7}},
                    {"protocol": {"minReaderVersion": 3, "minWriterVersion": 7,
                        "readerFeatures": ["adaptiveMetadata-preview"], "writerFeatures": ["adaptiveMetadata-preview"]}},
                    {"metaData": {"id": "tid", "format": {"provider": "parquet", "options": {}},
                        "schemaString": "{\"type\":\"struct\",\"fields\":[]}", "partitionColumns": [], "configuration": {}}},
                    {"txn": {"appId": "app", "version": 1}},
                    {"domainMetadata": {"domain": "d", "configuration": "c", "removed": false}},
                    {"sidecar": {"type": "txn", "path": "txn-v7.parquet", "sizeInBytes": 42, "modificationTime": 1700000000000}}
                ]
            }
        }"#;
        let hint: LastCheckpointHint = serde_json::from_slice(json).unwrap();
        let checkpoint = hint
            .amt_checkpoint
            .as_ref()
            .expect("amtCheckpoint present")
            .checkpoint
            .as_ref()
            .expect("checkpoint present");
        assert_eq!(checkpoint.version(), 7);
        assert_eq!(checkpoint.transactions.len(), 1);
        assert_eq!(checkpoint.transactions[0].app_id, "app");
        assert_eq!(checkpoint.domain_metadata.len(), 1);
        assert_eq!(checkpoint.domain_metadata[0].domain(), "d");
        assert_eq!(checkpoint.txn_sidecars.len(), 1);
        assert_eq!(checkpoint.txn_sidecars[0].path, "txn-v7.parquet");
        assert_eq!(checkpoint.txn_sidecars[0].size_in_bytes, 42);
        assert!(checkpoint.domain_metadata_sidecars.is_empty());
        let reparsed: LastCheckpointHint = serde_json::from_slice(&hint.to_json_bytes()).unwrap();
        assert_eq!(hint, reparsed);
    }

    /// Cross-check that the Delta-log `CheckpointAction` EngineData (de)serializer and its
    /// serde (used by the `_last_checkpoint` hint) agree on the checkpoint-action array wire
    /// form. The action the log path writes as JSON parses back through serde to the identical
    /// action, and the JSON serde writes parses back through the log path -- so a rename or
    /// element-shape change on either side is caught. Compared at the typed level (not raw JSON) so
    /// it is robust to the intended null-`tags` emission difference between the two writers.
    ///
    /// This exercises only the known element kinds; the two paths intentionally diverge on unknown
    /// elements (the serde path fails closed, the visitor skips them), which this test does not
    /// cover.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn checkpoint_action_cross_serializes_between_log_and_hint() -> KernelResult<()> {
        use crate::actions::ContentRoot;
        use crate::engine::to_json_bytes;
        use crate::engine_data::FilteredEngineData;
        use crate::unit_test_utils::parse_json_batch;

        // A fully-populated action: every element kind, both sidecar `type`s.
        let action = CheckpointAction {
            version: 7,
            content_root: ContentRoot::new("s3://bucket/manifest".to_string(), 512, 5),
            protocol: Protocol::new_unchecked(1, 2, None, None),
            metadata: Metadata::default(),
            transactions: vec![SetTransaction {
                app_id: "app".to_string(),
                version: 1,
                last_updated: None,
            }],
            domain_metadata: vec![DomainMetadata::new("d".to_string(), "c".to_string())],
            txn_sidecars: vec![Sidecar::new("txn.parquet".to_string(), 1, 2, None)],
            domain_metadata_sidecars: vec![Sidecar::new("dm.parquet".to_string(), 3, 4, None)],
        };

        // Log path -> JSON array -> serde: folds back to the identical typed action.
        let engine = SyncEngine::new();
        let data = action.clone().into_engine_data(&engine)?;
        let bytes = to_json_bytes(std::iter::once(Ok(
            FilteredEngineData::with_all_rows_selected(data),
        )))?;
        let commit: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
        let array = commit
            .get("checkpoint")
            .expect("checkpoint field present")
            .clone();
        let from_log: CheckpointAction = serde_json::from_value(array).unwrap();
        assert_eq!(from_log, action);

        // serde -> JSON array -> log path: reparses to the identical typed action.
        let array = serde_json::to_value(&action).unwrap();
        let commit = serde_json::json!({ "checkpoint": array }).to_string();
        let data = parse_json_batch(crate::arrow::array::StringArray::from(vec![commit]));
        let parsed = CheckpointAction::try_new_from_data(data.as_ref())?
            .expect("checkpoint action should round-trip through serde");
        assert_eq!(parsed, action);
        Ok(())
    }

    /// A writer may omit the optional prefetch: an `amtCheckpoint` carrying only the required
    /// `manifestCommitVersion` parses with `checkpoint` and `leaves` as `None` and round-trips
    /// unchanged. Guards the `Option`/`default` serde behavior of the prefetch fields -- a
    /// regression that made them required, or renamed them, would fail here rather than silently
    /// drop the prefetch on the happy path.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn amt_checkpoint_omitting_prefetch_parses_and_round_trips() {
        let json = br#"{
            "version": 6,
            "size": -1,
            "checkpointType": "AdaptiveMetadataTree",
            "amtCheckpoint": {"manifestCommitVersion": 6}
        }"#;
        let hint: LastCheckpointHint = serde_json::from_slice(json).unwrap();
        let amt = hint.amt_checkpoint.as_ref().expect("amtCheckpoint present");
        assert_eq!(amt.manifest_commit_version, 6);
        assert!(amt.checkpoint.is_none());
        assert!(amt.leaves.is_none());
        let reparsed: LastCheckpointHint = serde_json::from_slice(&hint.to_json_bytes()).unwrap();
        assert_eq!(hint, reparsed);
    }

    /// A multi-element `leaves` array parses to a `Vec` of the same length and round-trips,
    /// exercising the prefetch beyond the single-leaf happy path.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn amt_checkpoint_with_multiple_leaves_parses_and_round_trips() {
        let json = br#"{
            "version": 7,
            "size": -1,
            "checkpointType": "AdaptiveMetadataTree",
            "amtCheckpoint": {
                "manifestCommitVersion": 7,
                "leaves": [
                    {"contentType": 0, "location": "data/part-0.parquet", "recordCount": 3},
                    {"contentType": 0, "location": "data/part-1.parquet", "recordCount": 5}
                ]
            }
        }"#;
        let hint: LastCheckpointHint = serde_json::from_slice(json).unwrap();
        let amt = hint.amt_checkpoint.as_ref().expect("amtCheckpoint present");
        assert!(amt.checkpoint.is_none());
        assert_eq!(amt.leaves.as_ref().expect("leaves present").len(), 2);
        let reparsed: LastCheckpointHint = serde_json::from_slice(&hint.to_json_bytes()).unwrap();
        assert_eq!(hint, reparsed);
    }

    /// `applies_to` accepts the hint only for the checkpoint a segment actually selected: same
    /// version, and the identity the hint's fields imply must equal the selected checkpoint's.
    #[test]
    fn applies_to_matches_only_the_selected_checkpoint() {
        let root = Url::parse("memory:///_delta_log/").unwrap();
        let part = |name: &str| create_log_path(root.join(name).unwrap().as_str());

        let selected =
            "00000000000000000001.checkpoint.11111111-1111-1111-1111-111111111111.parquet";
        let other = "00000000000000000001.checkpoint.22222222-2222-2222-2222-222222222222.parquet";

        // V2 single-part: applies only when the version and the file name both match.
        let v2 = LastCheckpointHint {
            version: 1,
            v2_checkpoint: Some(LastCheckpointV2 {
                path: selected.to_string(),
                ..Default::default()
            }),
            ..Default::default()
        };
        assert!(v2.applies_to(&[part(selected)]));
        assert!(!v2.applies_to(&[part(other)]), "wrong v2 path");
        let v2_wrong_version = LastCheckpointHint {
            version: 2,
            ..v2.clone()
        };
        assert!(
            !v2_wrong_version.applies_to(&[part(selected)]),
            "wrong version"
        );

        // V1 multi-part: the part count comes from the selected part's file name.
        let mp1 = "00000000000000000001.checkpoint.0000000001.0000000002.parquet";
        let mp2 = "00000000000000000001.checkpoint.0000000002.0000000002.parquet";
        let v1_multi = LastCheckpointHint {
            version: 1,
            parts: Some(2),
            ..Default::default()
        };
        assert!(v1_multi.applies_to(&[part(mp1), part(mp2)]));
        let mp1_of_3 = "00000000000000000001.checkpoint.0000000001.0000000003.parquet";
        assert!(!v1_multi.applies_to(&[part(mp1_of_3)]), "wrong part count");

        // V1 single-part: an absent `parts` means one part.
        let v1_single = LastCheckpointHint {
            version: 1,
            ..Default::default()
        };
        let classic = "00000000000000000001.checkpoint.parquet";
        assert!(v1_single.applies_to(&[part(classic)]));

        // Naming has to match, not just part count: a 1-of-1 multi-part and a uuid-named checkpoint
        // each hold one part too.
        let one_of_one = "00000000000000000001.checkpoint.0000000001.0000000001.parquet";
        assert!(
            !v1_single.applies_to(&[part(one_of_one)]),
            "single-file hint applied to a multi-part checkpoint"
        );
        assert!(
            !v1_single.applies_to(&[part(selected)]),
            "single-file hint applied to a uuid-named checkpoint"
        );
        // Nor can a multi-part hint describe a single whole file.
        let v1_one_part = LastCheckpointHint {
            version: 1,
            parts: Some(1),
            ..Default::default()
        };
        assert!(v1_one_part.applies_to(&[part(one_of_one)]));
        assert!(
            !v1_one_part.applies_to(&[part(classic)]),
            "multi-part hint applied to a classic checkpoint"
        );

        assert!(!v1_single.applies_to(&[]), "no checkpoint selected");
    }

    /// An AMT hint's `parts` counts its leaves, not checkpoint files, so it never applies to a
    /// listed checkpoint even when the part counts coincide.
    #[cfg(feature = "adaptive-metadata-in-dev")]
    #[test]
    fn applies_to_rejects_amt_hint_for_listed_checkpoint() {
        let root = Url::parse("memory:///_delta_log/").unwrap();
        let part = |name: &str| create_log_path(root.join(name).unwrap().as_str());
        let amt = LastCheckpointHint {
            version: 1,
            parts: Some(2),
            checkpoint_type: Some(CheckpointType::AdaptiveMetadataTree),
            amt_checkpoint: Some(AmtCheckpoint {
                manifest_commit_version: 1,
                ..Default::default()
            }),
            ..Default::default()
        };
        let mp1 = "00000000000000000001.checkpoint.0000000001.0000000002.parquet";
        let mp2 = "00000000000000000001.checkpoint.0000000002.0000000002.parquet";
        assert!(!amt.applies_to(&[part(mp1), part(mp2)]));
    }

    /// `sidecarFiles` / `nonFileActions` are dropped only when their count exceeds the threshold --
    /// independently, and the whole field at once (never truncated); within-threshold fields are
    /// kept verbatim.
    #[test]
    fn drops_oversized_embedded_fields() {
        let sidecar = Sidecar {
            path: "s.parquet".to_string(),
            size_in_bytes: 1,
            modification_time: 0,
            tags: None,
        };
        let action = HintAction::Protocol(Protocol::default());
        let hint = |sidecars: usize, actions: usize| LastCheckpointHint {
            version: 1,
            v2_checkpoint: Some(LastCheckpointV2 {
                path: "cp".to_string(),
                sidecar_files: Some(vec![sidecar.clone(); sidecars]),
                non_file_actions: Some(vec![action.clone(); actions]),
                ..Default::default()
            }),
            ..Default::default()
        };

        // At the threshold: both fields kept.
        let v2 = hint(
            LAST_CHECKPOINT_SIDECARS_THRESHOLD,
            LAST_CHECKPOINT_NON_FILE_ACTIONS_THRESHOLD,
        )
        .drop_oversized_fields()
        .v2_checkpoint
        .unwrap();
        assert_eq!(
            v2.sidecar_files.unwrap().len(),
            LAST_CHECKPOINT_SIDECARS_THRESHOLD
        );
        assert_eq!(
            v2.non_file_actions.unwrap().len(),
            LAST_CHECKPOINT_NON_FILE_ACTIONS_THRESHOLD
        );

        // Over threshold: each field dropped independently.
        let v2 = hint(LAST_CHECKPOINT_SIDECARS_THRESHOLD + 1, 5)
            .drop_oversized_fields()
            .v2_checkpoint
            .unwrap();
        assert!(v2.sidecar_files.is_none(), "oversized sidecarFiles dropped");
        assert_eq!(v2.non_file_actions.unwrap().len(), 5, "nonFileActions kept");

        let v2 = hint(5, LAST_CHECKPOINT_NON_FILE_ACTIONS_THRESHOLD + 1)
            .drop_oversized_fields()
            .v2_checkpoint
            .unwrap();
        assert_eq!(v2.sidecar_files.unwrap().len(), 5, "sidecarFiles kept");
        assert!(
            v2.non_file_actions.is_none(),
            "oversized nonFileActions dropped"
        );
    }

    #[rstest]
    #[case::at_threshold(30, Some(30))]
    #[case::above_threshold(31, None)]
    fn reconstructed_hint_bounds_embedded_fields(
        #[case] count: usize,
        #[case] expected_count: Option<usize>,
    ) {
        let sidecar = Sidecar::new("s.parquet".to_string(), 1, 0, None);
        let action = HintAction::Protocol(Protocol::default());
        let hint = LastCheckpointHint::from_parts(
            1,
            1,
            None,
            None,
            None,
            None,
            None,
            None,
            Some(LastCheckpointV2::from_parts(
                "cp.parquet".to_string(),
                None,
                None,
                Some(vec![sidecar; count]),
                Some(vec![action; count]),
            )),
        )
        .unwrap();
        let v2 = hint.v2_checkpoint.unwrap();
        assert_eq!(v2.sidecar_files.as_ref().map(Vec::len), expected_count);
        assert_eq!(v2.non_file_actions.as_ref().map(Vec::len), expected_count);
    }

    #[test]
    fn reconstructed_hint_validates_checkpoint_schema() {
        let schema = r#"{"type":"struct","fields":[]}"#.to_string();
        let hint =
            LastCheckpointHint::from_parts(1, 1, None, None, None, Some(schema), None, None, None)
                .unwrap();
        assert!(hint.checkpoint_schema.is_some());

        assert!(LastCheckpointHint::from_parts(
            1,
            1,
            None,
            None,
            None,
            Some("not a schema".to_string()),
            None,
            None,
            None,
        )
        .is_err());
    }

    /// Returns the single `actions` element matching `extract`, asserting there is exactly one.
    fn one_action<'a, T: 'a>(
        actions: &'a [HintAction],
        extract: impl Fn(&'a HintAction) -> Option<&'a T>,
    ) -> &'a T {
        let mut matching = actions.iter().filter_map(extract);
        let found = matching.next().expect("expected a matching action");
        assert!(
            matching.next().is_none(),
            "expected exactly one matching action"
        );
        found
    }

    /// The `v2Checkpoint` hint parses from V2 checkpoint tables to its exact contents. Pins, per
    /// table, the checkpoint version, file path, sidecar paths, metadata id, created time, and
    /// configuration; and -- shared across these fixtures -- a `(3, 7)` protocol with
    /// V2Checkpoint/AppendOnly/ Invariants features and an unpartitioned parquet `id: long`
    /// schema. The non-file actions are exactly protocol + metadata + checkpointMetadata (no
    /// txn, no domainMetadata). Also checks the identity gate exposes a matched hint's sidecars
    /// but suppresses a mismatched one (`v2-classic-checkpoint-parquet`, whose hint names a
    /// UUID checkpoint while the segment selects the classic-named one).
    /// A table's expected `_last_checkpoint` identity and metadata for the case below.
    struct ExpectedHint {
        table: &'static str,
        version: u64,
        path: &'static str,
        sidecars: &'static [&'static str],
        metadata_id: &'static str,
        created_time: i64,
        config: &'static [(&'static str, &'static str)],
    }

    #[rstest]
    #[case::parquet_sidecars(ExpectedHint {
        table: "v2-checkpoints-parquet-with-sidecars",
        version: 6,
        path: "00000000000000000006.checkpoint.f15b9025-707a-4c73-aac0-31dfcbd29aa6.parquet",
        sidecars: &[
            "00000000000000000006.checkpoint.0000000001.0000000002.76931b15-ead3-480d-b86c-afe55a577fc3.parquet",
            "00000000000000000006.checkpoint.0000000002.0000000002.4367b29c-0e87-447f-8e81-9814cc01ad1f.parquet",
        ],
        metadata_id: "5a5afdfe-7d40-4109-bb92-29b051257e4c",
        created_time: 1739329708855,
        config: &[("delta.checkpointInterval", "1"), ("delta.checkpointPolicy", "v2")],
    })]
    #[case::json_sidecars(ExpectedHint {
        table: "v2-checkpoints-json-with-sidecars",
        version: 6,
        path: "00000000000000000006.checkpoint.2a15d0c6-8b11-4a98-bab4-957905d62f7f.json",
        sidecars: &[
            "00000000000000000006.checkpoint.0000000001.0000000002.19af1366-a425-47f4-8fa6-8d6865625573.parquet",
            "00000000000000000006.checkpoint.0000000002.0000000002.5008b69f-aa8a-4a66-9299-0733a56a7e63.parquet",
        ],
        metadata_id: "f571bf08-452e-4155-9f52-f793e630c55c",
        created_time: 1739329697356,
        config: &[("delta.checkpointInterval", "1"), ("delta.checkpointPolicy", "v2")],
    })]
    #[case::parquet_last_checkpoint(ExpectedHint {
        table: "v2-checkpoints-parquet-with-last-checkpoint",
        version: 0,
        path: "00000000000000000000.checkpoint.8516aa94-7099-4e71-92a0-d6d7e7bb3b2c.parquet",
        sidecars: &["00000000000000000000.checkpoint.0000000001.0000000001.c561300b-ad5f-49d4-a28d-9b3f4bb0331c.parquet"],
        metadata_id: "d3b78022-27fc-470d-a4ea-1b8b47fcc143",
        created_time: 1739329764309,
        config: &[("delta.checkpointPolicy", "v2")],
    })]
    #[case::json_last_checkpoint(ExpectedHint {
        table: "v2-checkpoints-json-with-last-checkpoint",
        version: 0,
        path: "00000000000000000000.checkpoint.0e42c15b-17cc-4918-990d-2ff76e918e4d.json",
        sidecars: &["00000000000000000000.checkpoint.0000000001.0000000001.9167a758-dd93-4e52-8636-7cf5776eb10f.parquet"],
        metadata_id: "f03ce383-0d09-4e1c-9446-8d80e1a59daa",
        created_time: 1739329763101,
        config: &[("delta.checkpointPolicy", "v2")],
    })]
    #[case::classic_parquet(ExpectedHint {
        table: "v2-classic-checkpoint-parquet",
        version: 1,
        path: "00000000000000000001.checkpoint.bfe7499d-715e-4d64-82a4-e6cdd2fc37af.parquet",
        sidecars: &["00000000000000000001.checkpoint.0000000001.0000000001.e2eb56f9-1c54-4a82-b122-de108e317c20.parquet"],
        metadata_id: "541a194a-df83-4f46-9adf-032a1275e82b",
        created_time: 1739329759409,
        config: &[("delta.checkpointPolicy", "v2")],
    })]
    #[case::classic_json(ExpectedHint {
        table: "v2-classic-checkpoint-json",
        version: 1,
        path: "00000000000000000001.checkpoint.6c750e24-bbc4-4618-8feb-7cd7d5b9e084.json",
        sidecars: &["00000000000000000001.checkpoint.0000000001.0000000001.c1bacf45-f3a9-4846-bd44-87cdacd4620f.parquet"],
        metadata_id: "29ef2045-59c5-4cf7-9d5d-2ba47e971d32",
        created_time: 1739313200623,
        config: &[("delta.checkpointPolicy", "v2")],
    })]
    fn v2_last_checkpoint_hint_contents(#[case] expected: ExpectedHint) -> KernelResult<()> {
        use crate::unit_test_utils::load_test_table;

        let ExpectedHint {
            table,
            version: expected_version,
            path: expected_path,
            sidecars: expected_sidecars,
            metadata_id: expected_metadata_id,
            created_time: expected_created_time,
            config: expected_config,
        } = expected;

        let (engine, snapshot, _tempdir) = load_test_table(table)?;
        let seg = snapshot.log_segment();
        let hint =
            LastCheckpointHint::try_read(engine.storage_handler().as_ref(), &seg.log_root, None)?
                .expect("table has a _last_checkpoint");
        let v2 = hint.v2_checkpoint.as_ref().expect("V2 checkpoint hint");

        // Version, checkpoint file path, and sidecar paths are this table's exact identity.
        assert_eq!(hint.version, expected_version, "{table}: version");
        assert_eq!(v2.path, expected_path, "{table}: v2 path");
        let sidecar_paths: Vec<&str> = v2
            .sidecar_files
            .as_ref()
            .expect("sidecarFiles present")
            .iter()
            .map(|s| s.path.as_str())
            .collect();
        assert_eq!(
            sidecar_paths.as_slice(),
            expected_sidecars,
            "{table}: sidecar paths"
        );

        // The non-file actions are exactly one protocol, one metadata, and one checkpointMetadata
        // -- no txn or domainMetadata.
        let actions = v2
            .non_file_actions
            .as_ref()
            .expect("nonFileActions present");
        assert!(
            !actions.iter().any(|a| matches!(a, HintAction::Txn(_))),
            "{table}: no txn"
        );
        assert!(
            !actions
                .iter()
                .any(|a| matches!(a, HintAction::DomainMetadata(_))),
            "{table}: no domain metadata"
        );

        // A (3, 7) protocol gated on the V2Checkpoint reader feature, with V2Checkpoint/AppendOnly/
        // Invariants on the writer side.
        let protocol = one_action(actions, |a| match a {
            HintAction::Protocol(p) => Some(p),
            _ => None,
        });
        assert_eq!(
            (protocol.min_reader_version(), protocol.min_writer_version()),
            (3, 7),
            "{table}: protocol version"
        );
        assert_eq!(
            protocol.reader_features(),
            Some([TableFeature::V2Checkpoint].as_slice()),
            "{table}: reader features"
        );
        assert_eq!(
            protocol.writer_features(),
            Some(
                [
                    TableFeature::V2Checkpoint,
                    TableFeature::AppendOnly,
                    TableFeature::Invariants
                ]
                .as_slice()
            ),
            "{table}: writer features"
        );

        // Metadata for an unnamed, unpartitioned parquet table with a single `id: long` column.
        let metadata = one_action(actions, |a| match a {
            HintAction::Metadata(m) => Some(m),
            _ => None,
        });
        assert_eq!(metadata.id(), expected_metadata_id, "{table}: metadata id");
        assert_eq!(metadata.name(), None, "{table}: metadata name");
        assert_eq!(
            metadata.description(),
            None,
            "{table}: metadata description"
        );
        assert_eq!(metadata.format_provider(), "parquet", "{table}: format");
        assert_eq!(
            metadata.parse_schema()?,
            schema! { nullable "id": LONG },
            "{table}: metadata schema"
        );
        assert!(
            metadata.partition_columns().is_empty(),
            "{table}: unpartitioned"
        );
        assert_eq!(
            metadata.created_time(),
            Some(expected_created_time),
            "{table}: created time"
        );
        let expected_config: HashMap<String, String> = expected_config
            .iter()
            .map(|(k, v)| (k.to_string(), v.to_string()))
            .collect();
        assert_eq!(
            metadata.configuration(),
            &expected_config,
            "{table}: configuration"
        );

        // checkpointMetadata records this checkpoint's version.
        let checkpoint_metadata = one_action(actions, |a| match a {
            HintAction::CheckpointMetadata(c) => Some(c),
            _ => None,
        });
        assert_eq!(
            checkpoint_metadata.version as u64, expected_version,
            "{table}: checkpointMetadata.version"
        );
        assert_eq!(
            checkpoint_metadata.tags, None,
            "{table}: checkpointMetadata.tags"
        );

        // Identity gate: a hint naming the selected checkpoint exposes its sidecars through the
        // accessor; one naming a different same-version checkpoint is fully suppressed.
        let selected = &seg
            .listed
            .checkpoint_parts
            .first()
            .expect("checkpoint present")
            .filename;
        if &v2.path == selected {
            assert_eq!(
                seg.checkpoint_hint_sidecars(),
                v2.sidecar_files.as_ref(),
                "{table}: matched hint exposes its sidecars"
            );
        } else {
            assert!(
                seg.checkpoint_hint_schema().is_none() && seg.checkpoint_hint_sidecars().is_none(),
                "{table}: mismatched hint ({}) must be suppressed",
                v2.path
            );
        }
        Ok(())
    }

    /// A storage handler whose every method panics, so a test can prove an operation never touched
    /// storage.
    struct NoIoStorageHandler;

    impl StorageHandler for NoIoStorageHandler {
        fn list_from(&self, _path: &Url) -> KernelResult<KernelResultIteratorStatic<FileMeta>> {
            panic!("list_from should not be called");
        }
        fn read_files(
            &self,
            _files: Vec<crate::FileSlice>,
        ) -> KernelResult<KernelResultIteratorStatic<bytes::Bytes>> {
            panic!("read_files should not be called");
        }
        fn put(&self, _path: &Url, _data: bytes::Bytes, _overwrite: bool) -> KernelResult<()> {
            panic!("put should not be called");
        }
        fn copy_atomic(&self, _src: &Url, _dest: &Url) -> KernelResult<()> {
            panic!("copy_atomic should not be called");
        }
        fn head(&self, _path: &Url) -> KernelResult<FileMeta> {
            panic!("head should not be called");
        }
        fn delete(&self, _path: &Url) -> KernelResult<()> {
            panic!("delete should not be called");
        }
    }

    // A cancelled token must surface as `Err(Cancelled)`, never swallowed as "no hint"
    // (`Ok(None)`). The pre-cancelled token short-circuits the default
    // `read_files_with_cancellation` before any read, so the panicking handler is never
    // touched.
    #[test]
    fn try_read_propagates_cancellation() {
        let log_root = Url::parse("memory:///_delta_log/").unwrap();
        let token: CancellationTokenRef =
            std::sync::Arc::new(crate::unit_test_utils::TestCancellationToken::cancelled());
        let result = LastCheckpointHint::try_read(&NoIoStorageHandler, &log_root, Some(&token));
        assert!(matches!(result, Err(KernelError::Cancelled)));
    }
}