From 32716f900da2baae84c1428501e03219349f40a3 Mon Sep 17 00:00:00 2001 From: Will Jones Date: Mon, 17 Aug 2026 17:22:22 -0700 Subject: [PATCH] fix(listing): paginate table listing instead of enumerating the database `ListingDatabase::list_tables` listed every table directory under the database prefix before applying `limit` and `page_token`. The cost of a request was set by the size of the database rather than the size of the page, so listing one table out of ten thousand took ten S3 round trips instead of one. List through `ObjectStore::read_dir_page`, which pushes the resume position and the page size into the store's list request. Stores with no paginated list API list the level in full and page it locally, which is what every store did before. Children that are not tables leave a page short of its limit, and one page is one request, so the listing asks again until the page is full or the database runs out. Two behaviour changes come with it: - `page_token` is opaque. It was a table name; it is now whatever resumes the store the database sits on, which for S3, GCS and Azure is a continuation token. Callers hand it back and do not construct or interpret one. Nothing validates it, so a token a caller invents resumes from the wrong place rather than failing. - Tables are reported in the order the store lists directories, which differs from sorting by name only between a name and one that extends it: `users-archive` now precedes `users`, because the `-` of `users-archive.lance` sorts below the `.` of `users.lance`. Pagination cannot report an order other than the one it resumes in. `table_names` is left on the full listing it has today: its `start_after` is a table name, which cannot be pushed into a store that resumes from a continuation token, and it is deprecated. A pushed-down listing does not pass through `WrappingObjectStore::wrap`, so every wrapper here says whether the pushdown survives it: the mirroring wrapper keeps it, since only writes are mirrored, and the test IO tracker gives it up rather than let a listing go around the counter. Co-Authored-By: Claude Opus 5 (1M context) --- rust/lancedb/src/connection.rs | 4 +- rust/lancedb/src/database/listing.rs | 274 +++++++++++++++--- rust/lancedb/src/io/object_store.rs | 11 + .../src/io/object_store/io_tracking.rs | 10 + rust/lancedb/src/table.rs | 9 + 5 files changed, 270 insertions(+), 38 deletions(-) diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs index 5f97967bc..17b7caebb 100644 --- a/rust/lancedb/src/connection.rs +++ b/rust/lancedb/src/connection.rs @@ -1731,7 +1731,9 @@ mod tests { let page = db.list_tables().limit(10).execute().await.unwrap(); assert_eq!(page.tables, names[..10]); - assert_eq!(page.page_token.as_deref(), Some(names[9].as_str())); + // The token is opaque and is not a table name: it is whatever resumes the store + // the database sits on. + assert!(page.page_token.is_some()); // Walking in pages has to reach every table exactly once, with nothing lost // at a page boundary. diff --git a/rust/lancedb/src/database/listing.rs b/rust/lancedb/src/database/listing.rs index 90ba0428d..d46891fc6 100644 --- a/rust/lancedb/src/database/listing.rs +++ b/rust/lancedb/src/database/listing.rs @@ -13,7 +13,7 @@ use lance::dataset::{ReadParams, WriteMode, builder::DatasetBuilder}; use lance::io::{ObjectStore, ObjectStoreParams, WrappingObjectStore}; use lance_datafusion::utils::StreamingWriteSource; use lance_file::version::LanceFileVersion; -use lance_io::object_store::{StorageOptionsAccessor, StorageOptionsProvider}; +use lance_io::object_store::{ReadDirOptions, StorageOptionsAccessor, StorageOptionsProvider}; use lance_table::io::commit::commit_handler_from_url; use object_store::local::LocalFileSystem; use snafu::ResultExt; @@ -281,6 +281,21 @@ impl std::fmt::Display for ListingDatabase { } const LANCE_EXTENSION: &str = "lance"; + +/// The table a listed child of the database names, or `None` if the child is not a table. +/// +/// A table is the directory `.lance`; a loose file or any other directory under the +/// database prefix belongs to something else. `dir_suffix` is `.lance`, built once by the +/// caller rather than per child. +/// The table a listed child directory holds, or `None` if it is not a table at all. +/// +/// Only directories are considered, so a loose object named like a table is not one. +fn table_name(location: &object_store::path::Path, dir_suffix: &str) -> Option { + location + .filename()? + .strip_suffix(dir_suffix) + .map(String::from) +} const ENGINE: &str = "engine"; const MIRRORED_STORE: &str = "mirroredStore"; @@ -988,51 +1003,71 @@ impl Database for ListingDatabase { Ok(f) } + /// List the tables in the database, a page at a time. + /// + /// The page token and the page size go into the object store's list request rather than + /// being applied to a full listing, so a page costs what the page holds and not what the + /// database holds. Stores with no paginated list API list the level in full and page it + /// locally, which is what every store did before. + /// + /// The token is opaque and is only meaningful to the store that issued it: it carries a + /// continuation token where the store has one. It is not a table name, and a caller must + /// not construct one. A page can be shorter than `limit` and still be followed by more, so + /// the token is what ends a walk. + /// + /// Tables come back in the order the store lists directories, which is by key. That + /// differs from sorting by name only between a name and one that extends it: + /// `users-archive` precedes `users`, because the `-` of `users-archive.lance` sorts below + /// the `.` of `users.lance`. async fn list_tables(&self, request: ListTablesRequest) -> Result { if request.id.as_ref().map(|v| !v.is_empty()).unwrap_or(false) { return self.namespace_database().list_tables(request).await; } - let mut f = self - .object_store - .read_dir(self.base_path.clone()) - .await? - .iter() - .map(Path::new) - .filter(|path| { - let is_lance = path - .extension() - .and_then(|e| e.to_str()) - .map(|e| e == LANCE_EXTENSION); - is_lance.unwrap_or(false) - }) - .filter_map(|p| p.file_stem().and_then(|s| s.to_str().map(String::from))) - .collect::>(); - f.sort(); + let limit = request.limit.map(|limit| limit.max(0) as usize); + let dir_suffix = format!(".{LANCE_EXTENSION}"); + let mut tables = Vec::new(); + let mut page_token = request.page_token.filter(|token| !token.is_empty()); - // Handle pagination with page_token - if let Some(ref page_token) = request.page_token { - let index = f - .iter() - .position(|name| name.as_str() > page_token.as_str()) - .unwrap_or(f.len()); - f.drain(0..index); + // A page of nothing: the store rejects a limit of zero, and no table was handed over + // for a token to resume after. + if limit == Some(0) { + return Ok(ListTablesResponse { + tables, + page_token: None, + }); } - // Determine if there's a next page. The token is the last name of this page, - // not the first of the next one: the next page resumes strictly after the - // token, so naming the next page's first entry would skip it. - let next_page_token = match request.limit { - Some(limit) if f.len() > limit as usize => { - f.truncate(limit as usize); - f.last().cloned() + loop { + // Ask only for what the page still has room for, so a database holding more + // than one page costs one request per page rather than one per table. + let listing = self + .object_store + .read_dir_page( + self.base_path.clone(), + ReadDirOptions { + page_token: page_token.take(), + limit: limit.map(|limit| limit - tables.len()), + }, + ) + .await?; + page_token = listing.page_token; + // Only child directories can be tables, and the store already separates them + // out, so the objects in the page are not looked at. + tables.extend( + listing + .result + .common_prefixes + .iter() + .filter_map(|location| table_name(location, &dir_suffix)), + ); + // Children that are not tables leave the page short of the limit, so keep + // going until the page is full or the database runs out. + if page_token.is_none() || limit.is_none_or(|limit| tables.len() >= limit) { + break; } - _ => None, - }; + } - Ok(ListTablesResponse { - tables: f, - page_token: next_page_token, - }) + Ok(ListTablesResponse { tables, page_token }) } async fn create_table(&self, request: CreateTableRequest) -> Result> { @@ -1298,6 +1333,171 @@ mod tests { use std::path::PathBuf; use tempfile::tempdir; + async fn create_tables(db: &ListingDatabase, names: &[&str]) { + let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)])); + for name in names { + db.create_table(CreateTableRequest { + name: name.to_string(), + namespace_path: vec![], + data: Box::new(RecordBatch::new_empty(schema.clone())) as Box, + mode: CreateTableMode::Create, + write_options: Default::default(), + location: None, + namespace_client: None, + }) + .await + .unwrap(); + } + } + + /// Every table in the database, taken `limit` at a time, which is how a caller walks a + /// listing: the token ends the walk, never a short page. + async fn walk(db: &ListingDatabase, limit: Option) -> Vec { + let mut seen = Vec::new(); + let mut page_token = None; + loop { + let page = db + .list_tables(ListTablesRequest { + limit, + page_token, + ..Default::default() + }) + .await + .unwrap(); + seen.extend(page.tables); + page_token = page.page_token; + if page_token.is_none() { + return seen; + } + assert!( + seen.len() < 100, + "the walk is serving tables more than once" + ); + } + } + + /// Paging with the returned token has to visit every table exactly once, whatever the + /// page size, with nothing lost or repeated at a boundary. + #[rstest::rstest] + #[tokio::test] + async fn test_list_tables_pages_over_every_table_once(#[values(1, 2, 3, 5, 10)] limit: i32) { + let (_tempdir, db) = setup_database().await; + create_tables(&db, &["a", "b", "c", "d", "e"]).await; + + assert_eq!(walk(&db, Some(limit)).await, vec!["a", "b", "c", "d", "e"]); + } + + /// The token is opaque: it is whatever resumes the store the database sits on, not a + /// table name. Callers hand it back and nothing else. + /// + /// Nothing validates a token, so one invented by a caller is read as a position rather + /// than refused — which is why the token has to come back from a previous page. + #[tokio::test] + async fn test_the_page_token_is_not_a_table_name() { + let (_tempdir, db) = setup_database().await; + create_tables(&db, &["a", "b", "c"]).await; + + let page = db + .list_tables(ListTablesRequest { + limit: Some(1), + ..Default::default() + }) + .await + .unwrap(); + + assert_eq!(page.tables, vec!["a"]); + let token = page.page_token.expect("two tables are still to come"); + assert_ne!(token, "a"); + + // Handing it back is the only thing a caller does with it, and it resumes. + let rest = db + .list_tables(ListTablesRequest { + page_token: Some(token), + ..Default::default() + }) + .await + .unwrap(); + assert_eq!(rest.tables, vec!["b", "c"]); + } + + /// A limit the listing does not fill leaves no token behind, so a caller paging by token + /// stops without asking for an empty page. + #[tokio::test] + async fn test_a_listing_that_runs_out_has_no_token() { + let (_tempdir, db) = setup_database().await; + create_tables(&db, &["a", "b"]).await; + + let page = db + .list_tables(ListTablesRequest { + limit: Some(10), + ..Default::default() + }) + .await + .unwrap(); + + assert_eq!(page.tables, vec!["a", "b"]); + assert_eq!(page.page_token, None); + } + + /// An empty page token means "from the start", which is how a client looping on a token + /// spells its first request. + #[tokio::test] + async fn test_an_empty_page_token_lists_from_the_start() { + let (_tempdir, db) = setup_database().await; + create_tables(&db, &["a", "b"]).await; + + let page = db + .list_tables(ListTablesRequest { + page_token: Some(String::new()), + ..Default::default() + }) + .await + .unwrap(); + + assert_eq!(page.tables, vec!["a", "b"]); + } + + /// Listing follows the order the object store lists directories in, so a name that + /// extends another comes first: the `-` of `users-archive.lance` sorts below the `.` of + /// `users.lance`. Pagination pushes its cursor into the list request, so it cannot report + /// an order other than the one it resumes in. + #[tokio::test] + async fn test_listing_order_follows_the_store_not_the_table_name() { + let (_tempdir, db) = setup_database().await; + create_tables(&db, &["users", "users-archive", "users.old"]).await; + + assert_eq!( + walk(&db, None).await, + vec!["users-archive", "users", "users.old"] + ); + // And paging reports the same order, so a walk sees each table once. + assert_eq!( + walk(&db, Some(1)).await, + vec!["users-archive", "users", "users.old"] + ); + } + + /// Only directories named `.lance` are tables; loose files and other directories + /// under the database prefix are not. A page spent on them is filled from the next one, + /// so a page holding only non-tables does not read as an empty database. + #[tokio::test] + async fn test_listing_ignores_non_table_children() { + let (tempdir, db) = setup_database().await; + create_tables(&db, &["real"]).await; + std::fs::write(tempdir.path().join("aaa-loose.lance"), b"not a table").unwrap(); + create_dir_all(tempdir.path().join("aaa-scratch")).unwrap(); + + let page = db + .list_tables(ListTablesRequest { + limit: Some(1), + ..Default::default() + }) + .await + .unwrap(); + + assert_eq!(page.tables, vec!["real"]); + } + async fn setup_database() -> (tempfile::TempDir, ListingDatabase) { let tempdir = tempdir().unwrap(); let uri = tempdir.path().to_str().unwrap(); diff --git a/rust/lancedb/src/io/object_store.rs b/rust/lancedb/src/io/object_store.rs index d594bd857..638523770 100644 --- a/rust/lancedb/src/io/object_store.rs +++ b/rust/lancedb/src/io/object_store.rs @@ -7,6 +7,7 @@ use std::{fmt::Formatter, sync::Arc}; use futures::{StreamExt, TryFutureExt, stream::BoxStream}; use lance::io::WrappingObjectStore; +use object_store::list::PaginatedListStore; use object_store::{ CopyOptions, Error, GetOptions, GetResult, ListResult, MultipartUpload, ObjectMeta, ObjectStore, ObjectStoreExt, PutMultipartOptions, PutOptions, PutPayload, PutResult, Result, @@ -187,6 +188,16 @@ impl WrappingObjectStore for MirroringObjectStoreWrapper { secondary: self.secondary.clone(), }) } + + // Only writes are mirrored, and a listing reads, so a pushed-down listing sees the same + // primary this wrapper would have read from. + fn wrap_paginated( + &self, + _store_prefix: &str, + original: Arc, + ) -> Option> { + Some(original) + } } // windows pathing can't be simply concatenated diff --git a/rust/lancedb/src/io/object_store/io_tracking.rs b/rust/lancedb/src/io/object_store/io_tracking.rs index bd4f8f54a..b2ebaecf5 100644 --- a/rust/lancedb/src/io/object_store/io_tracking.rs +++ b/rust/lancedb/src/io/object_store/io_tracking.rs @@ -57,6 +57,16 @@ impl WrappingObjectStore for IoStatsHolder { stats: self.0.clone(), }) } + + // This exists to count requests, so it gives up the pushdown rather than let a listing + // go around the counter. + fn wrap_paginated( + &self, + _store_prefix: &str, + _original: Arc, + ) -> Option> { + None + } } impl IoTrackingStore { diff --git a/rust/lancedb/src/table.rs b/rust/lancedb/src/table.rs index 0d8a8e8b9..c2303ab03 100644 --- a/rust/lancedb/src/table.rs +++ b/rust/lancedb/src/table.rs @@ -3787,6 +3787,15 @@ mod tests { self.called.store(true, Ordering::Relaxed); original } + + // Hands the store back untouched, so a listing has nothing to go around. + fn wrap_paginated( + &self, + _store_prefix: &str, + original: Arc, + ) -> Option> { + Some(original) + } } #[tokio::test]