diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0c3e8b06..be5e0a27 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -23,6 +23,22 @@ tag. Releases before 2.0.0 are recorded in the
   where it used to be a `404`. The page carries the colors, the type and the card layout of the
   documentation site, and it loads nothing from the network, so it also renders on a server with
   no route out. Swagger keeps its own path, so a bookmark to `/swagger` is unaffected.
+- **`browse_datasets` reads one directory level.** A folder view needs the files and the
+  sub-directories of one directory, and got them by listing the whole store and grouping the paths
+  in the browser. `browse_datasets([prefix[, offset[, limit]]])` reads a single level with one
+  delimiter request. On a bucket of 2 853 217 objects that measured 14 milliseconds, against 79.9
+  seconds to enumerate the same bucket. It shares its provider with `list_datasets`, so the columns
+  are the same, and a new `is_directory` column marks a sub-directory row. `GET
+  /api/browse-datasets?prefix=…` runs the same SQL and splits the rows into `folders` and
+  `datasets`, and the admin UI folder view calls it. A directory-shaped dataset (Zarr, Atlas) is a
+  folder at its own level; descend into it to see the dataset. Neither listing filters by grant
+  rules, as before. See
+  [Introspection](docs/docs/2.0.0-rc5/sql/table-functions-utility.md#browse-datasets) and
+  [the REST API](docs/docs/2.0.0-rc5/api/exploring-data.md#browse-one-directory-level).
+- **`list_datasets` takes a pattern, an offset and a limit.** All three were already accepted and
+  none were documented. `list_datasets([pattern[, offset[, limit]]])` globs (default `**/*`), skips
+  and caps, and the full row shape — `can_inspect`, `can_partial_explore`, `size`, `last_modified`
+  — is now written down beside it.
 - **`BEACON_TYPE_WIDENING_ON_CONFLICT` settles a column that no type holds.** A collection can
   type one column as a number in one file and as a string in another. No type holds both, so the
   schema merge refused the whole table and the table answered no query: `Incompatible types for
@@ -169,6 +185,36 @@ tag. Releases before 2.0.0 are recorded in the
 
 ### Changed
 
+- **A large S3 dataset listing is sharded, streamed and paged.** Beacon listed the whole datasets
+  store for every browse. A recursive listing is a chain of pages, and each page needs the
+  continuation token of the one before it, so a bucket of 2 853 217 objects cost 2854 strictly
+  sequential requests and 79.9 seconds. Three changes. Beacon asks for 5000 keys per page instead
+  of the server default of 1000, which removes 80% of the round trips. It then reads three
+  directory levels with a delimiter listing and walks each leaf prefix as its own page chain, 16 at
+  a time. The same walk now takes 19.4 seconds. And the listing streams: a plan node emits rows as
+  pages arrive, so `LIMIT 50` over a bucket of millions reads one page rather than all of it, and
+  the walk no longer holds every object in memory first. Two consequences. Shards interleave, so
+  objects no longer arrive sorted — `ObjectStore::list` never guaranteed an order, and a query that
+  needs one must say `ORDER BY`. And a directory-shaped dataset (Zarr, Atlas) reports no `size` or
+  `last_modified` in a recursive listing, because those are an aggregate over the directory and a
+  streamed row cannot wait for it. Raise `AWS_TIMEOUT` before raising the page size.
+- **The listing ran during query planning, on a worker thread.** `list_datasets` did its walk
+  inside `TableFunctionImpl::call`, a synchronous method, so it reached the store through
+  `block_in_place` and held a worker thread for the length of the walk — before anything decided to
+  read the result. The walk now belongs to the table provider, and it starts when the plan is
+  executed. `EXPLAIN SELECT … FROM list_datasets()` touches the store not at all, and names the
+  node and what it will list.
+- **A discovered Zarr store is a `*.zarr` directory.** Zarr v3 gives every group *and every array*
+  a `zarr.json`, so the marker alone does not say which one it belongs to. Beacon compared each
+  marker against the others to find the shallowest, which needs the whole listing in memory and
+  cannot be done as it streams. Discovery now reads the directory that holds the marker: only a
+  store carries the `.zarr` suffix. A store in a directory without the suffix stays out of
+  `list_datasets`. `read_zarr` is unaffected — the path already names the store, so that directory
+  needs no suffix. See [Zarr](docs/docs/2.0.0-rc5/formats/zarr.md#discovery-in-the-datasets-store).
+- **A dropped object-store error no longer shortens a dataset listing.** `list_datasets` discarded
+  a failure part-way through the walk, so a timeout on object 2 000 000 of 2 850 000 returned the
+  first two million as a complete listing. It now reports the error and names how far it got.
+
 - **A file statistics pass drains the queue, and one pass runs at a time.** A pass used to stop
   after one batch of `BEACON_FILE_STATS_BATCH_FILES` files, 10 000 by default. A fresh archive of
   a million files therefore needed 100 ticks, which is over 24 hours at the default interval of 900
diff --git a/Cargo.lock b/Cargo.lock
index b48d5802..4f73079e 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -1781,6 +1781,7 @@ dependencies = [
  "arrow 58.4.0",
  "arrow-flight",
  "arrow-schema 58.4.0",
+ "async-stream",
  "async-trait",
  "base64",
  "beacon-common",
@@ -1880,6 +1881,8 @@ name = "beacon-functions"
 version = "2.0.0-rc.5"
 dependencies = [
  "arrow 58.4.0",
+ "async-stream",
+ "async-trait",
  "beacon-arrow-bbf",
  "beacon-arrow-csv",
  "beacon-arrow-geoparquet",
diff --git a/beacon-clients/beacon-ts/src/client.ts b/beacon-clients/beacon-ts/src/client.ts
index cff7069b..10fa4428 100644
--- a/beacon-clients/beacon-ts/src/client.ts
+++ b/beacon-clients/beacon-ts/src/client.ts
@@ -323,6 +323,19 @@ export class BeaconClient {
     return this.http.fetchJson<T>("GET", "/api/dataset-schema", { query: { file } });
   }
 
+  /**
+   * Reads one directory level of the datasets store (`GET /api/browse-datasets`).
+   *
+   * Prefer this over {@link datasets} for a folder view. `datasets` globs, so it
+   * enumerates every object under the pattern before it answers; this reads a
+   * single level and its cost does not grow with the store below it.
+   */
+  browseDatasets<T = unknown>(prefix?: string): Promise<T> {
+    return this.http.fetchJson<T>("GET", "/api/browse-datasets", {
+      query: { prefix },
+    });
+  }
+
   /** Counts the total number of datasets (`GET /api/total-datasets`). */
   totalDatasets(): Promise<number> {
     return this.http.fetchJson<number>("GET", "/api/total-datasets");
diff --git a/beacon-clients/beacon-ts/test/client.test.ts b/beacon-clients/beacon-ts/test/client.test.ts
index 9e77291e..7bef8e56 100644
--- a/beacon-clients/beacon-ts/test/client.test.ts
+++ b/beacon-clients/beacon-ts/test/client.test.ts
@@ -175,6 +175,21 @@ describe("BeaconClient metadata paths", () => {
     expect(calls[0]!.url).toBe("http://beacon.test/api/table-schema?table_name=my+table");
   });
 
+  it("browses one directory level, and omits an absent prefix", async () => {
+    const { fn, calls } = stubFetch(() =>
+      jsonResponse({ prefix: "argo", folders: ["2024"], datasets: [] }),
+    );
+    const client = new BeaconClient({ url: "http://beacon.test", fetch: fn });
+
+    await client.browseDatasets("argo/floats");
+    expect(calls[0]!.url).toBe("http://beacon.test/api/browse-datasets?prefix=argo%2Ffloats");
+
+    // No prefix means the root, so the parameter is left off rather than sent
+    // empty.
+    await client.browseDatasets();
+    expect(calls[1]!.url).toBe("http://beacon.test/api/browse-datasets");
+  });
+
   it("respects basePath", async () => {
     const { fn, calls } = stubFetch(() => jsonResponse([]));
     const client = new BeaconClient({ url: "http://beacon.test", basePath: "/beacon", fetch: fn });
diff --git a/beacon-clients/beacon-web/src/pages/datasets.tsx b/beacon-clients/beacon-web/src/pages/datasets.tsx
index 1c96d5a9..17937b5b 100644
--- a/beacon-clients/beacon-web/src/pages/datasets.tsx
+++ b/beacon-clients/beacon-web/src/pages/datasets.tsx
@@ -119,38 +119,11 @@ function normalize(raw: unknown): DatasetItem[] {
   });
 }
 
-interface FolderEntry {
-  name: string;
-  count: number;
-}
-
-/** Splits the datasets under `prefix` into immediate sub-folders and files. */
-function browse(items: DatasetItem[], prefix: string, filterText: string) {
-  const folderCounts = new Map<string, number>();
-  const files: DatasetItem[] = [];
-  for (const it of items) {
-    if (prefix && !it.path.startsWith(prefix)) continue;
-    const rest = it.path.slice(prefix.length);
-    if (!rest) continue;
-    const slash = rest.indexOf("/");
-    if (slash >= 0) {
-      const folder = rest.slice(0, slash);
-      folderCounts.set(folder, (folderCounts.get(folder) ?? 0) + 1);
-    } else {
-      files.push(it);
-    }
-  }
-  const f = filterText.trim().toLowerCase();
-  let folders: FolderEntry[] = [...folderCounts.entries()]
-    .map(([name, count]) => ({ name, count }))
-    .sort((a, b) => a.name.localeCompare(b.name));
-  files.sort((a, b) => a.path.localeCompare(b.path));
-  let fileList = files;
-  if (f) {
-    folders = folders.filter((x) => x.name.toLowerCase().includes(f));
-    fileList = files.filter((x) => baseName(x.path).toLowerCase().includes(f));
-  }
-  return { folders, files: fileList };
+/** `GET /api/browse-datasets` — one directory level. */
+interface BrowseResponse {
+  prefix: string;
+  folders: string[];
+  datasets: unknown[];
 }
 
 const baseName = (p: string) => p.slice(p.lastIndexOf("/") + 1);
@@ -165,7 +138,7 @@ function parentPrefix(p: string): string {
 /** One rendered row: the up (..) entry, a sub-folder, or a file. */
 type ListRow =
   | { type: "up" }
-  | { type: "folder"; name: string; count: number }
+  | { type: "folder"; name: string; count?: number }
   | { type: "file"; item: DatasetItem; label: string };
 
 /** Fixed row height (px) and shared column grid — must match between header and rows. */
@@ -267,8 +240,8 @@ export function DatasetsPage() {
   const dragDepth = React.useRef(0);
 
   const refreshDatasets = React.useCallback(() => {
+    qc.invalidateQueries({ queryKey: ["datasets-browse"] });
     qc.invalidateQueries({ queryKey: ["datasets-all"] });
-    qc.invalidateQueries({ queryKey: ["total-datasets"] });
     // An upload or a delete moves the used space, so the disk values are stale.
     qc.invalidateQueries({ queryKey: ["dataset-storage"] });
   }, [qc]);
@@ -355,14 +328,31 @@ export function DatasetsPage() {
     );
   }
 
-  const totalQuery = useQuery({ queryKey: ["total-datasets"], queryFn: () => beacon.totalDatasets() });
-  const datasetsQuery = useQuery({
+  const searching = search.trim().length > 0;
+
+  // One directory level. This is the normal path, and its cost does not grow
+  // with the store: a delimiter listing of a 2.85-million-object bucket
+  // measured 14 ms, where enumerating the same bucket took 79.9 s.
+  const browseQuery = useQuery({
+    queryKey: ["datasets-browse", path],
+    queryFn: () => beacon.browseDatasets<BrowseResponse>(path || undefined),
+    enabled: !searching,
+  });
+
+  // Search is the one view that genuinely needs every path, so it is the one
+  // view that pays for them — and only once the user has typed something.
+  const searchQuery = useQuery({
     queryKey: ["datasets-all"],
     queryFn: async () => normalize(await beacon.datasets({ limit: 100_000 })),
+    enabled: searching,
   });
 
-  const items = datasetsQuery.data ?? [];
-  const searching = search.trim().length > 0;
+  // `/api/total-datasets` enumerates the whole store to produce one number, so
+  // it is not run on mount. The badge shows what the current view holds.
+  const datasetsQuery = searching ? searchQuery : browseQuery;
+  const items = searching
+    ? searchQuery.data ?? []
+    : normalize(browseQuery.data?.datasets ?? []);
   const crumbs = path ? path.replace(/\/$/, "").split("/") : [];
 
   // Unified, virtualizable row list. A non-empty search switches to a flat list
@@ -376,15 +366,22 @@ export function DatasetsPage() {
         .sort((a, b) => compareFiles(a, b, sort))
         .map((item) => ({ type: "file", item, label: item.path }));
     }
-    const { folders, files } = browse(items, path, "");
     const list: ListRow[] = [];
     if (path) list.push({ type: "up" });
     // Folders stay grouped on top (name-ordered); files follow the active sort.
-    for (const f of folders) list.push({ type: "folder", name: f.name, count: f.count });
+    // They come from the browse response, so no client-side grouping is needed
+    // and no count is known without descending into each one.
+    const folders = (browseQuery.data?.folders ?? []).map((name) => ({ name }));
+    const files = items;
+    for (const f of folders) list.push({ type: "folder", name: f.name });
     for (const file of [...files].sort((a, b) => compareFiles(a, b, sort)))
       list.push({ type: "file", item: file, label: baseName(file.path) });
     return list;
-  }, [items, path, search, searching, sort]);
+  }, [items, path, search, searching, sort, browseQuery.data]);
+
+  // What the badge counts: the folders and files of this view. The `..` entry is
+  // navigation, not content, so it is left out.
+  const shown = rows.filter((row) => row.type !== "up").length;
 
   function toggleSort(key: SortKey) {
     setSort((s) => (s.key === key ? { key, dir: s.dir === "asc" ? "desc" : "asc" } : { key, dir: "asc" }));
@@ -443,7 +440,7 @@ export function DatasetsPage() {
             <span className="truncate">{row.name}</span>
           </span>
           <span className="text-muted-foreground">
-            {row.count} item{row.count === 1 ? "" : "s"}
+            {row.count === undefined ? "" : `${row.count} item${row.count === 1 ? "" : "s"}`}
           </span>
           <span />
           <span />
@@ -530,9 +527,7 @@ export function DatasetsPage() {
       description="Browse the datasets store. Preview rows or inspect a file's schema."
       actions={
         <div className="flex items-center gap-2">
-          {typeof totalQuery.data === "number" && (
-            <Badge variant="secondary">{totalQuery.data} total</Badge>
-          )}
+          {shown > 0 && <Badge variant="secondary">{shown} here</Badge>}
           <Button size="sm" className="gap-1.5" onClick={() => openUpload([])}>
             <Upload className="h-4 w-4" />
             Upload
diff --git a/beacon-db/beacon-core/src/crawler/engine.rs b/beacon-db/beacon-core/src/crawler/engine.rs
index cb059bcb..997240f7 100644
--- a/beacon-db/beacon-core/src/crawler/engine.rs
+++ b/beacon-db/beacon-core/src/crawler/engine.rs
@@ -1,7 +1,7 @@
 //! The crawl engine: turn a [`CrawlerDefinition`] into registered external tables.
 //!
 //! Reuses Beacon's existing primitives end-to-end:
-//! - [`beacon_functions::file_formats::list_datasets::list_datasets`] for scan +
+//! - [`beacon_functions::listing::list_datasets`] for scan +
 //!   per-format classification,
 //! - [`ExternalTableDefinition::build_provider`] for schema inference + partition
 //!   validation (the same code path used when loading persisted tables),
@@ -19,7 +19,7 @@ use beacon_datafusion_ext::table_ext::{ExternalTable, ExternalTableDefinition, T
 use datafusion::prelude::SessionContext;
 use serde::{Deserialize, Serialize};
 
-use beacon_functions::file_formats::list_datasets::list_datasets;
+use beacon_functions::listing::list_datasets;
 
 use crate::statement_plan::{upgrade_session, SessionCell};
 
diff --git a/beacon-db/beacon-core/src/file_stats.rs b/beacon-db/beacon-core/src/file_stats.rs
index a78fd3fa..34683acf 100644
--- a/beacon-db/beacon-core/src/file_stats.rs
+++ b/beacon-db/beacon-core/src/file_stats.rs
@@ -72,7 +72,7 @@ use datafusion::datasource::listing::ListingTableUrl;
 use datafusion::execution::object_store::ObjectStoreUrl;
 use datafusion::prelude::SessionContext;
 use futures::StreamExt;
-use object_store::{path::Path, ObjectMeta, ObjectStore};
+use object_store::{path::Path, ObjectMeta};
 
 use crate::statement_plan::{upgrade_session, SessionCell};
 
@@ -450,7 +450,6 @@ pub struct FileStatsService {
     store: Arc<FileStatsStore>,
     collector: StatsCollector,
     session: SessionCell,
-    datasets_url: ObjectStoreUrl,
     config: FileStatsConfig,
     /// The timer, and the startup collection when one was asked for. Both are
     /// aborted on drop.
@@ -471,11 +470,13 @@ pub struct FileStatsService {
 }
 
 impl FileStatsService {
+    /// Takes no datasets store URL. Discovery resolves the scan prefix through
+    /// the listing factory on the session, so the store it reads is the store a
+    /// query would read.
     pub fn new(
         store: Arc<FileStatsStore>,
         analyzer: Arc<dyn FileAnalyzer>,
         session: SessionCell,
-        datasets_url: ObjectStoreUrl,
         config: FileStatsConfig,
     ) -> Arc<Self> {
         let collector = StatsCollector::new(
@@ -494,7 +495,6 @@ impl FileStatsService {
             store,
             collector,
             session,
-            datasets_url,
             config,
             tasks: parking_lot::Mutex::new(Vec::new()),
             pass_lock: Arc::new(tokio::sync::Mutex::new(())),
@@ -691,16 +691,23 @@ impl FileStatsService {
 
     /// The same, restricted to a prefix. `None` uses the configured scan prefix.
     async fn discover_under(&self, prefix: Option<&str>) -> anyhow::Result<usize> {
-        let session = self.session()?;
-        let store = session
-            .state()
-            .runtime_env()
-            .object_store(&self.datasets_url)
-            .map_err(|e| anyhow::anyhow!("datasets store unavailable: {e}"))?;
+        use beacon_datafusion_ext::listing_factory::ListingFactory;
 
+        let session = self.session()?;
+        let state = session.state();
         let scan_prefix = prefix.unwrap_or(self.config.scan_prefix.as_str());
-        let prefix = (!scan_prefix.is_empty()).then(|| Path::from(scan_prefix));
-        let mut listing = store.list(prefix.as_ref());
+
+        // Through the listing factory rather than the store directly, so the
+        // scan prefix resolves by the same rules a query would use: the
+        // configured datasets store, and a glob if one is given.
+        let factory = state
+            .config()
+            .get_extension::<ListingFactory>()
+            .ok_or_else(|| anyhow::anyhow!("the listing factory is not registered"))?;
+        let mut listing = factory
+            .listing(&state, scan_prefix)
+            .map_err(|e| anyhow::anyhow!("cannot resolve the scan prefix `{scan_prefix}`: {e}"))?
+            .stream();
 
         let mut batch: Vec<ObservedFile> = Vec::with_capacity(self.config.discovery_chunk);
         let mut total = 0usize;
diff --git a/beacon-db/beacon-core/src/runtime_builder.rs b/beacon-db/beacon-core/src/runtime_builder.rs
index 5e5aef86..3518cd71 100644
--- a/beacon-db/beacon-core/src/runtime_builder.rs
+++ b/beacon-db/beacon-core/src/runtime_builder.rs
@@ -449,13 +449,12 @@ async fn init_file_stats(
 
     let analyzer = Arc::new(crate::file_stats::FormatFileAnalyzer::new(
         session_cell.clone(),
-        datasets.clone(),
+        datasets,
     ));
     let service = crate::file_stats::FileStatsService::new(
         store,
         analyzer,
         session_cell,
-        datasets,
         builder.file_stats.clone(),
     );
     service.start();
diff --git a/beacon-db/beacon-core/tests/list_datasets_fn.rs b/beacon-db/beacon-core/tests/list_datasets_fn.rs
index e4841c71..622ffc37 100644
--- a/beacon-db/beacon-core/tests/list_datasets_fn.rs
+++ b/beacon-db/beacon-core/tests/list_datasets_fn.rs
@@ -163,3 +163,158 @@ async fn listing_composes_with_sql() {
     );
     assert_eq!(csv_count, 3, "the three CSVs should be countable through SQL");
 }
+
+// ---- browse_datasets -------------------------------------------------------
+
+/// Sorted file names from a `browse_datasets` invocation.
+async fn browsed(rt: &TestRuntime, args: &str) -> Vec<String> {
+    let mut names = column_strings(
+        &rt.sql(&format!(
+            "SELECT file_name FROM browse_datasets({args}) WHERE NOT is_directory ORDER BY file_name"
+        ))
+        .await,
+        0,
+    );
+    names.sort();
+    names
+}
+
+/// The browse reads one level. The root holds `a.csv` and `p.parquet`; the CSVs
+/// under `sub/` belong to the levels below and must not appear.
+#[tokio::test(flavor = "multi_thread")]
+async fn browse_reads_one_directory_level() {
+    let rt = seeded_runtime("browse_one_level").await;
+    assert_eq!(browsed(&rt, "''").await, vec!["a.csv", "p.parquet"]);
+}
+
+/// No arguments reads the root, the same as an empty prefix. The prefix
+/// defaults to the empty string, which names the root of the store.
+#[tokio::test(flavor = "multi_thread")]
+async fn browse_without_arguments_reads_the_root() {
+    let rt = seeded_runtime("browse_no_args").await;
+    assert_eq!(browsed(&rt, "").await, browsed(&rt, "''").await);
+    assert_eq!(browsed(&rt, "").await, vec!["a.csv", "p.parquet"]);
+}
+
+/// Descending shows that level and no deeper one.
+#[tokio::test(flavor = "multi_thread")]
+async fn browse_descends_by_prefix() {
+    let rt = seeded_runtime("browse_descend").await;
+    assert_eq!(browsed(&rt, "'sub'").await, vec!["sub/b.csv"]);
+    assert_eq!(browsed(&rt, "'sub/deep'").await, vec!["sub/deep/c.csv"]);
+}
+
+/// The whole point: a browse is not a recursive listing. `list_datasets` sees
+/// every CSV under the root, a browse of the root sees one.
+#[tokio::test(flavor = "multi_thread")]
+async fn browse_is_not_a_recursive_listing() {
+    let rt = seeded_runtime("browse_vs_list").await;
+    let recursive = names(&rt, "'**/*.csv'").await;
+    assert_eq!(recursive.len(), 3, "got {recursive:?}");
+
+    let one_level: Vec<String> = browsed(&rt, "''")
+        .await
+        .into_iter()
+        .filter(|n| n.ends_with(".csv"))
+        .collect();
+    assert_eq!(one_level, vec!["a.csv"]);
+}
+
+/// An empty directory returns no rows rather than failing.
+#[tokio::test(flavor = "multi_thread")]
+async fn browse_of_an_unknown_prefix_is_empty() {
+    let rt = seeded_runtime("browse_unknown").await;
+    assert!(browsed(&rt, "'nope'").await.is_empty());
+}
+
+/// The listing is inert until it is scanned, so building the plan must not
+/// touch the store. `EXPLAIN` plans without executing.
+#[tokio::test(flavor = "multi_thread")]
+async fn planning_a_listing_does_not_run_it() {
+    let rt = seeded_runtime("browse_plan_only").await;
+    let batches = rt.sql("EXPLAIN SELECT file_name FROM list_datasets()").await;
+    assert!(total_rows(&batches) > 0, "EXPLAIN should produce a plan");
+}
+
+/// Sub-directories come back as rows, flagged, so one query describes a level.
+#[tokio::test(flavor = "multi_thread")]
+async fn browse_reports_sub_directories_as_rows() {
+    let rt = seeded_runtime("browse_dirs").await;
+    let dirs = column_strings(
+        &rt.sql(
+            "SELECT file_name FROM browse_datasets('') \
+             WHERE is_directory ORDER BY file_name",
+        )
+        .await,
+        0,
+    );
+    assert_eq!(dirs, vec!["sub"]);
+}
+
+/// A recursive listing describes files, so the flag is never set there.
+#[tokio::test(flavor = "multi_thread")]
+async fn a_recursive_listing_reports_no_directories() {
+    let rt = seeded_runtime("list_no_dirs").await;
+    let dirs = column_strings(
+        &rt.sql("SELECT file_name FROM list_datasets() WHERE is_directory")
+            .await,
+        0,
+    );
+    assert!(dirs.is_empty(), "got {dirs:?}");
+}
+
+// ---- the streaming plan ----------------------------------------------------
+
+/// The listing is its own plan node, not a materialised table. `EXPLAIN` names
+/// it, and names what it will list.
+#[tokio::test(flavor = "multi_thread")]
+async fn a_listing_plans_as_its_own_node() {
+    let rt = seeded_runtime("plan_node").await;
+    let plan = column_strings(
+        &rt.sql("EXPLAIN SELECT file_name FROM list_datasets('**/*.csv')").await,
+        1,
+    )
+    .join("\n");
+    assert!(plan.contains("DatasetsExec"), "plan was:\n{plan}");
+    assert!(plan.contains("glob=**/*.csv"), "plan was:\n{plan}");
+}
+
+/// A browse names the level it reads.
+#[tokio::test(flavor = "multi_thread")]
+async fn a_browse_plans_as_a_level() {
+    let rt = seeded_runtime("plan_level").await;
+    let plan = column_strings(
+        &rt.sql("EXPLAIN SELECT file_name FROM browse_datasets('sub')").await,
+        1,
+    )
+    .join("\n");
+    assert!(plan.contains("level=sub"), "plan was:\n{plan}");
+}
+
+/// A `LIMIT` reaches the node, which is what lets it stop the walk rather than
+/// list everything and discard the tail.
+#[tokio::test(flavor = "multi_thread")]
+async fn a_limit_reaches_the_listing_node() {
+    let rt = seeded_runtime("plan_limit").await;
+    let plan = column_strings(
+        &rt.sql("EXPLAIN SELECT file_name FROM list_datasets() LIMIT 2").await,
+        1,
+    )
+    .join("\n");
+    assert!(plan.contains("limit=2"), "plan was:\n{plan}");
+
+    // And it returns that many.
+    let rows = rt.sql("SELECT file_name FROM list_datasets() LIMIT 2").await;
+    assert_eq!(total_rows(&rows), 2);
+}
+
+/// Rows survive the trip through the stream unchanged: the same four datasets,
+/// with their sizes and timestamps still attached.
+#[tokio::test(flavor = "multi_thread")]
+async fn streamed_rows_keep_their_object_metadata() {
+    let rt = seeded_runtime("stream_meta").await;
+    let sized = rt
+        .sql("SELECT file_name FROM list_datasets() WHERE size > 0 AND last_modified IS NOT NULL")
+        .await;
+    assert_eq!(total_rows(&sized), 4, "every dataset carries its object metadata");
+}
diff --git a/beacon-db/beacon-datafusion-ext/Cargo.toml b/beacon-db/beacon-datafusion-ext/Cargo.toml
index 8711034c..6f8d4538 100644
--- a/beacon-db/beacon-datafusion-ext/Cargo.toml
+++ b/beacon-db/beacon-datafusion-ext/Cargo.toml
@@ -5,6 +5,7 @@ edition = "2024"
 
 rust-version.workspace = true
 [dependencies]
+async-stream = { workspace = true }
 beacon-file-stats = { path = "../beacon-file-stats", features = ["datafusion"] }
 # Fingerprints a format's options for the schema cache. A fixed algorithm, so a
 # stored entry keeps its meaning across releases.
diff --git a/beacon-db/beacon-datafusion-ext/src/big_page_list.rs b/beacon-db/beacon-datafusion-ext/src/big_page_list.rs
new file mode 100644
index 00000000..c6521fd3
--- /dev/null
+++ b/beacon-db/beacon-datafusion-ext/src/big_page_list.rs
@@ -0,0 +1,664 @@
+//! A store wrapper that lists a prefix in parallel shards, with sized pages.
+//!
+//! A recursive listing is a chain of pages, and each page needs the
+//! continuation token of the one before it. Nothing overlaps, so the walk costs
+//! one round trip per page however fast the server is. On a bucket of 2 853 217
+//! objects that is 2854 strictly sequential requests.
+//!
+//! Two changes, both measured against a SeaweedFS bucket of that size. Every
+//! run below returned the full 2 853 217 objects.
+//!
+//! | Strategy                          | Time       | Requests |
+//! |-----------------------------------|------------|----------|
+//! | sequential, no `max-keys` (today) | 79.9 s     | 2854     |
+//! | sequential, `max-keys=5000`       | 64.3 s     | 571      |
+//! | 2-level shards, 16 ways           | 33.8 s     | 571      |
+//! | **3-level shards, 16 ways**       | **19.4 s** | 571      |
+//! | 3-level shards, 32 ways           | 17.7 s     | 571      |
+//!
+//! # Page size
+//!
+//! [`ObjectStore::list`] sends no `max-keys`, so a server applies its own
+//! default of 1000. [`DEFAULT_MAX_KEYS`] is 5000, well under the 65535 SeaweedFS
+//! will parse.
+//!
+//! 65535 is not a server limit, and asking for it is not an error. It is a
+//! timeout risk. `ClientOptions` gives a request 30 seconds by default, and one
+//! page of 65535 keys from a filer whose metadata is not yet in page cache took
+//! 6.5 s on its own; with 16 shards in flight, some requests crossed 30 s and
+//! failed. Warm, the same walk completed with no errors at all. A moderate page
+//! stays inside the budget whatever the cache is doing, and it already removes
+//! 80% of the round trips. Raise `AWS_TIMEOUT` before raising this.
+//!
+//! # Shards
+//!
+//! [`list`](ObjectStore::list) first walks `list_with_delimiter` down
+//! [`DEFAULT_FANOUT_DEPTH`] levels, which costs one request per directory seen
+//! and finished in under a second on the bucket above. Each leaf prefix is then
+//! walked as its own page chain, [`DEFAULT_CONCURRENCY`] at a time. Objects
+//! sitting at the intermediate levels are emitted too, so nothing is missed.
+//!
+//! Expansion stops at [`MAX_SHARDS`]. A bucket that is wide rather than deep
+//! would otherwise spend more requests finding shards than the walk saves, and
+//! past a few thousand shards the throughput curve is flat anyway.
+//!
+//! # Order
+//!
+//! Shards interleave, so objects do not arrive sorted. `ObjectStore::list`
+//! documents that the order of returned `ObjectMeta` is not guaranteed, and the
+//! sequential path was only incidentally ordered. A caller that needs order
+//! must sort.
+
+use std::fmt;
+use std::sync::Arc;
+
+use futures::stream::{BoxStream, StreamExt, TryStreamExt};
+use object_store::list::{PaginatedListOptions, PaginatedListStore};
+use object_store::path::{Path, DELIMITER};
+use object_store::{
+    CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore,
+    PutMultipartOptions, PutOptions, PutPayload, PutResult, Result as OsResult,
+};
+
+/// Keys requested per page. See the module docs for why this is not larger.
+pub const DEFAULT_MAX_KEYS: usize = 5000;
+
+/// Directory levels descended to find shards before walking them.
+pub const DEFAULT_FANOUT_DEPTH: usize = 3;
+
+/// Shard walks in flight. Throughput was flat from 16 upward.
+pub const DEFAULT_CONCURRENCY: usize = 16;
+
+/// Stop expanding once a level holds this many prefixes.
+pub const MAX_SHARDS: usize = 4096;
+
+/// A store that lists a prefix as parallel shards of sized pages.
+#[derive(Debug, Clone)]
+pub struct BigPageList<T> {
+    inner: Arc<T>,
+    max_keys: usize,
+    fanout_depth: usize,
+    concurrency: usize,
+}
+
+impl<T> BigPageList<T> {
+    /// Wrap `inner` with the measured defaults.
+    pub fn new(inner: T) -> Self {
+        Self {
+            inner: Arc::new(inner),
+            max_keys: DEFAULT_MAX_KEYS,
+            fanout_depth: DEFAULT_FANOUT_DEPTH,
+            concurrency: DEFAULT_CONCURRENCY,
+        }
+    }
+
+    /// Page size per request. Clamped to at least 1.
+    pub fn with_max_keys(mut self, max_keys: usize) -> Self {
+        self.max_keys = max_keys.max(1);
+        self
+    }
+
+    /// Directory levels descended before walking. `0` disables sharding, which
+    /// leaves a sequential walk with sized pages.
+    pub fn with_fanout_depth(mut self, depth: usize) -> Self {
+        self.fanout_depth = depth;
+        self
+    }
+
+    /// Shard walks in flight. Clamped to at least 1.
+    pub fn with_concurrency(mut self, concurrency: usize) -> Self {
+        self.concurrency = concurrency.max(1);
+        self
+    }
+}
+
+impl<T: fmt::Debug> fmt::Display for BigPageList<T> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(
+            f,
+            "BigPageList(max_keys={}, depth={}, concurrency={}, {:?})",
+            self.max_keys, self.fanout_depth, self.concurrency, self.inner
+        )
+    }
+}
+
+/// A directory prefix as the raw string a paginated listing wants.
+///
+/// [`PaginatedListStore::list_paginated`] states that it adds no trailing
+/// delimiter, where [`ObjectStore::list`] does. Without one the prefix matches
+/// by byte, not by directory: a shard of `a/one` also returns `a/one.txt` from
+/// the level above it and everything under a sibling `a/one2/`. Both are
+/// duplicates, because the level above is emitted separately and the sibling is
+/// a shard of its own.
+///
+/// An empty path selects the whole store, and `/` selects nothing, so it maps to
+/// `None` as it does in `ObjectStore::list`.
+fn shard_prefix(prefix: Option<&Path>) -> Option<String> {
+    let prefix = prefix?.as_ref();
+    (!prefix.is_empty()).then(|| format!("{prefix}{DELIMITER}"))
+}
+
+/// One prefix, walked to exhaustion as a lazy chain of sized pages.
+fn walk_shard<T>(
+    inner: Arc<T>,
+    prefix: Option<Path>,
+    max_keys: usize,
+) -> BoxStream<'static, OsResult<ObjectMeta>>
+where
+    T: PaginatedListStore + 'static,
+{
+    let prefix = shard_prefix(prefix.as_ref());
+    // `None` state means the previous page carried no continuation token.
+    futures::stream::try_unfold(Some(None::<String>), move |state| {
+        let inner = Arc::clone(&inner);
+        let prefix = prefix.clone();
+        async move {
+            let Some(token) = state else {
+                return Ok::<_, object_store::Error>(None);
+            };
+            let opts = PaginatedListOptions {
+                max_keys: Some(max_keys),
+                page_token: token,
+                ..Default::default()
+            };
+            let page = inner.list_paginated(prefix.as_deref(), opts).await?;
+            Ok(Some((
+                futures::stream::iter(page.result.objects.into_iter().map(Ok)),
+                page.page_token.map(Some),
+            )))
+        }
+    })
+    .try_flatten()
+    .boxed()
+}
+
+/// Descend `depth` directory levels from `root`.
+///
+/// Returns the leaf prefixes to walk and every object found at the levels
+/// above them, which belong to the listing just as much as the leaves do.
+async fn discover_shards<T>(
+    inner: &Arc<T>,
+    root: Option<Path>,
+    depth: usize,
+    concurrency: usize,
+) -> OsResult<(Vec<Option<Path>>, Vec<ObjectMeta>)>
+where
+    T: ObjectStore + 'static,
+{
+    let mut level = vec![root];
+    let mut above = Vec::new();
+
+    for _ in 0..depth {
+        if level.len() >= MAX_SHARDS {
+            break;
+        }
+        // Clone: a level with no children is itself the leaf level, and `level`
+        // must still hold it for the walk.
+        let expanded = futures::stream::iter(level.clone().into_iter().map(|p| {
+            let inner = Arc::clone(inner);
+            async move { inner.list_with_delimiter(p.as_ref()).await }
+        }))
+        .buffer_unordered(concurrency)
+        .try_collect::<Vec<ListResult>>()
+        .await?;
+
+        let mut kids = Vec::new();
+        let mut here = Vec::new();
+        for r in expanded {
+            kids.extend(r.common_prefixes.into_iter().map(Some));
+            here.extend(r.objects);
+        }
+        // No children means this level is the leaf level, and its own walk will
+        // return these objects. Keeping them would emit each one twice.
+        if kids.is_empty() {
+            break;
+        }
+        // Descending past this level, so nothing below will cover what sits
+        // directly in it.
+        above.extend(here);
+        level = kids;
+    }
+
+    Ok((level, above))
+}
+
+#[async_trait::async_trait]
+impl<T> ObjectStore for BigPageList<T>
+where
+    T: ObjectStore + PaginatedListStore + 'static,
+{
+    async fn put_opts(
+        &self,
+        location: &Path,
+        payload: PutPayload,
+        opts: PutOptions,
+    ) -> OsResult<PutResult> {
+        self.inner.put_opts(location, payload, opts).await
+    }
+
+    async fn put_multipart_opts(
+        &self,
+        location: &Path,
+        opts: PutMultipartOptions,
+    ) -> OsResult<Box<dyn MultipartUpload>> {
+        self.inner.put_multipart_opts(location, opts).await
+    }
+
+    async fn get_opts(&self, location: &Path, options: GetOptions) -> OsResult<GetResult> {
+        self.inner.get_opts(location, options).await
+    }
+
+    fn delete_stream(
+        &self,
+        locations: BoxStream<'static, OsResult<Path>>,
+    ) -> BoxStream<'static, OsResult<Path>> {
+        self.inner.delete_stream(locations)
+    }
+
+    /// Discover shards, then walk them concurrently.
+    ///
+    /// Discovery happens inside the stream, so building it costs nothing and a
+    /// caller that never polls never talks to the store. A single shard means
+    /// the prefix has no sub-directories worth splitting, and the walk is the
+    /// plain sequential one.
+    fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, OsResult<ObjectMeta>> {
+        let inner = Arc::clone(&self.inner);
+        let prefix = prefix.cloned();
+        let (max_keys, depth, concurrency) = (self.max_keys, self.fanout_depth, self.concurrency);
+
+        futures::stream::once(async move {
+            if depth == 0 {
+                return walk_shard(inner, prefix, max_keys);
+            }
+            let (shards, above) =
+                match discover_shards(&inner, prefix.clone(), depth, concurrency).await {
+                    Ok(found) => found,
+                    // Discovery is an optimization. A store that cannot answer a
+                    // delimiter listing still lists correctly the sequential way.
+                    Err(_) => return walk_shard(inner, prefix, max_keys),
+                };
+
+            if shards.len() <= 1 {
+                return walk_shard(inner, prefix, max_keys);
+            }
+
+            let walks = futures::stream::iter(
+                shards
+                    .into_iter()
+                    .map(move |p| walk_shard(Arc::clone(&inner), p, max_keys)),
+            )
+            .flatten_unordered(Some(concurrency));
+
+            futures::stream::iter(above.into_iter().map(Ok))
+                .chain(walks)
+                .boxed()
+        })
+        .flatten()
+        .boxed()
+    }
+
+    async fn list_with_delimiter(&self, prefix: Option<&Path>) -> OsResult<ListResult> {
+        self.inner.list_with_delimiter(prefix).await
+    }
+
+    async fn copy_opts(&self, from: &Path, to: &Path, options: CopyOptions) -> OsResult<()> {
+        self.inner.copy_opts(from, to, options).await
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeSet;
+    use std::sync::Mutex;
+
+    use futures::stream::TryStreamExt;
+    use object_store::{PutMultipartOptions, PutPayload};
+
+    use super::*;
+
+    /// A store built from a list of paths, with just enough behaviour to drive a
+    /// sharded listing: paginated listing that honours `max_keys`, and a
+    /// delimiter listing that reports one directory level.
+    ///
+    /// [`PaginatedListStore::list_paginated`] matches its prefix by byte and
+    /// adds no trailing delimiter, so this one does too. A fake that split on
+    /// directories instead would accept a prefix S3 rejects, and hide what
+    /// [`shard_prefix`] is for.
+    ///
+    /// It records the page sizes it was asked for, which is how the tests below
+    /// tell a sized page from a default one.
+    #[derive(Debug)]
+    struct FakeStore {
+        paths: Vec<Path>,
+        page_sizes: Mutex<Vec<Option<usize>>>,
+    }
+
+    impl FakeStore {
+        fn new(paths: &[&str]) -> Self {
+            let mut paths: Vec<Path> = paths.iter().map(|p| Path::from(*p)).collect();
+            paths.sort();
+            Self {
+                paths,
+                page_sizes: Mutex::new(Vec::new()),
+            }
+        }
+
+        fn meta(location: &Path) -> ObjectMeta {
+            ObjectMeta {
+                location: location.clone(),
+                last_modified: Default::default(),
+                size: 1,
+                e_tag: None,
+                version: None,
+            }
+        }
+
+        fn page_sizes(&self) -> Vec<Option<usize>> {
+            self.page_sizes.lock().unwrap().clone()
+        }
+    }
+
+    impl std::fmt::Display for FakeStore {
+        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+            write!(f, "FakeStore")
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl PaginatedListStore for FakeStore {
+        async fn list_paginated(
+            &self,
+            prefix: Option<&str>,
+            opts: PaginatedListOptions,
+        ) -> object_store::Result<object_store::list::PaginatedListResult> {
+            self.page_sizes.lock().unwrap().push(opts.max_keys);
+
+            let under: Vec<&Path> = self
+                .paths
+                .iter()
+                .filter(|p| match prefix {
+                    None => true,
+                    Some(prefix) => p.as_ref().starts_with(prefix),
+                })
+                .collect();
+
+            // The page token is the index of the next path to return.
+            let start: usize = opts
+                .page_token
+                .as_deref()
+                .map(|t| t.parse().unwrap())
+                .unwrap_or(0);
+            let size = opts.max_keys.unwrap_or(1000);
+            let end = (start + size).min(under.len());
+            let objects: Vec<ObjectMeta> = under[start..end].iter().map(|p| Self::meta(p)).collect();
+
+            Ok(object_store::list::PaginatedListResult {
+                result: ListResult {
+                    common_prefixes: Vec::new(),
+                    objects,
+                },
+                page_token: (end < under.len()).then(|| end.to_string()),
+            })
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl ObjectStore for FakeStore {
+        async fn put_opts(
+            &self,
+            _l: &Path,
+            _p: PutPayload,
+            _o: PutOptions,
+        ) -> OsResult<PutResult> {
+            unimplemented!("the tests only list")
+        }
+        async fn put_multipart_opts(
+            &self,
+            _l: &Path,
+            _o: PutMultipartOptions,
+        ) -> OsResult<Box<dyn MultipartUpload>> {
+            unimplemented!("the tests only list")
+        }
+        async fn get_opts(&self, _l: &Path, _o: GetOptions) -> OsResult<GetResult> {
+            unimplemented!("the tests only list")
+        }
+        fn delete_stream(
+            &self,
+            _l: BoxStream<'static, OsResult<Path>>,
+        ) -> BoxStream<'static, OsResult<Path>> {
+            unimplemented!("the tests only list")
+        }
+        fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, OsResult<ObjectMeta>> {
+            let prefix = prefix.map(|p| p.as_ref().to_string()).unwrap_or_default();
+            let objects: Vec<OsResult<ObjectMeta>> = self
+                .paths
+                .iter()
+                .filter(|p| prefix.is_empty() || p.as_ref().starts_with(&prefix))
+                .map(|p| Ok(Self::meta(p)))
+                .collect();
+            futures::stream::iter(objects).boxed()
+        }
+        async fn list_with_delimiter(&self, prefix: Option<&Path>) -> OsResult<ListResult> {
+            let base = prefix.map(|p| p.as_ref().to_string()).unwrap_or_default();
+            let mut common = BTreeSet::new();
+            let mut objects = Vec::new();
+            for path in &self.paths {
+                let full = path.as_ref();
+                let rest = if base.is_empty() {
+                    Some(full)
+                } else if full.starts_with(&base) && full.as_bytes().get(base.len()) == Some(&b'/') {
+                    Some(&full[base.len() + 1..])
+                } else {
+                    None
+                };
+                let Some(rest) = rest else { continue };
+                match rest.split_once('/') {
+                    Some((dir, _)) => {
+                        let joined = if base.is_empty() {
+                            dir.to_string()
+                        } else {
+                            format!("{base}/{dir}")
+                        };
+                        common.insert(Path::from(joined));
+                    }
+                    None => objects.push(Self::meta(path)),
+                }
+            }
+            Ok(ListResult {
+                common_prefixes: common.into_iter().collect(),
+                objects,
+            })
+        }
+        async fn copy_opts(&self, _f: &Path, _t: &Path, _o: CopyOptions) -> OsResult<()> {
+            unimplemented!("the tests only list")
+        }
+    }
+
+    /// A tree two levels deep, with an object at each level above the leaves so
+    /// the walk has something to miss if it only reads the leaves.
+    fn tree() -> Vec<&'static str> {
+        vec![
+            "top.txt",
+            "a/mid.txt",
+            "a/one/x1.txt",
+            "a/one/x2.txt",
+            "a/two/y1.txt",
+            "b/one/z1.txt",
+            "b/two/z2.txt",
+        ]
+    }
+
+    async fn listed(store: BigPageList<FakeStore>) -> Vec<String> {
+        let mut paths: Vec<String> = store
+            .list(None)
+            .map_ok(|meta| meta.location.to_string())
+            .try_collect::<Vec<_>>()
+            .await
+            .expect("the walk succeeds");
+        paths.sort();
+        paths
+    }
+
+    /// Every object comes back exactly once, including the ones above the leaf
+    /// directories the shards are taken from.
+    #[tokio::test]
+    async fn a_sharded_walk_returns_the_whole_tree() {
+        let expected: Vec<String> = {
+            let mut e: Vec<String> = tree().iter().map(|p| p.to_string()).collect();
+            e.sort();
+            e
+        };
+        let got = listed(BigPageList::new(FakeStore::new(&tree()))).await;
+        assert_eq!(got, expected);
+    }
+
+    /// Depth 0 turns sharding off. The result must not change.
+    #[tokio::test]
+    async fn a_sequential_walk_returns_the_same_tree() {
+        let sharded = listed(BigPageList::new(FakeStore::new(&tree()))).await;
+        let sequential =
+            listed(BigPageList::new(FakeStore::new(&tree())).with_fanout_depth(0)).await;
+        assert_eq!(sharded, sequential);
+    }
+
+    /// The page size reaches the store. Without it a server applies its own
+    /// default, which is the round-trip cost this type exists to remove.
+    #[tokio::test]
+    async fn the_page_size_reaches_the_store() {
+        let store = Arc::new(FakeStore::new(&tree()));
+        let listing = BigPageList {
+            inner: Arc::clone(&store),
+            max_keys: 3,
+            fanout_depth: 0,
+            concurrency: 1,
+        };
+        let mut walked = listing.list(None);
+        while walked.try_next().await.expect("the walk succeeds").is_some() {}
+
+        let sizes = store.page_sizes();
+        assert!(!sizes.is_empty(), "the store was asked for at least one page");
+        assert!(
+            sizes.iter().all(|s| *s == Some(3)),
+            "every page asked for 3 keys, got {sizes:?}"
+        );
+    }
+
+    /// A page size of zero would ask for nothing forever, so it is clamped.
+    #[tokio::test]
+    async fn a_page_size_below_one_is_clamped() {
+        let store = BigPageList::new(FakeStore::new(&tree())).with_max_keys(0);
+        assert_eq!(store.max_keys, 1);
+        // And it still terminates, one key at a time.
+        assert_eq!(listed(store).await.len(), tree().len());
+    }
+
+    /// Concurrency is clamped for the same reason: zero shards in flight would
+    /// never finish.
+    #[test]
+    fn concurrency_below_one_is_clamped() {
+        let store = BigPageList::new(FakeStore::new(&tree())).with_concurrency(0);
+        assert_eq!(store.concurrency, 1);
+    }
+
+    /// A prefix with no sub-directories has one shard, which is the plain
+    /// sequential walk. It must still return that directory's objects.
+    #[tokio::test]
+    async fn a_flat_prefix_walks_sequentially() {
+        let store = BigPageList::new(FakeStore::new(&["only/a.txt", "only/b.txt"]));
+        let mut paths: Vec<String> = store
+            .list(Some(&Path::from("only")))
+            .map_ok(|meta| meta.location.to_string())
+            .try_collect::<Vec<_>>()
+            .await
+            .expect("the walk succeeds");
+        paths.sort();
+        assert_eq!(paths, vec!["only/a.txt", "only/b.txt"]);
+    }
+
+    /// The delimiter listing is handed through untouched: it is one request, and
+    /// sharding it would mean sharding the answer it already gives.
+    #[tokio::test]
+    async fn a_delimiter_listing_passes_through() {
+        let store = BigPageList::new(FakeStore::new(&tree()));
+        let level = store.list_with_delimiter(None).await.expect("one level");
+        let dirs: Vec<String> = level
+            .common_prefixes
+            .iter()
+            .map(|p| p.to_string())
+            .collect();
+        assert_eq!(dirs, vec!["a", "b"]);
+        let files: Vec<String> = level.objects.iter().map(|o| o.location.to_string()).collect();
+        assert_eq!(files, vec!["top.txt"]);
+    }
+
+    /// No object appears twice. The leaf level's own walk covers what sits in
+    /// it, so a level that finds no children must not also contribute its
+    /// objects to the ones gathered above.
+    #[tokio::test]
+    async fn a_sharded_walk_repeats_nothing() {
+        let got = listed(BigPageList::new(FakeStore::new(&tree()))).await;
+        let unique: BTreeSet<&String> = got.iter().collect();
+        assert_eq!(got.len(), unique.len(), "duplicates in {got:?}");
+    }
+
+    /// An empty store yields nothing rather than hanging or erroring.
+    #[tokio::test]
+    async fn an_empty_store_yields_nothing() {
+        let store = BigPageList::new(FakeStore::new(&[]));
+        assert!(listed(store).await.is_empty());
+    }
+
+    /// A tree where names share a prefix: a directory `one`, a sibling
+    /// directory `one2`, and a file `one.txt` beside them.
+    fn shared_prefix_tree() -> Vec<&'static str> {
+        vec![
+            "a/one.txt",
+            "a/one/x.txt",
+            "a/one2/y.txt",
+            "a/onemore/z.txt",
+        ]
+    }
+
+    /// A shard is a directory, not a byte prefix.
+    ///
+    /// The shard of `a/one` must not also return `a/one.txt`, which the level
+    /// above already emitted, nor `a/one2/y.txt`, which the shard of `a/one2`
+    /// returns. Both come back twice when the trailing delimiter is missing.
+    #[tokio::test]
+    async fn a_shard_does_not_reach_a_sibling_that_shares_its_name() {
+        let got = listed(BigPageList::new(FakeStore::new(&shared_prefix_tree()))).await;
+        let mut expected: Vec<String> = shared_prefix_tree()
+            .iter()
+            .map(|p| p.to_string())
+            .collect();
+        expected.sort();
+        assert_eq!(got, expected, "every object exactly once");
+    }
+
+    /// The same rule for the prefix a caller asks for. Listing `a/one` names
+    /// that directory, so a sibling `a/one2` is not part of the answer.
+    #[tokio::test]
+    async fn a_listed_prefix_is_a_directory_not_a_byte_prefix() {
+        let store = BigPageList::new(FakeStore::new(&shared_prefix_tree()));
+        let mut paths: Vec<String> = store
+            .list(Some(&Path::from("a/one")))
+            .map_ok(|meta| meta.location.to_string())
+            .try_collect::<Vec<_>>()
+            .await
+            .expect("the walk succeeds");
+        paths.sort();
+        assert_eq!(paths, vec!["a/one/x.txt"]);
+    }
+
+    /// An empty prefix selects the whole store, so it reaches the store as
+    /// `None` rather than as a bare `/` that matches nothing.
+    #[test]
+    fn an_empty_prefix_selects_the_whole_store() {
+        assert_eq!(shard_prefix(None), None);
+        assert_eq!(shard_prefix(Some(&Path::from(""))), None);
+        assert_eq!(
+            shard_prefix(Some(&Path::from("a/one"))),
+            Some("a/one/".to_string())
+        );
+    }
+}
diff --git a/beacon-db/beacon-datafusion-ext/src/format_ext.rs b/beacon-db/beacon-datafusion-ext/src/format_ext.rs
index 764f5878..6da6b3be 100644
--- a/beacon-db/beacon-datafusion-ext/src/format_ext.rs
+++ b/beacon-db/beacon-datafusion-ext/src/format_ext.rs
@@ -18,6 +18,29 @@ pub trait FileFormatFactoryExt: FileFormatFactory + Send + Sync {
         &self,
         objects: &[ObjectMeta],
     ) -> datafusion::error::Result<Vec<DatasetMetadata>>;
+
+    /// Whether this format claims `object`, judged from that object alone.
+    ///
+    /// Every format decides per object: the file formats read an extension, and
+    /// Zarr reads the directory holding its marker. So the default asks
+    /// [`Self::discover_datasets`] about a listing of one, and a format needs to
+    /// override this only if it ever wants to see its objects together.
+    ///
+    /// This is what lets a listing classify as it streams, instead of holding
+    /// every object first. The size and timestamp come from `object` here, where
+    /// `discover_datasets` leaves them for a later pass over the whole listing.
+    fn classify_object(&self, object: &ObjectMeta) -> Option<DatasetMetadata> {
+        let mut found = self
+            .discover_datasets(std::slice::from_ref(object))
+            .ok()?
+            .into_iter()
+            .next()?;
+        if found.file_path == object.location.as_ref() {
+            found.size = Some(object.size);
+            found.last_modified = Some(object.last_modified);
+        }
+        Some(found)
+    }
     fn file_format_name(&self) -> String;
     fn list_with_file_extension(&self) -> bool {
         true
diff --git a/beacon-db/beacon-datafusion-ext/src/lib.rs b/beacon-db/beacon-datafusion-ext/src/lib.rs
index 62b4f41c..eef3a312 100644
--- a/beacon-db/beacon-datafusion-ext/src/lib.rs
+++ b/beacon-db/beacon-datafusion-ext/src/lib.rs
@@ -1,4 +1,5 @@
 pub mod analyzer_rules;
+pub mod big_page_list;
 pub mod consts;
 pub mod fast_object;
 pub mod format_ext;
diff --git a/beacon-db/beacon-datafusion-ext/src/listing_factory.rs b/beacon-db/beacon-datafusion-ext/src/listing_factory.rs
index 0dfd1688..42341033 100644
--- a/beacon-db/beacon-datafusion-ext/src/listing_factory.rs
+++ b/beacon-db/beacon-datafusion-ext/src/listing_factory.rs
@@ -1,5 +1,8 @@
 use std::{path::PathBuf, sync::Arc};
 
+use futures::stream::StreamExt;
+use object_store::ObjectMeta;
+
 use datafusion::{
     catalog::Session, datasource::listing::ListingTableUrl, execution::object_store::ObjectStoreUrl,
 };
@@ -206,9 +209,17 @@ impl ListingFactory {
         let mut objects = Vec::new();
         let mut entry_stream = listing_url.list_all_files(session, &store, "").await?;
         while let Some(entry) = entry_stream.next().await {
-            if let Ok(entry) = entry {
-                objects.push(entry);
-            }
+            // Propagate. Discarding the error here turned a transient object-store
+            // failure part-way through a walk into a short dataset list that looked
+            // complete: a timeout on object 2 000 000 of 2 850 000 reported success
+            // and lost the rest without a word.
+            let entry = entry.map_err(|e| {
+                DataFusionError::Execution(format!(
+                    "list_datasets: listing `{glob_path}` failed after {} objects: {e}",
+                    objects.len()
+                ))
+            })?;
+            objects.push(entry);
         }
 
         // Ask each file format which objects it owns and how to interpret them.
@@ -221,6 +232,167 @@ impl ListingFactory {
 
         Ok(datasets)
     }
+
+    /// Resolve `glob_path` into the listing it names.
+    ///
+    /// The only half of a listing that needs a session. [`ObjectListing`] holds
+    /// what the walk needs and can be read as many times as a caller wants, so a
+    /// plan resolves once here and reads at execute time.
+    pub fn listing(
+        &self,
+        session: &dyn Session,
+        glob_path: &str,
+    ) -> datafusion::error::Result<ObjectListing> {
+        use datafusion::error::DataFusionError;
+
+        let url = self.parse_listing_table_url(session, glob_path)?;
+        let store_url = url.object_store();
+        let store = session
+            .runtime_env()
+            .object_store(store_url.clone())
+            .map_err(|e| {
+                DataFusionError::Execution(format!(
+                    "listing: failed to get object store for {store_url}: {e}"
+                ))
+            })?;
+        Ok(ObjectListing { store, url })
+    }
+}
+
+/// A resolved listing: a store, and the URL selecting objects within it.
+///
+/// Holds no session, so it is `'static` and readable more than once. Path
+/// resolution happened when [`ListingFactory::listing`] built it, which is what
+/// lets every reader below inherit the same rules — the configured default
+/// store, a schemed path, a local directory, and the glob.
+#[derive(Debug, Clone)]
+pub struct ObjectListing {
+    store: Arc<dyn object_store::ObjectStore>,
+    url: ListingTableUrl,
+}
+
+impl ObjectListing {
+    /// The store the objects live in.
+    pub fn store(&self) -> &Arc<dyn object_store::ObjectStore> {
+        &self.store
+    }
+
+    /// The directory this listing addresses, relative to the store root. For a
+    /// glob that is the literal head, the part before the first wildcard.
+    pub fn prefix(&self) -> &object_store::path::Path {
+        self.url.prefix()
+    }
+
+    /// Every object the URL matches, as pages arrive.
+    ///
+    /// Yields each object as its page arrives and holds none of them, where
+    /// [`ListingFactory::list_datasets`] drains the whole walk into a `Vec`
+    /// first — about a gigabyte for a listing of 2 853 217 objects.
+    ///
+    /// Objects, not datasets: walking a store and deciding what a file *is* are
+    /// different jobs, and only the second needs to know about formats. A caller
+    /// that wants datasets classifies the stream itself.
+    ///
+    /// Stopping the stream stops the walk.
+    pub fn stream(&self) -> futures::stream::BoxStream<'static, datafusion::error::Result<ObjectMeta>> {
+        use datafusion::error::DataFusionError;
+        use object_store::ObjectStoreExt;
+
+        let store = Arc::clone(&self.store);
+        let url = self.url.clone();
+        let prefix = url.prefix().clone();
+        async_stream::try_stream! {
+            // A URL with no glob and no trailing slash names one object, not a
+            // directory, and a store lists a prefix at segment boundaries — so
+            // listing `obs/a.parquet` looks for a directory of that name and
+            // finds nothing. Ask for the object itself, and fall back to listing
+            // when it turns out to be a directory after all.
+            if !url.is_collection() {
+                match store.head(&prefix).await {
+                    Ok(meta) => {
+                        yield meta;
+                        return;
+                    }
+                    Err(object_store::Error::NotFound { .. }) => {}
+                    Err(e) => Err(DataFusionError::Execution(format!(
+                        "listing `{prefix}` failed: {e}"
+                    )))?,
+                }
+            }
+            let mut objects = store.list(Some(&prefix));
+            while let Some(object) = objects.next().await {
+                let object = object.map_err(|e| {
+                    DataFusionError::Execution(format!(
+                        "listing `{prefix}` failed part-way: {e}"
+                    ))
+                })?;
+                // The prefix is only the literal head of the glob, so the rest of
+                // the pattern is applied here, as a listing table does.
+                if url.contains(&object.location, false) {
+                    yield object;
+                }
+            }
+        }
+        .boxed()
+    }
+
+    /// One directory level: the sub-folder names, and the objects directly in it.
+    ///
+    /// A single delimiter request rather than a walk. The difference is not a
+    /// constant factor: against a SeaweedFS bucket of 2 853 217 objects the
+    /// recursive walk took 79.9 s and this took 14 ms.
+    ///
+    /// Folder names are relative to [`Self::prefix`], so a caller descends by
+    /// joining rather than by parsing. A directory-shaped dataset (Zarr) keeps
+    /// its marker *inside* its directory, so at this level it is a folder;
+    /// descending shows the marker. Naming it here would cost one request per
+    /// sub-folder, which is the recursive cost this avoids.
+    pub async fn level(&self) -> datafusion::error::Result<ObjectLevel> {
+        use datafusion::error::DataFusionError;
+
+        let base = self.prefix().clone();
+        let listed = self
+            .store
+            .list_with_delimiter(Some(&base))
+            .await
+            .map_err(|e| {
+                DataFusionError::Execution(format!("listing level `{base}` failed: {e}"))
+            })?;
+
+        let base_str = base.as_ref().to_string();
+        let mut folders: Vec<String> = listed
+            .common_prefixes
+            .iter()
+            .map(|p| {
+                let full = p.as_ref();
+                full.strip_prefix(&base_str)
+                    .map(|r| r.trim_start_matches('/').to_string())
+                    .unwrap_or_else(|| full.to_string())
+            })
+            .filter(|name| !name.is_empty())
+            .collect();
+        folders.sort();
+
+        let mut objects = listed.objects;
+        objects.sort_by(|a, b| a.location.cmp(&b.location));
+
+        Ok(ObjectLevel {
+            prefix: base_str,
+            folders,
+            objects,
+        })
+    }
+}
+
+/// One directory level of a store.
+#[derive(Debug, Clone)]
+pub struct ObjectLevel {
+    /// The directory this describes, relative to the store root.
+    pub prefix: String,
+    /// Immediate sub-folder names, relative to `prefix`, sorted.
+    pub folders: Vec<String>,
+    /// Objects sitting directly in `prefix`, sorted by path.
+    pub objects: Vec<ObjectMeta>,
 }
 
 /// Fill each dataset's `size` + `last_modified` from the object listing.
@@ -268,6 +440,140 @@ mod tests {
 
     use super::*;
 
+    // ---- ObjectListing ----------------------------------------------------
+
+    /// A session over a local directory holding `files`, and the dynamic factory
+    /// that resolves paths against it.
+    fn local_listing(files: &[(&str, &str)]) -> (tempfile::TempDir, SessionContext, ListingFactory) {
+        let dir = tempfile::tempdir().expect("temp dir");
+        for (path, body) in files {
+            let full = dir.path().join(path);
+            std::fs::create_dir_all(full.parent().unwrap()).unwrap();
+            std::fs::write(full, body).unwrap();
+        }
+        (dir, SessionContext::new(), ListingFactory::dynamic())
+    }
+
+    /// Paths the listing yields, relative to `root`, sorted.
+    async fn streamed(listing: &ObjectListing, root: &std::path::Path) -> Vec<String> {
+        use futures::stream::TryStreamExt;
+        // Anchor on the temp directory name rather than the whole root:
+        // `canonicalize` yields a verbatim prefix on Windows that the object
+        // path does not carry.
+        let anchor = format!("{}/", root.file_name().unwrap().to_string_lossy());
+        let mut paths: Vec<String> = listing
+            .stream()
+            .map_ok(|meta| {
+                let full = meta.location.as_ref().to_string();
+                match full.split_once(&anchor) {
+                    Some((_, rest)) => rest.to_string(),
+                    None => full,
+                }
+            })
+            .try_collect::<Vec<_>>()
+            .await
+            .expect("the walk succeeds");
+        paths.sort();
+        paths
+    }
+
+    /// A directory streams everything under it.
+    #[tokio::test]
+    async fn a_directory_streams_its_subtree() {
+        let (dir, ctx, factory) =
+            local_listing(&[("a.csv", "x"), ("sub/b.csv", "y"), ("sub/deep/c.csv", "z")]);
+        let root = std::fs::canonicalize(dir.path()).unwrap();
+        let listing = factory
+            .listing(&ctx.state(), &format!("{}/", root.display()))
+            .expect("the directory resolves");
+
+        assert_eq!(
+            streamed(&listing, &root).await,
+            vec!["a.csv", "sub/b.csv", "sub/deep/c.csv"]
+        );
+    }
+    /// A glob narrows the stream by extension, and crosses directories while it
+    /// does.
+    ///
+    /// `*` does not stop at a separator here: DataFusion matches listing globs
+    /// with the default `MatchOptions`, where `require_literal_separator` is
+    /// false. That is why one directory level needs its own function rather than
+    /// a cleverer pattern — see `browse_datasets`.
+    #[tokio::test]
+    async fn a_glob_narrows_the_stream_across_directories() {
+        let (dir, ctx, factory) =
+            local_listing(&[("a.csv", "x"), ("a.txt", "y"), ("sub/b.csv", "z")]);
+        let root = std::fs::canonicalize(dir.path()).unwrap();
+        let listing = factory
+            .listing(&ctx.state(), &format!("{}/*.csv", root.display()))
+            .expect("the glob resolves");
+
+        // The `.txt` is excluded, and the nested `.csv` is not.
+        assert_eq!(streamed(&listing, &root).await, vec!["a.csv", "sub/b.csv"]);
+    }
+
+    /// A path naming one file yields that file. A store lists a prefix at
+    /// segment boundaries, so listing `a.csv` would look for a directory of that
+    /// name and find nothing.
+    #[tokio::test]
+    async fn a_single_file_yields_itself() {
+        let (dir, ctx, factory) = local_listing(&[("a.csv", "x"), ("b.csv", "y")]);
+        let root = std::fs::canonicalize(dir.path()).unwrap();
+        let listing = factory
+            .listing(&ctx.state(), &format!("{}/a.csv", root.display()))
+            .expect("the file resolves");
+
+        assert_eq!(streamed(&listing, &root).await, vec!["a.csv"]);
+    }
+
+    /// A path that matches nothing is empty, not an error.
+    #[tokio::test]
+    async fn a_missing_path_streams_nothing() {
+        let (dir, ctx, factory) = local_listing(&[("a.csv", "x")]);
+        let root = std::fs::canonicalize(dir.path()).unwrap();
+        let listing = factory
+            .listing(&ctx.state(), &format!("{}/nope.csv", root.display()))
+            .expect("the path resolves");
+
+        assert!(streamed(&listing, &root).await.is_empty());
+    }
+
+    /// One level reports its sub-folders by name and its own objects, and stops.
+    #[tokio::test]
+    async fn a_level_reads_one_directory() {
+        let (dir, ctx, factory) =
+            local_listing(&[("a.csv", "x"), ("sub/b.csv", "y"), ("sub/deep/c.csv", "z")]);
+        let root = std::fs::canonicalize(dir.path()).unwrap();
+        let listing = factory
+            .listing(&ctx.state(), &format!("{}/", root.display()))
+            .expect("the directory resolves");
+
+        let level = listing.level().await.expect("one level");
+        assert_eq!(level.folders, vec!["sub"], "folders are named, not pathed");
+        let files: Vec<String> = level
+            .objects
+            .iter()
+            .map(|o| o.location.filename().unwrap_or_default().to_string())
+            .collect();
+        assert_eq!(files, vec!["a.csv"], "nothing from below this level");
+    }
+
+    /// The same listing reads more than once: a plan resolves it during `scan`
+    /// and reads it again for every execution.
+    #[tokio::test]
+    async fn a_listing_reads_more_than_once() {
+        let (dir, ctx, factory) = local_listing(&[("a.csv", "x"), ("b.csv", "y")]);
+        let root = std::fs::canonicalize(dir.path()).unwrap();
+        let listing = factory
+            .listing(&ctx.state(), &format!("{}/", root.display()))
+            .expect("the directory resolves");
+
+        let first = streamed(&listing, &root).await;
+        let second = streamed(&listing, &root).await;
+        assert_eq!(first, second);
+        assert_eq!(first.len(), 2);
+    }
+
     /// A configured factory whose default store maps to `root`. The store URL is
     /// irrelevant to the path-resolution these tests exercise (that reads only the
     /// root), so any valid URL will do.
diff --git a/beacon-db/beacon-datafusion-ext/src/listing_url_resolver.rs b/beacon-db/beacon-datafusion-ext/src/listing_url_resolver.rs
index b315f005..25759ac3 100644
--- a/beacon-db/beacon-datafusion-ext/src/listing_url_resolver.rs
+++ b/beacon-db/beacon-datafusion-ext/src/listing_url_resolver.rs
@@ -111,8 +111,17 @@ pub fn parse_listing_table_url(
         }
         // No default + no scheme → a local filesystem path.
         (None, None) => {
-            let abs = std::path::absolute(base_path.as_path()).map_err(|e| {
-                exec_datafusion_err!("failed to absolutize path {}: {e}", base_path.display())
+            // An empty path names the root of the store, which for a local store
+            // is the current directory. `std::path::absolute` refuses an empty
+            // path, so name the directory the way the glob branch already does.
+            // `browse_datasets()` arrives here: its default prefix is the root.
+            let base = if base_path.as_os_str().is_empty() {
+                std::path::Path::new(".")
+            } else {
+                base_path.as_path()
+            };
+            let abs = std::path::absolute(base).map_err(|e| {
+                exec_datafusion_err!("failed to absolutize path {}: {e}", base.display())
             })?;
             let file_url = Url::from_file_path(&abs).map_err(|_| {
                 exec_datafusion_err!("failed to build file URL for {}", abs.display())
@@ -611,4 +620,32 @@ mod tests {
         let url = Url::parse(store_url.as_str()).unwrap();
         assert_eq!(url.scheme(), "datasets");
     }
+
+    /// An empty path names the root of the configured store, which is the whole
+    /// store. `browse_datasets()` asks for exactly this.
+    #[test]
+    fn an_empty_path_is_the_root_of_the_configured_store() {
+        let registry = registry_with_datasets();
+        let store_url = ObjectStoreUrl::parse("datasets://").unwrap();
+        let url = parse_listing_table_url(Some(store_url), "", &registry).unwrap();
+
+        assert_eq!(url.prefix().as_ref(), "", "the whole store");
+        assert!(url.is_collection(), "a directory, not one object");
+        assert!(url.get_glob().is_none());
+    }
+
+    /// The same path in dynamic mode names the current directory, the root of
+    /// the local store. `std::path::absolute` refuses an empty path, so this
+    /// used to fail rather than list the directory.
+    #[test]
+    fn an_empty_path_is_the_current_directory_in_dynamic_mode() {
+        let registry = DefaultObjectStoreRegistry::new();
+        let url = parse_listing_table_url(None, "", &registry).unwrap();
+
+        assert!(url.as_str().starts_with("file:///"), "url={}", url.as_str());
+        assert!(url.is_collection(), "a directory, not one object");
+        // The same directory a lone glob resolves against.
+        let dot = parse_listing_table_url(None, ".", &registry).unwrap();
+        assert_eq!(url.prefix(), dot.prefix());
+    }
 }
diff --git a/beacon-db/beacon-datafusion-ext/src/object_store_registry.rs b/beacon-db/beacon-datafusion-ext/src/object_store_registry.rs
index 883ddb7c..632848b0 100644
--- a/beacon-db/beacon-datafusion-ext/src/object_store_registry.rs
+++ b/beacon-db/beacon-datafusion-ext/src/object_store_registry.rs
@@ -121,11 +121,14 @@ pub(crate) fn build_object_store(
                     }
                 }
             }
-            Arc::new(
+            // Large listing pages. `ObjectStore::list` sends no `max-keys`, so a
+            // server falls back to 1000 per page and a recursive walk becomes a
+            // long chain of sequential requests. See `big_page_list`.
+            Arc::new(crate::big_page_list::BigPageList::new(
                 builder
                     .build()
                     .map_err(|e| exec_datafusion_err!("failed to build S3 store for {url}: {e}"))?,
-            )
+            ))
         }
         "gs" => {
             let mut builder = GoogleCloudStorageBuilder::from_env().with_url(url.as_str());
diff --git a/beacon-db/beacon-db-py/python/beacondb/_beacondb.pyi b/beacon-db/beacon-db-py/python/beacondb/_beacondb.pyi
index 380210da..8bd94c64 100644
--- a/beacon-db/beacon-db-py/python/beacondb/_beacondb.pyi
+++ b/beacon-db/beacon-db-py/python/beacondb/_beacondb.pyi
@@ -146,6 +146,7 @@ class Connection:
     def read_tiff(self, *args: Any, **kwargs: Any) -> Relation: ...
     def read_bbf(self, *args: Any, **kwargs: Any) -> Relation: ...
     def list_datasets(self, *args: Any, **kwargs: Any) -> Relation: ...
+    def browse_datasets(self, *args: Any, **kwargs: Any) -> Relation: ...
     def table_functions(self) -> list[str]: ...
     # attach a remote Beacon instance as a catalog (queryable as name.schema.table, with pushdown)
     def attach(
diff --git a/beacon-db/beacon-db-py/src/connection.rs b/beacon-db/beacon-db-py/src/connection.rs
index 83fecd32..c9a0466e 100644
--- a/beacon-db/beacon-db-py/src/connection.rs
+++ b/beacon-db/beacon-db-py/src/connection.rs
@@ -731,6 +731,7 @@ impl Connection {
 /// here is not a real table function. A build without an optional format still lists its reader
 /// here; calling it fails at planning time, as it would for any unregistered function.
 const TABLE_FUNCTIONS: &[(&str, &[&str])] = &[
+    ("browse_datasets", &[]),
     ("list_datasets", &[]),
     ("read_arrow", &["glob_paths"]),
     ("read_arrow_schema", &["glob_paths"]),
diff --git a/beacon-db/beacon-file-formats/beacon-arrow-zarr/src/datafusion/mod.rs b/beacon-db/beacon-file-formats/beacon-arrow-zarr/src/datafusion/mod.rs
index 2f387cc1..404d4399 100644
--- a/beacon-db/beacon-file-formats/beacon-arrow-zarr/src/datafusion/mod.rs
+++ b/beacon-db/beacon-file-formats/beacon-arrow-zarr/src/datafusion/mod.rs
@@ -29,7 +29,9 @@ use zarrs::group::Group;
 use crate::{
     config::ZarrConfig,
     reader::schema_from_group_path,
-    util::{ZarrPath, ZarrStorage, is_zarr_v3_metadata, leaf_group_keys, top_level_zarr_meta_v3},
+    util::{
+        ZarrPath, ZarrStorage, is_zarr_store_root, leaf_group_keys, top_level_zarr_meta_v3,
+    },
 };
 
 pub mod source;
@@ -154,6 +156,8 @@ impl FileFormatFactoryExt for ZarrFormatFactory {
     /// depends on the whole store: an array added under it changes the schema
     /// while leaving the marker where it was.
     fn schema_units(&self, objects: &[ObjectMeta]) -> Vec<SchemaUnit> {
+        // The same rule `infer_schema` uses, so a cache entry keys on the marker
+        // that read the schema.
         units_over_stores(objects, &crate::util::top_level_zarr_meta_v3(objects))
     }
 
@@ -161,15 +165,12 @@ impl FileFormatFactoryExt for ZarrFormatFactory {
         &self,
         objects: &[ObjectMeta],
     ) -> datafusion::error::Result<Vec<DatasetMetadata>> {
-        let datasets: Vec<ObjectMeta> = objects
+        // A store is a `*.zarr` directory holding `zarr.json`. Decided per
+        // object, so this classifies a listing without holding it.
+        let zarr_paths: Vec<ZarrPath> = objects
             .iter()
-            .filter(|obj| is_zarr_v3_metadata(obj))
+            .filter(|obj| is_zarr_store_root(obj))
             .cloned()
-            .collect();
-
-        let top_level_datasets = top_level_zarr_meta_v3(&datasets);
-        let zarr_paths: Vec<ZarrPath> = top_level_datasets
-            .into_iter()
             .filter_map(|path| match ZarrPath::new_from_object_meta(path) {
                 Ok(zarr_path) => Some(zarr_path),
                 Err(e) => {
@@ -326,9 +327,11 @@ impl FileFormat for ZarrFormat {
     ) -> datafusion::error::Result<SchemaRef> {
         // The listing may include non-metadata objects — chunk data files such as
         // `<array>/c/0/0/0` — when the table is created without a `zarr.json`
-        // extension filter (e.g. via `read_zarr`). Select the top-level group
-        // metadata files and ignore the rest rather than erroring on the first
-        // chunk we encounter.
+        // extension filter (e.g. via `read_zarr`). Select the store roots and
+        // ignore the rest rather than erroring on the first chunk we encounter.
+        // The caller has already named the store, so the root is the shallowest
+        // marker here rather than a `*.zarr` directory. See `is_zarr_store_root`
+        // for why discovery cannot use this rule and this cannot use that one.
         let verified_objects = top_level_zarr_meta_v3(objects);
         if verified_objects.is_empty() {
             return Err(datafusion::error::DataFusionError::Execution(
diff --git a/beacon-db/beacon-file-formats/beacon-arrow-zarr/src/util.rs b/beacon-db/beacon-file-formats/beacon-arrow-zarr/src/util.rs
index 924bac20..6d4b38dc 100644
--- a/beacon-db/beacon-file-formats/beacon-arrow-zarr/src/util.rs
+++ b/beacon-db/beacon-file-formats/beacon-arrow-zarr/src/util.rs
@@ -1,10 +1,8 @@
 //! Zarr path handling and group-discovery helpers shared by the DataFusion
 //! integration.
 
-use std::{
-    collections::{HashMap, HashSet},
-    sync::Arc,
-};
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
 
 use object_store::{ObjectStore, ObjectStoreExt};
 use zarrs::group::Group;
@@ -121,8 +119,8 @@ pub fn path_parent(p: &object_store::path::Path) -> Option<object_store::path::P
 ///
 /// A bare `zarr.json` — no parent directory — counts too: a repository-backed
 /// store such as Icechunk exposes its root group there. Listing-based discovery
-/// is unaffected, because [`top_level_zarr_meta_v3`] keys groups by their parent
-/// directory and a root-level metadata file has none.
+/// is unaffected, because [`is_zarr_store_root`] names a store by the directory
+/// holding the marker and a root-level metadata file has none.
 pub fn is_zarr_v3_metadata(meta: &object_store::ObjectMeta) -> bool {
     let loc = meta.location.to_string().to_lowercase();
     loc == "zarr.json" || loc.ends_with("/zarr.json")
@@ -160,6 +158,44 @@ pub async fn leaf_group_keys(
     )
 }
 
+/// Whether `meta` is the root marker of a Zarr store, judged from its own path.
+///
+/// A store is a `*.zarr` directory holding `zarr.json` directly beside its
+/// arrays:
+///
+/// ```text
+/// gridded-example.zarr/zarr.json              the store
+/// gridded-example.zarr/lat/zarr.json          an array inside it
+/// ```
+///
+/// Zarr v3 gives every group *and every array* a `zarr.json`, so a marker alone
+/// says nothing about which one it is. The directory it sits in does: only the
+/// store carries the `.zarr` suffix.
+///
+/// This is decided per object, with no reference to any other. That is what lets
+/// discovery classify a listing as it streams, instead of holding every object
+/// to compare ancestors.
+///
+/// Discovery only. A caller that names a store — `read_zarr('some/dir')` — has
+/// already said which directory it means, and that directory need not carry the
+/// suffix, so those paths use [`top_level_zarr_meta_v3`] instead. The two
+/// questions differ: discovery asks "is this marker a store root, among many
+/// stores", and a read asks "which marker is the root of the one store I was
+/// given". Only the first can be answered from one object, and only the second
+/// has a listing small enough to compare.
+///
+/// A bare `zarr.json` at the store root has no directory to name it, so it is
+/// not a discovered dataset. Icechunk reaches its root group directly and does
+/// not come through discovery.
+pub fn is_zarr_store_root(meta: &object_store::ObjectMeta) -> bool {
+    if !is_zarr_v3_metadata(meta) {
+        return false;
+    }
+    path_parent(&meta.location)
+        .and_then(|parent| parent.filename().map(|name| name.to_lowercase()))
+        .is_some_and(|name| name.ends_with(".zarr"))
+}
+
 /// Return only the ObjectMeta entries corresponding to **top-level Zarr groups**.
 pub fn top_level_zarr_meta_v3(metas: &[object_store::ObjectMeta]) -> Vec<object_store::ObjectMeta> {
     let mut dir_to_meta: HashMap<object_store::path::Path, &object_store::ObjectMeta> =
@@ -260,56 +296,104 @@ mod tests {
         }
     }
 
+    #[test]
+    fn a_marker_in_a_zarr_directory_is_a_store() {
+        assert!(is_zarr_store_root(&meta("gridded-example.zarr/zarr.json")));
+        assert!(is_zarr_store_root(&meta("deep/nested/path/cube.zarr/zarr.json")));
+        // The suffix is matched case-insensitively, as the marker name is.
+        assert!(is_zarr_store_root(&meta("CUBE.ZARR/zarr.json")));
+    }
+
+    /// Every array in a v3 store also has a `zarr.json`. The directory between
+    /// it and the store is what tells them apart.
+    #[test]
+    fn a_marker_below_the_store_directory_is_an_array() {
+        assert!(!is_zarr_store_root(&meta("gridded-example.zarr/lat/zarr.json")));
+        assert!(!is_zarr_store_root(&meta("gridded-example.zarr/a/b/zarr.json")));
+    }
+
+    /// A directory without the suffix is not a store, however the marker looks.
+    #[test]
+    fn a_marker_outside_a_zarr_directory_is_not_a_store() {
+        assert!(!is_zarr_store_root(&meta("a/zarr.json")));
+        assert!(!is_zarr_store_root(&meta("root/zarr.json")));
+    }
+
+    /// A root-level marker has no directory to name it. Icechunk reaches its
+    /// root group directly rather than through discovery.
+    #[test]
+    fn a_bare_marker_is_not_a_discovered_store() {
+        assert!(!is_zarr_store_root(&meta("zarr.json")));
+    }
+
+    /// Non-markers are rejected before the directory is even considered.
+    #[test]
+    fn a_non_marker_is_never_a_store() {
+        assert!(!is_zarr_store_root(&meta("cube.zarr/not_zarr.json")));
+        assert!(!is_zarr_store_root(&meta("cube.zarr/data.nc")));
+    }
+
+    // ---- top_level_zarr_meta_v3 -------------------------------------------
+    //
+    // The other rule, for a caller that has already named the store. It reads a
+    // listing rather than one object, so it needs no `.zarr` suffix.
+
     fn extract_paths(metas: &[ObjectMeta]) -> Vec<String> {
-        metas.iter().map(|m| m.location.to_string()).collect()
+        let mut paths: Vec<String> = metas.iter().map(|m| m.location.to_string()).collect();
+        paths.sort();
+        paths
     }
 
+    /// One marker is its own store root.
     #[test]
-    fn test_single_top_level() {
+    fn a_single_marker_is_the_root() {
         let metas = vec![meta("a/zarr.json")];
-        let result = top_level_zarr_meta_v3(&metas);
-        assert_eq!(extract_paths(&result), vec!["a/zarr.json"]);
+        assert_eq!(extract_paths(&top_level_zarr_meta_v3(&metas)), vec!["a/zarr.json"]);
     }
 
+    /// A marker below another one describes an array or a sub-group, so only the
+    /// shallowest survives.
     #[test]
-    fn test_nested_group_dropped() {
+    fn a_marker_below_the_root_is_dropped() {
         let metas = vec![meta("a/zarr.json"), meta("a/b/zarr.json")];
-        let result = top_level_zarr_meta_v3(&metas);
-        assert_eq!(extract_paths(&result), vec!["a/zarr.json"]);
+        assert_eq!(extract_paths(&top_level_zarr_meta_v3(&metas)), vec!["a/zarr.json"]);
     }
 
+    /// Two unrelated stores keep a root each.
     #[test]
-    fn test_multiple_top_level() {
+    fn two_stores_keep_a_root_each() {
         let metas = vec![meta("a/zarr.json"), meta("b/zarr.json")];
-        let result = top_level_zarr_meta_v3(&metas);
-        let mut paths = extract_paths(&result);
-        paths.sort();
-        assert_eq!(paths, vec!["a/zarr.json", "b/zarr.json"]);
+        assert_eq!(
+            extract_paths(&top_level_zarr_meta_v3(&metas)),
+            vec!["a/zarr.json", "b/zarr.json"]
+        );
     }
 
+    /// Chunk files and look-alike names are not markers.
     #[test]
-    fn test_ignore_non_zarr_json() {
+    fn a_non_marker_is_never_a_root() {
         let metas = vec![
             meta("a/zarr.json"),
             meta("b/not_zarr.json"),
             meta("c/zarr.txt"),
         ];
-        let result = top_level_zarr_meta_v3(&metas);
-        assert_eq!(extract_paths(&result), vec!["a/zarr.json"]);
+        assert_eq!(extract_paths(&top_level_zarr_meta_v3(&metas)), vec!["a/zarr.json"]);
     }
 
+    /// However deep the nesting, only the shallowest ancestor of each store
+    /// survives.
     #[test]
-    fn deeply_nested_children_are_all_dropped() {
-        // Only the shallowest ancestor group survives.
+    fn every_marker_below_a_root_is_dropped() {
         let metas = vec![
             meta("root/zarr.json"),
             meta("root/a/zarr.json"),
             meta("root/a/b/zarr.json"),
             meta("other/zarr.json"),
         ];
-        let mut result = extract_paths(&top_level_zarr_meta_v3(&metas));
-        result.sort();
-        assert_eq!(result, vec!["other/zarr.json", "root/zarr.json"]);
+        assert_eq!(
+            extract_paths(&top_level_zarr_meta_v3(&metas)),
+            vec!["other/zarr.json", "root/zarr.json"]
+        );
     }
 
     #[test]
diff --git a/beacon-db/beacon-functions/Cargo.toml b/beacon-db/beacon-functions/Cargo.toml
index b972c904..51a36e62 100644
--- a/beacon-db/beacon-functions/Cargo.toml
+++ b/beacon-db/beacon-functions/Cargo.toml
@@ -20,6 +20,8 @@ proj = ["datafusion-spatial/proj"]
 proj-bundled = ["datafusion-spatial/proj-bundled"]
 
 [dependencies]
+async-stream = { workspace = true }
+async-trait = { workspace = true }
 datafusion = {workspace = true}
 arrow = {workspace = true}
 strsim = {workspace = true}
diff --git a/beacon-db/beacon-functions/src/file_formats/list_datasets.rs b/beacon-db/beacon-functions/src/file_formats/list_datasets.rs
deleted file mode 100644
index a71f330c..00000000
--- a/beacon-db/beacon-functions/src/file_formats/list_datasets.rs
+++ /dev/null
@@ -1,200 +0,0 @@
-use std::sync::{Arc, Weak};
-
-use arrow::{
-    array::{BooleanArray, StringArray, UInt64Array},
-    datatypes::{DataType, Field, Schema},
-    record_batch::RecordBatch,
-};
-use beacon_datafusion_ext::format_ext::{DatasetMetadata, FileFormatFactoryExt};
-use beacon_datafusion_ext::listing_factory::ListingFactory;
-use datafusion::{
-    catalog::{MemTable, TableFunctionImpl, TableProvider},
-    error::DataFusionError,
-    prelude::{Expr, SessionContext},
-    scalar::ScalarValue,
-};
-
-use crate::file_formats::BeaconTableFunctionImpl;
-
-/// Discover the datasets matching `pattern` (default `**/*`) under the datasets
-/// object store at `datasets_url`, asking each registered file format which
-/// objects it owns.
-pub async fn list_datasets(
-    session_ctx: &SessionContext,
-    file_formats: &[Arc<dyn FileFormatFactoryExt>],
-    offset: Option<usize>,
-    limit: Option<usize>,
-    search_pattern: Option<String>,
-) -> datafusion::error::Result<Vec<DatasetMetadata>> {
-    let state = session_ctx.state();
-    let listing_factory = state
-        .config()
-        .get_extension::<ListingFactory>()
-        .ok_or_else(|| {
-            DataFusionError::Execution(
-                "list_datasets: the listing factory is not registered on the session".to_string(),
-            )
-        })?;
-
-    // Discovery + object-metadata enrichment lives on the listing factory; this
-    // function only adds pagination on top.
-    let datasets = listing_factory
-        .list_datasets(
-            &state,
-            file_formats,
-            &search_pattern.unwrap_or_else(|| "**/*".to_string()),
-        )
-        .await?;
-
-    // Keep current pagination semantics to avoid behavior regressions.
-    // `saturating_sub`: an offset past the end must yield an empty page, not an
-    // underflow panic (`end` is clamped to `datasets.len()`, so it can be < start).
-    let start = offset.unwrap_or(0);
-    let end = limit.map(|l| start + l).unwrap_or(datasets.len());
-    let datasets = datasets
-        .into_iter()
-        .skip(start)
-        .take(end.saturating_sub(start))
-        .collect();
-
-    Ok(datasets)
-}
-
-pub struct ListDatasetsFunc {
-    runtime_handle: tokio::runtime::Handle,
-    session_ctx: Weak<SessionContext>,
-    file_formats: Vec<Arc<dyn FileFormatFactoryExt>>,
-}
-
-impl ListDatasetsFunc {
-    pub fn new(
-        runtime_handle: tokio::runtime::Handle,
-        session_ctx: Weak<SessionContext>,
-        file_formats: Vec<Arc<dyn FileFormatFactoryExt>>,
-    ) -> Self {
-        Self {
-            runtime_handle,
-            session_ctx,
-            file_formats,
-        }
-    }
-}
-
-impl std::fmt::Debug for ListDatasetsFunc {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "ListDatasetsFunc")
-    }
-}
-
-impl BeaconTableFunctionImpl for ListDatasetsFunc {
-    fn as_any(&self) -> &dyn std::any::Any {
-        self
-    }
-
-    fn name(&self) -> String {
-        "list_datasets".to_string()
-    }
-
-    fn description(&self) -> Option<String> {
-        Some(
-            "Lists the datasets stored in beacon. Optional arguments: \
-             list_datasets(pattern, offset, limit) — a glob (default '**/*'), \
-             a row offset, and a row limit."
-                .to_string(),
-        )
-    }
-}
-
-/// The full [`DatasetMetadata`] shape, so a caller gets everything discovery
-/// computed rather than just the name and format.
-pub fn list_datasets_schema() -> Arc<Schema> {
-    Arc::new(Schema::new(vec![
-        Field::new("file_name", DataType::Utf8, false),
-        Field::new("file_format", DataType::Utf8, false),
-        Field::new("can_inspect", DataType::Boolean, false),
-        Field::new("can_partial_explore", DataType::Boolean, false),
-        // Null when the size or timestamp could not be resolved.
-        Field::new("size", DataType::UInt64, true),
-        Field::new("last_modified", DataType::Utf8, true),
-    ]))
-}
-
-/// A `Utf8` literal argument, or `None` when absent.
-fn string_arg(args: &[Expr], index: usize) -> Option<String> {
-    match args.get(index) {
-        Some(Expr::Literal(ScalarValue::Utf8(value), _)) => value.clone(),
-        _ => None,
-    }
-}
-
-/// A non-negative integer literal argument, or `None` when absent.
-fn usize_arg(args: &[Expr], index: usize) -> Option<usize> {
-    match args.get(index) {
-        Some(Expr::Literal(scalar, _)) => match scalar {
-            ScalarValue::Int64(Some(v)) if *v >= 0 => Some(*v as usize),
-            ScalarValue::UInt64(Some(v)) => Some(*v as usize),
-            ScalarValue::Int32(Some(v)) if *v >= 0 => Some(*v as usize),
-            ScalarValue::UInt32(Some(v)) => Some(*v as usize),
-            _ => None,
-        },
-        _ => None,
-    }
-}
-
-impl TableFunctionImpl for ListDatasetsFunc {
-    /// `list_datasets([pattern[, offset[, limit]]])`.
-    ///
-    /// All three are optional and positional; omitting them recursively lists
-    /// everything, which is the historical behaviour.
-    fn call(&self, args: &[Expr]) -> datafusion::error::Result<Arc<dyn TableProvider>> {
-        let pattern = string_arg(args, 0).unwrap_or_else(|| "**/*".to_string());
-        let offset = usize_arg(args, 1);
-        let limit = usize_arg(args, 2);
-
-        let file_formats = self.file_formats.clone();
-        let session_ctx = self.session_ctx.upgrade().ok_or_else(|| {
-            datafusion::common::plan_datafusion_err!("session context has been dropped")
-        })?;
-
-        let datasets: Vec<DatasetMetadata> = tokio::task::block_in_place(|| {
-            self.runtime_handle.block_on(async move {
-                list_datasets(&session_ctx, &file_formats, offset, limit, Some(pattern)).await
-            })
-        })?;
-
-        let schema = list_datasets_schema();
-        let file_names: StringArray = datasets
-            .iter()
-            .map(|d| Some(d.file_path.as_str()))
-            .collect();
-        let formats: StringArray = datasets.iter().map(|d| Some(d.format.as_str())).collect();
-        let can_inspect = BooleanArray::from(
-            datasets.iter().map(|d| d.can_inspect).collect::<Vec<_>>(),
-        );
-        let can_partial_explore = BooleanArray::from(
-            datasets
-                .iter()
-                .map(|d| d.can_partial_explore)
-                .collect::<Vec<_>>(),
-        );
-        let sizes = UInt64Array::from(datasets.iter().map(|d| d.size).collect::<Vec<_>>());
-        let last_modified: StringArray = datasets
-            .iter()
-            .map(|d| d.last_modified.map(|ts| ts.to_rfc3339()))
-            .collect();
-
-        let batch = RecordBatch::try_new(
-            schema.clone(),
-            vec![
-                Arc::new(file_names),
-                Arc::new(formats),
-                Arc::new(can_inspect),
-                Arc::new(can_partial_explore),
-                Arc::new(sizes),
-                Arc::new(last_modified),
-            ],
-        )?;
-
-        Ok(Arc::new(MemTable::try_new(schema, vec![vec![batch]])?))
-    }
-}
diff --git a/beacon-db/beacon-functions/src/file_formats/mod.rs b/beacon-db/beacon-functions/src/file_formats/mod.rs
index fc4dbd22..d39dd493 100644
--- a/beacon-db/beacon-functions/src/file_formats/mod.rs
+++ b/beacon-db/beacon-functions/src/file_formats/mod.rs
@@ -10,7 +10,6 @@ use datafusion::prelude::SessionContext;
 pub use beacon_common::table_function::{parse_glob_paths_arg, BeaconTableFunctionImpl};
 
 // Cross-format table functions that don't belong to a single format crate.
-pub mod list_datasets;
 pub mod schema_function;
 
 use schema_function::SchemaTableFunc;
@@ -92,9 +91,11 @@ pub fn register_table_functions(
 
     let mut functions = readers;
     functions.extend(schema_funcs);
-    functions.push(Arc::new(list_datasets::ListDatasetsFunc::new(
-        runtime_handle,
-        Arc::downgrade(&session_ctx),
+    // Listing the datasets store: two functions over one provider.
+    functions.push(Arc::new(crate::listing::ListDatasetsFunc::new(
+        file_formats.clone(),
+    )));
+    functions.push(Arc::new(crate::listing::BrowseDatasetsFunc::new(
         file_formats,
     )));
     functions
diff --git a/beacon-db/beacon-functions/src/lib.rs b/beacon-db/beacon-functions/src/lib.rs
index 9ee45dc3..ffd95999 100644
--- a/beacon-db/beacon-functions/src/lib.rs
+++ b/beacon-db/beacon-functions/src/lib.rs
@@ -11,6 +11,7 @@ use crate::{
 
 pub mod blue_cloud;
 pub mod file_formats;
+pub mod listing;
 pub mod metadata;
 pub mod util;
 
diff --git a/beacon-db/beacon-functions/src/listing/args.rs b/beacon-db/beacon-functions/src/listing/args.rs
new file mode 100644
index 00000000..99ad444a
--- /dev/null
+++ b/beacon-db/beacon-functions/src/listing/args.rs
@@ -0,0 +1,28 @@
+//! Reading the positional literals a table function is called with.
+//!
+//! Both listing functions take the same shapes, so the readers live once.
+
+use datafusion::{prelude::Expr, scalar::ScalarValue};
+
+/// A `Utf8` literal argument, or `None` when absent.
+pub(super) fn string_arg(args: &[Expr], index: usize) -> Option<String> {
+    match args.get(index) {
+        Some(Expr::Literal(ScalarValue::Utf8(value), _)) => value.clone(),
+        _ => None,
+    }
+}
+
+/// A non-negative integer literal argument, or `None` when absent.
+pub(super) fn usize_arg(args: &[Expr], index: usize) -> Option<usize> {
+    match args.get(index) {
+        Some(Expr::Literal(scalar, _)) => match scalar {
+            ScalarValue::Int64(Some(v)) if *v >= 0 => Some(*v as usize),
+            ScalarValue::UInt64(Some(v)) => Some(*v as usize),
+            ScalarValue::Int32(Some(v)) if *v >= 0 => Some(*v as usize),
+            ScalarValue::UInt32(Some(v)) => Some(*v as usize),
+            _ => None,
+        },
+        _ => None,
+    }
+}
+
diff --git a/beacon-db/beacon-functions/src/listing/browse_datasets.rs b/beacon-db/beacon-functions/src/listing/browse_datasets.rs
new file mode 100644
index 00000000..40c911fa
--- /dev/null
+++ b/beacon-db/beacon-functions/src/listing/browse_datasets.rs
@@ -0,0 +1,70 @@
+//! `browse_datasets([prefix[, offset[, limit]]])`: one directory level.
+
+use std::sync::Arc;
+
+use beacon_datafusion_ext::format_ext::FileFormatFactoryExt;
+use datafusion::{
+    catalog::{TableFunctionImpl, TableProvider},
+    prelude::Expr,
+};
+
+use super::provider::{DatasetsTable, Listing};
+use crate::file_formats::BeaconTableFunctionImpl;
+use crate::listing::args::{string_arg, usize_arg};
+
+/// `browse_datasets([prefix[, offset[, limit]]])`: one directory level.
+///
+/// The same provider as `list_datasets`, stopped at one level. A folder view
+/// wants this: its cost is one delimiter request, so it does not grow with the
+/// size of the store below the prefix.
+pub struct BrowseDatasetsFunc {
+    file_formats: Vec<Arc<dyn FileFormatFactoryExt>>,
+}
+
+impl BrowseDatasetsFunc {
+    pub fn new(file_formats: Vec<Arc<dyn FileFormatFactoryExt>>) -> Self {
+        Self { file_formats }
+    }
+}
+
+impl std::fmt::Debug for BrowseDatasetsFunc {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "BrowseDatasetsFunc")
+    }
+}
+
+impl BeaconTableFunctionImpl for BrowseDatasetsFunc {
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    fn name(&self) -> String {
+        "browse_datasets".to_string()
+    }
+
+    fn description(&self) -> Option<String> {
+        Some(
+            "Lists one directory level of the datasets store. Optional arguments: \
+             browse_datasets(prefix, offset, limit) — a directory (default the root), \
+             a row offset, and a row limit. Sub-directories come back as rows with \
+             `is_directory` set. Unlike list_datasets it does not descend, so its \
+             cost does not grow with the size of the store below the prefix."
+                .to_string(),
+        )
+    }
+}
+
+impl TableFunctionImpl for BrowseDatasetsFunc {
+    fn call(&self, args: &[Expr]) -> datafusion::error::Result<Arc<dyn TableProvider>> {
+        let prefix = string_arg(args, 0).unwrap_or_default();
+        let offset = usize_arg(args, 1).unwrap_or(0);
+        let limit = usize_arg(args, 2);
+
+        Ok(Arc::new(DatasetsTable::new(
+            Listing::Level { prefix },
+            offset,
+            limit,
+            self.file_formats.clone(),
+        )))
+    }
+}
diff --git a/beacon-db/beacon-functions/src/listing/exec.rs b/beacon-db/beacon-functions/src/listing/exec.rs
new file mode 100644
index 00000000..f0647498
--- /dev/null
+++ b/beacon-db/beacon-functions/src/listing/exec.rs
@@ -0,0 +1,220 @@
+//! [`DatasetsExec`]: the plan node that turns a listing stream into result rows.
+//!
+//! A listing is a stream of paths from an object store, and a query wants
+//! batches of rows. Everything between those two is here.
+//!
+//! # Why not a `MemTable`
+//!
+//! Wrapping the finished listing in a `MemTable` was simpler, and it forced the
+//! whole walk to finish, and every row to exist at once, before the plan could
+//! start. On a store of 2 853 217 objects that is a two-figure number of seconds
+//! before the first row and about a gigabyte held while it happens.
+//!
+//! This node emits a batch as soon as it has [`BATCH_ROWS`] of them. First rows
+//! arrive in the time of the first listing page, and memory is bounded by one
+//! batch rather than by the store.
+//!
+//! # Limit
+//!
+//! `limit` stops the node, and because the source is a lazy stream, stopping the
+//! node stops the walk. A `LIMIT 50` over a bucket of millions reads one page,
+//! not all of it. That is the part a `MemTable` could not do however the limit
+//! was pushed down.
+
+use std::any::Any;
+use std::fmt;
+use std::sync::Arc;
+
+use arrow::{
+    array::{BooleanArray, StringArray, UInt64Array},
+    datatypes::SchemaRef,
+    record_batch::RecordBatch,
+};
+use datafusion::{
+    error::{DataFusionError, Result},
+    execution::{SendableRecordBatchStream, TaskContext},
+    physical_expr::EquivalenceProperties,
+    physical_plan::{
+        DisplayAs, DisplayFormatType, ExecutionPlan, Partitioning, PlanProperties,
+        execution_plan::{Boundedness, EmissionType},
+        stream::RecordBatchStreamAdapter,
+    },
+};
+use futures::stream::{BoxStream, StreamExt};
+
+use super::provider::Row;
+
+/// Rows gathered before a batch is emitted.
+///
+/// Small enough that the first rows reach the caller promptly, large enough that
+/// per-batch overhead stays out of the way.
+pub const BATCH_ROWS: usize = 8192;
+
+/// Builds the listing stream. Called once per partition, at execute time, so
+/// nothing touches the store until the plan runs.
+pub type RowStreamFactory =
+    Arc<dyn Fn() -> Result<BoxStream<'static, Result<Row>>> + Send + Sync>;
+
+/// Streams listing rows as record batches.
+#[derive(Clone)]
+pub struct DatasetsExec {
+    schema: SchemaRef,
+    rows: RowStreamFactory,
+    /// Rows to skip before emitting. Applied here so a page is never built and
+    /// thrown away.
+    offset: usize,
+    /// Rows to emit at most. `None` reads the listing out.
+    limit: Option<usize>,
+    /// How the plan prints itself, for `EXPLAIN`.
+    label: String,
+    properties: Arc<PlanProperties>,
+}
+
+impl DatasetsExec {
+    pub fn new(
+        schema: SchemaRef,
+        rows: RowStreamFactory,
+        offset: usize,
+        limit: Option<usize>,
+        label: String,
+    ) -> Self {
+        let properties = PlanProperties::new(
+            EquivalenceProperties::new(Arc::clone(&schema)),
+            // One partition: a listing is one walk, and splitting it would mean
+            // splitting the object store's own pagination.
+            Partitioning::UnknownPartitioning(1),
+            EmissionType::Incremental,
+            Boundedness::Bounded,
+        );
+        Self {
+            schema,
+            rows,
+            offset,
+            limit,
+            label,
+            properties: Arc::new(properties),
+        }
+    }
+}
+
+impl fmt::Debug for DatasetsExec {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "DatasetsExec({})", self.label)
+    }
+}
+
+impl DisplayAs for DatasetsExec {
+    fn fmt_as(&self, _t: DisplayFormatType, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "DatasetsExec: {}", self.label)?;
+        if self.offset > 0 {
+            write!(f, ", offset={}", self.offset)?;
+        }
+        if let Some(limit) = self.limit {
+            write!(f, ", limit={limit}")?;
+        }
+        Ok(())
+    }
+}
+
+impl ExecutionPlan for DatasetsExec {
+    fn name(&self) -> &str {
+        "DatasetsExec"
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+
+    fn properties(&self) -> &Arc<PlanProperties> {
+        &self.properties
+    }
+
+    fn children(&self) -> Vec<&Arc<dyn ExecutionPlan>> {
+        vec![]
+    }
+
+    fn with_new_children(
+        self: Arc<Self>,
+        _children: Vec<Arc<dyn ExecutionPlan>>,
+    ) -> Result<Arc<dyn ExecutionPlan>> {
+        Ok(self)
+    }
+
+    fn execute(
+        &self,
+        partition: usize,
+        _context: Arc<TaskContext>,
+    ) -> Result<SendableRecordBatchStream> {
+        if partition != 0 {
+            return Err(DataFusionError::Internal(format!(
+                "DatasetsExec has one partition; asked for {partition}"
+            )));
+        }
+
+        let schema = Arc::clone(&self.schema);
+        let rows = (self.rows)()?;
+        let batches = batch_rows(rows, Arc::clone(&schema), self.offset, self.limit);
+
+        Ok(Box::pin(RecordBatchStreamAdapter::new(schema, batches)))
+    }
+}
+
+/// Gather rows into batches, applying `offset` and `limit` as they pass.
+///
+/// The limit is applied before batching, so the stream behind it is dropped as
+/// soon as enough rows exist — which is what stops the walk.
+fn batch_rows(
+    rows: BoxStream<'static, Result<Row>>,
+    schema: SchemaRef,
+    offset: usize,
+    limit: Option<usize>,
+) -> BoxStream<'static, Result<RecordBatch>> {
+    let paged = rows.skip(offset);
+    let paged: BoxStream<'static, Result<Row>> = match limit {
+        Some(limit) => paged.take(limit).boxed(),
+        None => paged.boxed(),
+    };
+
+    async_stream::try_stream! {
+        let mut pending: Vec<Row> = Vec::with_capacity(BATCH_ROWS);
+        futures::pin_mut!(paged);
+        while let Some(row) = paged.next().await {
+            pending.push(row?);
+            if pending.len() >= BATCH_ROWS {
+                yield rows_batch(Arc::clone(&schema), &std::mem::take(&mut pending))?;
+                pending.reserve(BATCH_ROWS);
+            }
+        }
+        // The tail, and the empty case: a listing that matched nothing still has
+        // to produce a schema-shaped result rather than no stream at all.
+        if !pending.is_empty() {
+            yield rows_batch(Arc::clone(&schema), &pending)?;
+        }
+    }
+    .boxed()
+}
+
+/// Pack rows into one record batch.
+pub(super) fn rows_batch(schema: SchemaRef, rows: &[Row]) -> Result<RecordBatch> {
+    let file_names: StringArray = rows.iter().map(|r| Some(r.file_name.as_str())).collect();
+    let formats: StringArray = rows.iter().map(|r| Some(r.file_format.as_str())).collect();
+    let can_inspect = BooleanArray::from(rows.iter().map(|r| r.can_inspect).collect::<Vec<_>>());
+    let can_partial_explore =
+        BooleanArray::from(rows.iter().map(|r| r.can_partial_explore).collect::<Vec<_>>());
+    let sizes = UInt64Array::from(rows.iter().map(|r| r.size).collect::<Vec<_>>());
+    let last_modified: StringArray = rows.iter().map(|r| r.last_modified.clone()).collect();
+    let is_directory = BooleanArray::from(rows.iter().map(|r| r.is_directory).collect::<Vec<_>>());
+
+    Ok(RecordBatch::try_new(
+        schema,
+        vec![
+            Arc::new(file_names),
+            Arc::new(formats),
+            Arc::new(can_inspect),
+            Arc::new(can_partial_explore),
+            Arc::new(sizes),
+            Arc::new(last_modified),
+            Arc::new(is_directory),
+        ],
+    )?)
+}
diff --git a/beacon-db/beacon-functions/src/listing/list_datasets.rs b/beacon-db/beacon-functions/src/listing/list_datasets.rs
new file mode 100644
index 00000000..81daec43
--- /dev/null
+++ b/beacon-db/beacon-functions/src/listing/list_datasets.rs
@@ -0,0 +1,117 @@
+//! `list_datasets([pattern[, offset[, limit]]])`: the whole tree under a glob.
+
+use std::sync::Arc;
+
+use beacon_datafusion_ext::format_ext::{DatasetMetadata, FileFormatFactoryExt};
+use beacon_datafusion_ext::listing_factory::ListingFactory;
+use datafusion::{
+    catalog::{TableFunctionImpl, TableProvider},
+    error::DataFusionError,
+    prelude::{Expr, SessionContext},
+};
+
+use super::provider::{DatasetsTable, Listing};
+use crate::file_formats::BeaconTableFunctionImpl;
+use crate::listing::args::{string_arg, usize_arg};
+
+/// Discover the datasets matching `pattern` (default `**/*`) under the datasets
+/// object store at `datasets_url`, asking each registered file format which
+/// objects it owns.
+pub async fn list_datasets(
+    session_ctx: &SessionContext,
+    file_formats: &[Arc<dyn FileFormatFactoryExt>],
+    offset: Option<usize>,
+    limit: Option<usize>,
+    search_pattern: Option<String>,
+) -> datafusion::error::Result<Vec<DatasetMetadata>> {
+    let state = session_ctx.state();
+    let listing_factory = state
+        .config()
+        .get_extension::<ListingFactory>()
+        .ok_or_else(|| {
+            DataFusionError::Execution(
+                "list_datasets: the listing factory is not registered on the session".to_string(),
+            )
+        })?;
+
+    // Discovery + object-metadata enrichment lives on the listing factory; this
+    // function only adds pagination on top.
+    let datasets = listing_factory
+        .list_datasets(
+            &state,
+            file_formats,
+            &search_pattern.unwrap_or_else(|| "**/*".to_string()),
+        )
+        .await?;
+
+    // Keep current pagination semantics to avoid behavior regressions.
+    // `saturating_sub`: an offset past the end must yield an empty page, not an
+    // underflow panic (`end` is clamped to `datasets.len()`, so it can be < start).
+    let start = offset.unwrap_or(0);
+    let end = limit.map(|l| start + l).unwrap_or(datasets.len());
+    let datasets = datasets
+        .into_iter()
+        .skip(start)
+        .take(end.saturating_sub(start))
+        .collect();
+
+    Ok(datasets)
+}
+
+pub struct ListDatasetsFunc {
+    file_formats: Vec<Arc<dyn FileFormatFactoryExt>>,
+}
+
+impl ListDatasetsFunc {
+    pub fn new(file_formats: Vec<Arc<dyn FileFormatFactoryExt>>) -> Self {
+        Self { file_formats }
+    }
+}
+
+impl std::fmt::Debug for ListDatasetsFunc {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "ListDatasetsFunc")
+    }
+}
+
+impl BeaconTableFunctionImpl for ListDatasetsFunc {
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    fn name(&self) -> String {
+        "list_datasets".to_string()
+    }
+
+    fn description(&self) -> Option<String> {
+        Some(
+            "Lists the datasets stored in beacon, descending the whole tree. \
+             Optional arguments: list_datasets(pattern, offset, limit) — a glob \
+             (default '**/*'), a row offset, and a row limit. Use browse_datasets \
+             to read a single directory level instead."
+                .to_string(),
+        )
+    }
+}
+
+impl TableFunctionImpl for ListDatasetsFunc {
+    /// `list_datasets([pattern[, offset[, limit]]])`: every dataset the glob
+    /// matches, anywhere below it.
+    ///
+    /// All three are optional and positional; omitting them lists everything,
+    /// which is the historical behaviour. No I/O happens here: the arguments are
+    /// read and handed to a [`DatasetsTable`], which lists when it is scanned.
+    fn call(&self, args: &[Expr]) -> datafusion::error::Result<Arc<dyn TableProvider>> {
+        let pattern = string_arg(args, 0).unwrap_or_else(|| "**/*".to_string());
+        let offset = usize_arg(args, 1).unwrap_or(0);
+        let limit = usize_arg(args, 2);
+
+        Ok(Arc::new(DatasetsTable::new(
+            Listing::Glob { pattern },
+            offset,
+            limit,
+            self.file_formats.clone(),
+        )))
+    }
+}
+
diff --git a/beacon-db/beacon-functions/src/listing/mod.rs b/beacon-db/beacon-functions/src/listing/mod.rs
new file mode 100644
index 00000000..efa7c1e1
--- /dev/null
+++ b/beacon-db/beacon-functions/src/listing/mod.rs
@@ -0,0 +1,23 @@
+//! Listing the datasets store.
+//!
+//! Two table functions over one provider. `list_datasets` globs and descends the
+//! whole tree; `browse_datasets` reads one directory level. They differ only in
+//! how deep they go, so the provider is shared and each function is a thin
+//! reader of its own arguments.
+//!
+//! The difference is not a constant factor. One level is a single delimiter
+//! request; a recursive walk of a 2 853 217-object bucket took 79.9 s where the
+//! delimiter took 14 ms. A folder view wants the second, and used to pay for the
+//! first.
+
+pub mod browse_datasets;
+pub mod exec;
+pub mod list_datasets;
+pub mod provider;
+
+pub use browse_datasets::BrowseDatasetsFunc;
+pub use exec::DatasetsExec;
+pub use list_datasets::{list_datasets, ListDatasetsFunc};
+pub use provider::{list_datasets_schema, DatasetsTable, Listing, Row};
+
+pub(crate) mod args;
diff --git a/beacon-db/beacon-functions/src/listing/provider.rs b/beacon-db/beacon-functions/src/listing/provider.rs
new file mode 100644
index 00000000..200653e0
--- /dev/null
+++ b/beacon-db/beacon-functions/src/listing/provider.rs
@@ -0,0 +1,301 @@
+//! [`DatasetsTable`]: the table both listing functions return.
+//!
+//! # Nothing happens until it is scanned
+//!
+//! The listing used to run inside `TableFunctionImpl::call`, a *synchronous*
+//! trait method, so it reached the store through `block_in_place` + `block_on`
+//! and held a worker thread for the whole walk — during logical planning, before
+//! anything decided to read it. [`TableProvider::scan`] is async, so the walk
+//! belongs there, and even there it only builds the plan: the walk itself starts
+//! when the plan is executed. See [`super::exec`].
+
+use std::sync::Arc;
+
+use arrow::datatypes::{DataType, Field, Schema};
+use beacon_datafusion_ext::format_ext::{DatasetMetadata, FileFormatFactoryExt};
+use beacon_datafusion_ext::listing_factory::{ListingFactory, ObjectLevel};
+use datafusion::{
+    arrow::datatypes::SchemaRef,
+    catalog::{Session, TableProvider},
+    datasource::TableType,
+    error::DataFusionError,
+    physical_expr::PhysicalExpr,
+    physical_plan::{ExecutionPlan, expressions::Column, projection::ProjectionExec},
+    prelude::Expr,
+};
+use futures::stream::StreamExt;
+use object_store::ObjectMeta;
+
+use super::exec::{DatasetsExec, RowStreamFactory};
+
+/// The full [`DatasetMetadata`] shape, so a caller gets everything discovery
+/// computed rather than just the name and format.
+pub fn list_datasets_schema() -> Arc<Schema> {
+    Arc::new(Schema::new(vec![
+        Field::new("file_name", DataType::Utf8, false),
+        Field::new("file_format", DataType::Utf8, false),
+        Field::new("can_inspect", DataType::Boolean, false),
+        Field::new("can_partial_explore", DataType::Boolean, false),
+        // Null when the size or timestamp could not be resolved.
+        Field::new("size", DataType::UInt64, true),
+        Field::new("last_modified", DataType::Utf8, true),
+        // True for a sub-directory, which only a one-level listing reports. A
+        // recursive listing describes files, so this is always false there.
+        Field::new("is_directory", DataType::Boolean, false),
+    ]))
+}
+
+/// One row of a listing: a dataset, or a sub-directory of one.
+///
+/// The Rust shape of [`list_datasets_schema`], field for field. Public because
+/// [`DatasetsExec`] streams it, and a plan node that cannot name its own rows
+/// cannot be built by a caller.
+#[derive(Debug, Clone)]
+pub struct Row {
+    pub file_name: String,
+    pub file_format: String,
+    pub can_inspect: bool,
+    pub can_partial_explore: bool,
+    pub size: Option<u64>,
+    pub last_modified: Option<String>,
+    pub is_directory: bool,
+}
+
+impl From<DatasetMetadata> for Row {
+    fn from(d: DatasetMetadata) -> Self {
+        Self {
+            file_name: d.file_path,
+            file_format: d.format,
+            can_inspect: d.can_inspect,
+            can_partial_explore: d.can_partial_explore,
+            size: d.size,
+            last_modified: d.last_modified.map(|ts| ts.to_rfc3339()),
+            is_directory: false,
+        }
+    }
+}
+
+impl Row {
+    pub(super) fn directory(path: String) -> Self {
+        Self {
+            file_name: path,
+            file_format: String::new(),
+            can_inspect: false,
+            can_partial_explore: false,
+            size: None,
+            last_modified: None,
+            is_directory: true,
+        }
+    }
+}
+
+/// What a [`DatasetsTable`] enumerates.
+#[derive(Debug, Clone)]
+pub enum Listing {
+    /// Every dataset the glob matches, anywhere below it.
+    Glob { pattern: String },
+    /// One directory level: what is directly inside `prefix`.
+    Level { prefix: String },
+}
+
+/// The datasets of the store, as a table.
+///
+/// One provider, two table functions over it. `list_datasets` globs and
+/// descends; `browse_datasets` reads one directory level and reports its
+/// sub-directories as rows with `is_directory` set. They differ only in how deep
+/// they go, so they share everything below this point.
+///
+/// Two names rather than one with a depth argument. A depth argument would
+/// change what the *first* argument means — glob when absent, directory when set
+/// — and an argument that retypes another argument is a bad seam. It cannot be
+/// inferred from the glob either: DataFusion matches listing globs with the
+/// default `MatchOptions`, where `require_literal_separator` is false, so `sub/*`
+/// already matches `sub/deep/c.csv` and reading it as one level would silently
+/// change what existing patterns return.
+///
+/// The difference is not a constant factor. One level is a single delimiter
+/// request; a recursive walk of a 2 853 217-object bucket took 79.9 s where the
+/// delimiter took 14 ms.
+#[derive(Debug)]
+pub struct DatasetsTable {
+    listing: Listing,
+    offset: usize,
+    limit: Option<usize>,
+    file_formats: Vec<Arc<dyn FileFormatFactoryExt>>,
+    schema: SchemaRef,
+}
+
+impl DatasetsTable {
+    pub fn new(
+        listing: Listing,
+        offset: usize,
+        limit: Option<usize>,
+        file_formats: Vec<Arc<dyn FileFormatFactoryExt>>,
+    ) -> Self {
+        Self {
+            listing,
+            offset,
+            limit,
+            file_formats,
+            schema: list_datasets_schema(),
+        }
+    }
+
+    /// What this table enumerates. Query-time authorization reads it, so the
+    /// provider is recognised rather than treated as unintrospectable.
+    pub fn listing(&self) -> &Listing {
+        &self.listing
+    }
+
+    /// How this listing prints in `EXPLAIN`.
+    fn label(&self) -> String {
+        match &self.listing {
+            Listing::Glob { pattern } => format!("glob={pattern}"),
+            Listing::Level { prefix } if prefix.is_empty() => "level=<root>".to_string(),
+            Listing::Level { prefix } => format!("level={prefix}"),
+        }
+    }
+}
+
+/// The first format that claims `object`, as a row.
+///
+/// Order is registration order, and the first claim wins: a format that reads an
+/// extension another also reads would otherwise produce two rows for one file.
+fn classify(formats: &[Arc<dyn FileFormatFactoryExt>], object: &ObjectMeta) -> Option<Row> {
+    formats
+        .iter()
+        .find_map(|format| format.classify_object(object))
+        .map(Row::from)
+}
+
+/// The tighter of the function's own limit and the planner push-down.
+fn effective_limit(declared: Option<usize>, pushed: Option<usize>) -> Option<usize> {
+    match (declared, pushed) {
+        (Some(a), Some(b)) => Some(a.min(b)),
+        (a, b) => a.or(b),
+    }
+}
+
+/// One directory level, as rows. Directories first, so it reads like one.
+///
+/// Folder names arrive relative to the level, and a row carries a full path, so
+/// they are joined back on here.
+fn level_rows(formats: &[Arc<dyn FileFormatFactoryExt>], level: ObjectLevel) -> Vec<Row> {
+    let base = level.prefix.trim_end_matches('/').to_string();
+    let join = |name: &str| {
+        if base.is_empty() {
+            name.to_string()
+        } else {
+            format!("{base}/{name}")
+        }
+    };
+    level
+        .folders
+        .iter()
+        .map(|name| Row::directory(join(name)))
+        .chain(level.objects.iter().filter_map(|object| classify(formats, object)))
+        .collect()
+}
+
+#[async_trait::async_trait]
+impl TableProvider for DatasetsTable {
+    fn as_any(&self) -> &dyn std::any::Any {
+        self
+    }
+
+    fn schema(&self) -> SchemaRef {
+        Arc::clone(&self.schema)
+    }
+
+    fn table_type(&self) -> TableType {
+        TableType::Base
+    }
+
+    /// Builds the plan. The walk itself runs when the plan is executed.
+    ///
+    /// `limit` is the planner push-down, and here it bounds the walk as well as
+    /// the rows: a glob listing is a lazy stream, so the node stopping stops the
+    /// pages behind it. `LIMIT 50` over a bucket of millions reads one page.
+    async fn scan(
+        &self,
+        state: &dyn Session,
+        projection: Option<&Vec<usize>>,
+        _filters: &[Expr],
+        limit: Option<usize>,
+    ) -> datafusion::error::Result<Arc<dyn ExecutionPlan>> {
+        let schema = self.schema();
+        let factory = state
+            .config()
+            .get_extension::<ListingFactory>()
+            .ok_or_else(|| {
+                DataFusionError::Execution(
+                    "the listing factory is not registered on the session".to_string(),
+                )
+            })?;
+
+        // The two listings differ only here.
+        let rows: RowStreamFactory = match &self.listing {
+            // A glob streams. The path is resolved here, against this session.
+            // The handle holds no session, so each execute rebuilds the walk
+            // from it and the plan can run more than once.
+            Listing::Glob { pattern } => {
+                let listing = factory.listing(state, pattern)?;
+                let formats = self.file_formats.clone();
+                Arc::new(move || {
+                    let formats = formats.clone();
+                    Ok(listing
+                        .stream()
+                        .filter_map(move |object| {
+                            let formats = formats.clone();
+                            async move {
+                                match object {
+                                    // An object no format claims is not a dataset,
+                                    // and not an error either.
+                                    Ok(object) => classify(&formats, &object).map(Ok),
+                                    Err(e) => Some(Err(e)),
+                                }
+                            }
+                        })
+                        .boxed())
+                })
+            }
+            // One level is a single request. `scan` is async, so it is answered
+            // here and the rows are what the plan replays.
+            Listing::Level { prefix } => {
+                let level = factory.listing(state, prefix)?.level().await?;
+                let rows = Arc::new(level_rows(&self.file_formats, level));
+                Arc::new(move || {
+                    let rows = Arc::clone(&rows);
+                    Ok(futures::stream::iter((0..rows.len()).map(move |i| Ok(rows[i].clone())))
+                        .boxed())
+                })
+            }
+        };
+
+        let plan = Arc::new(DatasetsExec::new(
+            Arc::clone(&schema),
+            rows,
+            self.offset,
+            effective_limit(self.limit, limit),
+            self.label(),
+        ));
+
+        // The node produces the full row shape, so a projection sits above it.
+        match projection {
+            Some(projection) => {
+                let exprs: Vec<(Arc<dyn PhysicalExpr>, String)> = projection
+                    .iter()
+                    .map(|index| {
+                        let field = schema.field(*index);
+                        (
+                            Arc::new(Column::new(field.name(), *index)) as Arc<dyn PhysicalExpr>,
+                            field.name().to_string(),
+                        )
+                    })
+                    .collect();
+                Ok(Arc::new(ProjectionExec::try_new(exprs, plan)?))
+            }
+            None => Ok(plan),
+        }
+    }
+}
diff --git a/beacon-server/beacon-server/src/api.rs b/beacon-server/beacon-server/src/api.rs
index 188dff46..0d16ada9 100644
--- a/beacon-server/beacon-server/src/api.rs
+++ b/beacon-server/beacon-server/src/api.rs
@@ -62,6 +62,26 @@ pub struct DatasetInfo {
     pub last_modified: Option<String>,
 }
 
+/// One directory level of the datasets store.
+///
+/// What a folder view needs, and nothing more. `/api/list-datasets` walks the
+/// whole subtree under its pattern; this reads a single level, so its cost does
+/// not grow with the size of the store below it.
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, utoipa::ToSchema)]
+pub struct BrowseDatasetsResponse {
+    /// The directory this describes, relative to the datasets store root.
+    #[schema(example = "argo/floats")]
+    pub prefix: String,
+    /// Immediate sub-folder names, relative to `prefix`, sorted.
+    ///
+    /// A directory-shaped dataset (Zarr, Atlas) appears here rather than in
+    /// `datasets`: its marker lives inside the directory, one level down.
+    #[schema(example = json!(["2023", "2024"]))]
+    pub folders: Vec<String>,
+    /// Datasets sitting directly in `prefix`, sorted by path.
+    pub datasets: Vec<DatasetInfo>,
+}
+
 impl From<DatasetMetadata> for DatasetInfo {
     fn from(value: DatasetMetadata) -> Self {
         Self {
diff --git a/beacon-server/beacon-server/src/axum/client/datasets.rs b/beacon-server/beacon-server/src/axum/client/datasets.rs
index 2069dfdf..462d0175 100644
--- a/beacon-server/beacon-server/src/axum/client/datasets.rs
+++ b/beacon-server/beacon-server/src/axum/client/datasets.rs
@@ -7,7 +7,7 @@ use ::axum::{
     http::StatusCode,
     Extension, Json,
 };
-use crate::api::DatasetInfo;
+use crate::api::{BrowseDatasetsResponse, DatasetInfo};
 use beacon_core::AuthIdentity;
 use crate::server::{catalog, Server};
 use utoipa::{IntoParams, ToSchema};
@@ -178,4 +178,51 @@ pub(crate) async fn total_datasets(
             ))
         }
     }
-}
\ No newline at end of file
+}
+/// Query parameters for a single-level dataset browse.
+#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, ToSchema, IntoParams)]
+pub struct BrowseDatasetsQuery {
+    /// Directory to read, relative to the datasets store. Empty or absent means
+    /// the root.
+    pub prefix: Option<String>,
+}
+
+/// Reads one directory level of the datasets store.
+///
+/// The endpoint a folder view should call. `/api/list-datasets` enumerates every
+/// object under its pattern before it can answer, so its cost tracks the size of
+/// the store; this issues one delimiter request, so it does not. On a store of
+/// 2.85 million objects the difference measured 79.9 s against 14 ms.
+#[tracing::instrument(level = "info", skip(state))]
+#[utoipa::path(
+    tag = "datasets",
+    get,
+    path = "/api/browse-datasets",
+    params(BrowseDatasetsQuery),
+    responses(
+        (status = 200, description = "One directory level", body = BrowseDatasetsResponse),
+        (status = 500, description = "Failed to browse datasets"),
+    ),
+    security(
+        (),
+        ("basic-auth" = []),
+        ("bearer" = [])
+    )
+)]
+pub(crate) async fn browse_datasets(
+    State(state): State<Arc<Server>>,
+    Extension(identity): Extension<AuthIdentity>,
+    Query(query): Query<BrowseDatasetsQuery>,
+) -> Result<Json<BrowseDatasetsResponse>, (StatusCode, String)> {
+    let prefix = query.prefix.unwrap_or_default();
+    match catalog::browse_datasets(&state, &prefix, identity).await {
+        Ok(result) => Ok(Json(result)),
+        Err(err) => {
+            tracing::error!("Error browsing datasets: {:?}", err);
+            Err((
+                StatusCode::INTERNAL_SERVER_ERROR,
+                "Error browsing datasets".to_string(),
+            ))
+        }
+    }
+}
diff --git a/beacon-server/beacon-server/src/axum/client/mod.rs b/beacon-server/beacon-server/src/axum/client/mod.rs
index fc957675..eed8f6bd 100644
--- a/beacon-server/beacon-server/src/axum/client/mod.rs
+++ b/beacon-server/beacon-server/src/axum/client/mod.rs
@@ -39,6 +39,7 @@ pub(crate) fn setup_client_router() -> (Router<Arc<Server>>, utoipa::openapi::Op
         .routes(routes!(query::available_columns))
         .routes(routes!(datasets::datasets))
         .routes(routes!(datasets::list_datasets))
+        .routes(routes!(datasets::browse_datasets))
         .routes(routes!(datasets::list_dataset_schema))
         .routes(routes!(datasets::total_datasets))
         .routes(routes!(tables::list_tables))
diff --git a/beacon-server/beacon-server/src/server/catalog.rs b/beacon-server/beacon-server/src/server/catalog.rs
index 607f1de0..20089f65 100644
--- a/beacon-server/beacon-server/src/server/catalog.rs
+++ b/beacon-server/beacon-server/src/server/catalog.rs
@@ -214,6 +214,61 @@ pub(crate) async fn list_datasets(
         .collect())
 }
 
+/// One directory level of the datasets store.
+///
+/// Goes through the same SQL surface as [`list_datasets`], because it is the
+/// same provider: `browse_datasets` stops at one level and reports the
+/// sub-directories as rows. Splitting those rows out here is all this adds.
+pub(crate) async fn browse_datasets(
+    server: &Arc<Server>,
+    prefix: &str,
+    identity: AuthIdentity,
+) -> anyhow::Result<crate::api::BrowseDatasetsResponse> {
+    let sql = format!("SELECT * FROM browse_datasets({})", quote_literal(prefix));
+    let rows = query_rows(server, sql, identity).await?;
+
+    let is_dir = |row: &Value| {
+        row.get("is_directory")
+            .and_then(Value::as_bool)
+            .unwrap_or(false)
+    };
+
+    let folders = rows
+        .iter()
+        .filter(|row| is_dir(row))
+        // Rows carry the full path; a folder view wants the name below `prefix`.
+        .map(|row| {
+            let full = str_field(row, "file_name");
+            full.rsplit('/').next().unwrap_or(full).to_string()
+        })
+        .collect();
+
+    let datasets = rows
+        .iter()
+        .filter(|row| !is_dir(row))
+        .map(|row| DatasetInfo {
+            file_path: str_field(row, "file_name").to_string(),
+            format: str_field(row, "file_format").to_string(),
+            can_inspect: row.get("can_inspect").and_then(Value::as_bool).unwrap_or(false),
+            can_partial_explore: row
+                .get("can_partial_explore")
+                .and_then(Value::as_bool)
+                .unwrap_or(false),
+            size: row.get("size").and_then(Value::as_u64),
+            last_modified: row
+                .get("last_modified")
+                .and_then(Value::as_str)
+                .map(str::to_string),
+        })
+        .collect();
+
+    Ok(crate::api::BrowseDatasetsResponse {
+        prefix: prefix.trim_end_matches('/').to_string(),
+        folders,
+        datasets,
+    })
+}
+
 /// The `read_*` table function that reads a dataset file with the given
 /// extension, or `None` when the extension is not one beacon reads by path.
 fn read_function_for_extension(ext: &str) -> Option<&'static str> {
diff --git a/beacon-server/beacon-server/src/server/mod.rs b/beacon-server/beacon-server/src/server/mod.rs
index 48d053c7..9e4a706a 100644
--- a/beacon-server/beacon-server/src/server/mod.rs
+++ b/beacon-server/beacon-server/src/server/mod.rs
@@ -244,6 +244,11 @@ fn build_datasets_store(config: &Config) -> anyhow::Result<(Arc<dyn ObjectStore>
     let store = builder
         .build()
         .with_context(|| format!("failed to open the S3 datasets store for bucket '{bucket}'"))?;
+    // Shard the recursive listing and size its pages. This store is registered
+    // verbatim under `datasets://`, so it never passes through the registry that
+    // builds ad-hoc `s3://` stores — the wrapping has to happen here or the
+    // datasets store keeps the one-request-per-1000-objects walk.
+    let store = beacon_datafusion_ext::big_page_list::BigPageList::new(store);
 
     let base = s3
         .native_base_url()
diff --git a/beacon-server/beacon-server/tests/api_contract_http.rs b/beacon-server/beacon-server/tests/api_contract_http.rs
index 9cb8fe2b..ff6c52bd 100644
--- a/beacon-server/beacon-server/tests/api_contract_http.rs
+++ b/beacon-server/beacon-server/tests/api_contract_http.rs
@@ -54,6 +54,7 @@ const CLIENT_ENDPOINTS: &[(&str, &str)] = &[
     ("get", "/api/default-table-schema"),
     // -- datasets ------------------------------------------------------------
     ("get", "/api/list-datasets"),
+    ("get", "/api/browse-datasets"),
     ("get", "/api/dataset-schema"),
     ("get", "/api/total-datasets"),
     // -- functions & info ----------------------------------------------------
@@ -196,6 +197,7 @@ async fn admin_alias_get_endpoints_answer_successfully() {
         "/admin/api/tables",
         "/admin/api/tables-with-schema",
         "/admin/api/list-datasets",
+        "/admin/api/browse-datasets",
         "/admin/api/functions",
         "/admin/api/admin/check",
         "/admin/api/admin/auth/users",
@@ -372,6 +374,7 @@ async fn client_get_endpoints_answer_successfully() {
         ("/api/tables-with-schema", false),
         ("/api/default-table", false),
         ("/api/list-datasets", false),
+        ("/api/browse-datasets", false),
         ("/api/total-datasets", false),
         ("/api/functions", false),
         ("/api/table-functions", false),
diff --git a/docs/docs/2.0.0-rc5/api/exploring-data.md b/docs/docs/2.0.0-rc5/api/exploring-data.md
index 1544152f..7d2e9263 100644
--- a/docs/docs/2.0.0-rc5/api/exploring-data.md
+++ b/docs/docs/2.0.0-rc5/api/exploring-data.md
@@ -42,6 +42,52 @@ The optional query parameters are:
 GET /api/list-datasets?pattern=argo/**/*.nc&limit=50&offset=0
 ```
 
+This endpoint reads every object below `pattern`. The time increases with the size of the store.
+Use `/api/browse-datasets` for a folder view.
+
+### Browse one directory level
+
+```http
+GET /api/browse-datasets
+```
+
+Returns the files and the sub-directories of one directory. Beacon does not descend, so the time
+does not increase with the size of the store below the directory.
+
+The optional query parameter is:
+
+| Parameter | Description |
+| --------- | ----------- |
+| `prefix` | The directory to read. The default is the storage root |
+
+```http
+GET /api/browse-datasets?prefix=argo/floats
+```
+
+The response holds the directory, the sub-directory names and the datasets:
+
+```json
+{
+  "prefix": "argo/floats",
+  "folders": ["2023", "2024"],
+  "datasets": [
+    {
+      "file_path": "argo/floats/index.parquet",
+      "format": "parquet",
+      "can_inspect": true,
+      "can_partial_explore": true,
+      "size": 20480,
+      "last_modified": "2026-01-14T09:12:03Z"
+    }
+  ]
+}
+```
+
+::: info A directory-shaped dataset
+A Zarr store is a directory. Beacon reports it in `folders`, not in `datasets`. Descend into it to
+see the dataset.
+:::
+
 ### Dataset count
 
 ```http
diff --git a/docs/docs/2.0.0-rc5/formats/zarr.md b/docs/docs/2.0.0-rc5/formats/zarr.md
index 04a60e7e..3929c510 100644
--- a/docs/docs/2.0.0-rc5/formats/zarr.md
+++ b/docs/docs/2.0.0-rc5/formats/zarr.md
@@ -52,6 +52,21 @@ multi-dimensional store.
 - One external table can hold several Zarr stores. Use a glob pattern such as `sst/*/zarr.json`.
 - Beacon fully supports S3 and other object stores.
 
+### Discovery in the datasets store
+
+`list_datasets` and `browse_datasets` report a Zarr store as one dataset. A store is a directory
+with the `.zarr` suffix that holds `zarr.json`:
+
+```text
+sst/gridded-example.zarr/zarr.json      the store
+sst/gridded-example.zarr/lat/zarr.json  an array inside it
+```
+
+Zarr v3 gives every group and every array a `zarr.json`, so the marker alone does not name the
+store. The suffix does. A store in a directory without the suffix stays out of the listing.
+
+`read_zarr` needs no suffix. The path already names the store.
+
 ### Array attributes
 
 A Zarr array holds its attributes in the `attributes` section of its `zarr.json`. Beacon shows these
diff --git a/docs/docs/2.0.0-rc5/sql/table-functions-utility.md b/docs/docs/2.0.0-rc5/sql/table-functions-utility.md
index 387d41a5..d4f1d33a 100644
--- a/docs/docs/2.0.0-rc5/sql/table-functions-utility.md
+++ b/docs/docs/2.0.0-rc5/sql/table-functions-utility.md
@@ -50,23 +50,83 @@ WHERE data_type LIKE 'Timestamp%';
 ## `list_datasets`
 
 ```text
-list_datasets()
+list_datasets(pattern, offset, limit)
 ```
 
-Lists every file in the dataset storage root of Beacon. Returns one row for each file.
+Lists every file in the dataset storage root of Beacon. Returns one row for each file. Beacon
+descends the full tree below `pattern`.
+
+All three arguments are optional and positional:
+
+| Argument | Type | Description |
+| -------- | ---- | ----------- |
+| `pattern` | `TEXT` | A glob that filters the paths. The default is `**/*` |
+| `offset` | `INTEGER` | The number of rows to skip. The default is `0` |
+| `limit` | `INTEGER` | The maximum number of rows. The default is no limit |
+
+The output columns are:
 
 | Column | Type | Description |
 | ------ | ---- | ----------- |
 | `file_name` | `TEXT` | The path, relative to the storage root |
 | `file_format` | `TEXT` | The format that Beacon detects |
+| `can_inspect` | `BOOLEAN` | Whether Beacon reads a schema from this file |
+| `can_partial_explore` | `BOOLEAN` | Whether Beacon reads a part of this file |
+| `size` | `UBIGINT` | The size in bytes. Null when the store reports none |
+| `last_modified` | `TEXT` | The modification time, in RFC 3339. Null when the store reports none |
+| `is_directory` | `BOOLEAN` | Always false here. See `browse_datasets` |
 
 ```sql
 SELECT * FROM list_datasets()
 
 -- Find all NetCDF files
 SELECT file_name FROM list_datasets() WHERE file_format = 'nc'
+
+-- One page of one subtree
+SELECT file_name FROM list_datasets('argo/**/*.nc', 0, 50)
+```
+
+A `LIMIT` stops the walk. Beacon reads one page of the store for `LIMIT 50`, not the full store.
+
+## `browse_datasets`
+
+```text
+browse_datasets(prefix, offset, limit)
+```
+
+Lists one directory level of the dataset storage root. Returns one row for each file and one row
+for each sub-directory. Beacon does not descend.
+
+All three arguments are optional and positional:
+
+| Argument | Type | Description |
+| -------- | ---- | ----------- |
+| `prefix` | `TEXT` | The directory to read. The default is the storage root |
+| `offset` | `INTEGER` | The number of rows to skip. The default is `0` |
+| `limit` | `INTEGER` | The maximum number of rows. The default is no limit |
+
+The output columns are the columns of `list_datasets`. A sub-directory row sets `is_directory` and
+holds the directory path in `file_name`.
+
+Use this function for a folder view. One level costs one request, so the time does not increase
+with the size of the store below `prefix`. `list_datasets` reads every object below its pattern.
+
+```sql
+-- The root level
+SELECT * FROM browse_datasets()
+
+-- The files of one directory
+SELECT file_name FROM browse_datasets('argo') WHERE NOT is_directory
+
+-- The sub-directories of one directory
+SELECT file_name FROM browse_datasets('argo') WHERE is_directory
 ```
 
+::: info A directory-shaped dataset
+A Zarr store is a directory. At its own level Beacon reports it as a sub-directory. Descend into it
+to see the dataset row.
+:::
+
 ## `view_dataset_statistics`
 
 ```text