fix(listing): don't drop a table at a page boundary (#4040)

Listing tables a page at a time against a local database silently
skipped one table at every page boundary. `ListingDatabase::list_tables`
returned the first name of the *next* page as that page's token, but
resuming from a token drops every name at or before it — so the table
the token named was never handed to the caller. Walking `[a, b, c, d,
e]` with a limit of 2 returned `[a, b, d, e]`.

This PR returns the last name of the page as the token instead, which is
what resuming after the token expects.

This is reachable from Python today through
`db.list_tables(page_token=...)` on a local connection; it also affects
`len(db)` and `name in db`, which walk the pages. Remote and
namespace-backed connections page on the server and were never affected.

## Example

```python
db = lancedb.connect(tmp_path)
for name in ["a", "b", "c", "d", "e"]:
    db.create_table(name, [{"id": 1}])

names, token = [], None
while True:
    page = db.list_tables(page_token=token, limit=2)
    names += page.tables
    token = page.page_token
    if not token:
        break

# before: ['a', 'b', 'd', 'e']
# after:  ['a', 'b', 'c', 'd', 'e']
```

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

rust/lancedb/src/connection.rs

@@ -1679,6 +1679,50 @@ mod tests {
         assert_eq!(tables, names[..7]);
     }
 
+    #[tokio::test]
+    async fn test_list_tables_walks_page_boundaries() {
+        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(5);
+        for _ in 0..5 {
+            let name = uuid::Uuid::new_v4().to_string();
+            names.push(name.clone());
+            db.create_empty_table(name, schema.clone())
+                .execute()
+                .await
+                .unwrap();
+        }
+        names.sort();
+
+        // 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 page = db
+                .list_tables(ListTablesRequest {
+                    id: Some(Vec::new()),
+                    limit: Some(2),
+                    page_token,
+                    ..Default::default()
+                })
+                .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_open_table() {
         let tc = new_test_connection().await.unwrap();

rust/lancedb/src/database/listing.rs

@@ -974,17 +974,15 @@ impl Database for ListingDatabase {
             f.drain(0..index);
         }
 
-        // 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();
+        // 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);
-                Some(token)
-            } else {
-                None
+                f.last().cloned()
             }
-        } else {
-            None
+            _ => None,
         };
 
         Ok(ListTablesResponse {