delta_kernel/log_segment_files/

mod.rs

//! [`LogSegmentFiles`] is a struct holding the result of listing the delta log. Currently, it
//! exposes four APIs for listing:
//! 1. `list_commits`: Lists all commit files between the provided start and end versions.
//! 2. `list`: Lists all commit and checkpoint files between the provided start and end versions.
//! 3. `list_with_checkpoint_hint`: Lists all commit and checkpoint files after the provided
//!    checkpoint hint.
//! 4. `list_with_backward_checkpoint_scan`: Scans backward from an end version in 1000-version
//!    windows until a complete checkpoint is found or the log is exhausted.
//!
//! After listing, one can leverage the [`LogSegmentFiles`] to construct a [`LogSegment`].
//!
//! [`LogSegment`]: crate::log_segment::LogSegment

use std::collections::HashMap;

use delta_kernel_derive::internal_api;
use itertools::Itertools;
use tracing::{debug, info, instrument, warn};
use url::Url;

use crate::cancellation::{check_cancelled, CancellableIterator, CancellationTokenRef};
use crate::last_checkpoint_hint::LastCheckpointHint;
use crate::path::LogPathFileType::*;
use crate::path::{
    may_begin_listable_log_path, CheckpointInstance, LogPathFileType, ParsedLogPath,
};
use crate::{DeltaResult, Error, StorageHandler, Version};

#[cfg(test)]
mod tests;

/// Represents the set of log files found during a listing operation in the Delta log directory.
///
/// - `ascending_commit_files`: All commit and staged commit files found, sorted by version. May
///   contain gaps.
/// - `ascending_compaction_files`: All compaction commit files found, sorted by version.
/// - `checkpoint_parts`: All parts of the most recent complete checkpoint (all same version). Empty
///   if no checkpoint found. A version can hold several complete checkpoints; see
///   [`group_checkpoint_parts`] for which one this is.
/// - `latest_crc_file`: The CRC file with the highest version, only if version >= checkpoint
///   version.
/// - `latest_commit_file`: The commit file with the highest version, or `None` if no commits were
///   found. This field may be present even when `ascending_commit_files` is empty, such as when a
///   checkpoint subsumes all commits. In that case, it is retained because downstream code (e.g.
///   In-Commit Timestamp reading) needs access to the commit file at the snapshot version.
/// - `max_published_version`: The highest published commit file version, or `None` if no published
///   commits were found.
#[derive(Debug, Default, Clone, PartialEq, Eq)]
#[internal_api]
pub(crate) struct LogSegmentFiles {
    pub ascending_commit_files: Vec<ParsedLogPath>,
    pub ascending_compaction_files: Vec<ParsedLogPath>,
    pub checkpoint_parts: Vec<ParsedLogPath>,
    pub latest_crc_file: Option<ParsedLogPath>,
    pub latest_commit_file: Option<ParsedLogPath>,
    pub max_published_version: Option<Version>,
}

/// Returns a lazy iterator of [`ParsedLogPath`]s from the filesystem over versions
/// `[start_version, end_version]`. The iterator handles parsing, filtering out non-listable
/// files (e.g. dot-prefixed files), and stopping at `end_version`. It stops consuming the
/// underlying listing at the first path past the version-named region, so directories like
/// `_staged_commits/` and `_sidecars/` are never paged through.
///
/// This is a thin wrapper around [`StorageHandler::list_from`] that provides the standard
/// Delta log file discovery pipeline. Callers are responsible for handling the `log_tail`
/// (catalog-provided commits) and tracking `max_published_version`.
///
/// With a `cancellation_token`, the listing becomes cancellable: the engine may interrupt its own
/// I/O, and the returned iterator is polled against the token so cancellation arrives as a terminal
/// [`Error::Cancelled`] rather than an early end.
#[internal_api]
pub(crate) fn list_delta_log_from_storage(
    storage: &dyn StorageHandler,
    log_root: &Url,
    start_version: Version,
    end_version: Version,
    cancellation_token: Option<&CancellationTokenRef>,
) -> DeltaResult<impl Iterator<Item = DeltaResult<ParsedLogPath>>> {
    let start_from = log_root.join(&format!("{start_version:020}"))?;
    let log_root_str = log_root.to_string();
    let files = storage
        .list_from_with_cancellation(&start_from, cancellation_token.cloned())?
        // The listing is sorted by full path, so nothing relevant follows the first relative path
        // past the version-named region (see `may_begin_listable_log_path`). Stopping there avoids
        // paging through `_staged_commits/` and `_sidecars/`, which can hold thousands of files.
        // A path that doesn't strip the log_root prefix is kept; parsing discards it.
        // TODO(#2740): push the bound into the listing request itself.
        .take_while(move |meta_res| match meta_res {
            Ok(meta) => meta
                .location
                .as_str()
                .strip_prefix(&log_root_str)
                .is_none_or(may_begin_listable_log_path),
            Err(_) => true,
        })
        .map(|meta| ParsedLogPath::try_from(meta?))
        // NOTE: this filters out .crc files etc which start with "." - some engines
        // produce `.something.parquet.crc` corresponding to `something.parquet`. Kernel
        // doesn't care about these files. Critically, note these are _different_ than
        // normal `version.crc` files which are listed + captured normally. Additionally
        // we likely aren't even 'seeing' these files since lexicographically the string
        // "." comes before the string "0".
        .filter_map_ok(|path_opt| path_opt.filter(|p| p.should_list()))
        .take_while(move |path_res| match path_res {
            // discard any path with too-large version; keep errors
            Ok(path) => path.version <= end_version,
            Err(_) => true,
        });
    // Wrap the filtered pipeline so cancellation is checked as the iterator is consumed, outside
    // the version `take_while` above. Checked inside, a cancelled listing would end with `None` and
    // be indistinguishable from a complete one; outside, it surfaces as a terminal
    // `Error::Cancelled`.
    Ok(CancellableIterator::new(files, cancellation_token.cloned()))
}

/// Groups all checkpoint parts according to the checkpoint they belong to.
///
/// Several _complete_ checkpoints can legitimately share a version, say two multi-part checkpoints
/// with different part counts, or two uuid-named ones. Each gets its own [`CheckpointInstance`]
/// key, so the caller can pick a winner deterministically (see
/// `ListingAccumulator::select_checkpoint_for_group`).
///
/// `parts` must arrive in ascending file name order, which log listing provides: a multi-part
/// checkpoint only accumulates while its parts arrive in order.
#[internal_api]
fn group_checkpoint_parts(
    parts: Vec<ParsedLogPath>,
) -> HashMap<CheckpointInstance, Vec<ParsedLogPath>> {
    debug_assert!(
        parts.is_sorted_by_key(|p| &p.filename),
        "checkpoint parts must arrive in ascending file name order"
    );
    let mut checkpoints: HashMap<CheckpointInstance, Vec<ParsedLogPath>> = HashMap::new();
    for part_file in parts {
        match &part_file.file_type {
            // A single-file checkpoint is complete on its own. Keying uuid-named ones on file name
            // keeps two of them at the same version separate.
            ClassicCheckpoint | UuidCheckpoint => {
                if let Some(instance) = CheckpointInstance::of(&part_file) {
                    checkpoints.insert(instance, vec![part_file]);
                }
            }
            MultiPartCheckpoint {
                part_num: 1,
                num_parts,
            } => {
                // Start a new multi-part checkpoint
                checkpoints.insert(
                    CheckpointInstance::MultiPart {
                        num_parts: *num_parts,
                    },
                    vec![part_file],
                );
            }
            MultiPartCheckpoint {
                part_num,
                num_parts,
            } => {
                // Continue a multi-part checkpoint.
                // Checkpoint parts are required to be in-order from log listing to build
                // a multi-part checkpoint
                if let Some(part_files) = checkpoints.get_mut(&CheckpointInstance::MultiPart {
                    num_parts: *num_parts,
                }) {
                    if *part_num as usize == 1 + part_files.len() {
                        // Safe to append because all previous parts exist
                        part_files.push(part_file);
                    }
                }
            }
            Commit | StagedCommit | CompactedCommit { .. } | Crc | Unknown => {}
        }
    }
    checkpoints
}

/// Returns the version of the latest complete checkpoint in `files`, or `None` if no complete
/// checkpoint exists. Skips 0-byte checkpoint files so they don't count toward completeness.
fn find_complete_checkpoint_version(ascending_files: &[ParsedLogPath]) -> Option<Version> {
    ascending_files
        .iter()
        .filter(|f| f.is_checkpoint() && should_process_log_file(f))
        .chunk_by(|f| f.version)
        .into_iter()
        .filter_map(|(version, parts)| {
            let owned: Vec<ParsedLogPath> = parts.cloned().collect();
            group_checkpoint_parts(owned)
                .iter()
                .any(|(instance, part_files)| instance.is_complete(part_files))
                .then_some(version)
        })
        .last()
}

/// Validates a log file's size. Returns `true` to keep the file, `false` to skip it
/// (with a warning already emitted).
///
/// Compaction and checkpoint files are skipped when empty -- they have fallbacks
/// (individual commits, older checkpoints). Commit and CRC files are kept even
/// if empty; the warning ensures the corrupt file is identifiable in logs.
#[internal_api]
pub(crate) fn should_process_log_file(file: &ParsedLogPath) -> bool {
    if file.location.size > 0 {
        return true;
    }
    match file.file_type {
        // Commit files are kept even if 0 bytes -- the downstream JSON handler might
        // error, but the warning here ensures the corrupt file is identifiable in logs.
        // We don't skip commits because they are the source of truth for table state.
        Commit | StagedCommit => {
            warn!(
                "{:?} file is empty (0 bytes): {}",
                file.file_type, file.location.location,
            );
            return true;
        }
        CompactedCommit { .. } => {
            warn!(
                "Skipping empty (0 byte) compacted log file {}, \
                 falling back to individual commits",
                file.location.location,
            );
        }
        ClassicCheckpoint | UuidCheckpoint | MultiPartCheckpoint { .. } => {
            warn!(
                "Skipping empty (0 byte) checkpoint file: {}",
                file.location.location,
            );
        }
        // CRC files are optional and may report size 0 on some platforms.
        // Keep them -- the CRC reader handles empty/invalid content.
        Crc => {
            warn!("CRC file is empty (0 bytes): {}", file.location.location,);
            return true;
        }
        Unknown => return true,
    }
    false
}

/// Accumulates and groups log files during listing. Each "group" consists of all files that
/// share the same version number (e.g., commit, checkpoint parts, CRC files).
///
/// We need to group by version because:
/// 1. A version may have multiple checkpoint parts that must be collected before we can determine
///    if the checkpoint is complete
/// 2. If a complete checkpoint exists, we can discard all commits before it
///
/// Groups are flushed (processed) when we encounter a file with a different version or
/// reach EOF, at which point we check for complete checkpoints and update our state.
#[derive(Default)]
struct ListingAccumulator {
    /// The result being built up
    output: LogSegmentFiles,
    /// Staging area for checkpoint parts at the current version group; always empty when iteration
    /// ends
    pending_checkpoint_parts: Vec<ParsedLogPath>,
    /// End-version bound used in process_file() to filter CompactedCommit files
    // TODO(#2337): remove allow(dead_code) when log compaction is re-enabled
    #[allow(dead_code)]
    end_version: Option<Version>,
    /// The version of the current group being accumulated
    group_version: Option<Version>,
}

impl ListingAccumulator {
    fn process_file(&mut self, file: ParsedLogPath) {
        if !should_process_log_file(&file) {
            return;
        }
        match file.file_type {
            Commit | StagedCommit => self.output.ascending_commit_files.push(file),
            // TODO(#2337): re-enable log compaction once testing is sufficient
            // CompactedCommit { hi } if self.end_version.is_none_or(|end| hi <= end) => {
            //     self.output.ascending_compaction_files.push(file);
            // }
            // CompactedCommit { .. } => (), // Failed the bounds check above
            CompactedCommit { .. } => {
                debug!(
                    "Skipping unsupported log compaction file: {:?}",
                    file.location
                );
            }
            ClassicCheckpoint | UuidCheckpoint | MultiPartCheckpoint { .. } => {
                self.pending_checkpoint_parts.push(file)
            }
            Crc => {
                self.output.latest_crc_file.replace(file);
            }
            Unknown => {
                // It is possible that there are other files being stashed away into
                // _delta_log/  This is not necessarily forbidden, but something we
                // want to know about in a debugging scenario
                debug!(
                    "Found file {} with unknown file type {:?} at version {}",
                    file.filename, file.file_type, file.version
                );
            }
        }
    }

    /// Called before processing each new file. If `file_version` differs from the current
    /// `group_version`, finalizes the current group by calling `select_checkpoint_for_group`,
    /// then advances `group_version` to the new version. On the first call (when
    /// `group_version` is `None`), simply initializes it.
    fn maybe_flush_and_advance(&mut self, file_version: Version) {
        match self.group_version {
            Some(gv) if file_version != gv => {
                self.select_checkpoint_for_group(gv);
                self.group_version = Some(file_version);
            }
            None => {
                self.group_version = Some(file_version);
            }
            _ => {} // same version, no flush needed
        }
    }

    /// Selects this version's checkpoint. Any of a version's complete checkpoints (see
    /// [`group_checkpoint_parts`]) describes the same table state; the choice must be stable across
    /// processes (matching Delta-Spark), so we take the greatest in [`CheckpointInstance`] order.
    ///
    /// When a complete checkpoint exists we drop the commits/compactions collected so far, keeping
    /// only the latest commit.
    fn select_checkpoint_for_group(&mut self, version: Version) {
        let pending_checkpoint_parts = std::mem::take(&mut self.pending_checkpoint_parts);
        if let Some((_, complete_checkpoint)) = group_checkpoint_parts(pending_checkpoint_parts)
            .into_iter()
            .filter(|(instance, part_files)| instance.is_complete(part_files))
            .max_by(|(a, _), (b, _)| a.cmp(b))
        {
            self.output.checkpoint_parts = complete_checkpoint;
            // Keep the commit at the checkpoint version (if any) before clearing all older commits.
            self.output.latest_commit_file = self
                .output
                .ascending_commit_files
                .last()
                .filter(|c| c.version == version)
                .cloned();
            // Log replay only uses commits/compactions after a complete checkpoint
            self.output.ascending_commit_files.clear();
            self.output.ascending_compaction_files.clear();
            // Drop CRC file if older than checkpoint (CRC must be >= checkpoint version)
            if self
                .output
                .latest_crc_file
                .as_ref()
                .is_some_and(|crc| crc.version < version)
            {
                self.output.latest_crc_file = None;
            }
        }
    }
}

/// Number of versions covered by each backward-scan window in
/// `LogSegmentFiles::list_with_backward_checkpoint_scan`
const BACKWARD_SCAN_WINDOW_SIZE: u64 = 1000;

impl LogSegmentFiles {
    /// Assembles a `LogSegmentFiles` from `fs_files` (an iterator of files
    /// listed from storage) and `log_tail` (catalog-provided commits).
    ///
    /// - `fs_files`: files listed from storage in ascending version order
    /// - `log_tail`: list of commits that takes precedence over the filesystem ones
    /// - `start_version`: start version of the entire listing range provided; in practice, this is
    ///   the lower bound (inclusive) for log_tail entries included in the result
    /// - `end_version`: upper bound (inclusive) on versions to include, `None` means no bound
    pub(crate) fn build_log_segment_files(
        fs_files: impl Iterator<Item = DeltaResult<ParsedLogPath>>,
        log_tail: Vec<ParsedLogPath>,
        start_version: Version,
        end_version: Option<Version>,
    ) -> DeltaResult<Self> {
        // check log_tail is only commits
        // note that LogSegment checks no gaps/duplicates so we don't duplicate that here
        debug_assert!(
            log_tail.iter().all(|entry| entry.is_commit()),
            "log_tail should only contain commits"
        );

        let log_tail_start_version = log_tail.first().map(|f| f.version);
        let end = end_version.unwrap_or(Version::MAX);

        let mut acc = ListingAccumulator {
            end_version,
            ..Default::default()
        };

        // Phase 1: Stream filesystem files lazily (no collect).
        // We always list from the filesystem even when the log_tail covers the entire commit
        // range, because non-commit files (CRC, checkpoints, compactions) only exist on the
        // filesystem — the log_tail only provides commit files.
        for file_result in fs_files {
            let file = file_result?;

            // Track max published commit version from ALL filesystem Commit files,
            // including those that will be skipped because log_tail takes precedence.
            if matches!(file.file_type, LogPathFileType::Commit) {
                acc.output.max_published_version =
                    acc.output.max_published_version.max(Some(file.version));
            }

            // Skip filesystem commits at versions covered by the log_tail (the log_tail
            // is authoritative for commits). Non-commit files are always kept.
            if file.is_commit()
                && log_tail_start_version.is_some_and(|tail_start| file.version >= tail_start)
            {
                continue;
            }

            acc.maybe_flush_and_advance(file.version);
            acc.process_file(file);
        }

        // Phase 2: Process log_tail entries. We do this after Phase 1 because log_tail commits
        // start at log_tail_start_version and are in ascending version order — they always extend
        // (or overlap with, but supersede) the filesystem-listed commits. Processing them after
        // Phase 1 maintains ascending version order throughout, which is required by the checkpoint
        // grouping logic. Note that Phase 1 already skipped filesystem commits at log_tail
        // versions, so there's no duplication here.
        //
        // log_tail entries at versions before a checkpoint may still be included
        // here - LogSegment::try_new is the safeguard that filters those out unconditionally
        let filtered_log_tail = log_tail
            .into_iter()
            .filter(|entry| entry.version >= start_version && entry.version <= end);
        for file in filtered_log_tail {
            // Track max published version for published commits from the log_tail
            if matches!(file.file_type, LogPathFileType::Commit) {
                acc.output.max_published_version =
                    acc.output.max_published_version.max(Some(file.version));
            }

            acc.maybe_flush_and_advance(file.version);
            acc.process_file(file);
        }

        // Flush the final group
        if let Some(gv) = acc.group_version {
            acc.select_checkpoint_for_group(gv);
        }

        // Since ascending_commit_files is cleared at each checkpoint, if it's non-empty here
        // it contains only commits after the most recent checkpoint. The last element is the
        // highest version commit overall, so we update latest_commit_file to it. If it's empty,
        // we keep the value set at the checkpoint (if a commit existed at the checkpoint version),
        // or remains None.
        if let Some(commit_file) = acc.output.ascending_commit_files.last() {
            acc.output.latest_commit_file = Some(commit_file.clone());
        }

        Ok(acc.output)
    }

    pub(crate) fn ascending_commit_files(&self) -> &Vec<ParsedLogPath> {
        &self.ascending_commit_files
    }

    /// The staged (unpublished) commit files, in ascending version order.
    pub(crate) fn staged_commits(&self) -> impl Iterator<Item = &ParsedLogPath> {
        self.ascending_commit_files
            .iter()
            .filter(|f| f.file_type == LogPathFileType::StagedCommit)
    }

    pub(crate) fn ascending_commit_files_mut(&mut self) -> &mut Vec<ParsedLogPath> {
        &mut self.ascending_commit_files
    }

    pub(crate) fn checkpoint_parts(&self) -> &Vec<ParsedLogPath> {
        &self.checkpoint_parts
    }

    pub(crate) fn latest_commit_file(&self) -> &Option<ParsedLogPath> {
        &self.latest_commit_file
    }

    /// Iterator over every listed log path across all fields.
    pub(crate) fn iter_all_paths(&self) -> impl Iterator<Item = &ParsedLogPath> {
        self.ascending_commit_files
            .iter()
            .chain(&self.ascending_compaction_files)
            .chain(&self.checkpoint_parts)
            .chain(&self.latest_crc_file)
            .chain(&self.latest_commit_file)
    }

    /// Estimated heap size in bytes, best-effort estimate.
    pub(crate) fn estimated_heap_size_bytes(&self) -> usize {
        let vec_buffer_bytes = (self.ascending_commit_files.capacity()
            + self.ascending_compaction_files.capacity()
            + self.checkpoint_parts.capacity())
            * size_of::<ParsedLogPath>();
        let path_bytes: usize = self
            .iter_all_paths()
            .map(ParsedLogPath::estimated_heap_size_bytes)
            .sum();
        vec_buffer_bytes + path_bytes
    }

    /// List all commits between the provided `start_version` (inclusive) and `end_version`
    /// (inclusive). All other types are ignored.
    ///
    /// `log_tail` is a contiguous run of commits ending at the table's latest version. It takes
    /// precedence over the filesystem listing, and is required for catalog-managed tables, whose
    /// unbackfilled staged commits exist only here.
    pub(crate) fn list_commits(
        storage: &dyn StorageHandler,
        log_root: &Url,
        log_tail: Vec<ParsedLogPath>,
        start_version: Option<Version>,
        end_version: Option<Version>,
        cancellation_token: Option<&CancellationTokenRef>,
    ) -> DeltaResult<Self> {
        debug_assert!(
            log_tail.iter().all(|entry| entry.is_commit()),
            "log_tail should only contain commits"
        );
        let start = start_version.unwrap_or(0);
        let end = end_version.unwrap_or(Version::MAX);
        let fs_iter =
            list_delta_log_from_storage(storage, log_root, start, end, cancellation_token)?;

        let log_tail_start_version = log_tail.first().map(|f| f.version);
        let mut listed_commits = Vec::new();
        let mut max_published_version: Option<Version> = None;
        // Filesystem commits, skipping any covered by the log_tail.
        for file_result in fs_iter {
            let file = file_result?;
            if file.file_type != LogPathFileType::Commit {
                continue;
            }
            max_published_version = max_published_version.max(Some(file.version));
            if log_tail_start_version.is_some_and(|tail_start| file.version >= tail_start) {
                continue;
            }
            should_process_log_file(&file); // warns if 0 bytes
            listed_commits.push(file);
        }

        // Log_tail commits, extending the filesystem prefix in ascending order.
        for file in log_tail {
            if file.version < start || file.version > end {
                continue;
            }
            if file.file_type == LogPathFileType::Commit {
                max_published_version = max_published_version.max(Some(file.version));
            }
            listed_commits.push(file);
        }

        let latest_commit_file = listed_commits.last().cloned();
        Ok(LogSegmentFiles {
            ascending_commit_files: listed_commits,
            latest_commit_file,
            max_published_version,
            ..Default::default()
        })
    }

    /// List all commit and checkpoint files with versions above the provided `start_version`
    /// (inclusive). If successful, this returns a `LogSegmentFiles`.
    ///
    /// The `log_tail` is an optional sequence of commits provided by the caller, e.g. via
    /// [`SnapshotBuilder::with_log_tail`]. It may contain either published or staged commits. The
    /// `log_tail` must strictly adhere to being a 'tail' — a contiguous cover of versions `X..=Y`
    /// where `Y` is the latest version of the table. If it overlaps with commits listed from the
    /// filesystem, the `log_tail` will take precedence for commits; non-commit files (CRC,
    /// checkpoints, compactions) are always taken from the filesystem.
    // TODO: encode some of these guarantees in the output types. e.g. we could have:
    // - SortedCommitFiles: Vec<ParsedLogPath>, is_ascending: bool, end_version: Version
    // - CheckpointParts: Vec<ParsedLogPath>, checkpoint_version: Version (guarantee all same
    //   version)
    #[instrument(name = "log.list", skip_all, fields(start = ?start_version, end = ?end_version), err)]
    pub(crate) fn list(
        storage: &dyn StorageHandler,
        log_root: &Url,
        log_tail: Vec<ParsedLogPath>,
        start_version: Option<Version>,
        end_version: Option<Version>,
        cancellation_token: Option<&CancellationTokenRef>,
    ) -> DeltaResult<Self> {
        let start = start_version.unwrap_or(0);
        let end = end_version.unwrap_or(Version::MAX);
        let fs_iter =
            list_delta_log_from_storage(storage, log_root, start, end, cancellation_token)?;
        Self::build_log_segment_files(fs_iter, log_tail, start, end_version)
    }

    /// List all commit and checkpoint files after the provided checkpoint. It is guaranteed that
    /// all the returned [`ParsedLogPath`]s will have a version less than or equal to the
    /// `end_version`.
    ///
    /// The hint only tells us where to start listing; it never influences which checkpoint is
    /// selected at a version. A hint that turns out to describe a different checkpoint than the one
    /// selected is logged and ignored, not an error.
    pub(crate) fn list_with_checkpoint_hint(
        checkpoint_metadata: &LastCheckpointHint,
        storage: &dyn StorageHandler,
        log_root: &Url,
        log_tail: Vec<ParsedLogPath>,
        end_version: Option<Version>,
        cancellation_token: Option<&CancellationTokenRef>,
    ) -> DeltaResult<Self> {
        let listed_files = Self::list(
            storage,
            log_root,
            log_tail,
            Some(checkpoint_metadata.version),
            end_version,
            cancellation_token,
        )?;

        let Some(latest_checkpoint) = listed_files.checkpoint_parts.last() else {
            // The hint names a checkpoint that no longer exists, and because the listing started
            // at the hinted version, no checkpoint exists at or after it either. The log was
            // modified out of band: the checkpoint was deleted without clearing the hint, or the
            // table was dropped and recreated at the same path.
            //
            // Kernel fails rather than retrying the listing from version 0, because that recovery
            // is unsound. Listing only checks that the commits it finds are contiguous, not that
            // they start at version 0, so two distinct kinds of damage survive as a plausible but
            // wrong snapshot instead of an error:
            //   - A deleted log prefix leaves a contiguous suffix with no checkpoint. If commits
            //     0-40 and the checkpoint are removed but 41.. remain, listing from version 0
            //     replays 41 as if it were the start of history, dropping every action before it.
            //   - A recreated table yields a segment mixing files from two different tables. A
            //     table with a checkpoint at commit 3 is dropped and recreated at the same path;
            //     the drop leaves commits 0-2 behind, the new table writes its own 0, 1, 2, ...,
            //     and listing from version 0 sees one contiguous sequence whose low versions belong
            //     to two different tables and replays it as a single history.
            return Err(Error::invalid_checkpoint(
                "Had a _last_checkpoint hint but didn't find any checkpoints",
            ));
        };
        if latest_checkpoint.version != checkpoint_metadata.version {
            info!(
            "_last_checkpoint hint is out of date. _last_checkpoint version: {}. Using actual most recent: {}",
            checkpoint_metadata.version,
            latest_checkpoint.version
        );
        } else if !checkpoint_metadata.applies_to(&listed_files.checkpoint_parts) {
            // Expected whenever a writer checkpoints a version another writer already checkpointed
            // and leaves the hint alone. `applies_to` also makes `LogSegment::checkpoint_hint`
            // yield `None`, so this logs exactly when the hint's fields get dropped.
            debug!(
                version = checkpoint_metadata.version,
                hint_parts = checkpoint_metadata.parts.unwrap_or(1),
                selected_parts = listed_files.checkpoint_parts.len(),
                selected_checkpoint_part = %latest_checkpoint.filename,
                "_last_checkpoint hint describes a different checkpoint than the one selected at \
                 this version; using the checkpoint file's own fields"
            );
        }
        Ok(listed_files)
    }

    /// Returns a [`LogSegmentFiles`] ending at `end_version`, rooted at the most recent complete
    /// checkpoint at or before `end_version`, or rooted at version 0 if no checkpoint is found.
    ///
    /// To find the checkpoint without a full forward listing from version 0, this scans backward
    /// from `end_version` in windows of size [`BACKWARD_SCAN_WINDOW_SIZE`], stopping as soon as
    /// a complete checkpoint is found (or version 0 is reached).
    /// Then, all files from the windows that were scanned are combined with `log_tail` to produce a
    /// log segment rooted at the checkpoint version (or version 0 if no checkpoint) with all
    /// commits after the checkpoint version. A log_tail commit at exactly the checkpoint
    /// version may be included at this stage but will be filtered out by `LogSegment::try_new`.
    ///
    /// For example, given the desired end_version = 12500 and a checkpoint at v8900:
    /// - Window 1 [11501, 12501): no checkpoint -> continue
    /// - Window 2 [10501, 11501): no checkpoint -> continue
    /// - Window 3 [9501, 10501): no checkpoint -> continue
    /// - Window 4 [8501, 9501): checkpoint at v8900 found -> stop
    /// All files from windows 1-4 are combined with `log_tail` to produce a log segment
    /// rooted at the checkpoint at v8900 with all commits from v8901 to v12500.
    #[instrument(name = "log.list_with_backward_checkpoint_scan", skip_all, fields(end = end_version), err)]
    pub(crate) fn list_with_backward_checkpoint_scan(
        storage: &dyn StorageHandler,
        log_root: &Url,
        log_tail: Vec<ParsedLogPath>,
        end_version: Version,
        cancellation_token: Option<&CancellationTokenRef>,
    ) -> DeltaResult<Self> {
        // Scan backward in 1000-version windows, collecting ALL file types, until a complete
        // checkpoint is found or the log is exhausted.
        let mut windows: Vec<Vec<ParsedLogPath>> = Vec::new();
        let mut found_checkpoint_version: Option<Version> = None;
        // upper is the exclusive upper bound of the next window; adding 1 includes end_version
        // in the first window. The inclusive range passed to list_delta_log_from_storage is
        // [lower, upper - 1].
        let mut upper = end_version + 1;
        while upper > 0 {
            // Each window is collected eagerly, so check between windows too: a long backward scan
            // would otherwise keep going after cancellation.
            check_cancelled(cancellation_token)?;
            let lower = upper.saturating_sub(BACKWARD_SCAN_WINDOW_SIZE);
            let window_files: Vec<_> = list_delta_log_from_storage(
                storage,
                log_root,
                lower,
                upper - 1,
                cancellation_token,
            )?
            .try_collect()?;

            found_checkpoint_version = find_complete_checkpoint_version(&window_files);
            windows.push(window_files);

            if found_checkpoint_version.is_some() {
                break;
            }
            upper = lower;
        }

        let fs_iter = windows.into_iter().rev().flatten().map(Ok);
        let start = found_checkpoint_version.unwrap_or(0);
        Self::build_log_segment_files(fs_iter, log_tail, start, Some(end_version))
    }
}