feat(remote): route list_tables by what the server serves

`list_tables` asked for `/v2` unconditionally, which 404s against every server
that is not a recent Phalanx -- including `RestAdapter`, the namespace spec's
own reference implementation.

The connection now asks `/v1/version` once, before its first listing, and keeps
the one bit it needs: whether this server serves the `/v2` listing. Servers
that do not keep the listing they have always served, which is correct and
merely slower.

The answer has to be known before the first page rather than learned from it.
The two routes resume from different things, so a walk that started on one
cannot finish on the other, and a walk that learned from its own first page
would hand `/v2` a token `/v1` minted.

What is cached is that bit, not the `ServerVersion` it came from. A server that
sends no version header is indistinguishable from one running the oldest
version we know of, so caching the version would let a stripped header switch
off multivector, structural FTS, multipart write and blobs for the life of the
connection. A missing header can only cost a listing its pushdown.

`table_names` is untouched: it stays on `/v1`, where `page_token` is a table
name to resume after that its callers build themselves.

docs/src/js/classes/Connection.md

@@ -529,6 +529,71 @@ Child namespace names and
 
 ***
 
+### listTables()
+
+#### listTables(options)
+
+```ts
+abstract listTables(options?): Promise<ListTablesResponse>
+```
+
+List a page of tables in this database.
+
+Results may be paginated. To retrieve subsequent pages, pass the
+`pageToken` returned by a previous call. A page may be shorter than
+`limit` without being the last one, so walk until the response carries no
+page token:
+
+```ts
+const names = [];
+let pageToken = undefined;
+do {
+  const page = await conn.listTables({ pageToken, limit: 100 });
+  names.push(...page.tables);
+  pageToken = page.pageToken;
+} while (pageToken);
+```
+
+##### Parameters
+
+* **options?**: `Partial`&lt;[`ListTablesOptions`](../interfaces/ListTablesOptions.md)&gt;
+    Pagination options
+    (`pageToken`, `limit`).
+
+##### Returns
+
+`Promise`&lt;[`ListTablesResponse`](../interfaces/ListTablesResponse.md)&gt;
+
+Table names and an optional token
+  for fetching the next page.
+
+#### listTables(namespacePath, options)
+
+```ts
+abstract listTables(namespacePath?, options?): Promise<ListTablesResponse>
+```
+
+List a page of tables in this database.
+
+##### Parameters
+
+* **namespacePath?**: `string`[]
+    The namespace path to list tables from
+    (defaults to root namespace)
+
+* **options?**: `Partial`&lt;[`ListTablesOptions`](../interfaces/ListTablesOptions.md)&gt;
+    Pagination options
+    (`pageToken`, `limit`).
+
+##### Returns
+
+`Promise`&lt;[`ListTablesResponse`](../interfaces/ListTablesResponse.md)&gt;
+
+Table names and an optional token
+  for fetching the next page.
+
+***
+
 ### openTable()
 
 ```ts
@@ -590,7 +655,7 @@ a "not supported" error.
 
 ***
 
-### tableNames()
+### ~~tableNames()~~
 
 #### tableNames(options)
 
@@ -612,6 +677,10 @@ Tables will be returned in lexicographical order.
 
 `Promise`&lt;`string`[]&gt;
 
+##### Deprecated
+
+Use [Connection.listTables](Connection.md#listtables) instead.
+
 #### tableNames(namespacePath, options)
 
 ```ts
@@ -634,3 +703,7 @@ Tables will be returned in lexicographical order.
 ##### Returns
 
 `Promise`&lt;`string`[]&gt;
+
+##### Deprecated
+
+Use [Connection.listTables](Connection.md#listtables) instead.

docs/src/js/globals.md

@@ -94,6 +94,8 @@
 - [JobInfo](interfaces/JobInfo.md)
 - [ListNamespacesOptions](interfaces/ListNamespacesOptions.md)
 - [ListNamespacesResponse](interfaces/ListNamespacesResponse.md)
+- [ListTablesOptions](interfaces/ListTablesOptions.md)
+- [ListTablesResponse](interfaces/ListTablesResponse.md)
 - [LsmWriteSpec](interfaces/LsmWriteSpec.md)
 - [MergeBlocker](interfaces/MergeBlocker.md)
 - [MergeBranchResult](interfaces/MergeBranchResult.md)

docs/src/js/interfaces/ListTablesOptions.md

@@ -0,0 +1,33 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / ListTablesOptions
+
+# Interface: ListTablesOptions
+
+## Properties
+
+### limit?
+
+```ts
+optional limit: number;
+```
+
+An upper bound on how many tables to return.
+
+A page may hold fewer than this and still not be the last one, so continue
+while the response carries a page token rather than while pages are full.
+
+***
+
+### pageToken?
+
+```ts
+optional pageToken: string;
+```
+
+Token from a previous response for pagination.
+
+The token is opaque: it carries whatever the database needs to resume, and
+callers should not construct or interpret one.

docs/src/js/interfaces/ListTablesResponse.md

@@ -0,0 +1,23 @@
+[**@lancedb/lancedb**](../README.md) • **Docs**
+
+***
+
+[@lancedb/lancedb](../globals.md) / ListTablesResponse
+
+# Interface: ListTablesResponse
+
+## Properties
+
+### pageToken?
+
+```ts
+optional pageToken: string;
+```
+
+***
+
+### tables
+
+```ts
+tables: string[];
+```

docs/src/js/interfaces/TableNamesOptions.md

@@ -4,11 +4,16 @@
 
 [@lancedb/lancedb](../globals.md) / TableNamesOptions
 
-# Interface: TableNamesOptions
+# Interface: ~~TableNamesOptions~~
+
+## Deprecated
+
+Use [ListTablesOptions](ListTablesOptions.md) with [Connection.listTables](../classes/Connection.md#listtables)
+instead.
 
 ## Properties
 
-### limit?
+### ~~limit?~~
 
 ```ts
 optional limit: number;
@@ -18,7 +23,7 @@ An optional limit to the number of results to return.
 
 ***
 
-### startAfter?
+### ~~startAfter?~~
 
 ```ts
 optional startAfter: string;

nodejs/__test__/connection.test.ts

@@ -4,7 +4,13 @@
 import { readdirSync } from "fs";
 import { Field, Float64, Schema } from "apache-arrow";
 import * as tmp from "tmp";
-import { Connection, Table, connect, connectNamespace } from "../lancedb";
+import {
+  Connection,
+  ListTablesResponse,
+  Table,
+  connect,
+  connectNamespace,
+} from "../lancedb";
 import { LocalTable } from "../lancedb/table";
 
 describe("when connecting", () => {
@@ -129,6 +135,56 @@ describe("given a connection", () => {
     expect(tables).toEqual(["b", "c"]);
   });
 
+  it("should list tables with a page token", async () => {
+    const db = await connect(tmpDir.name);
+
+    await db.createTable("b", [{ id: 1 }]);
+    await db.createTable("a", [{ id: 1 }]);
+    await db.createTable("c", [{ id: 1 }]);
+
+    const all = await db.listTables();
+    expect(all.tables).toEqual(["a", "b", "c"]);
+    expect(all.pageToken).toBeUndefined();
+
+    const first = await db.listTables({ limit: 1 });
+    expect(first.tables).toEqual(["a"]);
+    expect(first.pageToken).toBeDefined();
+
+    const second = await db.listTables({
+      limit: 1,
+      pageToken: first.pageToken,
+    });
+    expect(second.tables).toEqual(["b"]);
+  });
+
+  it("should visit every table exactly once when paging", async () => {
+    const db = await connect(tmpDir.name);
+
+    const created = ["a", "b", "c", "d", "e"];
+    for (const name of created) {
+      await db.createTable(name, [{ id: 1 }]);
+    }
+
+    const seen: string[] = [];
+    let pageToken: string | undefined = undefined;
+    do {
+      const page: ListTablesResponse = await db.listTables({
+        limit: 2,
+        pageToken,
+      });
+      seen.push(...page.tables);
+      pageToken = page.pageToken;
+    } while (pageToken);
+
+    expect(seen.sort()).toEqual(created);
+  });
+
+  it("should reject listTables on a closed connection", async () => {
+    const db = await connect(tmpDir.name);
+    db.close();
+    await expect(db.listTables()).rejects.toThrow("Connection is closed");
+  });
+
   it("should create tables in v2 mode", async () => {
     const db = await connect(tmpDir.name);
     const data = [...Array(10000).keys()].map((i) => ({ id: i }));

nodejs/lancedb/connection.ts

@@ -25,12 +25,14 @@ import type {
   JobDescription,
   JobInfo,
   ListNamespacesResponse,
+  ListTablesResponse,
 } from "./native";
 export type {
   CreateNamespaceResponse,
   DescribeNamespaceResponse,
   DropNamespaceResponse,
   ListNamespacesResponse,
+  ListTablesResponse,
 };
 import { sanitizeTable } from "./sanitize";
 import { LocalTable, Table } from "./table";
@@ -128,6 +130,10 @@ export interface OpenTableOptions {
   indexCacheSize?: number;
 }
 
+/**
+ * @deprecated Use {@link ListTablesOptions} with {@link Connection.listTables}
+ * instead.
+ */
 export interface TableNamesOptions {
   /**
    * If present, only return names that come lexicographically after the
@@ -141,6 +147,23 @@ export interface TableNamesOptions {
   limit?: number;
 }
 
+export interface ListTablesOptions {
+  /**
+   * Token from a previous response for pagination.
+   *
+   * The token is opaque: it carries whatever the database needs to resume, and
+   * callers should not construct or interpret one.
+   */
+  pageToken?: string;
+  /**
+   * An upper bound on how many tables to return.
+   *
+   * A page may hold fewer than this and still not be the last one, so continue
+   * while the response carries a page token rather than while pages are full.
+   */
+  limit?: number;
+}
+
 export interface ListNamespacesOptions {
   /** Token from a previous response for pagination. */
   pageToken?: string;
@@ -225,6 +248,7 @@ export abstract class Connection {
    * @param {Partial<TableNamesOptions>} options - options to control the
    * paging / start point (backwards compatibility)
    *
+   * @deprecated Use {@link Connection.listTables} instead.
    */
   abstract tableNames(options?: Partial<TableNamesOptions>): Promise<string[]>;
   /**
@@ -235,12 +259,54 @@ export abstract class Connection {
    * @param {Partial<TableNamesOptions>} options - options to control the
    * paging / start point
    *
+   * @deprecated Use {@link Connection.listTables} instead.
    */
   abstract tableNames(
     namespacePath?: string[],
     options?: Partial<TableNamesOptions>,
   ): Promise<string[]>;
 
+  /**
+   * List a page of tables in this database.
+   *
+   * Results may be paginated. To retrieve subsequent pages, pass the
+   * `pageToken` returned by a previous call. A page may be shorter than
+   * `limit` without being the last one, so walk until the response carries no
+   * page token:
+   *
+   * ```ts
+   * const names = [];
+   * let pageToken = undefined;
+   * do {
+   *   const page = await conn.listTables({ pageToken, limit: 100 });
+   *   names.push(...page.tables);
+   *   pageToken = page.pageToken;
+   * } while (pageToken);
+   * ```
+   *
+   * @param {Partial<ListTablesOptions>} options - Pagination options
+   *   (`pageToken`, `limit`).
+   * @returns {Promise<ListTablesResponse>} Table names and an optional token
+   *   for fetching the next page.
+   */
+  abstract listTables(
+    options?: Partial<ListTablesOptions>,
+  ): Promise<ListTablesResponse>;
+  /**
+   * List a page of tables in this database.
+   *
+   * @param {string[]} namespacePath - The namespace path to list tables from
+   *   (defaults to root namespace)
+   * @param {Partial<ListTablesOptions>} options - Pagination options
+   *   (`pageToken`, `limit`).
+   * @returns {Promise<ListTablesResponse>} Table names and an optional token
+   *   for fetching the next page.
+   */
+  abstract listTables(
+    namespacePath?: string[],
+    options?: Partial<ListTablesOptions>,
+  ): Promise<ListTablesResponse>;
+
   /**
    * Open a table in the database.
    * @param {string} name - The name of the table
@@ -531,6 +597,29 @@ export class LocalConnection extends Connection {
     );
   }
 
+  async listTables(
+    namespacePathOrOptions?: string[] | Partial<ListTablesOptions>,
+    options?: Partial<ListTablesOptions>,
+  ): Promise<ListTablesResponse> {
+    // Detect if first argument is namespacePath array or options object
+    let namespacePath: string[] | undefined;
+    let listTablesOptions: Partial<ListTablesOptions> | undefined;
+
+    if (Array.isArray(namespacePathOrOptions)) {
+      namespacePath = namespacePathOrOptions;
+      listTablesOptions = options;
+    } else {
+      namespacePath = undefined;
+      listTablesOptions = namespacePathOrOptions;
+    }
+
+    return this.inner.listTables(
+      namespacePath ?? [],
+      listTablesOptions?.pageToken,
+      listTablesOptions?.limit,
+    );
+  }
+
   async openTable(
     name: string,
     namespacePath?: string[],

nodejs/lancedb/index.ts

@@ -74,11 +74,13 @@ export {
   Connection,
   CreateTableOptions,
   TableNamesOptions,
+  ListTablesOptions,
   OpenTableOptions,
   ListNamespacesOptions,
   CreateNamespaceOptions,
   DropNamespaceOptions,
   ListNamespacesResponse,
+  ListTablesResponse,
   CreateNamespaceResponse,
   DropNamespaceResponse,
   DescribeNamespaceResponse,

nodejs/src/connection.rs

@@ -36,6 +36,12 @@ pub struct ListNamespacesResponse {
     pub page_token: Option<String>,
 }
 
+#[napi(object)]
+pub struct ListTablesResponse {
+    pub tables: Vec<String>,
+    pub page_token: Option<String>,
+}
+
 #[napi(object)]
 pub struct CreateNamespaceResponse {
     pub properties: Option<HashMap<String, String>>,
@@ -189,6 +195,8 @@ impl Connection {
 
     /// List all tables in the dataset.
     #[napi(catch_unwind)]
+    // Deprecated in favour of `list_tables`, but still exposed to JavaScript.
+    #[allow(deprecated)]
     pub async fn table_names(
         &self,
         namespace_path: Option<Vec<String>>,
@@ -206,6 +214,29 @@ impl Connection {
         op.execute().await.default_error()
     }
 
+    /// List a page of tables in the database.
+    #[napi(catch_unwind)]
+    pub async fn list_tables(
+        &self,
+        namespace_path: Option<Vec<String>>,
+        page_token: Option<String>,
+        limit: Option<u32>,
+    ) -> napi::Result<ListTablesResponse> {
+        let mut op = self.get_inner()?.list_tables();
+        op = op.namespace(namespace_path.unwrap_or_default());
+        if let Some(page_token) = page_token {
+            op = op.page_token(page_token);
+        }
+        if let Some(limit) = limit {
+            op = op.limit(limit);
+        }
+        let resp = op.execute().await.default_error()?;
+        Ok(ListTablesResponse {
+            tables: resp.tables,
+            page_token: resp.page_token,
+        })
+    }
+
     /// Create table from a Apache Arrow IPC (file) buffer.
     ///
     /// Parameters:

python/src/connection.rs

@@ -121,6 +121,8 @@ impl Connection {
     }
 
     #[pyo3(signature = (namespace_path=None, start_after=None, limit=None))]
+    // Deprecated in favour of `list_tables`, but still exposed to Python.
+    #[allow(deprecated)]
     pub fn table_names(
         self_: PyRef<'_, Self>,
         namespace_path: Option<Vec<String>>,
@@ -522,14 +524,17 @@ impl Connection {
         let inner = self_.get_inner()?.clone();
         let py = self_.py();
         future_into_py(py, async move {
-            use lance_namespace::models::ListTablesRequest;
-            let request = ListTablesRequest {
-                id: namespace_path,
-                page_token,
-                limit: limit.map(|l| l as i32),
-                ..Default::default()
-            };
-            let response = inner.list_tables(request).await.infer_error()?;
+            let mut request = inner.list_tables();
+            if let Some(namespace_path) = namespace_path {
+                request = request.namespace(namespace_path);
+            }
+            if let Some(page_token) = page_token {
+                request = request.page_token(page_token);
+            }
+            if let Some(limit) = limit {
+                request = request.limit(limit);
+            }
+            let response = request.execute().await.infer_error()?;
             Python::attach(|py| -> PyResult<Py<PyDict>> {
                 let dict = PyDict::new(py);
                 dict.set_item("tables", response.tables)?;

rust/lancedb/examples/simple.rs

@@ -27,7 +27,7 @@ async fn main() -> Result<()> {
     // --8<-- [end:connect]
 
     // --8<-- [start:list_names]
-    println!("{:?}", db.table_names().execute().await?);
+    println!("{:?}", db.list_tables().execute().await?.tables);
     // --8<-- [end:list_names]
     let tbl = create_table(&db).await?;
     create_index(&tbl).await?;

rust/lancedb/src/connection.rs

@@ -73,11 +73,13 @@ fn set_storage_options_provider(
 }
 
 /// A builder for configuring a [`Connection::table_names`] operation
+#[deprecated(note = "Use Connection::list_tables instead")]
 pub struct TableNamesBuilder {
     parent: Arc<dyn Database>,
     request: TableNamesRequest,
 }
 
+#[allow(deprecated)]
 impl TableNamesBuilder {
     fn new(parent: Arc<dyn Database>) -> Self {
         Self {
@@ -115,6 +117,57 @@ impl TableNamesBuilder {
     }
 }
 
+/// A builder for configuring a [`Connection::list_tables`] operation
+pub struct ListTablesBuilder {
+    parent: Arc<dyn Database>,
+    request: ListTablesRequest,
+}
+
+impl ListTablesBuilder {
+    fn new(parent: Arc<dyn Database>) -> Self {
+        Self {
+            parent,
+            request: ListTablesRequest {
+                // The root namespace is an empty path, not an absent one: a
+                // namespace-backed database rejects a request that names no namespace.
+                id: Some(Vec::new()),
+                ..Default::default()
+            },
+        }
+    }
+
+    /// Resume listing from a previous page.
+    ///
+    /// Pass the `page_token` from the previous [`ListTablesResponse`]. The token is
+    /// opaque: it carries whatever the database needs to resume, and callers should
+    /// not construct or interpret one. A response whose token is `None` or empty is
+    /// the end of the listing.
+    pub fn page_token(mut self, page_token: impl Into<String>) -> Self {
+        self.request.page_token = Some(page_token.into());
+        self
+    }
+
+    /// An upper bound on how many tables to return.
+    ///
+    /// A page may hold fewer than this and still not be the last one, so continue
+    /// while the response carries a page token rather than while pages are full.
+    pub fn limit(mut self, limit: u32) -> Self {
+        self.request.limit = Some(i32::try_from(limit).unwrap_or(i32::MAX));
+        self
+    }
+
+    /// Set the namespace path to list tables from. Defaults to the root namespace.
+    pub fn namespace(mut self, namespace_path: Vec<String>) -> Self {
+        self.request.id = Some(namespace_path);
+        self
+    }
+
+    /// Execute the list tables operation
+    pub async fn execute(self) -> Result<ListTablesResponse> {
+        self.parent.clone().list_tables(self.request).await
+    }
+}
+
 #[derive(Clone, Debug)]
 pub struct OpenTableBuilder {
     parent: Arc<dyn Database>,
@@ -414,7 +467,9 @@ impl Connection {
     /// under creation, may contain only uncommitted storage, or may be concurrently
     /// dropped before it is opened.
     ///
-    /// The parameters `page_token` and `limit` can be used to paginate the results
+    /// The parameters `start_after` and `limit` can be used to paginate the results
+    #[deprecated(note = "Use Connection::list_tables instead")]
+    #[allow(deprecated)]
     pub fn table_names(&self) -> TableNamesBuilder {
         TableNamesBuilder::new(self.internal.clone())
     }
@@ -652,9 +707,32 @@ impl Connection {
         self.internal.namespace_client_config().await
     }
 
-    /// List tables with pagination support
-    pub async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
-        self.internal.list_tables(request).await
+    /// List the tables in the database, a page at a time
+    ///
+    /// ```
+    /// # use lancedb::Connection;
+    /// # async fn list_all(conn: &Connection) -> Result<Vec<String>, lancedb::Error> {
+    /// let mut names = Vec::new();
+    /// let mut token = None;
+    /// loop {
+    ///     let mut request = conn.list_tables().limit(100);
+    ///     if let Some(token) = token {
+    ///         request = request.page_token(token);
+    ///     }
+    ///     let page = request.execute().await?;
+    ///     names.extend(page.tables);
+    ///     // A page may be short without being the last one, so the token is what ends
+    ///     // the walk.
+    ///     token = page.page_token.filter(|token| !token.is_empty());
+    ///     if token.is_none() {
+    ///         break;
+    ///     }
+    /// }
+    /// # Ok(names)
+    /// # }
+    /// ```
+    pub fn list_tables(&self) -> ListTablesBuilder {
+        ListTablesBuilder::new(self.internal.clone())
     }
 
     /// Get the in-memory embedding registry.
@@ -1310,6 +1388,8 @@ mod test_utils {
 }
 
 #[cfg(test)]
+// `table_names` is deprecated but still supported, so its tests still call it.
+#[allow(deprecated)]
 mod tests {
     use arrow_schema::{DataType, Field, Schema};
     use lance_testing::datagen::{BatchGenerator, IncrementingInt32};
@@ -1652,6 +1732,75 @@ mod tests {
         assert_eq!(tables, names[..7]);
     }
 
+    #[tokio::test]
+    async fn test_list_tables_paginates() {
+        let tc = new_test_connection().await.unwrap();
+        if tc.is_remote {
+            // What resumes a page is the server's to decide, and asserting it here would be
+            // asserting the server's contract rather than this one.
+            return;
+        }
+        let db = tc.connection;
+        let schema = Arc::new(Schema::new(vec![Field::new("x", DataType::Int32, false)]));
+        let mut names = Vec::with_capacity(25);
+        for _ in 0..25 {
+            let name = uuid::Uuid::new_v4().to_string();
+            names.push(name.clone());
+            db.create_empty_table(name, schema.clone())
+                .execute()
+                .await
+                .unwrap();
+        }
+        names.sort();
+
+        let page = db.list_tables().limit(10).execute().await.unwrap();
+        assert_eq!(page.tables, names[..10]);
+        // The token is opaque and is not a table name: it is whatever resumes the store
+        // the database sits on, so a caller checks that there is one, not what it says.
+        assert!(page.page_token.is_some());
+
+        // Walking in pages has to reach every table exactly once, with nothing lost
+        // at a page boundary.
+        let mut seen = Vec::with_capacity(names.len());
+        let mut page_token = None;
+        loop {
+            let mut request = db.list_tables().limit(10);
+            if let Some(token) = page_token {
+                request = request.page_token(token);
+            }
+            let page = request.execute().await.unwrap();
+            seen.extend(page.tables);
+            page_token = page.page_token.filter(|token| !token.is_empty());
+            if page_token.is_none() {
+                break;
+            }
+        }
+        assert_eq!(seen, names);
+    }
+
+    #[tokio::test]
+    async fn test_list_tables_exhausted_has_no_token() {
+        let tc = new_test_connection().await.unwrap();
+        let db = tc.connection;
+        let schema = Arc::new(Schema::new(vec![Field::new("x", DataType::Int32, false)]));
+        for i in 0..3 {
+            db.create_empty_table(format!("table{i}"), schema.clone())
+                .execute()
+                .await
+                .unwrap();
+        }
+
+        // A limit the listing does not fill leaves no token behind.
+        let page = db.list_tables().limit(10).execute().await.unwrap();
+        assert_eq!(page.tables.len(), 3);
+        assert_eq!(page.page_token, None);
+
+        // Neither does one that exactly exhausts it.
+        let page = db.list_tables().limit(3).execute().await.unwrap();
+        assert_eq!(page.tables.len(), 3);
+        assert_eq!(page.page_token, None);
+    }
+
     #[tokio::test]
     async fn test_open_table() {
         let tc = new_test_connection().await.unwrap();

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 `<name>.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<String> {
+    location
+        .filename()?
+        .strip_suffix(dir_suffix)
+        .map(String::from)
+}
 const ENGINE: &str = "engine";
 const MIRRORED_STORE: &str = "mirroredStore";
 
@@ -988,53 +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<ListTablesResponse> {
         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::<Vec<String>>();
-        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
-        let next_page_token = if let Some(limit) = request.limit {
-            if f.len() > limit as usize {
-                let token = f[limit as usize].clone();
-                f.truncate(limit as usize);
-                Some(token)
-            } else {
-                None
+        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;
             }
-        } else {
-            None
-        };
+        }
 
-        Ok(ListTablesResponse {
-            tables: f,
-            page_token: next_page_token,
-        })
+        Ok(ListTablesResponse { tables, page_token })
     }
 
     async fn create_table(&self, request: CreateTableRequest) -> Result<Arc<dyn BaseTable>> {
@@ -1307,6 +1340,171 @@ mod tests {
     use tokio::sync::Barrier;
     use tokio::time::timeout;
 
+    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<dyn Scannable>,
+                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<i32>) -> Vec<String> {
+        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 `<name>.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();

rust/lancedb/src/database/namespace.rs

@@ -621,7 +621,10 @@ impl Database for LanceNamespaceDatabase {
 }
 
 #[cfg(test)]
-#[cfg(not(windows))] // TODO: support windows for lance-namespace
+#[cfg(not(windows))]
+// TODO: support windows for lance-namespace
+// `table_names` is deprecated but still supported, so its tests still call it.
+#[allow(deprecated)]
 mod tests {
     use super::*;
     use crate::connect_namespace;

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<dyn PaginatedListStore>,
+    ) -> Option<Arc<dyn PaginatedListStore>> {
+        Some(original)
+    }
 }
 
 // windows pathing can't be simply concatenated

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<dyn object_store::list::PaginatedListStore>,
+    ) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
+        None
+    }
 }
 
 impl IoTrackingStore {

rust/lancedb/src/remote/db.rs

@@ -33,7 +33,7 @@ use super::client::{
     ClientConfig, HeaderProvider, HttpSend, RequestResultExt, RestfulLanceDbClient, Sender,
 };
 use super::table::RemoteTable;
-use super::util::parse_server_version;
+use super::util::{SERVER_VERSION_HEADER, parse_server_version};
 use super::{ARROW_STREAM_CONTENT_TYPE, extract_job_id};
 
 // Request structure for the remote clone table API
@@ -86,6 +86,10 @@ impl ServerVersion {
     pub fn support_blobs(&self) -> bool {
         self.0 >= semver::Version::new(0, 5, 0)
     }
+
+    pub fn support_paginated_list_tables(&self) -> bool {
+        self.0 >= semver::Version::new(0, 6, 0)
+    }
 }
 
 pub const OPT_REMOTE_PREFIX: &str = "remote_database_";
@@ -207,6 +211,14 @@ pub struct RemoteDatabase<S: HttpSend = Sender> {
     namespace_context_provider: Option<Arc<dyn DynamicContextProvider>>,
     /// TLS configuration for mTLS support
     tls_config: Option<super::client::TlsConfig>,
+    /// Whether this server serves the `/v2` table listing, learned once per connection.
+    ///
+    /// This holds the one answer it is asked for rather than the server version it was
+    /// derived from. A server that sends no version header is indistinguishable from one
+    /// running the oldest version we know of, and caching that as a version would let a
+    /// stripped header switch off multivector, structural FTS, multipart write and blobs for
+    /// the life of the connection. A missing header can only cost a listing its pushdown.
+    serves_paginated_list: tokio::sync::OnceCell<bool>,
 }
 
 #[derive(Clone)]
@@ -325,11 +337,45 @@ impl RemoteDatabase {
             namespace_headers,
             namespace_context_provider,
             tls_config: client_config.tls_config,
+            serves_paginated_list: tokio::sync::OnceCell::new(),
         })
     }
 }
 
 impl<S: HttpSend> RemoteDatabase<S> {
+    /// Whether this server serves the `/v2` table listing, asking it once per connection.
+    ///
+    /// The answer has to be known before the first listing rather than learned from it. The
+    /// two listing routes resume from different things, so a walk that started on one cannot
+    /// finish on the other, and a walk that learned the answer from its own first page would
+    /// do exactly that: page one goes to `/v1` for want of an answer, and page two, now
+    /// holding one, hands `/v2` a token `/v1` minted.
+    ///
+    /// `/v1/version` is the question. A server too old to serve it still answers, because the
+    /// version header is on its 404 as well, and a server that is not Phalanx sends no header
+    /// at all, which is the answer for every implementation of the namespace spec: they serve
+    /// `/v1` only.
+    async fn serves_paginated_list(&self) -> Result<bool> {
+        self.serves_paginated_list
+            .get_or_try_init(|| async {
+                let req = self.client.get("/v1/version");
+                // Deliberately not `check_response`: a 404 is a useful answer here, and its
+                // headers carry the version just as a 200's do.
+                let (request_id, rsp) = self.client.send(req).await?;
+                let Some(version) = rsp.headers().get(SERVER_VERSION_HEADER) else {
+                    return Ok(false);
+                };
+                let version = version.to_str().map_err(|e| Error::Http {
+                    source: e.into(),
+                    request_id,
+                    status_code: Some(rsp.status()),
+                })?;
+                Ok(ServerVersion::parse(version)?.support_paginated_list_tables())
+            })
+            .await
+            .copied()
+    }
+
     async fn submit_drop_table(
         &self,
         name: &str,
@@ -366,6 +412,7 @@ mod test_utils {
                 namespace_headers: HashMap::new(),
                 namespace_context_provider: None,
                 tls_config: None,
+                serves_paginated_list: tokio::sync::OnceCell::new(),
             }
         }
 
@@ -388,6 +435,7 @@ mod test_utils {
                 namespace_headers: config.extra_headers.clone(),
                 namespace_context_provider,
                 tls_config: config.tls_config.clone(),
+                serves_paginated_list: tokio::sync::OnceCell::new(),
             }
         }
     }
@@ -676,9 +724,17 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
     async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
         let namespace_parts = request.id.as_deref().unwrap_or(&[]);
         let namespace_id = build_namespace_identifier(namespace_parts, &self.client.id_delimiter);
-        let mut req = self
-            .client
-            .get(&format!("/v1/namespace/{}/table/list", namespace_id));
+        // v1 takes a table name in `page_token` to resume after, which `table_names` callers
+        // build themselves from the last name they saw, so its meaning cannot change. Only a
+        // token the server minted can resume a listing in the store, and that contract needs a
+        // route no name-passing caller reaches. Servers that do not serve it keep the listing
+        // they have always served, which is correct and merely slower.
+        let path = if self.serves_paginated_list().await? {
+            format!("/v2/namespace/{}/table/list", namespace_id)
+        } else {
+            format!("/v1/namespace/{}/table/list", namespace_id)
+        };
+        let mut req = self.client.get(&path);
 
         if let Some(limit) = request.limit {
             req = req.query(&[("limit", limit)]);
@@ -1134,8 +1190,10 @@ impl From<StorageOptions> for RemoteOptions {
 }
 
 #[cfg(test)]
+// `table_names` is deprecated but still supported, so its tests still call it.
+#[allow(deprecated)]
 mod tests {
-    use super::{NamespaceHeaderProviderContext, build_cache_key};
+    use super::{NamespaceHeaderProviderContext, SERVER_VERSION_HEADER, build_cache_key};
     use std::collections::HashMap;
     use std::sync::atomic::{AtomicUsize, Ordering};
     use std::sync::{Arc, OnceLock};
@@ -1664,6 +1722,125 @@ mod tests {
         assert_eq!(names, vec!["table1", "table2"]);
     }
 
+    /// Answer the capability probe with a version, or with no header at all for `None`.
+    fn version_probe(version: Option<&str>) -> http::Response<String> {
+        let builder = http::Response::builder().status(if version.is_some() { 200 } else { 404 });
+        match version {
+            Some(version) => builder
+                .header(SERVER_VERSION_HEADER, version)
+                .body(format!(r#"{{"version": "{version}"}}"#))
+                .unwrap(),
+            None => builder.body(String::new()).unwrap(),
+        }
+    }
+
+    #[tokio::test]
+    async fn test_list_tables_uses_the_opaque_token_route() {
+        // `table_names` keeps /v1, where `page_token` is a table name to resume after. Only
+        // /v2 round-trips a server-minted token, so that is where this API has to go.
+        let conn = Connection::new_with_handler(|request| {
+            if request.url().path() == "/v1/version" {
+                return version_probe(Some("0.6.0"));
+            }
+            assert_eq!(request.url().path(), "/v2/namespace/$/table/list");
+            assert_eq!(request.url().query(), None);
+
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"tables": ["table1"], "page_token": "opaque=="}"#.to_string())
+                .unwrap()
+        });
+        let page = conn.list_tables().execute().await.unwrap();
+        assert_eq!(page.tables, vec!["table1"]);
+        assert_eq!(page.page_token.as_deref(), Some("opaque=="));
+    }
+
+    #[tokio::test]
+    async fn test_list_tables_resumes_with_the_server_token() {
+        let conn = Connection::new_with_handler(|request| {
+            if request.url().path() == "/v1/version" {
+                return version_probe(Some("0.6.0"));
+            }
+            assert_eq!(request.url().path(), "/v2/namespace/ns1$ns2/table/list");
+            assert_eq!(request.url().query(), Some("page_token=opaque%3D%3D"));
+
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"tables": ["table2"]}"#.to_string())
+                .unwrap()
+        });
+        let page = conn
+            .list_tables()
+            .namespace(vec!["ns1".to_string(), "ns2".to_string()])
+            .page_token("opaque==")
+            .execute()
+            .await
+            .unwrap();
+        assert_eq!(page.tables, vec!["table2"]);
+        assert_eq!(page.page_token, None);
+    }
+
+    #[tokio::test]
+    async fn test_a_server_without_the_route_keeps_the_v1_listing() {
+        // Every implementation of the namespace spec is this case: no version header, and only
+        // /v1 to serve. Asking it for /v2 would 404 a listing that /v1 can answer.
+        let conn = Connection::new_with_handler(|request| {
+            if request.url().path() == "/v1/version" {
+                return version_probe(None);
+            }
+            assert_eq!(request.url().path(), "/v1/namespace/$/table/list");
+
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"tables": ["table1"]}"#.to_string())
+                .unwrap()
+        });
+        let page = conn.list_tables().execute().await.unwrap();
+        assert_eq!(page.tables, vec!["table1"]);
+    }
+
+    #[tokio::test]
+    async fn test_a_server_older_than_the_route_keeps_the_v1_listing() {
+        let conn = Connection::new_with_handler(|request| {
+            if request.url().path() == "/v1/version" {
+                return version_probe(Some("0.5.0"));
+            }
+            assert_eq!(request.url().path(), "/v1/namespace/$/table/list");
+
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"tables": ["table1"]}"#.to_string())
+                .unwrap()
+        });
+        assert_eq!(
+            conn.list_tables().execute().await.unwrap().tables,
+            vec!["table1"]
+        );
+    }
+
+    #[tokio::test]
+    async fn test_the_probe_is_asked_once_per_connection() {
+        // A walk resumes from what its first page returned, so every page of it has to reach
+        // the same route. Re-asking per request would let a rolling deploy answer differently
+        // mid-walk and hand /v2 a token /v1 minted.
+        let probes = Arc::new(AtomicUsize::new(0));
+        let counted = probes.clone();
+        let conn = Connection::new_with_handler(move |request| {
+            if request.url().path() == "/v1/version" {
+                counted.fetch_add(1, Ordering::SeqCst);
+                return version_probe(Some("0.6.0"));
+            }
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"tables": ["table1"]}"#.to_string())
+                .unwrap()
+        });
+        for _ in 0..3 {
+            conn.list_tables().execute().await.unwrap();
+        }
+        assert_eq!(probes.load(Ordering::SeqCst), 1);
+    }
+
     #[tokio::test]
     async fn test_table_names_with_nested_namespace() {
         // When namespace is vec!["ns1", "ns2"], should use /v1/namespace/ns1$ns2/table/list
@@ -2272,10 +2449,9 @@ mod tests {
 
             // List tables in the child namespace
             let list_response = conn
-                .list_tables(ListTablesRequest {
-                    id: Some(namespace.clone()),
-                    ..Default::default()
-                })
+                .list_tables()
+                .namespace(namespace.clone())
+                .execute()
                 .await
                 .expect("Failed to list tables");
             assert_eq!(list_response.tables, vec!["test_table"]);
@@ -2346,10 +2522,9 @@ mod tests {
 
             // List tables in the child namespace
             let list_response = conn
-                .list_tables(ListTablesRequest {
-                    id: Some(namespace.clone()),
-                    ..Default::default()
-                })
+                .list_tables()
+                .namespace(namespace.clone())
+                .execute()
                 .await
                 .unwrap();
             assert_eq!(list_response.tables.len(), 3);

rust/lancedb/src/remote/util.rs

@@ -47,10 +47,15 @@ pub fn stream_as_body(data: SendableRecordBatchStream) -> Result<reqwest::Body>
     Ok(reqwest::Body::wrap_stream(stream))
 }
 
+/// The response header a Phalanx server stamps its version onto.
+///
+/// A global layer adds it, so it is on error responses as well as successful ones.
+pub const SERVER_VERSION_HEADER: &str = "phalanx-version";
+
 pub fn parse_server_version(req_id: &str, rsp: &Response) -> Result<ServerVersion> {
     let version = rsp
         .headers()
-        .get("phalanx-version")
+        .get(SERVER_VERSION_HEADER)
         .map(|v| {
             let v = v.to_str().map_err(|e| crate::Error::Http {
                 source: e.into(),

rust/lancedb/src/table.rs

@@ -4058,6 +4058,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<dyn object_store::list::PaginatedListStore>,
+        ) -> Option<Arc<dyn object_store::list::PaginatedListStore>> {
+            Some(original)
+        }
     }
 
     #[tokio::test]

rust/lancedb/tests/object_store_test.rs

@@ -117,8 +117,8 @@ async fn test_minio_lifecycle() -> Result<()> {
     let row_count = table.count_rows(None).await?;
     assert_eq!(row_count, 3);
 
-    let table_names = db.table_names().execute().await?;
-    assert_eq!(table_names, vec!["test_table"]);
+    let tables = db.list_tables().execute().await?.tables;
+    assert_eq!(tables, vec!["test_table"]);
 
     // Re-open the table
     let table = db.open_table("test_table").execute().await?;