From 9300fd49b78b7150fb61c6040c4757ce95e8ee5e Mon Sep 17 00:00:00 2001 From: Robin Kooyman Date: Thu, 20 Aug 2026 00:35:43 +0200 Subject: [PATCH 01/21] Request large listing pages from S3, and repair the S3 back-compat flag `ObjectStore::list` sends no `max-keys`, so a server applies its own default of 1000 per page. A recursive walk of a large bucket becomes a long chain of strictly sequential requests, and the listing is pure wait rather than work. Measured against a SeaweedFS bucket of 2 853 217 objects: the walk cost 2854 requests, and the same walk with max-keys=65535 cost 44. `BigPageList` wraps a `PaginatedListStore` and re-expresses `list` over `list_paginated` with an explicit page size. The stream stays lazy, so a caller that stops early still stops paying. Every other method delegates. AWS caps max-keys at 1000 and ignores a larger value, so this is inert there. Also: `s3_data_lake_deprecated` read `BEACON_S3_DATASETS`, the new name, which left the back-compat path dead. A 1.x deployment carrying only `BEACON_S3_DATA_LAKE=true` fell back to the local datasets directory and reported an empty store. --- .../src/big_page_list.rs | 144 ++++++++++++++++++ beacon-db/beacon-datafusion-ext/src/lib.rs | 1 + .../src/object_store_registry.rs | 7 +- beacon-server/beacon-server-config/src/lib.rs | 6 +- 4 files changed, 155 insertions(+), 3 deletions(-) create mode 100644 beacon-db/beacon-datafusion-ext/src/big_page_list.rs 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..08d7c988 --- /dev/null +++ b/beacon-db/beacon-datafusion-ext/src/big_page_list.rs @@ -0,0 +1,144 @@ +//! A store wrapper that asks for large listing pages. +//! +//! [`ObjectStore::list`] sends no `max-keys`, so a server applies its own +//! default of 1000 keys per page. A recursive listing of a large bucket is +//! therefore a long chain of small, strictly sequential requests: each page +//! needs the continuation token of the one before it, so nothing overlaps. +//! +//! Measured against a SeaweedFS bucket holding 2 853 217 objects, the walk took +//! 2854 requests. The same walk with `max-keys=65535` took 44. On a co-located +//! deployment the saving is the per-request round trip, which is the whole cost +//! of the walk: the listing is pure wait, not work. +//! +//! [`BigPageList`] wraps a store that implements [`PaginatedListStore`] and +//! re-expresses `list` over it with an explicit page size. Every other method +//! delegates untouched. +//! +//! # Page size +//! +//! [`DEFAULT_MAX_KEYS`] is 65535. AWS caps `max-keys` at 1000 and ignores +//! anything larger, so this is a no-op there rather than an error. SeaweedFS +//! parses the parameter as a `uint16` and honours the whole range. A server +//! that returns fewer keys than asked is already the normal case, and the +//! continuation token handles it. + +use std::fmt; +use std::sync::Arc; + +use futures::stream::{BoxStream, StreamExt, TryStreamExt}; +use object_store::list::{PaginatedListOptions, PaginatedListStore}; +use object_store::path::Path; +use object_store::{ + CopyOptions, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore, + PutMultipartOptions, PutOptions, PutPayload, PutResult, Result as OsResult, +}; + +/// Keys requested per listing page. See the module docs for why this value. +pub const DEFAULT_MAX_KEYS: usize = 65535; + +/// A store whose `list` asks for [`DEFAULT_MAX_KEYS`] keys per page. +#[derive(Debug, Clone)] +pub struct BigPageList { + inner: Arc, + max_keys: usize, +} + +impl BigPageList { + /// Wrap `inner`, requesting [`DEFAULT_MAX_KEYS`] keys per page. + pub fn new(inner: T) -> Self { + Self::with_max_keys(inner, DEFAULT_MAX_KEYS) + } + + /// The same, with the page size named. + pub fn with_max_keys(inner: T, max_keys: usize) -> Self { + Self { + inner: Arc::new(inner), + max_keys: max_keys.max(1), + } + } +} + +impl fmt::Display for BigPageList { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "BigPageList(max_keys={}, {:?})", self.max_keys, self.inner) + } +} + +#[async_trait::async_trait] +impl ObjectStore for BigPageList +where + T: ObjectStore + PaginatedListStore + 'static, +{ + async fn put_opts( + &self, + location: &Path, + payload: PutPayload, + opts: PutOptions, + ) -> OsResult { + self.inner.put_opts(location, payload, opts).await + } + + async fn put_multipart_opts( + &self, + location: &Path, + opts: PutMultipartOptions, + ) -> OsResult> { + self.inner.put_multipart_opts(location, opts).await + } + + async fn get_opts(&self, location: &Path, options: GetOptions) -> OsResult { + self.inner.get_opts(location, options).await + } + + fn delete_stream( + &self, + locations: BoxStream<'static, OsResult>, + ) -> BoxStream<'static, OsResult> { + self.inner.delete_stream(locations) + } + + /// The recursive listing, page by page, with an explicit `max-keys`. + /// + /// The stream stays lazy: a page is fetched only when the consumer has + /// drained the one before it, so a caller that stops early (a `LIMIT`, a + /// prefix match) stops paying. That mirrors the stream `list` returns. + fn list(&self, prefix: Option<&Path>) -> BoxStream<'static, OsResult> { + let inner = Arc::clone(&self.inner); + let prefix = prefix.cloned(); + let max_keys = self.max_keys; + + futures::stream::try_unfold(Some(None::), move |state| { + let inner = Arc::clone(&inner); + let prefix = prefix.clone(); + async move { + // `None` state means the previous page was the last one. + 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_ref().map(|p| p.as_ref()), opts) + .await?; + let next = page.page_token.map(Some); + Ok(Some(( + futures::stream::iter(page.result.objects.into_iter().map(Ok)), + next, + ))) + } + }) + .try_flatten() + .boxed() + } + + async fn list_with_delimiter(&self, prefix: Option<&Path>) -> OsResult { + 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 + } +} diff --git a/beacon-db/beacon-datafusion-ext/src/lib.rs b/beacon-db/beacon-datafusion-ext/src/lib.rs index a3734334..5b4fc32a 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/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-server/beacon-server-config/src/lib.rs b/beacon-server/beacon-server-config/src/lib.rs index 05acf589..916fe2f6 100644 --- a/beacon-server/beacon-server-config/src/lib.rs +++ b/beacon-server/beacon-server-config/src/lib.rs @@ -356,7 +356,11 @@ struct RawConfig { // Former name of `BEACON_S3_DATASETS`, kept so existing deployments keep working. // `Config::load` warns when it is the one that turned the S3 store on. Remove it // one major version after 2.0. - #[envconfig(from = "BEACON_S3_DATASETS", default = "false")] + // + // This read the *new* name until 2.0.0-rc.4, which made the whole back-compat + // path dead: a 1.x deployment carrying only `BEACON_S3_DATA_LAKE=true` fell + // silently back to the local `datasets/` directory and reported an empty store. + #[envconfig(from = "BEACON_S3_DATA_LAKE", default = "false")] s3_data_lake_deprecated: bool, #[envconfig(from = "BEACON_S3_BUCKET")] s3_bucket: Option, From 1a0890b32616fdc50e5d024745c97ddefabf7a1b Mon Sep 17 00:00:00 2001 From: Robin Kooyman Date: Thu, 20 Aug 2026 00:58:45 +0200 Subject: [PATCH 02/21] Shard a recursive listing across sub-directories, and size the pages A page chain is strictly sequential: each page needs the continuation token of the one before it, so a walk costs one round trip per page however fast the server is. `list` now descends `list_with_delimiter` three levels to find shards, then walks each shard as its own page chain, 16 at a time. Discovery costs one request per directory seen and finished in under a second. Objects at the intermediate levels are emitted too, so nothing is missed. A prefix with no sub-directories falls back to the sequential walk, as does a store that cannot answer a delimiter listing. Measured against a SeaweedFS bucket of 2 853 217 objects, warm and back to back. Every run returned the full count: sequential, no max-keys (before) 79.9s 2854 requests sequential, max-keys=5000 64.3s 571 requests 2-level shards, 16 ways 33.8s 3-level shards, 16 ways 19.4s 3-level shards, 32 ways 17.7s DEFAULT_MAX_KEYS is 5000, not the 65535 SeaweedFS will parse. Larger pages measured worse and they broke: at 65535 with 8 to 16 shards in flight, SeaweedFS failed whole responses with a body error and the walk silently returned 2 288 498 of 2 853 217 objects. Shards interleave, so results are no longer incidentally sorted. `ObjectStore::list` does not guarantee order. --- .../src/big_page_list.rs | 248 ++++++++++++++---- 1 file changed, 193 insertions(+), 55 deletions(-) diff --git a/beacon-db/beacon-datafusion-ext/src/big_page_list.rs b/beacon-db/beacon-datafusion-ext/src/big_page_list.rs index 08d7c988..fca38e9e 100644 --- a/beacon-db/beacon-datafusion-ext/src/big_page_list.rs +++ b/beacon-db/beacon-datafusion-ext/src/big_page_list.rs @@ -1,26 +1,49 @@ -//! A store wrapper that asks for large listing pages. +//! A store wrapper that lists a prefix in parallel shards, with sized pages. //! -//! [`ObjectStore::list`] sends no `max-keys`, so a server applies its own -//! default of 1000 keys per page. A recursive listing of a large bucket is -//! therefore a long chain of small, strictly sequential requests: each page -//! needs the continuation token of the one before it, so nothing overlaps. +//! 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. //! -//! Measured against a SeaweedFS bucket holding 2 853 217 objects, the walk took -//! 2854 requests. The same walk with `max-keys=65535` took 44. On a co-located -//! deployment the saving is the per-request round trip, which is the whole cost -//! of the walk: the listing is pure wait, not work. +//! Two changes, both measured against a SeaweedFS bucket of that size. Every +//! run below returned the full 2 853 217 objects. //! -//! [`BigPageList`] wraps a store that implements [`PaginatedListStore`] and -//! re-expresses `list` over it with an explicit page size. Every other method -//! delegates untouched. +//! | 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 //! -//! [`DEFAULT_MAX_KEYS`] is 65535. AWS caps `max-keys` at 1000 and ignores -//! anything larger, so this is a no-op there rather than an error. SeaweedFS -//! parses the parameter as a `uint16` and honours the whole range. A server -//! that returns fewer keys than asked is already the normal case, and the -//! continuation token handles it. +//! [`ObjectStore::list`] sends no `max-keys`, so a server applies its own +//! default of 1000. [`DEFAULT_MAX_KEYS`] is 5000, not the 65535 SeaweedFS will +//! parse. Larger pages measured *worse*, and they broke: at 65535 with 8 to 16 +//! shards in flight, SeaweedFS failed whole responses with a body error and the +//! walk silently returned 2 288 498 of 2 853 217 objects. A moderate page keeps +//! the response small enough to survive concurrency, and it already removes 80% +//! of the round trips. +//! +//! # 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; @@ -33,35 +56,145 @@ use object_store::{ PutMultipartOptions, PutOptions, PutPayload, PutResult, Result as OsResult, }; -/// Keys requested per listing page. See the module docs for why this value. -pub const DEFAULT_MAX_KEYS: usize = 65535; +/// 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; -/// A store whose `list` asks for [`DEFAULT_MAX_KEYS`] keys per page. +/// 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 { inner: Arc, max_keys: usize, + fanout_depth: usize, + concurrency: usize, } impl BigPageList { - /// Wrap `inner`, requesting [`DEFAULT_MAX_KEYS`] keys per page. + /// Wrap `inner` with the measured defaults. pub fn new(inner: T) -> Self { - Self::with_max_keys(inner, DEFAULT_MAX_KEYS) - } - - /// The same, with the page size named. - pub fn with_max_keys(inner: T, max_keys: usize) -> Self { Self { inner: Arc::new(inner), - max_keys: max_keys.max(1), + 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 fmt::Display for BigPageList { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - write!(f, "BigPageList(max_keys={}, {:?})", self.max_keys, self.inner) + write!( + f, + "BigPageList(max_keys={}, depth={}, concurrency={}, {:?})", + self.max_keys, self.fanout_depth, self.concurrency, self.inner + ) + } +} + +/// One prefix, walked to exhaustion as a lazy chain of sized pages. +fn walk_shard( + inner: Arc, + prefix: Option, + max_keys: usize, +) -> BoxStream<'static, OsResult> +where + T: PaginatedListStore + 'static, +{ + // `None` state means the previous page carried no continuation token. + futures::stream::try_unfold(Some(None::), 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_ref().map(|p| p.as_ref()), 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( + inner: &Arc, + root: Option, + depth: usize, + concurrency: usize, +) -> OsResult<(Vec>, Vec)> +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::>() + .await?; + + let mut kids = Vec::new(); + for r in expanded { + kids.extend(r.common_prefixes.into_iter().map(Some)); + above.extend(r.objects); + } + if kids.is_empty() { + break; + } + level = kids; } + + Ok((level, above)) } #[async_trait::async_trait] @@ -97,40 +230,45 @@ where self.inner.delete_stream(locations) } - /// The recursive listing, page by page, with an explicit `max-keys`. + /// Discover shards, then walk them concurrently. /// - /// The stream stays lazy: a page is fetched only when the consumer has - /// drained the one before it, so a caller that stops early (a `LIMIT`, a - /// prefix match) stops paying. That mirrors the stream `list` returns. + /// 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> { let inner = Arc::clone(&self.inner); let prefix = prefix.cloned(); - let max_keys = self.max_keys; - - futures::stream::try_unfold(Some(None::), move |state| { - let inner = Arc::clone(&inner); - let prefix = prefix.clone(); - async move { - // `None` state means the previous page was the last one. - 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 (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), }; - let page = inner - .list_paginated(prefix.as_ref().map(|p| p.as_ref()), opts) - .await?; - let next = page.page_token.map(Some); - Ok(Some(( - futures::stream::iter(page.result.objects.into_iter().map(Ok)), - next, - ))) + + 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() }) - .try_flatten() + .flatten() .boxed() } From 7153912e2c1c44162c8868ec7505dfaae6c97e37 Mon Sep 17 00:00:00 2001 From: Robin Kooyman Date: Thu, 20 Aug 2026 01:04:11 +0200 Subject: [PATCH 03/21] Propagate listing errors, and correct why the page size is capped `list_datasets` matched `if let Ok(entry)` over the listing stream and dropped everything else. A transient object-store failure part-way through a walk therefore produced 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. The error now propagates, naming how far the walk got. The page-size note blamed SeaweedFS for failing large responses. It is not a server limit. `ClientOptions` gives a request 30 seconds by default; one page of 65535 keys off a cold filer took 6.5s on its own, and with 16 shards in flight some requests crossed 30s. Re-run warm, the same walk returned all 2 853 217 objects with no errors, at both the default timeout and a raised one. 5000 stays the default because it holds whatever the filer's cache is doing; `AWS_TIMEOUT` is the knob to raise first for anyone who wants larger pages. --- .../beacon-datafusion-ext/src/big_page_list.rs | 16 ++++++++++------ .../beacon-datafusion-ext/src/listing_factory.rs | 14 +++++++++++--- 2 files changed, 21 insertions(+), 9 deletions(-) diff --git a/beacon-db/beacon-datafusion-ext/src/big_page_list.rs b/beacon-db/beacon-datafusion-ext/src/big_page_list.rs index fca38e9e..4f108df8 100644 --- a/beacon-db/beacon-datafusion-ext/src/big_page_list.rs +++ b/beacon-db/beacon-datafusion-ext/src/big_page_list.rs @@ -19,12 +19,16 @@ //! # Page size //! //! [`ObjectStore::list`] sends no `max-keys`, so a server applies its own -//! default of 1000. [`DEFAULT_MAX_KEYS`] is 5000, not the 65535 SeaweedFS will -//! parse. Larger pages measured *worse*, and they broke: at 65535 with 8 to 16 -//! shards in flight, SeaweedFS failed whole responses with a body error and the -//! walk silently returned 2 288 498 of 2 853 217 objects. A moderate page keeps -//! the response small enough to survive concurrency, and it already removes 80% -//! of the round trips. +//! 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 //! diff --git a/beacon-db/beacon-datafusion-ext/src/listing_factory.rs b/beacon-db/beacon-datafusion-ext/src/listing_factory.rs index 0dfd1688..2a6b5bff 100644 --- a/beacon-db/beacon-datafusion-ext/src/listing_factory.rs +++ b/beacon-db/beacon-datafusion-ext/src/listing_factory.rs @@ -206,9 +206,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. From 78e81ec0d25e423d763b3ceeabc35ac82fa96057 Mon Sep 17 00:00:00 2001 From: Robin Kooyman Date: Thu, 20 Aug 2026 01:28:57 +0200 Subject: [PATCH 04/21] Browse the datasets store one directory at a time The datasets page enumerated the whole store to draw a folder. It asked for 100 000 rows, grouped them into folders in the browser, and asked `/api/total-datasets` for a count that also walked everything. Both ran on mount and again after every upload. Nothing about a folder view needs the subtree, and the gap is not a constant factor. Against a SeaweedFS bucket of 2 853 217 objects the recursive walk took 79.9s and one delimiter request took 14 ms. `GET /api/browse-datasets?prefix=` returns one level: sub-folder names and the datasets sitting directly in it. `ListingFactory::browse_datasets` does the delimiter listing and classifies only what is at that level; `Runtime::browse_datasets` exposes it without going through SQL, since there is no pattern to plan and no rows to page. The page now browses per folder. Search is the one view that genuinely needs every path, so it is the one view that pays for them, and only once something has been typed. The total badge counts the current folder rather than walking the store. A directory-shaped dataset (Zarr, Atlas) keeps its marker inside its directory, so at this level it reads as a folder; descending shows the marker. Classifying it in place would cost one request per sub-folder, which is the recursive cost this removes. Folder item counts are gone for the same reason. Authorization matches `list_datasets`: authenticated, no per-path grant filtering. `GRANT ... ON PATH` still governs reading a dataset. Moving both listings onto granted paths is a separate change and has to move them together. --- beacon-clients/beacon-ts/src/client.ts | 13 +++ .../beacon-web/src/pages/datasets.tsx | 81 ++++++++--------- beacon-db/beacon-core/src/runtime.rs | 42 +++++++++ beacon-db/beacon-core/src/runtime_builder.rs | 3 +- .../src/listing_factory.rs | 86 +++++++++++++++++++ beacon-server/beacon-server/src/api.rs | 20 +++++ .../beacon-server/src/axum/client/datasets.rs | 51 ++++++++++- .../beacon-server/src/axum/client/mod.rs | 1 + .../beacon-server/src/server/catalog.rs | 18 ++++ 9 files changed, 268 insertions(+), 47 deletions(-) diff --git a/beacon-clients/beacon-ts/src/client.ts b/beacon-clients/beacon-ts/src/client.ts index 9c832c57..5a6defe4 100644 --- a/beacon-clients/beacon-ts/src/client.ts +++ b/beacon-clients/beacon-ts/src/client.ts @@ -310,6 +310,19 @@ export class BeaconClient { return this.http.fetchJson("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(prefix?: string): Promise { + return this.http.fetchJson("GET", "/api/browse-datasets", { + query: { prefix }, + }); + } + /** Counts the total number of datasets (`GET /api/total-datasets`). */ totalDatasets(): Promise { return this.http.fetchJson("GET", "/api/total-datasets"); diff --git a/beacon-clients/beacon-web/src/pages/datasets.tsx b/beacon-clients/beacon-web/src/pages/datasets.tsx index 358bf6ce..26fc5ea3 100644 --- a/beacon-clients/beacon-web/src/pages/datasets.tsx +++ b/beacon-clients/beacon-web/src/pages/datasets.tsx @@ -118,38 +118,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(); - 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); @@ -164,7 +137,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. */ @@ -266,8 +239,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"] }); }, [qc]); const deleteMutation = useMutation({ @@ -352,14 +325,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(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 @@ -373,15 +363,18 @@ 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]); function toggleSort(key: SortKey) { setSort((s) => (s.key === key ? { key, dir: s.dir === "asc" ? "desc" : "asc" } : { key, dir: "asc" })); @@ -440,7 +433,7 @@ export function DatasetsPage() { {row.name} - {row.count} item{row.count === 1 ? "" : "s"} + {row.count === undefined ? "" : `${row.count} item${row.count === 1 ? "" : "s"}`} @@ -527,8 +520,8 @@ export function DatasetsPage() { description="Browse the datasets store. Preview rows or inspect a file's schema." actions={
- {typeof totalQuery.data === "number" && ( - {totalQuery.data} total + {rows.length > 0 && ( + {items.length} here )}