fix(remote): stop table_names inventing a page token for a namespace (#4039)

`table_names` paginates using a `start_after` table name. This works for
the `/v1/table` endpoint, which guarantees table-order. But the
`/v1/namespace/{id}/table/list` does not. We change that caller to
instead collect all table names, sort, and apply the pagination locally.

We are deprecating this API, so this is just an interim fix. For good
performance, users should move to the `list_tables` API instead, which
uses opaque tokens that don't rely on lexical sorting.

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

rust/lancedb/src/remote/db.rs

@@ -344,6 +344,62 @@ impl<S: HttpSend> RemoteDatabase<S> {
         self.table_cache.remove(&cache_key).await;
         Ok((request_id, resp))
     }
+
+    /// Collect the tables of a namespace in name order, for `table_names`.
+    ///
+    /// `table_names` promises name order and resumes after a table name, but the namespace
+    /// route's `page_token` is opaque -- it belongs to the store the listing walks, and a
+    /// token this client invented would resume from the wrong place. So the whole namespace is
+    /// walked by handing each response's token straight back, and the name semantics are
+    /// applied here. Constructing no token is what makes this work against a server on either
+    /// side of the change: it only ever repeats what the server said.
+    ///
+    /// This is the cost `table_names` already paid -- the server used to enumerate and sort the
+    /// namespace on every request -- and it is why `list_tables` replaces it.
+    async fn table_names_in_namespace(
+        &self,
+        request: &TableNamesRequest,
+    ) -> Result<(Vec<String>, ServerVersion)> {
+        let namespace_id =
+            build_namespace_identifier(&request.namespace_path, &self.client.id_delimiter);
+        let path = format!("/v1/namespace/{}/table/list", namespace_id);
+
+        let mut names = Vec::new();
+        // Every page reports the same server, so keep the first page's version.
+        let mut version: Option<ServerVersion> = None;
+        let mut page_token: Option<String> = None;
+        loop {
+            let mut req = self.client.get(&path);
+            if let Some(ref token) = page_token {
+                req = req.query(&[("page_token", token)]);
+            }
+            let (request_id, rsp) = self.client.send_with_retry(req, None, true).await?;
+            let rsp = self.client.check_response(&request_id, rsp).await?;
+            if version.is_none() {
+                version = Some(parse_server_version(&request_id, &rsp)?);
+            }
+            let response: ListTablesResponse = rsp.json().await.err_to_http(request_id)?;
+            names.extend(response.tables);
+            // An empty token is the end of the listing, not a token to send back: a server
+            // that reads an empty token as "start from the beginning" would hand back the
+            // first page again.
+            match response.page_token.filter(|token| !token.is_empty()) {
+                // A server that repeated a token would never finish; treat that as the end
+                // rather than looping on it.
+                Some(token) if Some(&token) != page_token.as_ref() => page_token = Some(token),
+                _ => break,
+            }
+        }
+
+        names.sort();
+        if let Some(ref start_after) = request.start_after {
+            names.retain(|name| name > start_after);
+        }
+        if let Some(limit) = request.limit {
+            names.truncate(limit as usize);
+        }
+        Ok((names, version.unwrap_or_default()))
+    }
 }
 
 #[cfg(all(test, feature = "remote"))]
@@ -621,29 +677,29 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
     }
 
     async fn table_names(&self, request: TableNamesRequest) -> Result<Vec<String>> {
-        let mut req = if !request.namespace_path.is_empty() {
-            let namespace_id =
-                build_namespace_identifier(&request.namespace_path, &self.client.id_delimiter);
-            self.client
-                .get(&format!("/v1/namespace/{}/table/list", namespace_id))
+        let (tables, version) = if request.namespace_path.is_empty() {
+            // The flat route resumes after a table name and orders by name, which is exactly
+            // what `start_after` means, so the server does the paging.
+            let mut req = self.client.get("/v1/table/");
+            if let Some(limit) = request.limit {
+                req = req.query(&[("limit", limit)]);
+            }
+            if let Some(ref start_after) = request.start_after {
+                req = req.query(&[("page_token", start_after)]);
+            }
+            let (request_id, rsp) = self.client.send_with_retry(req, None, true).await?;
+            let rsp = self.client.check_response(&request_id, rsp).await?;
+            let version = parse_server_version(&request_id, &rsp)?;
+            let tables = rsp
+                .json::<ListTablesResponse>()
+                .await
+                .err_to_http(request_id)?
+                .tables;
+            (tables, version)
         } else {
-            self.client.get("/v1/table/")
+            self.table_names_in_namespace(&request).await?
         };
 
-        if let Some(limit) = request.limit {
-            req = req.query(&[("limit", limit)]);
-        }
-        if let Some(start_after) = request.start_after {
-            req = req.query(&[("page_token", start_after)]);
-        }
-        let (request_id, rsp) = self.client.send_with_retry(req, None, true).await?;
-        let rsp = self.client.check_response(&request_id, rsp).await?;
-        let version = parse_server_version(&request_id, &rsp)?;
-        let tables = rsp
-            .json::<ListTablesResponse>()
-            .await
-            .err_to_http(request_id)?
-            .tables;
         for table in &tables {
             let table_identifier =
                 build_table_identifier(table, &request.namespace_path, &self.client.id_delimiter);
@@ -1227,6 +1283,101 @@ mod tests {
         assert_eq!(names, vec!["table1", "table2"]);
     }
 
+    #[tokio::test]
+    async fn test_table_names_in_a_namespace_never_invents_a_page_token() {
+        // The namespace route's token belongs to the store, so `table_names` cannot build one
+        // from `start_after`. It walks the namespace on the server's own tokens and applies the
+        // name semantics itself, which is what keeps it working either side of the change.
+        let page = Arc::new(AtomicUsize::new(0));
+        let conn = Connection::new_with_handler(move |request| {
+            assert_eq!(request.url().path(), "/v1/namespace/ns/table/list");
+            let query = request.url().query().unwrap_or("");
+            assert!(
+                !query.contains("page_token=users"),
+                "a table name must never be sent as a page token: {query}"
+            );
+            match page.fetch_add(1, Ordering::SeqCst) {
+                0 => {
+                    assert!(
+                        !query.contains("page_token"),
+                        "the walk starts with no token"
+                    );
+                    http::Response::builder()
+                        .status(200)
+                        .body(r#"{"tables": ["users", "orders"], "page_token": "opaque-1"}"#)
+                        .unwrap()
+                }
+                _ => {
+                    assert!(query.contains("page_token=opaque-1"));
+                    http::Response::builder()
+                        .status(200)
+                        .body(r#"{"tables": ["widgets"]}"#)
+                        .unwrap()
+                }
+            }
+        });
+
+        let names = conn
+            .table_names()
+            .namespace(vec!["ns".to_string()])
+            .start_after("users")
+            .execute()
+            .await
+            .unwrap();
+        // Name order, resumed after "users": "orders" sorts before it and is dropped.
+        assert_eq!(names, vec!["widgets"]);
+    }
+
+    #[tokio::test]
+    async fn test_table_names_in_a_namespace_stops_on_a_repeated_token() {
+        // A server that handed back the token it was given would never finish the walk.
+        let conn = Connection::new_with_handler(|_request| {
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"tables": ["a"], "page_token": "same"}"#)
+                .unwrap()
+        });
+
+        let names = conn
+            .table_names()
+            .namespace(vec!["ns".to_string()])
+            .execute()
+            .await
+            .unwrap();
+        // The guard bounds the walk instead of letting it run forever. The repeat is the
+        // server breaking the token contract and is not papered over here.
+        assert_eq!(names, vec!["a", "a"]);
+    }
+
+    #[tokio::test]
+    async fn test_table_names_in_a_namespace_stops_on_an_empty_token() {
+        // An empty token ends the listing. Sending it back would ask a server that reads it
+        // as "start from the beginning" for the first page a second time, and every name on
+        // that page would be collected twice.
+        let requests = Arc::new(AtomicUsize::new(0));
+        let seen = requests.clone();
+        let conn = Connection::new_with_handler(move |request| {
+            seen.fetch_add(1, Ordering::SeqCst);
+            assert!(
+                !request.url().query().unwrap_or("").contains("page_token"),
+                "an empty token must never be sent back"
+            );
+            http::Response::builder()
+                .status(200)
+                .body(r#"{"tables": ["a"], "page_token": ""}"#)
+                .unwrap()
+        });
+
+        let names = conn
+            .table_names()
+            .namespace(vec!["ns".to_string()])
+            .execute()
+            .await
+            .unwrap();
+        assert_eq!(names, vec!["a"]);
+        assert_eq!(requests.load(Ordering::SeqCst), 1);
+    }
+
     #[tokio::test]
     async fn test_table_names_pagination() {
         let conn = Connection::new_with_handler(|request| {