diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 8265d22c0..3f4216bc8 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -344,6 +344,62 @@ impl RemoteDatabase { 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, 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 = None; + let mut page_token: Option = 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 Database for RemoteDatabase { } async fn table_names(&self, request: TableNamesRequest) -> Result> { - 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::() + .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::() - .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| {