feat: Add list with backward checkpoint scan (delta-io#2174)

## 🥞 Stacked PR
Use this
[link](https://github.com/delta-io/delta-kernel-rs/pull/2174/files) to
review incremental changes.
-
[stack/refactor-list-new](delta-io#2173)
[[Files
changed](https://github.com/delta-io/delta-kernel-rs/pull/2173/files)]
[MERGED]
-
[**stack/backward-listing-new**](delta-io#2174)
[[Files
changed](https://github.com/delta-io/delta-kernel-rs/pull/2174/files)]

---------
## What changes are proposed in this pull request?

This PR improves snapshot loading for time travel by replacing the full
forward listing fallback with a backward checkpoint scan. Previously,
when a _last_checkpoint hint was unavailable or pointed past the
requested version, kernel fell back to listing the entire log from v0.
The new strategy lists from the filesystem from the target version in
1000-version windows backwards, stopping as soon as a complete
checkpoint is found. It keeps track of the files as it lists backwards
and once a window with a checkpoint is found, it calls the new
build_log_segment_files() function (refactored list() logic - added in
previous PR in the stack) with the files.

This PR introduces the following changes:
- New list_with_backward_checkpoint_scan() method on LogSegmentFiles:
scans backward from an end_version in 1000-version windows, stopping at
the window with the first complete checkpoint
- New has_complete_checkpoint_in() helper: given a slice of
ParsedLogPath files, returns true if at least one complete checkpoint is
present. This is used by the backward scan loop to decide whether to
stop.
  - Updated for_snapshot_impl case dispatch
- Updated build_log_segment_files(): when called with start_version:
None, derives the lower bound from the most recent complete checkpoint
found during the filesystem phase rather than requiring the caller to
supply it explicitly. This allows list_with_backward_checkpoint_scan()
to pass all scanned files in a single call and have the
(checkpoint-derived) lower bound computed automatically.
  
## How was this change tested?

For new functions introduced in log_segment_files:
- Tests for list_with_backward_checkpoint_scan() behaviour
- Window loop arithmetic (for listing from file system 1000 versions at
a time)
- has_complete_checkpoint_in() edge cases

Existing and new tests that cover the new path/case in
for_snapshot_impl():
- build_snapshot_without_checkpoints (second sub-test, ttv=Some(2))
- build_snapshot_with_checkpoint_greater_than_time_travel_version
- build_snapshot_time_travel_no_checkpoint_falls_back_to_v0 (added with
this PR)
- build_snapshot_time_travel_no_hint_checkpoint_at_end_version_included
(added with this PR)
- test_compaction_in_version_range
- test_compaction_out_of_version_range

Initially opted for an [implementation similar to delta kernel
java](https://sourcegraph.prod.databricks-corp.com/delta-io/delta/-/blob/kernel/kernel-api/src/main/java/io/delta/kernel/internal/checkpoints/Checkpointer.java?L182),
where we have a function that finds the last complete checkpoint
version, returns the version, and then we call the normal list()
function from there (my [initial
implementation](delta-io#2118)).
This improves performance in the case where there is a nearby checkpoint
to find, but in the case that there are no checkpoints, this makes the
process of building a snapshot 2x slower (because we essentially list
from the filesystem the same files twice). Because of this, I opted for
an implementation that keeps track of the files as we list backwards.

Benchmarking results:
<img width="2266" height="838" alt="image"
src="https://github.com/user-attachments/assets/4ed1b7d5-f640-4f42-8324-3f83b21fadbb"
/>
The first row of this picture shows the current state of this code, the
second row is the initial implementation that causes a regression, and
the third row is the current implementation.

Author: lorenarosati

kernel/src/log_segment.rs

@@ -298,20 +298,38 @@ impl LogSegment {
             .as_ref()
             .and_then(|hint| hint.checkpoint_schema.clone());
 
-        let listed_files = match (checkpoint_hint, time_travel_version) {
-            (Some(cp), None) => {
-                LogSegmentFiles::list_with_checkpoint_hint(&cp, storage, &log_root, log_tail, None)?
-            }
-            (Some(cp), Some(end_version)) if cp.version <= end_version => {
-                LogSegmentFiles::list_with_checkpoint_hint(
-                    &cp,
-                    storage,
-                    &log_root,
-                    log_tail,
-                    Some(end_version),
-                )?
-            }
-            _ => LogSegmentFiles::list(storage, &log_root, log_tail, None, time_travel_version)?,
+        // The end_version is the time_travel_version, if present
+        // TODO: When max catalog version is implemented, we would use that as end_version if
+        // time_travel_version is not present
+        let end_version = time_travel_version;
+
+        // Keep the hint only if it points at or before end_version, or if there is no end_version bound
+        let usable_hint = checkpoint_hint.filter(|cp| end_version.is_none_or(|v| cp.version <= v));
+
+        // Cases:
+        //
+        // 1. usable_hint present, end_version is Some  --> list_with_checkpoint_hint from hint.version TO end_version
+        // 2. usable_hint present, end_version is None  --> list_with_checkpoint_hint from hint.version unbounded
+        // 3. no usable_hint,      end_version is Some  --> backward-scan for checkpoint before end_version,
+        //                                                  list from that checkpoint TO end_version
+        //                                                  (falls back to v0 if no checkpoint found)
+        // 4. no usable_hint,      end_version is None  --> list from v0 unbounded
+
+        let listed_files = match (usable_hint, end_version) {
+            // Cases 1 and 2
+            (Some(cp), end_version) => LogSegmentFiles::list_with_checkpoint_hint(
+                &cp,
+                storage,
+                &log_root,
+                log_tail,
+                end_version,
+            )?,
+            // Case 3
+            (None, Some(end)) => LogSegmentFiles::list_with_backward_checkpoint_scan(
+                storage, &log_root, log_tail, end,
+            )?,
+            // Case 4
+            (None, None) => LogSegmentFiles::list(storage, &log_root, log_tail, None, None)?,
         };
 
         LogSegment::try_new(

kernel/src/log_segment/tests.rs

@@ -829,6 +829,62 @@ async fn build_snapshot_with_start_checkpoint_and_time_travel_version() {
     assert_eq!(log_segment.listed.ascending_commit_files[0].version, 4);
 }
 
+#[rstest::rstest]
+#[case::no_hint(None)]
+#[case::stale_hint(Some(LastCheckpointHint {
+    version: 10, // stale: 10 > end_version 5, so it is discarded
+    size: 10,
+    parts: None,
+    size_in_bytes: None,
+    num_of_add_files: None,
+    checkpoint_schema: None,
+    checksum: None,
+    tags: None,
+}))]
+#[tokio::test]
+async fn build_snapshot_time_travel_no_checkpoint_falls_back_to_v0(
+    #[case] hint: Option<LastCheckpointHint>,
+) {
+    let paths: Vec<Path> = (0..=5).map(|v| delta_path_for_version(v, "json")).collect();
+    let (storage, log_root) = build_log_with_paths_and_checkpoint(&paths, None).await;
+
+    let log_segment =
+        LogSegment::for_snapshot_impl(storage.as_ref(), log_root, vec![], hint, Some(5)).unwrap();
+
+    let commit_files = log_segment.listed.ascending_commit_files;
+    let checkpoint_parts = log_segment.listed.checkpoint_parts;
+
+    assert_eq!(checkpoint_parts.len(), 0);
+    let versions = commit_files.into_iter().map(|x| x.version).collect_vec();
+    assert_eq!(versions, vec![0, 1, 2, 3, 4, 5]);
+}
+
+#[tokio::test]
+async fn build_snapshot_time_travel_no_hint_checkpoint_at_end_version_included() {
+    let (storage, log_root) = build_log_with_paths_and_checkpoint(
+        &[
+            delta_path_for_version(0, "json"),
+            delta_path_for_version(1, "json"),
+            delta_path_for_version(2, "json"),
+            delta_path_for_version(3, "json"),
+            delta_path_for_version(4, "json"),
+            delta_path_for_version(5, "json"),
+            delta_path_for_version(5, "checkpoint.parquet"),
+        ],
+        None,
+    )
+    .await;
+
+    let log_segment =
+        LogSegment::for_snapshot_impl(storage.as_ref(), log_root, vec![], None, Some(5)).unwrap();
+
+    let commit_files = log_segment.listed.ascending_commit_files;
+    let checkpoint_parts = log_segment.listed.checkpoint_parts;
+    assert_eq!(checkpoint_parts.len(), 1);
+    assert_eq!(checkpoint_parts[0].version, 5);
+    assert_eq!(commit_files.len(), 0);
+}
+
 #[tokio::test]
 async fn build_table_changes_with_commit_versions() {
     let (storage, log_root) = build_log_with_paths_and_checkpoint(