diff --git a/default-engine/src/filesystem.rs b/default-engine/src/filesystem.rs
index 7642779e80..586661be9c 100644
--- a/default-engine/src/filesystem.rs
+++ b/default-engine/src/filesystem.rs
@@ -1,8 +1,9 @@
 use std::sync::Arc;
 
 use bytes::Bytes;
+use delta_kernel::object_store::list::{PaginatedListOptions, PaginatedListStore};
 use delta_kernel::object_store::path::Path;
-use delta_kernel::object_store::{self, DynObjectStore, ObjectStoreExt as _, PutMode};
+use delta_kernel::object_store::{self, DynObjectStore, ObjectMeta, ObjectStoreExt as _, PutMode};
 use delta_kernel::{
     CancellationTokenRef, DeltaResult, DeltaResultIteratorStatic, Error, FileMeta, FileSlice,
     StorageHandler,
@@ -14,17 +15,35 @@ use url::Url;
 use crate::executor::TaskExecutor;
 use crate::UrlExt;
 
-#[derive(Debug)]
 pub struct ObjectStoreStorageHandler<E: TaskExecutor> {
     inner: Arc<DynObjectStore>,
+    paginated: Option<Arc<dyn PaginatedListStore>>,
+    ordered: bool,
     task_executor: Arc<E>,
     readahead: usize,
 }
 
+impl<E: TaskExecutor> std::fmt::Debug for ObjectStoreStorageHandler<E> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ObjectStoreStorageHandler")
+            .field("inner", &self.inner)
+            .field("paginated", &self.paginated.is_some())
+            .field("readahead", &self.readahead)
+            .finish_non_exhaustive()
+    }
+}
+
 impl<E: TaskExecutor> ObjectStoreStorageHandler<E> {
-    pub(crate) fn new(store: Arc<DynObjectStore>, task_executor: Arc<E>) -> Self {
+    pub(crate) fn new(
+        store: Arc<DynObjectStore>,
+        paginated: Option<Arc<dyn PaginatedListStore>>,
+        ordered: bool,
+        task_executor: Arc<E>,
+    ) -> Self {
         Self {
             inner: store,
+            paginated,
+            ordered,
             task_executor,
             readahead: 10,
         }
@@ -46,13 +65,23 @@ impl<E: TaskExecutor> ObjectStoreStorageHandler<E> {
 /// [`MeteredStorageHandler`]: delta_kernel::metrics::MeteredStorageHandler
 async fn list_from_impl(
     store: Arc<DynObjectStore>,
+    paginated: Option<Arc<dyn PaginatedListStore>>,
+    ordered: bool,
     path: Url,
+    end: Option<Url>,
 ) -> DeltaResult<BoxStream<'static, DeltaResult<FileMeta>>> {
+    if let Some(end) = &end {
+        delta_kernel::path::validate_list_range(&path, end)?;
+    }
     // The offset is used for list-after; the prefix is used to restrict the listing to a specific
     // directory. Unfortunately, `Path` provides no easy way to check whether a name is
     // directory-like, because it strips trailing /, so we're reduced to manually checking the
     // original URL.
     let offset = Path::from_url_path(path.path())?;
+    let end = end.map(|end| Path::from_url_path(end.path())).transpose()?;
+    if end.as_ref().is_some_and(|end| end <= &offset) {
+        return Ok(stream::empty().boxed());
+    }
     let prefix = if path.path().ends_with('/') {
         offset.clone()
     } else {
@@ -65,10 +94,35 @@ async fn list_from_impl(
         Path::from_iter(parts)
     };
 
-    let has_ordered_listing = supports_ordered_listing(&path);
+    if let Some(paginated) = paginated {
+        return list_paginated(paginated, path, prefix, offset, ordered, end).await;
+    }
+    let stop_before = end.clone();
 
+    // `list_with_offset` lets capable stores push down the offset but recursively lists
+    // descendants.
     let stream = store
         .list_with_offset(Some(&prefix), &offset)
+        .take_while(move |result| {
+            futures::future::ready(
+                !ordered
+                    || result.as_ref().map_or(true, |meta| {
+                        stop_before.as_ref().is_none_or(|end| &meta.location < end)
+                    }),
+            )
+        })
+        // Apply the caller's bound before hiding descendants. Without a bound, sorted paths can
+        // interleave direct children and descendants: a, b/x, c. Encountering b/x does not
+        // establish that all direct children have been listed.
+        .try_filter(move |meta| {
+            futures::future::ready(
+                end.as_ref().is_none_or(|end| &meta.location < end)
+                    && meta
+                        .location
+                        .prefix_match(&prefix)
+                        .is_some_and(|parts| parts.count() == 1),
+            )
+        })
         .map(move |meta| {
             let meta = meta?;
             let mut location = path.clone();
@@ -80,10 +134,11 @@ async fn list_from_impl(
             })
         });
 
-    if !has_ordered_listing {
-        // Local filesystem doesn't return sorted list - need to collect and sort
+    if !ordered {
         let mut items: Vec<_> = stream.try_collect().await?;
-        items.sort_unstable();
+        items.sort_unstable_by(|left, right| {
+            delta_kernel::path::compare_listing_paths(&left.location, &right.location)
+        });
         Ok(Box::pin(stream::iter(
             items.into_iter().map(Ok::<FileMeta, delta_kernel::Error>),
         )))
@@ -92,6 +147,85 @@ async fn list_from_impl(
     }
 }
 
+/// Lists one directory level, pushing the offset only for globally ordered stores.
+async fn list_paginated(
+    store: Arc<dyn PaginatedListStore>,
+    base_url: Url,
+    prefix: Path,
+    offset: Path,
+    ordered: bool,
+    end: Option<Path>,
+) -> DeltaResult<BoxStream<'static, DeltaResult<FileMeta>>> {
+    let request_prefix = (!prefix.as_ref().is_empty()).then(|| format!("{}/", prefix.as_ref()));
+    let request_offset = (ordered && offset != prefix).then(|| offset.to_string());
+    let first_options = PaginatedListOptions {
+        offset: request_offset,
+        delimiter: Some("/".into()),
+        ..Default::default()
+    };
+
+    let pages: BoxStream<'static, object_store::Result<Vec<ObjectMeta>>> =
+        stream::try_unfold(Some(first_options), move |options| {
+            let store = store.clone();
+            let request_prefix = request_prefix.clone();
+            async move {
+                let Some(options) = options else {
+                    return Ok::<_, object_store::Error>(None);
+                };
+                let result = store
+                    .list_paginated(request_prefix.as_deref(), options.clone())
+                    .await?;
+                let next_options = result.page_token.map(|page_token| PaginatedListOptions {
+                    page_token: Some(page_token),
+                    offset: None,
+                    ..options
+                });
+                Ok(Some((result.result.objects, next_options)))
+            }
+        })
+        .boxed();
+
+    let stop_before = end.clone();
+    let stream = pages
+        .map_ok(|objects| stream::iter(objects.into_iter().map(Ok::<_, object_store::Error>)))
+        .try_flatten()
+        .take_while(move |result| {
+            futures::future::ready(
+                !ordered
+                    || result.as_ref().map_or(true, |meta| {
+                        stop_before.as_ref().is_none_or(|end| &meta.location < end)
+                    }),
+            )
+        })
+        .try_filter(move |meta| {
+            futures::future::ready(
+                meta.location > offset && end.as_ref().is_none_or(|end| &meta.location < end),
+            )
+        })
+        .map_ok(move |meta| file_meta(&base_url, meta))
+        .err_into::<delta_kernel::Error>();
+
+    if ordered {
+        Ok(stream.boxed())
+    } else {
+        let mut items: Vec<_> = stream.try_collect().await?;
+        items.sort_unstable_by(|left, right| {
+            delta_kernel::path::compare_listing_paths(&left.location, &right.location)
+        });
+        Ok(stream::iter(items.into_iter().map(Ok)).boxed())
+    }
+}
+
+fn file_meta(base_url: &Url, meta: ObjectMeta) -> FileMeta {
+    let mut location = base_url.clone();
+    location.set_path(&format!("/{}", meta.location.as_ref()));
+    FileMeta {
+        location,
+        last_modified: meta.last_modified.timestamp_millis(),
+        size: meta.size,
+    }
+}
+
 /// Native async implementation for read_files
 async fn read_files_impl(
     store: Arc<DynObjectStore>,
@@ -198,7 +332,13 @@ impl<E: TaskExecutor> StorageHandler for ObjectStoreStorageHandler<E> {
         path: &Url,
         cancellation_token: Option<CancellationTokenRef>,
     ) -> DeltaResult<DeltaResultIteratorStatic<FileMeta>> {
-        let future = list_from_impl(self.inner.clone(), path.clone());
+        let future = list_from_impl(
+            self.inner.clone(),
+            self.paginated.clone(),
+            self.ordered,
+            path.clone(),
+            None,
+        );
         let iter = super::stream_future_to_cancellable_iter(
             self.task_executor.clone(),
             future,
@@ -207,6 +347,26 @@ impl<E: TaskExecutor> StorageHandler for ObjectStoreStorageHandler<E> {
         Ok(iter)
     }
 
+    fn list_range(
+        &self,
+        start: &Url,
+        end: &Url,
+        cancellation_token: Option<CancellationTokenRef>,
+    ) -> DeltaResult<DeltaResultIteratorStatic<FileMeta>> {
+        let future = list_from_impl(
+            self.inner.clone(),
+            self.paginated.clone(),
+            self.ordered,
+            start.clone(),
+            Some(end.clone()),
+        );
+        super::stream_future_to_cancellable_iter(
+            self.task_executor.clone(),
+            future,
+            cancellation_token,
+        )
+    }
+
     /// Read data specified by the start and end offset from the file.
     ///
     /// This will return the data in the same order as the provided file slices.
@@ -277,30 +437,278 @@ impl<E: TaskExecutor> StorageHandler for ObjectStoreStorageHandler<E> {
 ///   "This page shows you how to list the [objects](https://cloud.google.com/storage/docs/objects)
 ///   stored in your Cloud Storage buckets, which are ordered in the list lexicographically by
 ///   name."
-fn supports_ordered_listing(url: &Url) -> bool {
+pub(crate) fn supports_ordered_listing(url: &Url) -> bool {
+    let path_style_bucket =
+        if url.scheme() == "https" && url.host_str().is_some_and(|host| host.starts_with("s3.")) {
+            url.path_segments().and_then(|mut segments| segments.next())
+        } else {
+            None
+        };
     !((url.scheme() == "file")
         // S3 Directory Buckets
         || url.domain().map(|d| d.contains("--x-s3")).unwrap_or(false)
         // S3 Directory Bucket Access Points
-        || url.domain().map(|d| d.contains("-xa-s3")).unwrap_or(false))
+        || url.domain().map(|d| d.contains("-xa-s3")).unwrap_or(false)
+        || path_style_bucket.is_some_and(|bucket| bucket.contains("--x-s3") || bucket.contains("-xa-s3")))
 }
 
 #[cfg(test)]
 mod tests {
     use std::ops::Range;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::sync::Mutex;
     use std::time::Duration;
 
+    use delta_kernel::log_segment_files::list_delta_log_from_storage;
+    use delta_kernel::object_store::list::PaginatedListResult;
     use delta_kernel::object_store::local::LocalFileSystem;
     use delta_kernel::object_store::memory::InMemory;
+    use delta_kernel::object_store::{
+        CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectStore,
+        PutMultipartOptions, PutOptions, PutPayload, PutResult,
+    };
     use delta_kernel::Engine as _;
     use delta_kernel_default_engine_test_utils::current_time_duration;
     use itertools::Itertools;
+    use rstest::rstest;
     use test_utils::delta_path_for_version;
+    use wiremock::matchers::{method, path, query_param, query_param_is_missing};
+    use wiremock::{Mock, MockServer, ResponseTemplate};
 
     use super::*;
     use crate::executor::tokio::TokioBackgroundExecutor;
+    use crate::storage::EngineStore;
     use crate::DefaultEngineBuilder;
 
+    #[derive(Debug)]
+    struct RecordingOffsetStore {
+        inner: InMemory,
+        list_requests: Mutex<Vec<(Option<Path>, Path)>>,
+        upstream_pulls: Arc<AtomicUsize>,
+        pages: Vec<Vec<&'static str>>,
+        page_requests: AtomicUsize,
+        recursive_keys: Vec<&'static str>,
+        fail_at: Option<Path>,
+    }
+
+    impl RecordingOffsetStore {
+        fn new() -> Self {
+            Self {
+                inner: InMemory::new(),
+                list_requests: Mutex::new(Vec::new()),
+                upstream_pulls: Arc::new(AtomicUsize::new(0)),
+                pages: Vec::new(),
+                page_requests: AtomicUsize::new(0),
+                recursive_keys: Vec::new(),
+                fail_at: None,
+            }
+        }
+    }
+
+    #[derive(Debug, PartialEq)]
+    struct PaginatedRequest {
+        prefix: Option<String>,
+        offset: Option<String>,
+        delimiter: Option<String>,
+        page_token: Option<String>,
+    }
+
+    #[derive(Debug, Default)]
+    struct UnorderedPaginatedStore {
+        requests: Mutex<Vec<PaginatedRequest>>,
+    }
+
+    #[async_trait::async_trait]
+    impl PaginatedListStore for UnorderedPaginatedStore {
+        async fn list_paginated(
+            &self,
+            prefix: Option<&str>,
+            options: PaginatedListOptions,
+        ) -> object_store::Result<PaginatedListResult> {
+            self.requests.lock().unwrap().push(PaginatedRequest {
+                prefix: prefix.map(ToOwned::to_owned),
+                offset: options.offset.clone(),
+                delimiter: options.delimiter.as_deref().map(ToOwned::to_owned),
+                page_token: options.page_token.clone(),
+            });
+            let (locations, page_token) = match options.page_token.as_deref() {
+                None => (
+                    vec![
+                        "table/_delta_log/00000000000000000012.json",
+                        "table/_delta_log/00000000000000000010.json",
+                        "table/_delta_log/00000000000000000009.json",
+                    ],
+                    Some("second-page".to_string()),
+                ),
+                Some("second-page") => (vec!["table/_delta_log/00000000000000000011.json"], None),
+                Some(token) => panic!("Unexpected page token: {token}"),
+            };
+            let objects = locations
+                .into_iter()
+                .map(|location| ObjectMeta {
+                    location: Path::from(location),
+                    last_modified: chrono::Utc::now(),
+                    size: 1,
+                    e_tag: None,
+                    version: None,
+                })
+                .collect();
+            Ok(PaginatedListResult {
+                result: ListResult {
+                    common_prefixes: vec![Path::from("table/_delta_log/_sidecars")],
+                    objects,
+                },
+                page_token,
+            })
+        }
+    }
+
+    impl std::fmt::Display for RecordingOffsetStore {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            write!(f, "RecordingOffsetStore")
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl PaginatedListStore for RecordingOffsetStore {
+        async fn list_paginated(
+            &self,
+            prefix: Option<&str>,
+            options: PaginatedListOptions,
+        ) -> object_store::Result<PaginatedListResult> {
+            assert_eq!(prefix, Some("_delta_log/"));
+            assert_eq!(options.delimiter.as_deref(), Some("/"));
+            assert!(options.offset.is_none());
+            self.page_requests.fetch_add(1, Ordering::Relaxed);
+            let index = options
+                .page_token
+                .as_deref()
+                .map(|token| token.parse::<usize>().unwrap())
+                .unwrap_or(0);
+            let mut objects = Vec::new();
+            for key in &self.pages[index] {
+                objects.push(self.inner.head(&Path::from(*key)).await?);
+            }
+            Ok(PaginatedListResult {
+                result: ListResult {
+                    common_prefixes: Vec::new(),
+                    objects,
+                },
+                page_token: (index + 1 < self.pages.len()).then(|| (index + 1).to_string()),
+            })
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl ObjectStore for RecordingOffsetStore {
+        async fn put_opts(
+            &self,
+            location: &Path,
+            payload: PutPayload,
+            options: PutOptions,
+        ) -> object_store::Result<PutResult> {
+            self.inner.put_opts(location, payload, options).await
+        }
+
+        async fn put_multipart_opts(
+            &self,
+            location: &Path,
+            options: PutMultipartOptions,
+        ) -> object_store::Result<Box<dyn MultipartUpload>> {
+            self.inner.put_multipart_opts(location, options).await
+        }
+
+        async fn get_opts(
+            &self,
+            location: &Path,
+            options: GetOptions,
+        ) -> object_store::Result<GetResult> {
+            self.inner.get_opts(location, options).await
+        }
+
+        fn list(
+            &self,
+            prefix: Option<&Path>,
+        ) -> BoxStream<'static, object_store::Result<ObjectMeta>> {
+            self.inner.list(prefix)
+        }
+
+        fn list_with_offset(
+            &self,
+            prefix: Option<&Path>,
+            offset: &Path,
+        ) -> BoxStream<'static, object_store::Result<ObjectMeta>> {
+            self.list_requests
+                .lock()
+                .unwrap()
+                .push((prefix.cloned(), offset.clone()));
+            let upstream_pulls = self.upstream_pulls.clone();
+            let listing = if self.recursive_keys.is_empty() {
+                self.inner.list_with_offset(prefix, offset)
+            } else {
+                let offset = offset.clone();
+                let prefix = prefix.cloned();
+                stream::iter(self.recursive_keys.clone())
+                    .map(|key| Path::parse(key).unwrap())
+                    .filter(move |location| {
+                        futures::future::ready(
+                            location > &offset
+                                && prefix
+                                    .as_ref()
+                                    .is_none_or(|prefix| location.prefix_match(prefix).is_some()),
+                        )
+                    })
+                    .map(|location| {
+                        Ok(ObjectMeta {
+                            location,
+                            last_modified: chrono::Utc::now(),
+                            size: 1,
+                            e_tag: None,
+                            version: None,
+                        })
+                    })
+                    .boxed()
+            };
+            let fail_at = self.fail_at.clone();
+            listing
+                .map(move |result| {
+                    upstream_pulls.fetch_add(1, Ordering::Relaxed);
+                    let meta = result?;
+                    if fail_at.as_ref() == Some(&meta.location) {
+                        return Err(object_store::Error::Generic {
+                            store: "test",
+                            source: "injected listing failure".into(),
+                        });
+                    }
+                    Ok(meta)
+                })
+                .boxed()
+        }
+
+        async fn list_with_delimiter(
+            &self,
+            prefix: Option<&Path>,
+        ) -> object_store::Result<ListResult> {
+            self.inner.list_with_delimiter(prefix).await
+        }
+
+        fn delete_stream(
+            &self,
+            locations: BoxStream<'static, object_store::Result<Path>>,
+        ) -> BoxStream<'static, object_store::Result<Path>> {
+            self.inner.delete_stream(locations)
+        }
+
+        async fn copy_opts(
+            &self,
+            from: &Path,
+            to: &Path,
+            options: CopyOptions,
+        ) -> object_store::Result<()> {
+            self.inner.copy_opts(from, to, options).await
+        }
+    }
+
     fn setup_test() -> (
         tempfile::TempDir,
         Arc<LocalFileSystem>,
@@ -309,7 +717,7 @@ mod tests {
         let tmp = tempfile::tempdir().unwrap();
         let store = Arc::new(LocalFileSystem::new());
         let executor = Arc::new(TokioBackgroundExecutor::new());
-        let handler = ObjectStoreStorageHandler::new(store.clone(), executor);
+        let handler = ObjectStoreStorageHandler::new(store.clone(), None, false, executor);
         (tmp, store, handler)
     }
 
@@ -356,7 +764,7 @@ mod tests {
 
         let store = Arc::new(LocalFileSystem::new());
         let executor = Arc::new(TokioBackgroundExecutor::new());
-        let storage = ObjectStoreStorageHandler::new(store, executor);
+        let storage = ObjectStoreStorageHandler::new(store, None, false, executor);
 
         let mut slices: Vec<FileSlice> = Vec::new();
 
@@ -387,7 +795,7 @@ mod tests {
             .unwrap();
 
         let executor = Arc::new(TokioBackgroundExecutor::new());
-        let storage = ObjectStoreStorageHandler::new(store, executor);
+        let storage = ObjectStoreStorageHandler::new(store, None, false, executor);
         let file_url = Url::parse("memory:///hello%2C%20world%21").unwrap();
 
         let read_back: Vec<Bytes> = storage
@@ -413,7 +821,7 @@ mod tests {
         let engine = DefaultEngineBuilder::new(store).build();
         let files: Vec<_> = engine
             .storage_handler()
-            .list_from(&table_root.join("_delta_log").unwrap().join("0").unwrap())
+            .list_from(&table_root.join("_delta_log/").unwrap().join("0").unwrap())
             .unwrap()
             .try_collect()
             .unwrap();
@@ -443,7 +851,7 @@ mod tests {
         let engine = DefaultEngineBuilder::new(store).build();
         let files = engine
             .storage_handler()
-            .list_from(&url.join("_delta_log").unwrap().join("0").unwrap())
+            .list_from(&url.join("_delta_log/").unwrap().join("0").unwrap())
             .unwrap();
         let mut len = 0;
         for (file, expected) in files.zip(expected_names.iter()) {
@@ -462,6 +870,450 @@ mod tests {
         assert_eq!(len, 10, "list_from should have returned 10 files");
     }
 
+    #[tokio::test]
+    async fn list_from_applies_offset_and_excludes_nested_files() {
+        let store = Arc::new(RecordingOffsetStore::new());
+        for key in [
+            "_delta_log/00000000000000000000.json",
+            "_delta_log/00000000000000000001.json",
+            "_delta_log/00000000000000000001.json/child",
+            "_delta_log/00000000000000000002.json",
+            "_delta_log/_staged_commits/00000000000000000003.uuid.json",
+            "_delta_log/z.txt",
+        ] {
+            store
+                .put(&Path::from(key), Bytes::from_static(b"x").into())
+                .await
+                .unwrap();
+        }
+
+        let executor = Arc::new(TokioBackgroundExecutor::new());
+        let handler = ObjectStoreStorageHandler::new(store.clone(), None, true, executor);
+        let start = Url::parse("memory:///_delta_log/00000000000000000001.json").unwrap();
+
+        let locations: Vec<_> = handler
+            .list_from(&start)
+            .unwrap()
+            .map(|result| result.unwrap().location.path().to_string())
+            .collect();
+
+        assert_eq!(
+            locations,
+            vec!["/_delta_log/00000000000000000002.json", "/_delta_log/z.txt"]
+        );
+        assert_eq!(
+            *store.list_requests.lock().unwrap(),
+            vec![(
+                Some(Path::from("_delta_log")),
+                Path::from("_delta_log/00000000000000000001.json")
+            )]
+        );
+    }
+
+    #[rstest]
+    #[case::first_version(0, vec![0], 2)]
+    #[case::finite_range(1, vec![0, 1], 3)]
+    #[case::unbounded_version(u64::MAX, vec![0, 1, 2], 4)]
+    #[tokio::test]
+    async fn bounded_ordered_fallback_log_discovery_stops_with_or_without_hint(
+        #[values(false, true)] include_hint: bool,
+        #[case] end_version: u64,
+        #[case] expected_versions: Vec<u64>,
+        #[case] expected_pulls: usize,
+    ) {
+        let store = Arc::new(RecordingOffsetStore::new());
+        let mut keys: Vec<_> = (0..3)
+            .map(|version| format!("_delta_log/{version:020}.json"))
+            .collect();
+        keys.extend(
+            (0..1000).map(|version| format!("_delta_log/_staged_commits/{version:020}.uuid.json")),
+        );
+        if include_hint {
+            keys.push("_delta_log/_last_checkpoint".into());
+        }
+        for key in keys {
+            store
+                .put(&Path::from(key), Bytes::from_static(b"x").into())
+                .await
+                .unwrap();
+        }
+        let engine = DefaultEngineBuilder::new(EngineStore::from_ordered(store.clone())).build();
+        let storage = engine.storage_handler();
+        let log_root = Url::parse("s3://bucket/_delta_log/").unwrap();
+        let versions: Vec<_> =
+            list_delta_log_from_storage(storage.as_ref(), &log_root, 0, end_version, None)
+                .unwrap()
+                .map(|result| result.unwrap().version)
+                .collect();
+        assert_eq!(versions, expected_versions);
+        assert_eq!(store.upstream_pulls.load(Ordering::Relaxed), expected_pulls);
+        assert_eq!(store.page_requests.load(Ordering::Relaxed), 0);
+    }
+
+    #[tokio::test]
+    async fn custom_paginated_store_sorts_across_pages_before_kernel_stops_listing() {
+        let store = Arc::new(RecordingOffsetStore {
+            pages: vec![
+                vec![
+                    "_delta_log/00000000000000000000.json",
+                    "_delta_log/_last_checkpoint",
+                ],
+                vec!["_delta_log/00000000000000000001.json"],
+            ],
+            ..RecordingOffsetStore::new()
+        });
+        for key in store.pages.iter().flatten() {
+            store
+                .put(&Path::from(*key), Bytes::from_static(b"x").into())
+                .await
+                .unwrap();
+        }
+        let engine = DefaultEngineBuilder::new(EngineStore::from_paginated(store.clone())).build();
+        let storage = engine.storage_handler();
+        let log_root = Url::parse("s3://bucket/_delta_log/").unwrap();
+        let versions: Vec<_> =
+            list_delta_log_from_storage(storage.as_ref(), &log_root, 0, u64::MAX, None)
+                .unwrap()
+                .map(|result| result.unwrap().version)
+                .collect();
+        assert_eq!(versions, vec![0, 1]);
+        assert_eq!(store.page_requests.load(Ordering::Relaxed), 2);
+    }
+
+    #[rstest]
+    #[case::all_direct("", None, vec!["a", "c", "d"], 5)]
+    #[case::interleaved("", Some("d"), vec!["a", "c"], 4)]
+    #[case::exclusive_start("a", Some("d"), vec!["c"], 3)]
+    #[case::exclusive_end("", Some("c"), vec!["a"], 3)]
+    #[case::empty("c", Some("c"), vec![], 0)]
+    #[case::reversed("d", Some("c"), vec![], 0)]
+    fn bounded_listing_preserves_interleaved_children_and_exclusive_bounds(
+        #[values(false, true)] ordered: bool,
+        #[case] start: &str,
+        #[case] end: Option<&str>,
+        #[case] expected: Vec<&str>,
+        #[case] ordered_pulls: usize,
+    ) {
+        let store = Arc::new(RecordingOffsetStore {
+            recursive_keys: vec!["dir/a", "dir/b/x", "dir/c", "dir/d", "dir/z/x"],
+            ..RecordingOffsetStore::new()
+        });
+        let engine_store = if ordered {
+            EngineStore::from_ordered(store.clone())
+        } else {
+            EngineStore::plain(store.clone())
+        };
+        let engine = DefaultEngineBuilder::new(engine_store).build();
+        let storage = engine.storage_handler();
+        let root = Url::parse("s3://bucket/dir/").unwrap();
+        let start = root.join(start).unwrap();
+        let files = match end {
+            Some(end) => storage.list_range(&start, &root.join(end).unwrap(), None),
+            None => storage.list_from(&start),
+        }
+        .unwrap()
+        .collect::<DeltaResult<Vec<_>>>()
+        .unwrap();
+        let expected: Vec<_> = expected
+            .into_iter()
+            .map(|name| root.join(name).unwrap())
+            .collect();
+        assert_eq!(
+            files
+                .into_iter()
+                .map(|file| file.location)
+                .collect::<Vec<_>>(),
+            expected
+        );
+        if ordered || ordered_pulls == 0 {
+            assert_eq!(store.upstream_pulls.load(Ordering::Relaxed), ordered_pulls);
+        }
+    }
+
+    #[test]
+    fn bounded_unknown_order_listing_does_not_stop_at_a_descendant_past_the_bound() {
+        let store = Arc::new(RecordingOffsetStore {
+            recursive_keys: vec!["dir/z/x", "dir/c", "dir/b/x", "dir/a"],
+            ..RecordingOffsetStore::new()
+        });
+        let engine = DefaultEngineBuilder::new(store.clone()).build();
+        let root = Url::parse("s3://bucket/dir/").unwrap();
+        let files = engine
+            .storage_handler()
+            .list_range(&root, &root.join("d").unwrap(), None)
+            .unwrap()
+            .collect::<DeltaResult<Vec<_>>>()
+            .unwrap();
+        assert_eq!(
+            files
+                .into_iter()
+                .map(|file| file.location)
+                .collect::<Vec<_>>(),
+            vec![root.join("a").unwrap(), root.join("c").unwrap()]
+        );
+        assert_eq!(store.upstream_pulls.load(Ordering::Relaxed), 4);
+    }
+
+    #[rstest]
+    fn bounded_listing_compares_decoded_names_in_escaped_directories(
+        #[values(false, true)] ordered: bool,
+    ) {
+        let store = Arc::new(RecordingOffsetStore {
+            recursive_keys: vec!["a b/0", "a b/z", "a b/é"],
+            ..RecordingOffsetStore::new()
+        });
+        let engine_store = if ordered {
+            EngineStore::from_ordered(store)
+        } else {
+            EngineStore::plain(store)
+        };
+        let engine = DefaultEngineBuilder::new(engine_store).build();
+        let start = Url::parse("s3://bucket/a%20b/").unwrap();
+        let files = engine
+            .storage_handler()
+            .list_range(&start, &start.join("%C3%A9").unwrap(), None)
+            .unwrap()
+            .collect::<DeltaResult<Vec<_>>>()
+            .unwrap();
+        assert_eq!(
+            files
+                .into_iter()
+                .map(|meta| meta.location)
+                .collect::<Vec<_>>(),
+            vec![start.join("0").unwrap(), start.join("z").unwrap()]
+        );
+    }
+
+    #[rstest]
+    #[case("s3://another/dir/d")]
+    #[case("s3://bucket/other/d")]
+    #[case("gs://bucket/dir/d")]
+    fn bounded_listing_rejects_mismatched_directories_without_io(#[case] end: &str) {
+        let store = Arc::new(RecordingOffsetStore::new());
+        let engine = DefaultEngineBuilder::new(EngineStore::from_ordered(store.clone())).build();
+        assert!(engine
+            .storage_handler()
+            .list_range(
+                &Url::parse("s3://bucket/dir/a").unwrap(),
+                &Url::parse(end).unwrap(),
+                None,
+            )
+            .is_err());
+        assert!(store.list_requests.lock().unwrap().is_empty());
+    }
+
+    #[rstest]
+    fn bounded_listing_propagates_source_errors(#[values(false, true)] ordered: bool) {
+        let store = Arc::new(RecordingOffsetStore {
+            recursive_keys: vec!["dir/a", "dir/b", "dir/c"],
+            fail_at: Some(Path::from("dir/b")),
+            ..RecordingOffsetStore::new()
+        });
+        let engine_store = if ordered {
+            EngineStore::from_ordered(store)
+        } else {
+            EngineStore::plain(store)
+        };
+        let engine = DefaultEngineBuilder::new(engine_store).build();
+        let root = Url::parse("s3://bucket/dir/").unwrap();
+        let result = engine
+            .storage_handler()
+            .list_range(&root, &root.join("d").unwrap(), None)
+            .and_then(|files| files.collect::<DeltaResult<Vec<_>>>());
+        assert!(result
+            .unwrap_err()
+            .to_string()
+            .contains("injected listing failure"));
+    }
+
+    #[tokio::test]
+    async fn bounded_ordered_paginated_listing_stops_before_requesting_next_page() {
+        let store = Arc::new(RecordingOffsetStore {
+            pages: vec![vec!["_delta_log/a", "_delta_log/c"], vec!["_delta_log/d"]],
+            ..RecordingOffsetStore::new()
+        });
+        for name in ["_delta_log/a", "_delta_log/c", "_delta_log/d"] {
+            store
+                .inner
+                .put(&Path::from(name), Bytes::new().into())
+                .await
+                .unwrap();
+        }
+        let engine =
+            DefaultEngineBuilder::new(EngineStore::from_ordered_paginated(store.clone())).build();
+        let start = Url::parse("s3://bucket/_delta_log/").unwrap();
+        let locations: Vec<_> = engine
+            .storage_handler()
+            .list_range(&start, &start.join("c").unwrap(), None)
+            .unwrap()
+            .map(|result| result.unwrap().location)
+            .collect();
+        assert_eq!(locations, vec![start.join("a").unwrap()]);
+        assert_eq!(store.page_requests.load(Ordering::Relaxed), 1);
+    }
+
+    #[rstest]
+    #[case::ordinary_bucket("bucket")]
+    #[case::directory_bucket("bucket--usw2-az1--x-s3")]
+    #[case::directory_bucket_access_point("access-point-usw2-az1-xa-s3")]
+    #[tokio::test]
+    async fn unordered_paginated_listing_filters_offset_and_sorts_all_pages(
+        #[case] bucket: &str,
+        #[values(false, true)] bounded: bool,
+    ) {
+        let paginated = Arc::new(UnorderedPaginatedStore::default());
+        let executor = Arc::new(TokioBackgroundExecutor::new());
+        let handler = ObjectStoreStorageHandler::new(
+            Arc::new(InMemory::new()),
+            Some(paginated.clone()),
+            false,
+            executor,
+        );
+        let start = Url::parse(&format!(
+            "s3://{bucket}/table/_delta_log/00000000000000000010.json"
+        ))
+        .unwrap();
+
+        let end = start.join("00000000000000000012.json").unwrap();
+        let listing = if bounded {
+            handler.list_range(&start, &end, None)
+        } else {
+            handler.list_from(&start)
+        };
+        let locations: Vec<_> = listing
+            .unwrap()
+            .map(|result| result.unwrap().location.path().to_string())
+            .collect();
+
+        assert_eq!(
+            locations,
+            if bounded {
+                vec!["/table/_delta_log/00000000000000000011.json"]
+            } else {
+                vec![
+                    "/table/_delta_log/00000000000000000011.json",
+                    "/table/_delta_log/00000000000000000012.json",
+                ]
+            }
+        );
+        assert_eq!(
+            *paginated.requests.lock().unwrap(),
+            vec![
+                PaginatedRequest {
+                    prefix: Some("table/_delta_log/".to_string()),
+                    offset: None,
+                    delimiter: Some("/".to_string()),
+                    page_token: None,
+                },
+                PaginatedRequest {
+                    prefix: Some("table/_delta_log/".to_string()),
+                    offset: None,
+                    delimiter: Some("/".to_string()),
+                    page_token: Some("second-page".to_string()),
+                },
+            ]
+        );
+    }
+
+    #[rstest]
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn azure_listing_pushes_directory_and_offset_and_remains_lazy(
+        #[values(true, false)] use_url_factory: bool,
+    ) {
+        const OFFSET: &str = "table/_delta_log/00000000000000000010.json";
+        const NEXT: &str = "table/_delta_log/00000000000000000011.json";
+        const NEXT_PAGE: &str = "table/_delta_log/00000000000000000012.json";
+
+        let server = MockServer::start().await;
+        let body = format!(
+            "<EnumerationResults><Blobs>\
+             <Blob><Name>table/_delta_log/00000000000000000009.json</Name><Properties>\
+             <Last-Modified>Thu, 01 Jul 2021 10:44:59 GMT</Last-Modified>\
+             <Content-Length>1</Content-Length><Content-Type>application/json</Content-Type>\
+             </Properties></Blob>\
+             <Blob><Name>{OFFSET}</Name><Properties>\
+             <Last-Modified>Thu, 01 Jul 2021 10:44:59 GMT</Last-Modified>\
+             <Content-Length>1</Content-Length><Content-Type>application/json</Content-Type>\
+             </Properties></Blob>\
+             <Blob><Name>{NEXT}</Name><Properties>\
+             <Last-Modified>Thu, 01 Jul 2021 10:44:59 GMT</Last-Modified>\
+             <Content-Length>1</Content-Length><Content-Type>application/json</Content-Type>\
+             </Properties></Blob>\
+             <BlobPrefix><Name>table/_delta_log/_staged_commits/</Name></BlobPrefix>\
+             </Blobs><NextMarker>page-2</NextMarker></EnumerationResults>"
+        );
+        Mock::given(method("GET"))
+            .and(path("/container"))
+            .and(query_param("prefix", "table/_delta_log/"))
+            .and(query_param("delimiter", "/"))
+            .and(query_param("startFrom", OFFSET))
+            .and(query_param_is_missing("marker"))
+            .respond_with(ResponseTemplate::new(200).set_body_raw(body, "application/xml"))
+            .expect(1)
+            .mount(&server)
+            .await;
+        Mock::given(method("GET"))
+            .and(path("/container"))
+            .and(query_param("prefix", "table/_delta_log/"))
+            .and(query_param("delimiter", "/"))
+            .and(query_param("marker", "page-2"))
+            .and(query_param_is_missing("startFrom"))
+            .respond_with(ResponseTemplate::new(200).set_body_raw(
+                format!(
+                    "<EnumerationResults><Blobs>\
+                     <Blob><Name>{NEXT_PAGE}</Name><Properties>\
+                     <Last-Modified>Thu, 01 Jul 2021 10:44:59 GMT</Last-Modified>\
+                     <Content-Length>1</Content-Length><Content-Type>application/json</Content-Type>\
+                     </Properties></Blob>\
+                     </Blobs></EnumerationResults>"
+                ),
+                "application/xml",
+            ))
+            .expect(1)
+            .mount(&server)
+            .await;
+
+        let table_url =
+            Url::parse("abfss://container@account.dfs.core.windows.net/table/").unwrap();
+        let store = if use_url_factory {
+            let options = vec![
+                ("endpoint", server.uri()),
+                ("allow_http", "true".to_string()),
+                ("skip_signature", "true".to_string()),
+            ];
+            EngineStore::from_url_opts(&table_url, options).unwrap()
+        } else {
+            EngineStore::from_ordered_paginated(Arc::new(
+                object_store::azure::MicrosoftAzureBuilder::new()
+                    .with_url(table_url.as_str())
+                    .with_endpoint(server.uri())
+                    .with_allow_http(true)
+                    .with_skip_signature(true)
+                    .build()
+                    .unwrap(),
+            ))
+        };
+        let engine = DefaultEngineBuilder::new(store).build();
+        let start = table_url
+            .join("_delta_log/00000000000000000010.json")
+            .unwrap();
+        let mut files = engine.storage_handler().list_from(&start).unwrap();
+
+        assert!(server.received_requests().await.unwrap().is_empty());
+        assert_eq!(
+            files.next().unwrap().unwrap().location,
+            table_url.join(&format!("/{NEXT}")).unwrap()
+        );
+        assert_eq!(server.received_requests().await.unwrap().len(), 1);
+        assert_eq!(
+            files.next().unwrap().unwrap().location,
+            table_url.join(&format!("/{NEXT_PAGE}")).unwrap()
+        );
+        assert!(files.next().is_none());
+        assert_eq!(server.received_requests().await.unwrap().len(), 2);
+        server.verify().await;
+    }
+
     #[tokio::test]
     async fn test_copy() {
         let (tmp, store, handler) = setup_test();
@@ -610,6 +1462,9 @@ mod tests {
         let listed = handler.list_from_with_cancellation(&url, Some(token.clone()));
         assert!(matches!(listed, Err(Error::Cancelled)));
 
+        let bounded = handler.list_range(&url, &url.join("z").unwrap(), Some(token.clone()));
+        assert!(matches!(bounded, Err(Error::Cancelled)));
+
         let read = handler.read_files_with_cancellation(vec![(url, None)], Some(token));
         assert!(matches!(read, Err(Error::Cancelled)));
     }
diff --git a/default-engine/src/lib.rs b/default-engine/src/lib.rs
index a075b808b5..282546ca2f 100644
--- a/default-engine/src/lib.rs
+++ b/default-engine/src/lib.rs
@@ -29,6 +29,7 @@ use self::executor::TaskExecutor;
 use self::filesystem::ObjectStoreStorageHandler;
 use self::json::DefaultJsonHandler;
 use self::parquet::DefaultParquetHandler;
+use self::storage::EngineStore;
 
 pub mod executor;
 pub mod file_stream;
@@ -210,9 +211,8 @@ pub struct DefaultEngine<E: TaskExecutor> {
 ///     .with_task_executor(Arc::new(TokioBackgroundExecutor::new()))
 ///     .build();
 /// ```
-#[derive(Debug)]
 pub struct DefaultEngineBuilder<E> {
-    object_store: Arc<DynObjectStore>,
+    store: EngineStore,
     /// The state is either [`DefaultTaskExecutor`] or `Arc<E>` with a custom task executor.
     task_executor: E,
     /// Read-path I/O concurrency config applied to the JSON and Parquet handlers. `None` fields
@@ -220,6 +220,15 @@ pub struct DefaultEngineBuilder<E> {
     io_config: ReadIoConfig,
 }
 
+impl<E> std::fmt::Debug for DefaultEngineBuilder<E> {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("DefaultEngineBuilder")
+            .field("store", &self.store)
+            .field("io_config", &self.io_config)
+            .finish_non_exhaustive()
+    }
+}
+
 /// Read-path I/O tuning for [`DefaultEngine`]'s JSON and Parquet handlers.
 ///
 /// Both knobs default to the handlers' built-in defaults when left unset.
@@ -237,10 +246,35 @@ struct ReadIoConfig {
 pub struct DefaultTaskExecutor;
 
 impl DefaultEngineBuilder<DefaultTaskExecutor> {
-    /// Create a new [`DefaultEngineBuilder`] instance with the default executor.
-    pub fn new(object_store: Arc<DynObjectStore>) -> Self {
+    /// Create a new [`DefaultEngineBuilder`] with the default executor.
+    ///
+    /// `store` supplies the engine's object operations and optional paginated listing support.
+    /// Use [`EngineStore::from_url_opts`] or [`EngineStore::from_paginated`] for cloud stores.
+    ///
+    /// # Listing performance
+    ///
+    /// Passing an `Arc<S>` directly selects [`EngineStore::plain`], even when `S` is a concrete
+    /// cloud store that implements paginated listing. The results still exclude nested files,
+    /// but listing collects and sorts the recursive results. Use [`EngineStore::from_ordered`]
+    /// for a globally ordered recursive source to allow streaming and bounded early stopping.
+    /// [`EngineStore::from_paginated`] avoids listing descendants but collects and sorts all pages.
+    /// Use [`EngineStore::from_ordered_paginated`] for a store that guarantees global listing order
+    /// to retain streaming and offset pushdown. [`EngineStore::from_url_opts`] selects this for
+    /// ordered built-in cloud backends automatically.
+    ///
+    /// ```no_run
+    /// # use std::sync::Arc;
+    /// # use delta_kernel::object_store::aws::AmazonS3Builder;
+    /// # use delta_kernel_default_engine::{storage::EngineStore, DefaultEngineBuilder};
+    /// # fn example() -> Result<(), Box<dyn std::error::Error>> {
+    /// let store = Arc::new(AmazonS3Builder::from_env().with_bucket_name("bucket").build()?);
+    /// let engine = DefaultEngineBuilder::new(EngineStore::from_ordered_paginated(store)).build();
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn new(store: impl Into<EngineStore>) -> Self {
         Self {
-            object_store,
+            store: store.into(),
             task_executor: DefaultTaskExecutor,
             io_config: ReadIoConfig::default(),
         }
@@ -249,7 +283,7 @@ impl DefaultEngineBuilder<DefaultTaskExecutor> {
     /// Build the [`DefaultEngine`] instance.
     pub fn build(self) -> DefaultEngine<executor::tokio::TokioBackgroundExecutor> {
         let task_executor = Arc::new(executor::tokio::TokioBackgroundExecutor::new());
-        DefaultEngine::new_with_opts(self.object_store, task_executor, self.io_config)
+        DefaultEngine::new_with_opts(self.store, task_executor, self.io_config)
     }
 }
 
@@ -262,7 +296,7 @@ impl<E> DefaultEngineBuilder<E> {
         task_executor: Arc<F>,
     ) -> DefaultEngineBuilder<Arc<F>> {
         DefaultEngineBuilder {
-            object_store: self.object_store,
+            store: self.store,
             task_executor,
             io_config: self.io_config,
         }
@@ -293,7 +327,7 @@ impl<E> DefaultEngineBuilder<E> {
 impl<E: TaskExecutor> DefaultEngineBuilder<Arc<E>> {
     /// Build the [`DefaultEngine`] instance.
     pub fn build(self) -> DefaultEngine<E> {
-        DefaultEngine::new_with_opts(self.object_store, self.task_executor, self.io_config)
+        DefaultEngine::new_with_opts(self.store, self.task_executor, self.io_config)
     }
 }
 
@@ -309,13 +343,16 @@ impl DefaultEngine<executor::tokio::TokioBackgroundExecutor> {
 }
 
 impl<E: TaskExecutor> DefaultEngine<E> {
-    fn new_with_opts(
-        object_store: Arc<DynObjectStore>,
-        task_executor: Arc<E>,
-        io_config: ReadIoConfig,
-    ) -> Self {
+    fn new_with_opts(store: EngineStore, task_executor: Arc<E>, io_config: ReadIoConfig) -> Self {
+        let EngineStore {
+            object_store,
+            paginated,
+            ordered,
+        } = store;
         let raw_storage: Arc<dyn StorageHandler> = Arc::new(ObjectStoreStorageHandler::new(
             object_store.clone(),
+            paginated,
+            ordered,
             task_executor.clone(),
         ));
 
diff --git a/default-engine/src/storage.rs b/default-engine/src/storage.rs
index c7bf7bf8b9..aadb638f13 100644
--- a/default-engine/src/storage.rs
+++ b/default-engine/src/storage.rs
@@ -1,11 +1,178 @@
 use std::collections::HashMap;
 use std::sync::{Arc, LazyLock, RwLock};
 
+use delta_kernel::object_store::aws::{AmazonS3, AmazonS3Builder, AmazonS3ConfigKey};
+use delta_kernel::object_store::azure::{AzureConfigKey, MicrosoftAzure, MicrosoftAzureBuilder};
+use delta_kernel::object_store::gcp::{
+    GoogleCloudStorage, GoogleCloudStorageBuilder, GoogleConfigKey,
+};
+use delta_kernel::object_store::list::PaginatedListStore;
 use delta_kernel::object_store::path::Path;
-use delta_kernel::object_store::{self, Error, ObjectStore};
-use delta_kernel::Error as DeltaError;
+use delta_kernel::object_store::{self, DynObjectStore, Error, ObjectStore, ObjectStoreScheme};
+use delta_kernel::{DeltaResult, Error as DeltaError};
 use url::Url;
 
+/// Object-store handles used by a [`DefaultEngine`](crate::DefaultEngine).
+///
+/// The optional paginated handle lets cloud backends apply the directory delimiter in storage
+/// requests. Globally ordered stores can also push down the starting offset. Without it, listing
+/// uses [`ObjectStore::list_with_offset`] and removes nested descendants client-side.
+pub struct EngineStore {
+    pub(crate) object_store: Arc<DynObjectStore>,
+    pub(crate) paginated: Option<Arc<dyn PaginatedListStore>>,
+    pub(crate) ordered: bool,
+}
+
+impl std::fmt::Debug for EngineStore {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("EngineStore")
+            .field("object_store", &self.object_store)
+            .field("paginated", &self.paginated.is_some())
+            .field("ordered", &self.ordered)
+            .finish()
+    }
+}
+
+impl EngineStore {
+    /// Create a store without provider-specific paginated listing support.
+    ///
+    /// Offset pushdown remains available through [`ObjectStore::list_with_offset`], but stores
+    /// whose offset listing is recursive may still retrieve nested descendants. Results are
+    /// collected and sorted because ordering is unknown. Use [`Self::from_ordered`] when the
+    /// source guarantees globally ordered listings.
+    ///
+    /// # Performance
+    ///
+    /// Listing uses a recursive stream and filters out nested files locally. Finding the next
+    /// direct child or reaching the end can consume a large nested subtree, even though none of
+    /// its files appear in the returned results. On cloud storage, this can require additional
+    /// listing requests.
+    ///
+    /// This also applies to cloud stores passed to the engine as ordinary `Arc` handles. Use
+    /// [`Self::from_paginated`] or [`Self::from_url_opts`] to retain provider-side shallow listing
+    /// where supported.
+    pub fn plain(object_store: Arc<DynObjectStore>) -> Self {
+        Self {
+            object_store,
+            paginated: None,
+            ordered: false,
+        }
+    }
+
+    /// Create a recursive store whose offset listing is globally ordered by full UTF-8 path.
+    ///
+    /// `object_store.list_with_offset` must preserve this order across all pages. The result
+    /// streams direct children and can stop at a caller-supplied upper bound before filtering
+    /// descendants. Violating the ordering guarantee can silently omit files. Use
+    /// [`Self::plain`] when ordering is unknown, including S3 Express directory buckets.
+    /// Unbounded listings can still consume entire descendant subtrees.
+    pub fn from_ordered(object_store: Arc<DynObjectStore>) -> Self {
+        Self {
+            object_store,
+            paginated: None,
+            ordered: true,
+        }
+    }
+
+    /// Create a store that supports provider-specific paginated listing.
+    ///
+    /// The same store handles ordinary object operations and paginated listing, preventing the two
+    /// handles from referring to different backends. `store` must honor the requested delimiter:
+    /// objects must be direct children and subdirectories must be returned as common prefixes.
+    ///
+    /// Results are collected across all pages and sorted before being returned. No starting
+    /// offset is sent to the provider. Use [`Self::from_ordered_paginated`] only when `store`
+    /// guarantees globally ordered listing, including across page boundaries.
+    pub fn from_paginated<S: ObjectStore + PaginatedListStore + 'static>(store: Arc<S>) -> Self {
+        Self {
+            object_store: store.clone(),
+            paginated: Some(store),
+            ordered: false,
+        }
+    }
+
+    /// Create a store whose paginated listings are globally lexicographically ordered by path.
+    ///
+    /// `store` must honor delimiters as in [`Self::from_paginated`] and support starting offsets.
+    /// Every object's path must sort after all paths on preceding pages; ordering within each
+    /// individual page is not sufficient. Violating this contract can cause log discovery to
+    /// miss commits. Do not use this constructor for S3 Express directory buckets.
+    ///
+    /// Listings stream without collecting all pages, and the starting offset is sent with the
+    /// first request. Use [`Self::from_paginated`] when ordering is unknown.
+    pub fn from_ordered_paginated<S: ObjectStore + PaginatedListStore + 'static>(
+        store: Arc<S>,
+    ) -> Self {
+        Self {
+            object_store: store.clone(),
+            paginated: Some(store),
+            ordered: true,
+        }
+    }
+
+    /// Create a store for `url` using the provider-specific `options`.
+    ///
+    /// Built-in S3, GCS, and Azure stores retain paginated listing support. Registered custom URL
+    /// handlers and other built-in stores use [`Self::plain`].
+    /// Ordered cloud backends stream results; S3 Express listings are collected and sorted.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the URL, storage configuration, or custom URL handler is invalid.
+    pub fn from_url_opts<I, K, V>(url: &Url, options: I) -> DeltaResult<Self>
+    where
+        I: IntoIterator<Item = (K, V)>,
+        K: AsRef<str>,
+        V: Into<String>,
+    {
+        let has_custom_handler = URL_REGISTRY
+            .read()
+            .map(|handlers| handlers.contains_key(url.scheme()))
+            .unwrap_or(false);
+        if has_custom_handler {
+            return Ok(Self::plain(store_from_url_opts(url, options)?));
+        }
+
+        let (scheme, _path) = ObjectStoreScheme::parse(url).map_err(object_store::Error::from)?;
+        let options = options
+            .into_iter()
+            .map(|(key, value)| (key.as_ref().to_string(), value.into()));
+        macro_rules! paginated {
+            ($builder:expr, $ordered:expr) => {{
+                let store = Arc::new(build_cloud_store($builder, url, options)?);
+                if $ordered {
+                    Self::from_ordered_paginated(store)
+                } else {
+                    Self::from_paginated(store)
+                }
+            }};
+        }
+        Ok(match scheme {
+            ObjectStoreScheme::AmazonS3 => paginated!(
+                AmazonS3Builder::new(),
+                crate::filesystem::supports_ordered_listing(url)
+            ),
+            ObjectStoreScheme::GoogleCloudStorage => {
+                paginated!(GoogleCloudStorageBuilder::new(), true)
+            }
+            ObjectStoreScheme::MicrosoftAzure => paginated!(MicrosoftAzureBuilder::new(), true),
+            _ => Self::plain(store_from_url_opts(url, options)?),
+        })
+    }
+}
+
+impl From<Arc<DynObjectStore>> for EngineStore {
+    fn from(object_store: Arc<DynObjectStore>) -> Self {
+        Self::plain(object_store)
+    }
+}
+
+impl<S: ObjectStore + 'static> From<Arc<S>> for EngineStore {
+    fn from(object_store: Arc<S>) -> Self {
+        Self::plain(object_store)
+    }
+}
+
 /// Alias for convenience
 type ClosureReturn = Result<(Box<dyn ObjectStore>, Path), Error>;
 /// This type alias makes it easier to reference the handler closure(s)
@@ -39,6 +206,7 @@ pub fn insert_url_handler(
 /// Create an [`ObjectStore`] from a URL.
 ///
 /// Returns an `Arc<dyn ObjectStore>` ready to use with [`crate::DefaultEngine`].
+/// For cloud engines, use [`EngineStore::from_url_opts`] to retain non-recursive offset listing.
 ///
 /// This function checks for custom URL handlers registered via [`insert_url_handler`]
 /// before falling back to [`object_store`]'s default behavior.
@@ -62,6 +230,7 @@ pub fn store_from_url(url: &Url) -> delta_kernel::DeltaResult<Arc<dyn ObjectStor
 /// Create an [`ObjectStore`] from a URL with custom options.
 ///
 /// Returns an `Arc<dyn ObjectStore>` ready to use with [`crate::DefaultEngine`].
+/// For cloud engines, use [`EngineStore::from_url_opts`] to retain non-recursive offset listing.
 ///
 /// This function checks for custom URL handlers registered via [`insert_url_handler`]
 /// before falling back to [`object_store`]'s default behavior.
@@ -108,6 +277,60 @@ where
     Ok(Arc::new(store))
 }
 
+/// Builds a concrete cloud store without erasing its [`PaginatedListStore`] implementation.
+fn build_cloud_store<B: CloudBuilder>(
+    builder: B,
+    url: &Url,
+    options: impl IntoIterator<Item = (String, String)>,
+) -> DeltaResult<B::Store> {
+    let builder = options.into_iter().fold(
+        builder.with_url(url.to_string()),
+        |builder, (key, value)| match key.to_ascii_lowercase().parse() {
+            Ok(config_key) => builder.with_config(config_key, value),
+            Err(_) => builder,
+        },
+    );
+    Ok(builder.build()?)
+}
+
+trait CloudBuilder: Sized {
+    type ConfigKey: std::str::FromStr;
+    type Store: ObjectStore + PaginatedListStore;
+
+    fn with_url(self, url: String) -> Self;
+    fn with_config(self, key: Self::ConfigKey, value: String) -> Self;
+    fn build(self) -> object_store::Result<Self::Store>;
+}
+
+macro_rules! impl_cloud_builder {
+    ($builder:ty, $key:ty, $store:ty) => {
+        impl CloudBuilder for $builder {
+            type ConfigKey = $key;
+            type Store = $store;
+
+            fn with_url(self, url: String) -> Self {
+                self.with_url(url)
+            }
+
+            fn with_config(self, key: Self::ConfigKey, value: String) -> Self {
+                self.with_config(key, value)
+            }
+
+            fn build(self) -> object_store::Result<Self::Store> {
+                self.build()
+            }
+        }
+    };
+}
+
+impl_cloud_builder!(AmazonS3Builder, AmazonS3ConfigKey, AmazonS3);
+impl_cloud_builder!(
+    GoogleCloudStorageBuilder,
+    GoogleConfigKey,
+    GoogleCloudStorage
+);
+impl_cloud_builder!(MicrosoftAzureBuilder, AzureConfigKey, MicrosoftAzure);
+
 #[cfg(test)]
 mod tests {
     use std::collections::HashMap;
@@ -115,8 +338,9 @@ mod tests {
     use delta_kernel::object_store::path::Path;
     use delta_kernel::object_store::{self, ObjectStore};
     use hdfs_native_object_store::HdfsObjectStoreBuilder;
+    use rstest::rstest;
 
-    use super::{insert_url_handler, store_from_url_opts, URL_REGISTRY};
+    use super::{insert_url_handler, store_from_url_opts, EngineStore, URL_REGISTRY};
     use crate::*;
 
     /// Example funciton of doing testing of a custom [HdfsObjectStore] construction
@@ -175,4 +399,28 @@ mod tests {
             }
         }
     }
+
+    #[rstest]
+    #[case("s3://bucket/table", true)]
+    #[case("s3://bucket/table--x-s3", true)]
+    #[case("s3://bucket--usw2-az1--x-s3/table", false)]
+    #[case("s3://access-point-usw2-az1-xa-s3/table", false)]
+    #[case("https://s3.us-west-2.amazonaws.com/bucket/table", true)]
+    #[case("https://bucket.s3.us-west-2.amazonaws.com/table--x-s3", true)]
+    #[case(
+        "https://s3.us-west-2.amazonaws.com/bucket--usw2-az1--x-s3/table",
+        false
+    )]
+    #[case(
+        "https://bucket--usw2-az1--x-s3.s3.us-west-2.amazonaws.com/table",
+        false
+    )]
+    #[case("gs://bucket/table", true)]
+    #[case("abfss://container@account.dfs.core.windows.net/table", true)]
+    fn cloud_url_factory_preserves_listing_capabilities(#[case] url: &str, #[case] ordered: bool) {
+        let url = Url::parse(url).unwrap();
+        let store = EngineStore::from_url_opts(&url, HashMap::<String, String>::new()).unwrap();
+        assert!(store.paginated.is_some());
+        assert_eq!(store.ordered, ordered, "{url}");
+    }
 }
diff --git a/docs/user-guide/src/concepts/engine_trait.md b/docs/user-guide/src/concepts/engine_trait.md
index 81976a1400..74033ae370 100644
--- a/docs/user-guide/src/concepts/engine_trait.md
+++ b/docs/user-guide/src/concepts/engine_trait.md
@@ -38,7 +38,7 @@ write commit and checkpoint files.
 
 | Method | Purpose |
 |--------|---------|
-| `list_from(path)` | List files lexicographically after `path` in the same directory |
+| `list_from(path)` | List files lexicographically after `path` in the same directory, non-recursively |
 | `read_files(files)` | Read byte ranges from one or more files |
 | `copy_atomic(src, dst)` | Atomically copy a file (used for publishing commits) |
 | `put(path, data, overwrite)` | Write raw bytes to a path (fails if `overwrite` is false and file exists) |
diff --git a/docs/user-guide/src/connector/implementing_engine.md b/docs/user-guide/src/connector/implementing_engine.md
index bec49cec9b..06a71962fa 100644
--- a/docs/user-guide/src/connector/implementing_engine.md
+++ b/docs/user-guide/src/connector/implementing_engine.md
@@ -46,9 +46,10 @@ pub trait StorageHandler {
 
 ### Key contracts
 
-- **`list_from`**: Results must be sorted lexicographically by path. If the path ends with
-  `/`, list all files in that directory. Otherwise, list files lexicographically greater than
-  the given path in the same directory.
+- **`list_from`**: Results must be sorted lexicographically by path. Lists a single directory
+  level: nested files and subdirectory entries are excluded. If the path ends with `/`, list the
+  files directly within that directory. Otherwise, list files directly within the same directory
+  that sort lexicographically after the given path.
 
 - **`copy_atomic`**: Must fail with `Error::FileAlreadyExists` if the destination exists.
   This is used for commit publishing in catalog-managed tables.
diff --git a/ffi/src/lib.rs b/ffi/src/lib.rs
index e6eae0fdfc..11a8660701 100644
--- a/ffi/src/lib.rs
+++ b/ffi/src/lib.rs
@@ -18,8 +18,6 @@ use delta_kernel::history_manager::{
     get_earliest_commit as kernel_get_earliest_commit,
     latest_version_as_of as kernel_latest_version_as_of, CommitAt, HistoryCommitType,
 };
-#[cfg(feature = "default-engine-base")]
-use delta_kernel::object_store::ObjectStore;
 use delta_kernel::schema::Schema;
 use delta_kernel::snapshot::{CheckpointWriteResult, Snapshot, SnapshotHint, SnapshotRef};
 use delta_kernel::{DeltaResult, Engine, EngineData, FileStats, LogPath, Version};
@@ -1007,28 +1005,31 @@ fn get_default_engine_impl(
     io_config: IoConcurrencyConfig,
     allocate_error: AllocateErrorFn,
 ) -> DeltaResult<Handle<SharedExternEngine>> {
-    use delta_kernel_default_engine::storage::store_from_url_opts;
+    use delta_kernel_default_engine::storage::EngineStore;
 
-    let store = match object_store_backend {
-        ObjectStoreBackend::UrlScheme => store_from_url_opts(&url, options)?,
-        ObjectStoreBackend::Rest(rest) => {
-            rest_engine::build_rest_object_store(&url, &options, rest.as_ref())?
-        }
-    };
+    let store =
+        match object_store_backend {
+            ObjectStoreBackend::UrlScheme => EngineStore::from_url_opts(&url, options)?,
+            ObjectStoreBackend::Rest(rest) => EngineStore::plain(
+                rest_engine::build_rest_object_store(&url, &options, rest.as_ref())?,
+            ),
+        };
     build_engine_from_store(store, executor_config, io_config, allocate_error)
 }
 
-/// Assemble a default engine from a pre-built [`ObjectStore`], applying executor and read-path I/O
+/// Assemble a default engine from pre-built storage handles, applying executor and read-path I/O
 /// tuning. Shared by the URL-scheme and REST engine builder paths.
 #[cfg(feature = "default-engine-base")]
 pub(crate) fn build_engine_from_store(
-    store: Arc<dyn ObjectStore>,
+    store: impl Into<delta_kernel_default_engine::storage::EngineStore>,
     executor_config: Option<MultithreadedExecutorConfig>,
     io_config: IoConcurrencyConfig,
     allocate_error: AllocateErrorFn,
 ) -> DeltaResult<Handle<SharedExternEngine>> {
     use delta_kernel_default_engine::DefaultEngineBuilder;
 
+    let store = store.into();
+
     // The builder is generic over the executor type, so apply the shared I/O config via a generic
     // helper to both branches without naming the concrete builder type.
     fn apply_io_config<E>(
@@ -2082,7 +2083,7 @@ mod tests {
 
     use delta_kernel::object_store::memory::InMemory;
     use delta_kernel::object_store::path::Path;
-    use delta_kernel::object_store::{DynObjectStore, ObjectStoreExt as _};
+    use delta_kernel::object_store::{DynObjectStore, ObjectStore, ObjectStoreExt as _};
     use delta_kernel::schema::schema_ref;
     use delta_kernel_default_engine::executor::tokio::TokioMultiThreadExecutor;
     use delta_kernel_default_engine::DefaultEngineBuilder;
diff --git a/kernel/examples/common/src/lib.rs b/kernel/examples/common/src/lib.rs
index 9a4bb5b5fd..1562a96b27 100644
--- a/kernel/examples/common/src/lib.rs
+++ b/kernel/examples/common/src/lib.rs
@@ -8,12 +8,12 @@ use delta_kernel::arrow::array::RecordBatch;
 use delta_kernel::object_store::aws::AmazonS3Builder;
 use delta_kernel::object_store::azure::MicrosoftAzureBuilder;
 use delta_kernel::object_store::gcp::GoogleCloudStorageBuilder;
-use delta_kernel::object_store::{DynObjectStore, ObjectStoreScheme};
+use delta_kernel::object_store::ObjectStoreScheme;
 use delta_kernel::scan::Scan;
 use delta_kernel::schema::MetadataColumnSpec;
 use delta_kernel::{DeltaResult, SnapshotRef};
 use delta_kernel_default_engine::executor::tokio::TokioBackgroundExecutor;
-use delta_kernel_default_engine::storage::store_from_url_opts;
+use delta_kernel_default_engine::storage::EngineStore;
 use delta_kernel_default_engine::{DefaultEngine, DefaultEngineBuilder};
 use url::Url;
 
@@ -134,18 +134,29 @@ pub fn get_engine(
         })?;
         use ObjectStoreScheme::*;
         let url_str = url.to_string();
-        let store: Arc<DynObjectStore> = match scheme {
-            AmazonS3 => Arc::new(AmazonS3Builder::from_env().with_url(url_str).build()?),
-            GoogleCloudStorage => Arc::new(
-                GoogleCloudStorageBuilder::from_env()
-                    .with_url(url_str)
-                    .build()?,
-            ),
-            MicrosoftAzure => Arc::new(
-                MicrosoftAzureBuilder::from_env()
-                    .with_url(url_str)
-                    .build()?,
-            ),
+        let bucket = if url.scheme() == "https"
+            && url.host_str().is_some_and(|host| host.starts_with("s3."))
+        {
+            url.path_segments().and_then(|mut segments| segments.next())
+        } else {
+            url.host_str()
+        };
+        let ordered = scheme != AmazonS3
+            || !bucket.is_some_and(|bucket| bucket.contains("--x-s3") || bucket.contains("-xa-s3"));
+        macro_rules! engine_store {
+            ($builder:ty) => {{
+                let store = Arc::new(<$builder>::from_env().with_url(url_str).build()?);
+                if ordered {
+                    EngineStore::from_ordered_paginated(store)
+                } else {
+                    EngineStore::from_paginated(store)
+                }
+            }};
+        }
+        let store = match scheme {
+            AmazonS3 => engine_store!(AmazonS3Builder),
+            GoogleCloudStorage => engine_store!(GoogleCloudStorageBuilder),
+            MicrosoftAzure => engine_store!(MicrosoftAzureBuilder),
             Local | Memory | Http => {
                 return Err(delta_kernel::Error::Generic(format!(
                     "Scheme {scheme:?} doesn't support getting credentials from environment"
@@ -158,13 +169,13 @@ pub fn get_engine(
                 )));
             }
         };
-        Ok(DefaultEngineBuilder::new(Arc::new(store)).build())
+        Ok(DefaultEngineBuilder::new(store).build())
     } else if !args.option.is_empty() {
         let opts = args.option.iter().map(|option| {
             let parts: Vec<&str> = option.split("=").collect();
             (parts[0].to_ascii_lowercase(), parts[1])
         });
-        Ok(DefaultEngineBuilder::new(store_from_url_opts(url, opts)?).build())
+        Ok(DefaultEngineBuilder::new(EngineStore::from_url_opts(url, opts)?).build())
     } else {
         let mut options = if let Some(ref region) = args.region {
             HashMap::from([("region", region.clone())])
@@ -174,7 +185,7 @@ pub fn get_engine(
         if args.public {
             options.insert("skip_signature", "true".to_string());
         }
-        Ok(DefaultEngineBuilder::new(store_from_url_opts(url, options)?).build())
+        Ok(DefaultEngineBuilder::new(EngineStore::from_url_opts(url, options)?).build())
     }
 }
 
diff --git a/kernel/src/engine/sync/storage.rs b/kernel/src/engine/sync/storage.rs
index 4e175287a6..b4414028ec 100644
--- a/kernel/src/engine/sync/storage.rs
+++ b/kernel/src/engine/sync/storage.rs
@@ -34,26 +34,36 @@ impl StorageHandler for SyncStorageHandler {
     fn list_from(&self, url_path: &Url) -> DeltaResult<DeltaResultIteratorStatic<FileMeta>> {
         let (store, base_url, offset) = resolve_scope(self.store.as_ref(), url_path)?;
 
-        // For directory URLs, prefix == offset and the offset acts as a lower bound that still
-        // includes everything underneath (since `dir/a` > `dir` lexicographically). For file
-        // URLs, the prefix is the parent directory and the offset filters out files at-or-below.
+        // Directory URLs list the directory itself (prefix == offset). File URLs list the parent
+        // directory after the file. The retain below narrows the raw listing to direct children.
         let prefix = if url_path.path().ends_with('/') {
             offset.clone()
         } else {
             let mut parts: Vec<_> = offset.parts().collect();
-            parts.pop();
+            if parts.pop().is_none() {
+                return Err(Error::generic(format!(
+                    "Offset path must not be a root directory. Got: '{url_path}'"
+                )));
+            }
             Path::from_iter(parts)
         };
 
-        // LocalFileSystem and InMemory do not return sorted listings, so we collect and sort
-        // to give callers a deterministic order.
+        // `list_with_offset` recurses, so filter to direct children (one path part left after the
+        // prefix).
         let mut metas: Vec<_> = futures::executor::block_on(
             store
                 .list_with_offset(Some(&prefix), &offset)
                 .collect::<Vec<_>>(),
         )
         .into_iter()
-        .collect::<Result<_, _>>()?;
+        .collect::<Result<Vec<_>, _>>()?;
+        metas.retain(|meta| {
+            meta.location
+                .prefix_match(&prefix)
+                .is_some_and(|parts| parts.count() == 1)
+        });
+        // LocalFileSystem and InMemory do not return sorted listings, so sort for a deterministic
+        // order.
         metas.sort_unstable_by(|a, b| a.location.cmp(&b.location));
 
         let iter = metas.into_iter().map(move |meta| {
@@ -218,10 +228,9 @@ mod tests {
         Ok(())
     }
 
-    /// `list_from` against an [`ObjectStore`] must walk subdirectories, matching the local FS
-    /// implementation that uses `read_dir` recursively.
+    /// `list_from` is single-directory: files nested in subdirectories are excluded.
     #[tokio::test]
-    async fn list_from_store_is_recursive() {
+    async fn list_from_store_is_single_directory() {
         let store = std::sync::Arc::new(InMemory::new());
         store
             .put(
@@ -250,7 +259,15 @@ mod tests {
             .iter()
             .map(|f| f.location.path().to_string())
             .collect();
-        assert_eq!(names, vec!["/a/file1.json", "/a/sub/file2.json"]);
+        assert_eq!(names, vec!["/a/file1.json"]);
+    }
+
+    #[test]
+    fn list_from_root_offset_errors() {
+        let storage = SyncStorageHandler::new(Some(std::sync::Arc::new(InMemory::new())));
+        // Authority-only URL: empty path, not directory-like, so the offset has no parent.
+        let url = Url::parse("memory://bucket").unwrap();
+        assert!(matches!(storage.list_from(&url), Err(Error::Generic(_))));
     }
 
     #[test]
diff --git a/kernel/src/lib.rs b/kernel/src/lib.rs
index cf82be0ca0..d21fea0e48 100644
--- a/kernel/src/lib.rs
+++ b/kernel/src/lib.rs
@@ -546,20 +546,22 @@ pub(crate) fn create_row(
 /// file system where the Delta table is present. Connector implementation of
 /// this trait can hide filesystem specific details from Delta Kernel.
 pub trait StorageHandler: AsAny {
-    /// Recursively list files whose full path is lexicographically greater than (UTF-8 sorting)
-    /// the given `path`, restricted to descendants of `path`'s parent directory. The result must
-    /// be sorted by the full path (UTF-8 byte order).
+    /// List files directly within `path`'s parent directory whose full path is lexicographically
+    /// greater than (UTF-8 sorting) the given `path`. The result must be sorted by the full path
+    /// (UTF-8 byte order).
     ///
-    /// The listing is **recursive**: files in nested subdirectories are included, not just files
-    /// directly under the parent. For example, listing from `dir/0001.json` may return
-    /// `dir/0002.json`, `dir/sub/0003.json`, and `dir/sub/nested/0004.json`, all interleaved
-    /// in lexicographic order.
+    /// The listing is a **single directory level**: files in nested subdirectories are not
+    /// included, and subdirectories are not returned as entries. For example, listing from
+    /// `dir/0001.json` returns direct children like `dir/0002.checkpoint.parquet` and
+    /// `dir/0002.json` (in lexicographic order), but nothing under `dir/sub/`.
+    /// Delta log discovery defensively tolerates recursive descendants, but that compatibility
+    /// behavior is not part of this contract and implementations must not rely on it.
     ///
     /// The parent directory is derived from `path`:
     /// - If `path` is directory-like (ends with `/`), the parent is `path` itself and the result
-    ///   contains all files at or below that directory.
-    /// - Otherwise, the parent is the directory containing `path`, and only files (at any depth
-    ///   under that parent) whose full path sorts strictly greater than `path` are returned.
+    ///   contains the files directly within it.
+    /// - Otherwise, the parent is the directory containing `path`, and only files directly within
+    ///   that parent whose full path sorts strictly greater than `path` are returned.
     fn list_from(&self, path: &Url) -> DeltaResult<DeltaResultIteratorStatic<FileMeta>>;
 
     /// Cancellation-aware variant of [`list_from`](Self::list_from).
@@ -585,6 +587,42 @@ pub trait StorageHandler: AsAny {
         self.list_from(path)
     }
 
+    /// List direct children with full paths strictly between `start` and `end`, in UTF-8 order.
+    ///
+    /// Both bounds are exclusive. `start` determines the directory as in [`Self::list_from`];
+    /// `end` must name a path in that same directory. An empty or reversed interval returns no
+    /// files. Comparisons use percent-decoded path bytes, not URL serialization. Cancellation
+    /// follows [`Self::list_from_with_cancellation`].
+    ///
+    /// The default implementation bounds an already shallow, sorted listing. Engines backed by
+    /// ordered recursive listings should apply `end` before removing descendants, so hidden
+    /// subtrees beyond the bound do not need to be exhausted.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error for bounds in different directories, cancellation, or listing failures.
+    fn list_range(
+        &self,
+        start: &Url,
+        end: &Url,
+        cancellation_token: Option<CancellationTokenRef>,
+    ) -> DeltaResult<DeltaResultIteratorStatic<FileMeta>> {
+        check_cancelled(cancellation_token.as_ref())?;
+        path::validate_list_range(start, end)?;
+        if path::compare_listing_paths(end, start).is_le() {
+            return Ok(Box::new(std::iter::empty()));
+        }
+        let end = end.clone();
+        Ok(Box::new(
+            self.list_from_with_cancellation(start, cancellation_token)?
+                .take_while(move |result| {
+                    result.as_ref().map_or(true, |meta| {
+                        path::compare_listing_paths(&meta.location, &end).is_lt()
+                    })
+                }),
+        ))
+    }
+
     /// Read data specified by the start and end offset from the file.
     fn read_files(&self, files: Vec<FileSlice>) -> DeltaResult<DeltaResultIteratorStatic<Bytes>>;
 
diff --git a/kernel/src/log_segment_files/mod.rs b/kernel/src/log_segment_files/mod.rs
index 3ed16bb732..816a51da53 100644
--- a/kernel/src/log_segment_files/mod.rs
+++ b/kernel/src/log_segment_files/mod.rs
@@ -58,11 +58,13 @@ pub(crate) struct LogSegmentFiles {
 
 /// 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.
+/// files (e.g. dot-prefixed files), and stopping at `end_version`. The upper bound lets ordered
+/// engines stop before filtering descendants past the version-named region. Engines that do not
+/// push down the bound may still page through `_staged_commits/` and `_sidecars/` internally.
+/// This also defensively tolerates a recursive storage listing by treating its first nested path
+/// past the version-named region as the end of the relevant listing.
 ///
-/// This is a thin wrapper around [`StorageHandler::list_from`] that provides the standard
+/// This is a thin wrapper around [`StorageHandler::list_range`] 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`.
 ///
@@ -78,14 +80,17 @@ pub(crate) fn list_delta_log_from_storage(
     cancellation_token: Option<&CancellationTokenRef>,
 ) -> DeltaResult<impl Iterator<Item = DeltaResult<ParsedLogPath>>> {
     let start_from = log_root.join(&format!("{start_version:020}"))?;
+    let end_before = match end_version.checked_add(1) {
+        Some(version) => log_root.join(&format!("{version:020}"))?,
+        None => log_root.join("./:")?,
+    };
     let log_root_str = log_root.to_string();
     let files = storage
-        .list_from_with_cancellation(&start_from, cancellation_token.cloned())?
+        .list_range(&start_from, &end_before, 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
diff --git a/kernel/src/log_segment_files/tests.rs b/kernel/src/log_segment_files/tests.rs
index 82c8a198bb..6b8cc74447 100644
--- a/kernel/src/log_segment_files/tests.rs
+++ b/kernel/src/log_segment_files/tests.rs
@@ -197,6 +197,46 @@ impl StorageHandler for CountingStorageHandler {
     }
 }
 
+/// Simulates a connector whose `list_from` output includes recursive descendants.
+struct RecursiveListingHandler {
+    files: Vec<FileMeta>,
+    items_pulled: Arc<AtomicU32>,
+}
+
+impl StorageHandler for RecursiveListingHandler {
+    fn list_from(&self, _path: &Url) -> DeltaResult<DeltaResultIteratorStatic<FileMeta>> {
+        let items_pulled = self.items_pulled.clone();
+        let iter = self.files.clone().into_iter().map(move |file| {
+            items_pulled.fetch_add(1, Ordering::Relaxed);
+            Ok(file)
+        });
+        Ok(Box::new(iter))
+    }
+
+    fn read_files(
+        &self,
+        _files: Vec<crate::FileSlice>,
+    ) -> DeltaResult<DeltaResultIteratorStatic<bytes::Bytes>> {
+        panic!("read_files should not be called during listing");
+    }
+
+    fn put(&self, _path: &Url, _data: bytes::Bytes, _overwrite: bool) -> DeltaResult<()> {
+        panic!("put should not be called during listing");
+    }
+
+    fn copy_atomic(&self, _src: &Url, _dest: &Url) -> DeltaResult<()> {
+        panic!("copy_atomic should not be called during listing");
+    }
+
+    fn head(&self, _path: &Url) -> DeltaResult<FileMeta> {
+        panic!("head should not be called during listing");
+    }
+
+    fn delete(&self, _path: &Url) -> DeltaResult<()> {
+        panic!("delete should not be called during listing");
+    }
+}
+
 /// Helper to call `LogSegmentFiles::list()` and destructure the result for assertions.
 /// Returns (ascending_commit_files, ascending_compaction_files, checkpoint_parts,
 ///          latest_crc_file, latest_commit_file, max_published_version).
@@ -478,14 +518,13 @@ async fn test_listing_omits_staged_commits() {
 }
 
 #[tokio::test]
-async fn test_listing_stops_at_first_staged_commit_without_consuming_the_rest() {
+async fn test_listing_does_not_yield_staged_commits() {
     let mut log_files = vec![
         (0, LogPathFileType::Commit, CommitSource::Filesystem),
         (1, LogPathFileType::Commit, CommitSource::Filesystem),
         (2, LogPathFileType::Commit, CommitSource::Filesystem),
     ];
-    // Staged commits sort after every version-named file ('_' > '9'), so a sorted listing
-    // reaches them only after all relevant files. None should be consumed beyond the first.
+    // Staged commits sort after every version-named file ('_' > '9') but live in a subdirectory.
     log_files
         .extend((0..100).map(|v| (v, LogPathFileType::StagedCommit, CommitSource::Filesystem)));
 
@@ -498,19 +537,53 @@ async fn test_listing_stops_at_first_staged_commit_without_consuming_the_rest()
     assert_eq!(commits.len(), 3);
     assert_eq!(latest_commit.unwrap().version, 2);
     assert_eq!(max_pub, Some(2));
-    // 3 commits plus the single staged commit that stops the listing
-    assert_eq!(storage.items_listed(), 4);
+    assert_eq!(storage.items_listed(), 3);
 }
 
-// Any path past the version-named region stops the listing, not just `_staged_commits/`:
-// checkpoint sidecars under `_sidecars/` and non-underscore names whose first byte sorts
-// past '9' (e.g. 'Z'). Both sentinels sort before `_staged_commits/`, so no staged commit
-// is ever consumed.
+#[test]
+fn list_delta_log_from_storage_tolerates_recursive_results() {
+    let log_root = Url::parse("memory:///_delta_log/").unwrap();
+    let mut files: Vec<_> = [
+        "00000000000000000000.json",
+        "00000000000000000001.json",
+        "00000000000000000002.json",
+        "_sidecars/016ae953-37a9-438e-8683-9a9a4a79a395.parquet",
+        "_staged_commits/00000000000000000003.uuid.json",
+    ]
+    .into_iter()
+    .map(|path| FileMeta::new(log_root.join(path).unwrap(), 0, 1))
+    .collect();
+    files.sort_unstable();
+
+    let items_pulled = Arc::new(AtomicU32::new(0));
+    let storage = RecursiveListingHandler {
+        files,
+        items_pulled: items_pulled.clone(),
+    };
+
+    let listed: Vec<_> = list_delta_log_from_storage(&storage, &log_root, 0, Version::MAX, None)
+        .unwrap()
+        .map(Result::unwrap)
+        .collect();
+
+    assert_eq!(
+        listed.iter().map(|path| path.version).collect::<Vec<_>>(),
+        vec![0, 1, 2]
+    );
+    // The first nested path terminates the listing without pulling later recursive descendants.
+    assert_eq!(items_pulled.load(Ordering::Relaxed), 4);
+}
+
+// Nested paths are excluded by `StorageHandler`; direct paths past the version-named region stop
+// the listing.
 #[rstest]
-#[case::sidecar("_delta_log/_sidecars/016ae953-37a9-438e-8683-9a9a4a79a395.parquet")]
-#[case::non_underscore_sentinel("_delta_log/Zsentinel")]
+#[case::sidecar("_delta_log/_sidecars/016ae953-37a9-438e-8683-9a9a4a79a395.parquet", 3)]
+#[case::non_underscore_sentinel("_delta_log/Zsentinel", 4)]
 #[tokio::test]
-async fn test_listing_stops_at_first_non_version_named_path(#[case] sentinel_path: &str) {
+async fn test_listing_excludes_nested_and_stops_at_direct_non_version_path(
+    #[case] sentinel_path: &str,
+    #[case] expected_items_listed: u32,
+) {
     let mut log_files = vec![
         (0, LogPathFileType::Commit, CommitSource::Filesystem),
         (1, LogPathFileType::Commit, CommitSource::Filesystem),
@@ -527,8 +600,7 @@ async fn test_listing_stops_at_first_non_version_named_path(#[case] sentinel_pat
     assert_eq!(commits.len(), 3);
     assert_eq!(latest_commit.unwrap().version, 2);
     assert_eq!(max_pub, Some(2));
-    // 3 commits plus the sentinel that stops the listing
-    assert_eq!(storage.items_listed(), 4);
+    assert_eq!(storage.items_listed(), expected_items_listed);
 }
 
 #[tokio::test]
diff --git a/kernel/src/metrics/metered_storage.rs b/kernel/src/metrics/metered_storage.rs
index 2c67a71fd5..09997ef749 100644
--- a/kernel/src/metrics/metered_storage.rs
+++ b/kernel/src/metrics/metered_storage.rs
@@ -72,6 +72,21 @@ impl StorageHandler for MeteredStorageHandler {
         )))
     }
 
+    fn list_range(
+        &self,
+        start: &Url,
+        end: &Url,
+        cancellation_token: Option<CancellationTokenRef>,
+    ) -> DeltaResult<DeltaResultIteratorStatic<FileMeta>> {
+        let begun = Instant::now();
+        let inner = self.inner.list_range(start, end, cancellation_token)?;
+        Ok(Box::new(MetricsIterator::<_, FileMeta>::new(
+            inner,
+            StorageListCompleted::NAME,
+            begun,
+        )))
+    }
+
     fn read_files(&self, files: Vec<FileSlice>) -> DeltaResult<DeltaResultIteratorStatic<Bytes>> {
         let start = Instant::now();
         let inner = self.inner.read_files(files)?;
@@ -168,6 +183,41 @@ mod tests {
         Url::parse("memory:///_delta_log/").unwrap()
     }
 
+    #[rstest::rstest]
+    #[case("0", "%5B", vec!["1"])]
+    #[case("%5B", "%5B", vec![])]
+    #[case("%5B", "0", vec![])]
+    fn default_list_range_compares_decoded_paths(
+        #[case] start: &str,
+        #[case] end: &str,
+        #[case] expected: Vec<&str>,
+    ) {
+        let storage = StubStorageHandler {
+            list_results: vec![
+                fake_file_meta("1"),
+                fake_file_meta("%5B"),
+                fake_file_meta("z"),
+            ],
+            read_results: vec![],
+        };
+        let listed: Vec<_> = storage
+            .list_range(
+                &fake_url().join(start).unwrap(),
+                &fake_url().join(end).unwrap(),
+                None,
+            )
+            .unwrap()
+            .map(|result| result.unwrap().location)
+            .collect();
+        assert_eq!(
+            listed,
+            expected
+                .into_iter()
+                .map(|name| fake_file_meta(name).location)
+                .collect::<Vec<_>>()
+        );
+    }
+
     fn fake_file_meta(name: &str) -> FileMeta {
         FileMeta {
             location: Url::parse(&format!("memory:///_delta_log/{name}")).unwrap(),
@@ -182,8 +232,8 @@ mod tests {
         (reporter, guard)
     }
 
-    #[test]
-    fn list_from_emits_storage_list_completed() {
+    #[rstest::rstest]
+    fn listing_emits_one_storage_list_completed(#[values(false, true)] bounded: bool) {
         let (reporter, _guard) = install_capture();
         let inner: Arc<dyn StorageHandler> = Arc::new(StubStorageHandler {
             list_results: vec![
@@ -194,10 +244,26 @@ mod tests {
         });
         let storage = MeteredStorageHandler::new(inner);
 
-        let iter = storage.list_from(&fake_url()).unwrap();
+        let iter = if bounded {
+            storage.list_range(
+                &fake_url(),
+                &fake_url().join("00000000000000000001.json").unwrap(),
+                None,
+            )
+        } else {
+            storage.list_from(&fake_url())
+        }
+        .unwrap();
         let _: Vec<_> = iter.collect();
 
         let events = reporter.events();
+        assert_eq!(
+            events
+                .iter()
+                .filter(|event| matches!(event, MetricEvent::StorageListCompleted(_)))
+                .count(),
+            1
+        );
         let listed = events
             .iter()
             .find(|e| matches!(e, MetricEvent::StorageListCompleted(_)))
@@ -205,7 +271,7 @@ mod tests {
         let MetricEvent::StorageListCompleted(e) = listed else {
             unreachable!();
         };
-        assert_eq!(e.num_files, 2);
+        assert_eq!(e.num_files, if bounded { 1 } else { 2 });
     }
 
     #[test]
@@ -264,9 +330,21 @@ mod tests {
     #[derive(Default)]
     struct TokenCapturingStorageHandler {
         seen: std::sync::Mutex<Option<CancellationTokenRef>>,
+        range: std::sync::Mutex<Option<(Url, Url)>>,
     }
 
     impl StorageHandler for TokenCapturingStorageHandler {
+        fn list_range(
+            &self,
+            start: &Url,
+            end: &Url,
+            cancellation_token: Option<CancellationTokenRef>,
+        ) -> DeltaResult<DeltaResultIteratorStatic<FileMeta>> {
+            *self.seen.lock().unwrap() = cancellation_token;
+            *self.range.lock().unwrap() = Some((start.clone(), end.clone()));
+            Ok(Box::new(std::iter::empty()))
+        }
+
         fn list_from(&self, _path: &Url) -> DeltaResult<DeltaResultIteratorStatic<FileMeta>> {
             Ok(Box::new(std::iter::empty()))
         }
@@ -316,15 +394,23 @@ mod tests {
     // The wrapper forwards to the inner cancellation-aware methods and passes the token through by
     // identity, so a caller can still downcast it to recover what it supplied.
     #[rstest::rstest]
-    #[case::list(true)]
-    #[case::read(false)]
-    fn forwards_cancellation_token_by_identity(#[case] list: bool) {
+    #[case::list("list")]
+    #[case::range("range")]
+    #[case::read("read")]
+    fn forwards_cancellation_token_by_identity(#[case] operation: &str) {
         let stub = Arc::new(TokenCapturingStorageHandler::default());
         let storage = MeteredStorageHandler::new(stub.clone());
         let token: CancellationTokenRef =
             Arc::new(crate::unit_test_utils::TestCancellationToken::default());
 
-        if list {
+        if operation == "range" {
+            let end = fake_url().join("z").unwrap();
+            let iter = storage
+                .list_range(&fake_url(), &end, Some(token.clone()))
+                .unwrap();
+            let _: Vec<_> = iter.collect();
+            assert_eq!(*stub.range.lock().unwrap(), Some((fake_url(), end)));
+        } else if operation == "list" {
             let iter = storage
                 .list_from_with_cancellation(&fake_url(), Some(token.clone()))
                 .unwrap();
diff --git a/kernel/src/path.rs b/kernel/src/path.rs
index 16b6f168db..9042f02a48 100644
--- a/kernel/src/path.rs
+++ b/kernel/src/path.rs
@@ -1,9 +1,11 @@
 //! Utilities to make working with directory and file paths easier
 
+use std::cmp::Ordering;
 use std::slice;
 use std::str::FromStr;
 
 use delta_kernel_derive::internal_api;
+use percent_encoding::percent_decode_str;
 use url::Url;
 use uuid::Uuid;
 
@@ -164,6 +166,34 @@ impl AsUrl for Url {
     }
 }
 
+/// Compare `left` and `right` by their percent-decoded path bytes, ignoring other URL components.
+///
+/// Returns their lexicographic ordering without allocating decoded strings.
+#[internal_api]
+pub(crate) fn compare_listing_paths(left: &Url, right: &Url) -> Ordering {
+    percent_decode_str(left.path()).cmp(percent_decode_str(right.path()))
+}
+
+/// Validate that `end` belongs to the directory listed from `start`.
+///
+/// Query strings and fragments do not affect directory identity. Returns an error when the
+/// bounds select different directories or their parent URLs cannot be resolved.
+#[internal_api]
+pub(crate) fn validate_list_range(start: &Url, end: &Url) -> DeltaResult<()> {
+    let mut start_directory = start.join(".")?;
+    let mut end_directory = end.join(".")?;
+    start_directory.set_query(None);
+    start_directory.set_fragment(None);
+    end_directory.set_query(None);
+    end_directory.set_fragment(None);
+    if start_directory != end_directory {
+        return Err(Error::generic(
+            "Listing bounds must belong to the same directory",
+        ));
+    }
+    Ok(())
+}
+
 fn path_contains_delta_log_dir(mut path_segments: std::str::Split<'_, char>) -> bool {
     path_segments.any(|p| p == DELTA_LOG_DIR)
 }
diff --git a/kernel/src/plans/ir/operation.rs b/kernel/src/plans/ir/operation.rs
index da9ecb7ed7..a4c37d26f5 100644
--- a/kernel/src/plans/ir/operation.rs
+++ b/kernel/src/plans/ir/operation.rs
@@ -37,7 +37,7 @@ impl Operation {
 /// is documented on the variant in terms of [`PlanResult`](crate::plans::PlanResult).
 #[derive(Debug)]
 pub enum IoOperation {
-    /// Recursively list files at the given URL.
+    /// List files directly within the directory of the given URL (single level).
     ///
     /// Should return a [`PlanResult::FileMeta`](crate::plans::PlanResult::FileMeta) with one entry
     /// per file. See [`StorageHandler::list_from`] for more details on the ordering contract.