mirror of
https://github.com/lancedb/lancedb.git
synced 2026-10-12 23:02:28 +00:00
fix: align table listing pagination across backends (#4412)
Fixes #4379 Remote `table_names` discarded the continuation token after the first root-listing request, silently returning the server's default page of 10 names. Remote `list_tables` also inherited that server default, while local listings had no SDK page limit. Python's synchronous and namespace wrappers still imposed a default limit of 10 names. The namespace adapter also discarded continuation tokens, so an unlimited REST namespace listing could stop at the server's page cap. Follow server-provided root listing tokens until the requested name limit or the end of the listing, preserving `start_after`, cache population, and guards against empty or repeated tokens. The namespace adapter collects every server page, then sorts names and applies the name cursor and total limit: namespace tokens are opaque, and page order need not be name order. Empty tokens end the listing; token cycles return an error. Python sync and namespace `table_names` wrappers now default to `None`, matching the asynchronous and TypeScript APIs. Namespace overrides accept the base connection's optional limit, avoiding import-time signature errors on Python 3.10 and 3.11. Set the `list_tables` default to 100 at the shared connection boundary, including namespaces, and return empty results for a zero limit without sending an invalid server request. Update the SDK documentation to describe these defaults. `list_tables()` is now a paginated call in both modes: callers with more than 100 tables must follow the returned token or supply an explicit larger limit. `table_names()` without a limit returns every name. Namespace table-name calls enumerate the namespace before applying name cursors and limits; use `list_tables()` for paginated access. Validation: - Regression cases reproduce the original root-listing truncation and cover 130 tables, opaque tokens, short pages, explicit limits, `start_after`, and zero limits. - Relevant Rust remote database, connection, and local listing test groups: 175 passed. Namespace adapter tests: 23 passed, including opaque tokens, unordered and empty pages, name cursors, limits, zero requests, and token cycles. - Rebuilt the Python extension; namespace and local/mocked remote listing tests using `uv.lock`: 46 passed on Python 3.10 and 46 passed on Python 3.13. - REST namespace regressions reproduced the truncation before the fix and now return all 12 names from a server capped at 10 per response, for sync/async connections at root and child paths. They also cover unordered pages, name cursors, explicit limits, and zero requests. The exact Python 3.10 import error also reproduced before the optional-signature fix. - TypeScript connection test file: 38 passed; `pnpm build`, `pnpm lint`, and `pnpm run docs` passed. - `cargo check --quiet --features remote --tests --examples` and `cargo clippy --quiet --features remote --tests --examples` passed. - Rust formatting, Python Ruff formatting/lint, and diff whitespace checks passed. <!-- lance-gatekeeper-fix:v1 agent=43f92f1f05ac9ca74229af590945cb88 generation=1 --> --------- Co-authored-by: Gatefixer <313497061+lancedb-gatefixer[bot]@users.noreply.github.com>
This commit is contained in:
@@ -15,6 +15,7 @@ optional limit: number;
|
||||
```
|
||||
|
||||
An upper bound on how many tables to return.
|
||||
Defaults to 100. Zero returns an empty page without a continuation token.
|
||||
|
||||
A page may hold fewer than this and still not be the last one, so keep
|
||||
going while the response carries a page token rather than while pages are
|
||||
|
||||
@@ -19,7 +19,7 @@ instead.
|
||||
optional limit: number;
|
||||
```
|
||||
|
||||
An optional limit to the number of results to return.
|
||||
The maximum number of names to return. Omitted returns all names; zero returns none.
|
||||
|
||||
***
|
||||
|
||||
|
||||
@@ -1,7 +1,8 @@
|
||||
// SPDX-License-Identifier: Apache-2.0
|
||||
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
|
||||
|
||||
import { readdirSync } from "fs";
|
||||
import { mkdirSync, readdirSync } from "fs";
|
||||
import { join } from "path";
|
||||
import { Field, Float64, Schema } from "apache-arrow";
|
||||
import * as tmp from "tmp";
|
||||
import {
|
||||
@@ -180,6 +181,32 @@ describe("given a connection", () => {
|
||||
expect(seen).toEqual(created);
|
||||
});
|
||||
|
||||
it("should list all names and use a default page size of 100", async () => {
|
||||
const names = Array.from(
|
||||
{ length: 130 },
|
||||
(_, i) => `t${i.toString().padStart(3, "0")}`,
|
||||
);
|
||||
for (const name of names) {
|
||||
mkdirSync(join(tmpDir.name, `${name}.lance`));
|
||||
}
|
||||
await expect(db.tableNames()).resolves.toEqual(names);
|
||||
await expect(db.tableNames({ limit: 0 })).resolves.toEqual([]);
|
||||
const first = await db.listTables();
|
||||
expect(first.tables).toEqual(names.slice(0, 100));
|
||||
expect(first.pageToken).toBeDefined();
|
||||
const second = await db.listTables({ pageToken: first.pageToken });
|
||||
expect(second.tables).toEqual(names.slice(100));
|
||||
expect(second.pageToken).toBeUndefined();
|
||||
await expect(db.listTables({ limit: 200 })).resolves.toEqual({
|
||||
tables: names,
|
||||
pageToken: undefined,
|
||||
});
|
||||
await expect(db.listTables({ limit: 0 })).resolves.toEqual({
|
||||
tables: [],
|
||||
pageToken: undefined,
|
||||
});
|
||||
});
|
||||
|
||||
it("should list tables in a namespace", async () => {
|
||||
const db = await connect(tmpDir.name, {
|
||||
// biome-ignore lint/style/useNamingConvention: opaque backend property key, must match Rust
|
||||
|
||||
@@ -150,7 +150,7 @@ export interface TableNamesOptions {
|
||||
* the last table name from the previous page.
|
||||
*/
|
||||
startAfter?: string;
|
||||
/** An optional limit to the number of results to return. */
|
||||
/** The maximum number of names to return. Omitted returns all names; zero returns none. */
|
||||
limit?: number;
|
||||
}
|
||||
|
||||
@@ -164,6 +164,7 @@ export interface ListTablesOptions {
|
||||
pageToken?: string;
|
||||
/**
|
||||
* An upper bound on how many tables to return.
|
||||
* Defaults to 100. Zero returns an empty page without a continuation token.
|
||||
*
|
||||
* A page may hold fewer than this and still not be the last one, so keep
|
||||
* going while the response carries a page token rather than while pages are
|
||||
|
||||
+20
-14
@@ -317,7 +317,7 @@ class DBConnection(EnforceOverrides):
|
||||
page_token: Optional[str] = None,
|
||||
limit: Optional[int] = None,
|
||||
) -> ListTablesResponse:
|
||||
"""List all tables in this database with pagination support.
|
||||
"""List a page of tables in this database.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
@@ -328,7 +328,8 @@ class DBConnection(EnforceOverrides):
|
||||
Token for pagination. Use the token from a previous response
|
||||
to get the next page of results.
|
||||
limit: int, optional
|
||||
The maximum number of results to return.
|
||||
The maximum number of results to return, default 100.
|
||||
Zero returns an empty page without a continuation token.
|
||||
|
||||
Returns
|
||||
-------
|
||||
@@ -343,7 +344,7 @@ class DBConnection(EnforceOverrides):
|
||||
def table_names(
|
||||
self,
|
||||
page_token: Optional[str] = None,
|
||||
limit: int = 10,
|
||||
limit: Optional[int] = None,
|
||||
*,
|
||||
namespace_path: Optional[List[str]] = None,
|
||||
) -> Iterable[str]:
|
||||
@@ -357,8 +358,9 @@ class DBConnection(EnforceOverrides):
|
||||
page_token: str, optional
|
||||
The token to use for pagination. If not present, start from the beginning.
|
||||
Typically, this token is last table name from the previous page.
|
||||
limit: int, default 10
|
||||
The size of the page to return.
|
||||
limit: int, optional
|
||||
The maximum number of tables to return. None returns all tables;
|
||||
zero returns an empty list.
|
||||
|
||||
Returns
|
||||
-------
|
||||
@@ -1490,7 +1492,7 @@ class LanceDBConnection(DBConnection):
|
||||
page_token: Optional[str] = None,
|
||||
limit: Optional[int] = None,
|
||||
) -> ListTablesResponse:
|
||||
"""List all tables in this database with pagination support.
|
||||
"""List a page of tables in this database.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
@@ -1501,7 +1503,8 @@ class LanceDBConnection(DBConnection):
|
||||
Token for pagination. Use the token from a previous response
|
||||
to get the next page of results.
|
||||
limit: int, optional
|
||||
The maximum number of results to return.
|
||||
The maximum number of results to return, default 100.
|
||||
Zero returns an empty page without a continuation token.
|
||||
|
||||
Returns
|
||||
-------
|
||||
@@ -1520,7 +1523,7 @@ class LanceDBConnection(DBConnection):
|
||||
def table_names(
|
||||
self,
|
||||
page_token: Optional[str] = None,
|
||||
limit: int = 10,
|
||||
limit: Optional[int] = None,
|
||||
*,
|
||||
namespace_path: Optional[List[str]] = None,
|
||||
) -> Iterable[str]:
|
||||
@@ -1535,8 +1538,9 @@ class LanceDBConnection(DBConnection):
|
||||
The namespace to list tables in.
|
||||
page_token: str, optional
|
||||
The token to use for pagination.
|
||||
limit: int, default 10
|
||||
The maximum number of tables to return.
|
||||
limit: int, optional
|
||||
The maximum number of tables to return. None returns all tables;
|
||||
zero returns an empty list.
|
||||
|
||||
Returns
|
||||
-------
|
||||
@@ -2366,7 +2370,7 @@ class AsyncConnection(object):
|
||||
page_token: Optional[str] = None,
|
||||
limit: Optional[int] = None,
|
||||
) -> ListTablesResponse:
|
||||
"""List all tables in this database with pagination support.
|
||||
"""List a page of tables in this database.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
@@ -2377,7 +2381,8 @@ class AsyncConnection(object):
|
||||
Token for pagination. Use the token from a previous response
|
||||
to get the next page of results.
|
||||
limit: int, optional
|
||||
The maximum number of results to return.
|
||||
The maximum number of results to return, default 100.
|
||||
Zero returns an empty page without a continuation token.
|
||||
|
||||
Returns
|
||||
-------
|
||||
@@ -2414,8 +2419,9 @@ class AsyncConnection(object):
|
||||
|
||||
This can be combined with limit to implement pagination by setting this to
|
||||
the last table name from the previous page.
|
||||
limit: int, default 10
|
||||
The number of results to return.
|
||||
limit: int, optional
|
||||
The maximum number of tables to return. None returns all tables;
|
||||
zero returns an empty list.
|
||||
|
||||
Returns
|
||||
-------
|
||||
|
||||
@@ -522,13 +522,15 @@ class LanceNamespaceDBConnection(DBConnection):
|
||||
def table_names(
|
||||
self,
|
||||
page_token: Optional[str] = None,
|
||||
limit: int = 10,
|
||||
limit: Optional[int] = None,
|
||||
*,
|
||||
namespace_path: Optional[List[str]] = None,
|
||||
) -> Iterable[str]:
|
||||
"""
|
||||
List table names in the database.
|
||||
|
||||
If limit is None, return all table names; zero returns an empty list.
|
||||
|
||||
.. deprecated::
|
||||
Use :meth:`list_tables` instead, which provides proper pagination support.
|
||||
"""
|
||||
@@ -1157,13 +1159,15 @@ class AsyncLanceNamespaceDBConnection:
|
||||
async def table_names(
|
||||
self,
|
||||
page_token: Optional[str] = None,
|
||||
limit: int = 10,
|
||||
limit: Optional[int] = None,
|
||||
*,
|
||||
namespace_path: Optional[List[str]] = None,
|
||||
) -> Iterable[str]:
|
||||
"""
|
||||
List table names in the namespace.
|
||||
|
||||
If limit is None, return all table names; zero returns an empty list.
|
||||
|
||||
.. deprecated::
|
||||
Use :meth:`list_tables` instead, which provides proper pagination support.
|
||||
"""
|
||||
|
||||
@@ -379,7 +379,7 @@ class RemoteDBConnection(DBConnection):
|
||||
page_token: Optional[str] = None,
|
||||
limit: Optional[int] = None,
|
||||
) -> ListTablesResponse:
|
||||
"""List all tables in this database with pagination support.
|
||||
"""List a page of tables in this database.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
@@ -390,7 +390,8 @@ class RemoteDBConnection(DBConnection):
|
||||
Token for pagination. Use the token from a previous response
|
||||
to get the next page of results.
|
||||
limit: int, optional
|
||||
The maximum number of results to return.
|
||||
The maximum number of results to return, default 100.
|
||||
Zero returns an empty page without a continuation token.
|
||||
|
||||
Returns
|
||||
-------
|
||||
@@ -409,7 +410,7 @@ class RemoteDBConnection(DBConnection):
|
||||
def table_names(
|
||||
self,
|
||||
page_token: Optional[str] = None,
|
||||
limit: int = 10,
|
||||
limit: Optional[int] = None,
|
||||
*,
|
||||
namespace_path: Optional[List[str]] = None,
|
||||
) -> Iterable[str]:
|
||||
@@ -425,8 +426,9 @@ class RemoteDBConnection(DBConnection):
|
||||
Empty list represents root namespace.
|
||||
page_token: str
|
||||
The last token to start the new page.
|
||||
limit: int, default 10
|
||||
The maximum number of tables to return for each page.
|
||||
limit: int, optional
|
||||
The maximum number of tables to return. None returns all tables;
|
||||
zero returns an empty list.
|
||||
|
||||
Returns
|
||||
-------
|
||||
|
||||
@@ -245,6 +245,40 @@ def test_db_contains_and_len_include_all_table_name_pages(tmp_db: lancedb.DBConn
|
||||
assert "does_not_exist" not in tmp_db
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_table_listing_defaults_and_zero_limit(tmp_path):
|
||||
names = [f"t{i:03}" for i in range(130)]
|
||||
for name in names:
|
||||
(tmp_path / f"{name}.lance").mkdir()
|
||||
|
||||
db = lancedb.connect(tmp_path)
|
||||
assert db.table_names() == names
|
||||
assert db.table_names(limit=None) == names
|
||||
assert db.table_names(limit=0) == []
|
||||
first = db.list_tables()
|
||||
assert first.tables == names[:100]
|
||||
assert first.page_token
|
||||
second = db.list_tables(page_token=first.page_token)
|
||||
assert second.tables == names[100:]
|
||||
assert second.page_token is None
|
||||
assert db.list_tables(limit=200).tables == names
|
||||
empty = db.list_tables(limit=0)
|
||||
assert empty.tables == []
|
||||
assert empty.page_token is None
|
||||
|
||||
adb = await lancedb.connect_async(tmp_path)
|
||||
assert await adb.table_names() == names
|
||||
assert await adb.table_names(limit=0) == []
|
||||
first = await adb.list_tables()
|
||||
assert first.tables == names[:100]
|
||||
second = await adb.list_tables(page_token=first.page_token)
|
||||
assert second.tables == names[100:]
|
||||
assert second.page_token is None
|
||||
empty = await adb.list_tables(limit=0)
|
||||
assert empty.tables == []
|
||||
assert empty.page_token is None
|
||||
|
||||
|
||||
def test_db_contains_stops_after_matching_table_page(
|
||||
tmp_db: lancedb.DBConnection, monkeypatch
|
||||
):
|
||||
|
||||
@@ -6,6 +6,10 @@
|
||||
import tempfile
|
||||
import shutil
|
||||
import importlib
|
||||
import http.server
|
||||
import json
|
||||
import threading
|
||||
from urllib.parse import parse_qs, unquote, urlparse
|
||||
import pytest
|
||||
import pyarrow as pa
|
||||
import lancedb
|
||||
@@ -73,6 +77,132 @@ def _namespace_lance_table(namespace_client: _NamespaceClient) -> LanceTable:
|
||||
return table
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("namespace_path", [[], ["test_ns"]])
|
||||
async def test_namespace_table_names_defaults_and_limits(tmp_path, namespace_path):
|
||||
db = lancedb.connect_namespace("dir", {"root": str(tmp_path)})
|
||||
if namespace_path:
|
||||
db.create_namespace(namespace_path)
|
||||
names = [f"t{idx:03}" for idx in range(12)]
|
||||
schema = pa.schema([pa.field("id", pa.int64())])
|
||||
for name in names:
|
||||
db.create_table(name, schema=schema, namespace_path=namespace_path)
|
||||
|
||||
assert list(db.table_names(namespace_path=namespace_path)) == names
|
||||
assert list(db.table_names(limit=None, namespace_path=namespace_path)) == names
|
||||
assert list(db.table_names(limit=0, namespace_path=namespace_path)) == []
|
||||
assert list(db.table_names(limit=2, namespace_path=namespace_path)) == names[:2]
|
||||
assert (
|
||||
list(
|
||||
db.table_names(page_token=names[1], limit=2, namespace_path=namespace_path)
|
||||
)
|
||||
== names[2:4]
|
||||
)
|
||||
|
||||
adb = lancedb.connect_namespace_async("dir", {"root": str(tmp_path)})
|
||||
assert list(await adb.table_names(namespace_path=namespace_path)) == names
|
||||
assert (
|
||||
list(await adb.table_names(limit=None, namespace_path=namespace_path)) == names
|
||||
)
|
||||
assert list(await adb.table_names(limit=0, namespace_path=namespace_path)) == []
|
||||
assert (
|
||||
list(await adb.table_names(limit=2, namespace_path=namespace_path)) == names[:2]
|
||||
)
|
||||
assert (
|
||||
list(
|
||||
await adb.table_names(
|
||||
page_token=names[1], limit=2, namespace_path=namespace_path
|
||||
)
|
||||
)
|
||||
== names[2:4]
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def paginated_rest_namespace():
|
||||
names = [f"t{idx:03}" for idx in reversed(range(12))]
|
||||
requests = []
|
||||
|
||||
class Handler(http.server.BaseHTTPRequestHandler):
|
||||
def do_GET(self):
|
||||
url = urlparse(self.path)
|
||||
query = parse_qs(url.query)
|
||||
requests.append((unquote(url.path), query))
|
||||
token = query.get("page_token", [None])[0]
|
||||
if token not in {None, "opaque-next"}:
|
||||
self.send_error(400, "Unknown page token")
|
||||
return
|
||||
start = 0 if token is None else 10
|
||||
body = {"tables": names[start : start + 10]}
|
||||
if start == 0:
|
||||
body["page_token"] = "opaque-next"
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.end_headers()
|
||||
self.wfile.write(json.dumps(body).encode())
|
||||
|
||||
def log_message(self, *args):
|
||||
pass
|
||||
|
||||
with http.server.HTTPServer(("127.0.0.1", 0), Handler) as server:
|
||||
thread = threading.Thread(target=server.serve_forever)
|
||||
thread.start()
|
||||
try:
|
||||
yield {"uri": f"http://127.0.0.1:{server.server_port}"}, requests
|
||||
finally:
|
||||
server.shutdown()
|
||||
thread.join()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("namespace_path", [[], ["n"]])
|
||||
async def test_rest_namespace_table_names_collects_pages(
|
||||
paginated_rest_namespace, namespace_path
|
||||
):
|
||||
properties, requests = paginated_rest_namespace
|
||||
names = [f"t{idx:03}" for idx in range(12)]
|
||||
db = lancedb.connect_namespace("rest", properties)
|
||||
first = db.list_tables(namespace_path=namespace_path)
|
||||
second = db.list_tables(namespace_path=namespace_path, page_token=first.page_token)
|
||||
assert len(first.tables) == 10
|
||||
assert first.page_token == "opaque-next"
|
||||
assert sorted(first.tables + second.tables) == names
|
||||
assert second.page_token is None
|
||||
|
||||
assert list(db.table_names(namespace_path=namespace_path)) == names
|
||||
assert list(db.table_names(limit=None, namespace_path=namespace_path)) == names
|
||||
assert list(db.table_names(limit=11, namespace_path=namespace_path)) == names[:11]
|
||||
assert (
|
||||
list(db.table_names(page_token="t005", limit=2, namespace_path=namespace_path))
|
||||
== names[6:8]
|
||||
)
|
||||
|
||||
adb = lancedb.connect_namespace_async("rest", properties)
|
||||
assert list(await adb.table_names(namespace_path=namespace_path)) == names
|
||||
assert (
|
||||
list(await adb.table_names(limit=None, namespace_path=namespace_path)) == names
|
||||
)
|
||||
assert (
|
||||
list(await adb.table_names(limit=11, namespace_path=namespace_path))
|
||||
== names[:11]
|
||||
)
|
||||
assert (
|
||||
list(
|
||||
await adb.table_names(
|
||||
page_token="t005", limit=2, namespace_path=namespace_path
|
||||
)
|
||||
)
|
||||
== names[6:8]
|
||||
)
|
||||
|
||||
request_count = len(requests)
|
||||
assert list(db.table_names(limit=0, namespace_path=namespace_path)) == []
|
||||
assert list(await adb.table_names(limit=0, namespace_path=namespace_path)) == []
|
||||
assert len(requests) == request_count
|
||||
identifier = "$" if not namespace_path else "n"
|
||||
assert all(path == f"/v1/namespace/{identifier}/table/list" for path, _ in requests)
|
||||
|
||||
|
||||
class TestNamespaceConnection:
|
||||
"""Test namespace-based LanceDB connection using DirectoryNamespace."""
|
||||
|
||||
|
||||
@@ -13,6 +13,7 @@ import threading
|
||||
import time
|
||||
from unittest.mock import MagicMock, patch
|
||||
import uuid
|
||||
from urllib.parse import parse_qs, urlparse
|
||||
from packaging.version import Version
|
||||
|
||||
import lancedb
|
||||
@@ -105,6 +106,67 @@ async def mock_lancedb_connection_async(handler, **client_config):
|
||||
handle.join()
|
||||
|
||||
|
||||
def paginated_table_listing_handler(request):
|
||||
url = urlparse(request.path)
|
||||
assert url.path in {
|
||||
"/v1/table/",
|
||||
"/v1/namespace/$/table/list",
|
||||
"/v1/namespace/n/table/list",
|
||||
}
|
||||
query = parse_qs(url.query)
|
||||
limit = int(query.get("limit", [10])[0])
|
||||
assert limit > 0
|
||||
token = query.get("page_token", ["page-0"])[0]
|
||||
start = int(token.removeprefix("page-"))
|
||||
end = min(start + limit, 130)
|
||||
body = {"tables": [f"t{i:03}" for i in range(start, end)]}
|
||||
if end < 130:
|
||||
body["page_token"] = f"page-{end}"
|
||||
request.send_response(200)
|
||||
request.send_header("Content-Type", "application/json")
|
||||
request.end_headers()
|
||||
request.wfile.write(json.dumps(body).encode())
|
||||
|
||||
|
||||
@pytest.mark.parametrize("namespace", [[], ["n"]])
|
||||
def test_remote_table_listing_defaults_and_zero_limit(namespace):
|
||||
names = [f"t{i:03}" for i in range(130)]
|
||||
with mock_lancedb_connection(paginated_table_listing_handler) as db:
|
||||
assert db.table_names(namespace_path=namespace) == names
|
||||
assert db.table_names(limit=None, namespace_path=namespace) == names
|
||||
assert db.table_names(limit=0, namespace_path=namespace) == []
|
||||
first = db.list_tables(namespace_path=namespace)
|
||||
assert first.tables == names[:100]
|
||||
assert first.page_token == "page-100"
|
||||
second = db.list_tables(namespace_path=namespace, page_token=first.page_token)
|
||||
assert second.tables == names[100:]
|
||||
assert second.page_token is None
|
||||
assert db.list_tables(namespace_path=namespace, limit=200).tables == names
|
||||
empty = db.list_tables(namespace_path=namespace, limit=0)
|
||||
assert empty.tables == []
|
||||
assert empty.page_token is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("namespace", [[], ["n"]])
|
||||
async def test_async_remote_table_listing_defaults_and_zero_limit(namespace):
|
||||
names = [f"t{i:03}" for i in range(130)]
|
||||
async with mock_lancedb_connection_async(paginated_table_listing_handler) as db:
|
||||
assert await db.table_names(namespace_path=namespace) == names
|
||||
assert await db.table_names(limit=0, namespace_path=namespace) == []
|
||||
first = await db.list_tables(namespace_path=namespace)
|
||||
assert first.tables == names[:100]
|
||||
assert first.page_token == "page-100"
|
||||
second = await db.list_tables(
|
||||
namespace_path=namespace, page_token=first.page_token
|
||||
)
|
||||
assert second.tables == names[100:]
|
||||
assert second.page_token is None
|
||||
empty = await db.list_tables(namespace_path=namespace, limit=0)
|
||||
assert empty.tables == []
|
||||
assert empty.page_token is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_remote_db():
|
||||
def handler(request):
|
||||
|
||||
@@ -106,7 +106,9 @@ impl TableNamesBuilder {
|
||||
self
|
||||
}
|
||||
|
||||
/// The maximum number of table names to return
|
||||
/// The maximum number of table names to return.
|
||||
///
|
||||
/// Without a limit, all names are returned. Zero returns an empty list.
|
||||
pub fn limit(mut self, limit: u32) -> Self {
|
||||
self.request.limit = Some(limit);
|
||||
self
|
||||
@@ -553,7 +555,8 @@ 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
|
||||
/// Without a limit, all names are returned. The parameters `start_after` and
|
||||
/// `limit` can be used to paginate the results.
|
||||
pub fn table_names(&self) -> TableNamesBuilder {
|
||||
TableNamesBuilder::new(self.internal.clone())
|
||||
}
|
||||
@@ -1185,7 +1188,21 @@ impl Connection {
|
||||
}
|
||||
|
||||
/// List tables with pagination support
|
||||
pub async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
|
||||
///
|
||||
/// The default limit is 100 tables per page for both local and remote connections,
|
||||
/// including namespaces. Zero returns an empty page without a continuation token.
|
||||
/// Follow the response's opaque `page_token` until it is absent to retrieve every table;
|
||||
/// a page can contain fewer than the limit even when more tables remain.
|
||||
pub async fn list_tables(&self, mut request: ListTablesRequest) -> Result<ListTablesResponse> {
|
||||
// Apply the SDK's page size here so every backend receives the same request.
|
||||
let limit = request.limit.get_or_insert(100);
|
||||
if *limit <= 0 {
|
||||
return Ok(ListTablesResponse {
|
||||
context: None,
|
||||
tables: Vec::new(),
|
||||
page_token: None,
|
||||
});
|
||||
}
|
||||
self.internal.list_tables(request).await
|
||||
}
|
||||
|
||||
@@ -2271,6 +2288,61 @@ mod tests {
|
||||
assert_eq!(seen, names);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_table_listing_defaults_and_zero_limit() {
|
||||
let tempdir = tempfile::tempdir().unwrap();
|
||||
// Listing discovers physical entries without opening the datasets.
|
||||
let names: Vec<_> = (0..130).map(|i| format!("t{i:03}")).collect();
|
||||
for name in &names {
|
||||
std::fs::create_dir(tempdir.path().join(format!("{name}.lance"))).unwrap();
|
||||
}
|
||||
let db = connect(tempdir.path().to_str().unwrap())
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(db.table_names().execute().await.unwrap(), names);
|
||||
assert!(
|
||||
db.table_names()
|
||||
.limit(0)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.is_empty()
|
||||
);
|
||||
|
||||
let first = db.list_tables(ListTablesRequest::default()).await.unwrap();
|
||||
assert_eq!(first.tables, names[..100]);
|
||||
assert!(first.page_token.is_some());
|
||||
let second = db
|
||||
.list_tables(ListTablesRequest {
|
||||
page_token: first.page_token,
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(second.tables, names[100..]);
|
||||
assert!(second.page_token.is_none());
|
||||
|
||||
let all = db
|
||||
.list_tables(ListTablesRequest {
|
||||
limit: Some(200),
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(all.tables, names);
|
||||
assert!(all.page_token.is_none());
|
||||
let empty = db
|
||||
.list_tables(ListTablesRequest {
|
||||
limit: Some(0),
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(empty.tables.is_empty());
|
||||
assert!(empty.page_token.is_none());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_open_table() {
|
||||
let tc = new_test_connection().await.unwrap();
|
||||
|
||||
@@ -288,16 +288,44 @@ impl Database for LanceNamespaceDatabase {
|
||||
}
|
||||
|
||||
async fn table_names(&self, request: TableNamesRequest) -> Result<Vec<String>> {
|
||||
let ns_request = ListTablesRequest {
|
||||
id: Some(request.namespace_path),
|
||||
page_token: request.start_after,
|
||||
limit: request.limit.map(|l| l as i32),
|
||||
..Default::default()
|
||||
};
|
||||
if request.limit == Some(0) {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
|
||||
let response = self.namespace.list_tables(ns_request).await?;
|
||||
// Namespace tokens are opaque, and pages need not arrive in name order.
|
||||
// Collect the listing before applying the name cursor and overall limit.
|
||||
let mut names = Vec::new();
|
||||
let mut page_token = None;
|
||||
let mut seen_tokens = HashSet::new();
|
||||
loop {
|
||||
let response = self
|
||||
.namespace
|
||||
.list_tables(ListTablesRequest {
|
||||
id: Some(request.namespace_path.clone()),
|
||||
page_token,
|
||||
..Default::default()
|
||||
})
|
||||
.await?;
|
||||
names.extend(response.tables);
|
||||
let Some(token) = response.page_token.filter(|token| !token.is_empty()) else {
|
||||
break;
|
||||
};
|
||||
if !seen_tokens.insert(token.clone()) {
|
||||
return Err(Error::Runtime {
|
||||
message: "Namespace table listing response repeated a page_token".into(),
|
||||
});
|
||||
}
|
||||
page_token = Some(token);
|
||||
}
|
||||
|
||||
Ok(response.tables)
|
||||
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)
|
||||
}
|
||||
|
||||
async fn list_tables(&self, request: ListTablesRequest) -> Result<ListTablesResponse> {
|
||||
@@ -589,6 +617,135 @@ mod tests {
|
||||
use futures::TryStreamExt;
|
||||
use tempfile::tempdir;
|
||||
|
||||
#[derive(Debug)]
|
||||
struct ScriptedNamespace {
|
||||
pages: Vec<ListTablesResponse>,
|
||||
requests: Mutex<Vec<ListTablesRequest>>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl LanceNamespace for ScriptedNamespace {
|
||||
fn namespace_id(&self) -> String {
|
||||
"scripted".into()
|
||||
}
|
||||
|
||||
async fn list_tables(
|
||||
&self,
|
||||
request: ListTablesRequest,
|
||||
) -> lance_core::Result<ListTablesResponse> {
|
||||
let mut requests = self.requests.lock().unwrap();
|
||||
let page = self.pages[requests.len()].clone();
|
||||
requests.push(request);
|
||||
Ok(page)
|
||||
}
|
||||
}
|
||||
|
||||
fn namespace_with_pages(
|
||||
pages: Vec<ListTablesResponse>,
|
||||
) -> (LanceNamespaceDatabase, Arc<ScriptedNamespace>) {
|
||||
let namespace = Arc::new(ScriptedNamespace {
|
||||
pages,
|
||||
requests: Mutex::new(Vec::new()),
|
||||
});
|
||||
let db = LanceNamespaceDatabase::from_namespace_client(
|
||||
namespace.clone(),
|
||||
"scripted".into(),
|
||||
HashMap::new(),
|
||||
HashMap::new(),
|
||||
None,
|
||||
None,
|
||||
HashSet::new(),
|
||||
);
|
||||
(db, namespace)
|
||||
}
|
||||
|
||||
fn listing_page(names: &[&str], token: Option<&str>) -> ListTablesResponse {
|
||||
ListTablesResponse {
|
||||
tables: names.iter().map(|name| (*name).into()).collect(),
|
||||
page_token: token.map(String::from),
|
||||
..Default::default()
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[allow(deprecated)]
|
||||
async fn test_table_names_collects_namespace_pages() {
|
||||
for namespace_path in [vec![], vec!["child".into()]] {
|
||||
for (start_after, limit, expected) in [
|
||||
(None, None, vec!["a", "b", "c", "d"]),
|
||||
(Some("a".into()), Some(2), vec!["b", "c"]),
|
||||
] {
|
||||
let (db, namespace) = namespace_with_pages(vec![
|
||||
listing_page(&["d", "b"], Some("opaque-1")),
|
||||
listing_page(&[], Some("opaque-2")),
|
||||
listing_page(&["c", "a"], None),
|
||||
]);
|
||||
let names = db
|
||||
.table_names(TableNamesRequest {
|
||||
namespace_path: namespace_path.clone(),
|
||||
start_after,
|
||||
limit,
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(names, expected);
|
||||
|
||||
let requests = namespace.requests.lock().unwrap();
|
||||
assert_eq!(requests.len(), 3);
|
||||
for request in requests.iter() {
|
||||
assert_eq!(request.id.as_ref(), Some(&namespace_path));
|
||||
assert_eq!(request.limit, None);
|
||||
}
|
||||
assert_eq!(requests[0].page_token, None);
|
||||
assert_eq!(requests[1].page_token.as_deref(), Some("opaque-1"));
|
||||
assert_eq!(requests[2].page_token.as_deref(), Some("opaque-2"));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[allow(deprecated)]
|
||||
async fn test_table_names_zero_limit_skips_namespace_request() {
|
||||
let (db, namespace) = namespace_with_pages(vec![]);
|
||||
assert!(
|
||||
db.table_names(TableNamesRequest {
|
||||
limit: Some(0),
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap()
|
||||
.is_empty()
|
||||
);
|
||||
assert!(namespace.requests.lock().unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[allow(deprecated)]
|
||||
async fn test_table_names_stops_on_empty_namespace_token() {
|
||||
let (db, namespace) = namespace_with_pages(vec![listing_page(&["a"], Some(""))]);
|
||||
assert_eq!(
|
||||
db.table_names(TableNamesRequest::default()).await.unwrap(),
|
||||
vec!["a"]
|
||||
);
|
||||
assert_eq!(namespace.requests.lock().unwrap().len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[allow(deprecated)]
|
||||
async fn test_table_names_rejects_namespace_token_cycle() {
|
||||
let (db, namespace) = namespace_with_pages(vec![
|
||||
listing_page(&["a"], Some("one")),
|
||||
listing_page(&["b"], Some("two")),
|
||||
listing_page(&["c"], Some("one")),
|
||||
]);
|
||||
let error = db
|
||||
.table_names(TableNamesRequest::default())
|
||||
.await
|
||||
.unwrap_err();
|
||||
assert!(error.to_string().contains("repeated a page_token"));
|
||||
assert_eq!(namespace.requests.lock().unwrap().len(), 3);
|
||||
}
|
||||
|
||||
/// Helper function to create test data
|
||||
fn create_test_data() -> RecordBatch {
|
||||
let schema = Arc::new(Schema::new(vec![
|
||||
|
||||
+176
-19
@@ -1350,25 +1350,49 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
|
||||
}
|
||||
|
||||
async fn table_names(&self, request: TableNamesRequest) -> Result<Vec<String>> {
|
||||
// A zero limit is an empty listing, even on servers that reject limit=0.
|
||||
if request.limit == Some(0) {
|
||||
return Ok(Vec::new());
|
||||
}
|
||||
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)]);
|
||||
// what `start_after` means. Subsequent cursors come from the server and may be
|
||||
// opaque. Follow them until the requested limit or the end of the listing.
|
||||
let mut tables = Vec::new();
|
||||
let mut version = None;
|
||||
let mut page_token = request.start_after.clone();
|
||||
let mut seen_tokens = HashSet::new();
|
||||
if let Some(ref token) = page_token {
|
||||
seen_tokens.insert(token.clone());
|
||||
}
|
||||
if let Some(ref start_after) = request.start_after {
|
||||
req = req.query(&[("page_token", start_after)]);
|
||||
loop {
|
||||
let mut req = self.client.get("/v1/table/");
|
||||
if let Some(limit) = request.limit {
|
||||
req = req.query(&[("limit", limit as usize - tables.len())]);
|
||||
}
|
||||
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)?;
|
||||
tables.extend(response.tables);
|
||||
if let Some(limit) = request.limit
|
||||
&& tables.len() >= limit as usize
|
||||
{
|
||||
tables.truncate(limit as usize);
|
||||
break;
|
||||
}
|
||||
// Empty or repeated tokens must not restart the listing or loop forever.
|
||||
match response.page_token.filter(|token| !token.is_empty()) {
|
||||
Some(token) if seen_tokens.insert(token.clone()) => page_token = Some(token),
|
||||
_ => break,
|
||||
}
|
||||
}
|
||||
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)
|
||||
(tables, version.unwrap_or_default())
|
||||
} else {
|
||||
self.table_names_in_namespace(&request).await?
|
||||
};
|
||||
@@ -2136,6 +2160,133 @@ mod tests {
|
||||
assert_eq!(names, vec!["table1", "table2"]);
|
||||
}
|
||||
|
||||
#[rstest::rstest]
|
||||
#[case(None, None)]
|
||||
#[case(Some("t009"), None)]
|
||||
#[case(None, Some(200))]
|
||||
#[case(Some("t009"), Some(15))]
|
||||
#[tokio::test]
|
||||
async fn test_table_names_follows_server_pages(
|
||||
#[case] start_after: Option<&str>,
|
||||
#[case] limit: Option<u32>,
|
||||
) {
|
||||
let conn = Connection::new_with_handler(|request| {
|
||||
assert_eq!(request.method(), &reqwest::Method::GET);
|
||||
assert_eq!(request.url().path(), "/v1/table/");
|
||||
let query: HashMap<_, _> = request.url().query_pairs().collect();
|
||||
let start = query
|
||||
.get("page_token")
|
||||
.map(|token| token[1..4].parse::<usize>().unwrap() + 1)
|
||||
.unwrap_or(0);
|
||||
let limit = query
|
||||
.get("limit")
|
||||
.map(|limit| limit.parse::<usize>().unwrap())
|
||||
.unwrap_or(10);
|
||||
// A server may return a short page even when more tables remain.
|
||||
let end = (start + limit.min(10)).min(130);
|
||||
let tables: Vec<_> = (start..end).map(|i| format!("t{i:03}")).collect();
|
||||
let page_token = (end < 130).then(|| format!("t{:03}.lance/", end - 1));
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(serde_json::json!({"tables": tables, "page_token": page_token}).to_string())
|
||||
.unwrap()
|
||||
});
|
||||
let mut op = conn.table_names();
|
||||
if let Some(start_after) = start_after {
|
||||
op = op.start_after(start_after);
|
||||
}
|
||||
if let Some(limit) = limit {
|
||||
op = op.limit(limit);
|
||||
}
|
||||
let start = if start_after.is_some() { 10 } else { 0 };
|
||||
let end = limit.map(|limit| (start + limit).min(130)).unwrap_or(130);
|
||||
let expected: Vec<_> = (start..end).map(|i| format!("t{i:03}")).collect();
|
||||
assert_eq!(op.execute().await.unwrap(), expected);
|
||||
}
|
||||
|
||||
#[rstest::rstest]
|
||||
#[case(vec![])]
|
||||
#[case(vec!["ns".to_string()])]
|
||||
#[tokio::test]
|
||||
async fn test_table_listing_zero_limit_never_sends_a_request(#[case] namespace: Vec<String>) {
|
||||
let conn = Connection::new_with_handler(|_| -> http::Response<String> {
|
||||
panic!("a zero limit must not be sent to the server")
|
||||
});
|
||||
assert!(
|
||||
conn.table_names()
|
||||
.namespace(namespace.clone())
|
||||
.limit(0)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap()
|
||||
.is_empty()
|
||||
);
|
||||
let page = conn
|
||||
.list_tables(lance_namespace::models::ListTablesRequest {
|
||||
id: Some(namespace),
|
||||
limit: Some(0),
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(page.tables.is_empty());
|
||||
assert!(page.page_token.is_none());
|
||||
}
|
||||
|
||||
#[rstest::rstest]
|
||||
#[case(vec![], "/v1/namespace/$/table/list")]
|
||||
#[case(vec!["ns".to_string()], "/v1/namespace/ns/table/list")]
|
||||
#[tokio::test]
|
||||
async fn test_list_tables_default_page_size(
|
||||
#[case] namespace: Vec<String>,
|
||||
#[case] path: &'static str,
|
||||
) {
|
||||
let conn = Connection::new_with_handler(move |request| {
|
||||
assert_eq!(request.url().path(), path);
|
||||
let query: HashMap<_, _> = request.url().query_pairs().collect();
|
||||
assert_eq!(query.get("limit").map(|limit| limit.as_ref()), Some("100"));
|
||||
let start = match query.get("page_token") {
|
||||
None => 0,
|
||||
Some(token) => {
|
||||
assert_eq!(token, "opaque-token");
|
||||
100
|
||||
}
|
||||
};
|
||||
let end = (start + 100).min(130);
|
||||
let tables: Vec<_> = (start..end).map(|i| format!("t{i:03}")).collect();
|
||||
http::Response::builder()
|
||||
.status(200)
|
||||
.body(
|
||||
serde_json::json!({
|
||||
"tables": tables,
|
||||
"page_token": (end < 130).then_some("opaque-token")
|
||||
})
|
||||
.to_string(),
|
||||
)
|
||||
.unwrap()
|
||||
});
|
||||
let first = conn
|
||||
.list_tables(lance_namespace::models::ListTablesRequest {
|
||||
id: Some(namespace.clone()),
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
assert_eq!(first.tables.len(), 100);
|
||||
assert_eq!(first.page_token.as_deref(), Some("opaque-token"));
|
||||
let second = conn
|
||||
.list_tables(lance_namespace::models::ListTablesRequest {
|
||||
id: Some(namespace),
|
||||
page_token: first.page_token,
|
||||
..Default::default()
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
let expected: Vec<_> = (100..130).map(|i| format!("t{i:03}")).collect();
|
||||
assert_eq!(second.tables, expected);
|
||||
assert!(second.page_token.is_none());
|
||||
}
|
||||
|
||||
#[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
|
||||
@@ -2181,8 +2332,11 @@ mod tests {
|
||||
assert_eq!(names, vec!["widgets"]);
|
||||
}
|
||||
|
||||
#[rstest::rstest]
|
||||
#[case(vec![])]
|
||||
#[case(vec!["ns".to_string()])]
|
||||
#[tokio::test]
|
||||
async fn test_table_names_in_a_namespace_stops_on_a_repeated_token() {
|
||||
async fn test_table_names_stops_on_a_repeated_token(#[case] namespace: Vec<String>) {
|
||||
// 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()
|
||||
@@ -2193,7 +2347,7 @@ mod tests {
|
||||
|
||||
let names = conn
|
||||
.table_names()
|
||||
.namespace(vec!["ns".to_string()])
|
||||
.namespace(namespace)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
@@ -2202,8 +2356,11 @@ mod tests {
|
||||
assert_eq!(names, vec!["a", "a"]);
|
||||
}
|
||||
|
||||
#[rstest::rstest]
|
||||
#[case(vec![])]
|
||||
#[case(vec!["ns".to_string()])]
|
||||
#[tokio::test]
|
||||
async fn test_table_names_in_a_namespace_stops_on_an_empty_token() {
|
||||
async fn test_table_names_stops_on_an_empty_token(#[case] namespace: Vec<String>) {
|
||||
// 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.
|
||||
@@ -2223,7 +2380,7 @@ mod tests {
|
||||
|
||||
let names = conn
|
||||
.table_names()
|
||||
.namespace(vec!["ns".to_string()])
|
||||
.namespace(namespace)
|
||||
.execute()
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
Reference in New Issue
Block a user