diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index cd872621b..4ac2c3070 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -232,7 +232,10 @@ jobs: ALL_FEATURES=`cargo metadata --format-version=1 --no-deps \ | jq -r '.packages[] | .features | keys | .[]' \ | grep -v s3-test | sort | uniq | paste -s -d "," -` - cargo test --profile ci --features $ALL_FEATURES --locked + # Run doctests before test binaries fill the runner disk. Examples are + # already built by the Linux job, so avoid retaining them here. + cargo test --profile ci --features $ALL_FEATURES --locked --doc + cargo test --profile ci --features $ALL_FEATURES --locked --lib --tests windows: strategy: diff --git a/Cargo.lock b/Cargo.lock index bcc7526c8..d0c3dc408 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -332,6 +332,34 @@ dependencies = [ "num-traits", ] +[[package]] +name = "arrow-flight" +version = "58.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b2dbe34824c639e43136af8f106992792ab456540d54b880bc320a3192502d2e" +dependencies = [ + "arrow-arith", + "arrow-array", + "arrow-buffer", + "arrow-cast", + "arrow-data", + "arrow-ipc", + "arrow-ord", + "arrow-row", + "arrow-schema", + "arrow-select", + "arrow-string", + "base64 0.22.1", + "bytes", + "futures", + "once_cell", + "paste", + "prost", + "prost-types", + "tonic", + "tonic-prost", +] + [[package]] name = "arrow-ipc" version = "58.4.0" @@ -1129,7 +1157,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f" dependencies = [ "async-trait", - "axum-core", + "axum-core 0.4.5", "bytes", "futures-util", "http 1.5.0", @@ -1138,7 +1166,7 @@ dependencies = [ "hyper 1.9.0", "hyper-util", "itoa", - "matchit", + "matchit 0.7.3", "memchr", "mime", "percent-encoding", @@ -1156,6 +1184,31 @@ dependencies = [ "tracing", ] +[[package]] +name = "axum" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "31b698c5f9a010f6573133b09e0de5408834d0c82f8d7475a89fc1867a71cd90" +dependencies = [ + "axum-core 0.5.6", + "bytes", + "futures-util", + "http 1.5.0", + "http-body 1.1.0", + "http-body-util", + "itoa", + "matchit 0.8.4", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "sync_wrapper", + "tower", + "tower-layer", + "tower-service", +] + [[package]] name = "axum-core" version = "0.4.5" @@ -1177,6 +1230,24 @@ dependencies = [ "tracing", ] +[[package]] +name = "axum-core" +version = "0.5.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08c78f31d7b1291f7ee735c1c6780ccde7785daae9a9206026862dab7d8792d1" +dependencies = [ + "bytes", + "futures-core", + "http 1.5.0", + "http-body 1.1.0", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", +] + [[package]] name = "backoff" version = "0.4.0" @@ -5272,7 +5343,7 @@ dependencies = [ "arrow-ipc", "arrow-schema", "async-trait", - "axum", + "axum 0.7.9", "base64 0.22.1", "bytes", "chrono", @@ -5412,6 +5483,7 @@ dependencies = [ "arrow-buffer", "arrow-cast", "arrow-data", + "arrow-flight", "arrow-ipc", "arrow-ord", "arrow-schema", @@ -5467,6 +5539,7 @@ dependencies = [ "polars", "polars-arrow", "pprof 0.14.1", + "prost", "rand 0.9.5", "random_word", "regex", @@ -5483,6 +5556,7 @@ dependencies = [ "test-log", "tokenizers", "tokio", + "tonic", "url", "urlencoding", "uuid", @@ -5540,6 +5614,7 @@ dependencies = [ "serde_json", "snafu 0.8.9", "tokio", + "uuid", ] [[package]] @@ -5860,6 +5935,12 @@ version = "0.7.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0e7465ac9959cc2b1404e8e2367b43684a6d13790fe23056cc8c6c5a6b7bcb94" +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + [[package]] name = "matrixmultiply" version = "0.3.10" @@ -7735,6 +7816,7 @@ dependencies = [ "pyo3-build-config", "pyo3-ffi", "pyo3-macros", + "uuid", ] [[package]] @@ -10087,6 +10169,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ac2a5518c70fa84342385732db33fb3f44bc4cc748936eb5833d2df34d6445ef" dependencies = [ "async-trait", + "axum 0.8.9", "base64 0.22.1", "bytes", "h2 0.4.16", @@ -10098,9 +10181,11 @@ dependencies = [ "hyper-util", "percent-encoding", "pin-project", + "rustls-native-certs", "socket2 0.6.3", "sync_wrapper", "tokio", + "tokio-rustls 0.26.4", "tokio-stream", "tower", "tower-layer", diff --git a/Cargo.toml b/Cargo.toml index eba5a421a..63ebeb75a 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,6 +39,7 @@ arrow-ord = "58.0.0" arrow-schema = "58.0.0" arrow-select = "58.0.0" arrow-cast = "58.0.0" +arrow-flight = { version = "58.0.0", features = ["flight-sql-experimental"] } async-trait = "0" bytes = "1" datafusion = { version = "54.0.0", default-features = false } @@ -71,7 +72,8 @@ serde = "1" serde_json = "1" tempfile = "3.5.0" tokio = { version = "1.23", features = ["rt-multi-thread", "sync"] } -uuid = { version = "1.7.0", features = ["v4"] } +tonic = { version = "0.14", features = ["tls-native-roots", "tls-ring"] } +uuid = { version = "1.7.0", features = ["v4", "v7"] } chrono = { version = "0.4", default-features = false, features = ["clock"] } [profile.ci] diff --git a/docs/src/python/python.md b/docs/src/python/python.md index 5f359f4b7..f2bc72b3a 100644 --- a/docs/src/python/python.md +++ b/docs/src/python/python.md @@ -28,6 +28,59 @@ is also an [asynchronous API client](#connections-asynchronous). ::: lancedb.Session +## Remote SQL + +Submit SQL against a remote LanceDB database through the connection. +The connected database and `default_namespace_path=["public"]` are used for +unqualified tables. Fully qualified references can still query other databases +and namespaces available to the same deployment. `execute_query` returns a +reader as soon as its initial result stream is available. `execute_query_async` +returns a query handle immediately; use it to inspect progress, open a reader, +or cancel the query. The SQL client is initialized by the first query and +retained for the lifetime of the remote connection. Query ids are random, +connection-scoped references rather than encoded SQL or durable resume tokens: + +```python +import lancedb + +db = lancedb.connect( + "db://analytics", + api_key="ldb_...", + host_override="https://api.example.com", + sql_host_override="grpc+tls://sql.example.com:10026", +) +reader = db.execute_query( + """ + SELECT events.id, accounts.name + FROM analytics.public.events AS events + JOIN users.public.accounts AS accounts ON events.user_id = accounts.id + """, + default_namespace_path=["public"], +) +for batch in reader: + print(batch.num_rows) + +query = db.execute_query_async("SELECT * FROM events") +print(query.id) +print(query.describe().status) +for batch in query.reader(): + print(batch.num_rows) + +# The async connection exposes the same lifecycle without blocking: +# async_db = await lancedb.connect_async( +# "db://analytics", +# api_key="ldb_...", +# host_override="https://api.example.com", +# sql_host_override="grpc+tls://sql.example.com:10026", +# ) +# reader = await async_db.execute_query("SELECT * FROM events") +# query = await async_db.execute_query_async("SELECT * FROM events") +# description = await async_db.describe_query(query.id) +# async for batch in await query.reader(): +# print(batch.num_rows) +# await query.cancel() +``` + ## Namespaces (Synchronous) A namespace-backed connection resolves tables through a @@ -102,6 +155,12 @@ listing a storage directory. ::: lancedb.job.AsyncJob +::: lancedb.sql.Query + +::: lancedb.sql.AsyncQuery + +::: lancedb.sql.QueryDescription + ## Materialized Views (Synchronous) ::: lancedb.materialized_view.MaterializedView diff --git a/python/Cargo.toml b/python/Cargo.toml index 98ae17fef..850752925 100644 --- a/python/Cargo.toml +++ b/python/Cargo.toml @@ -28,7 +28,7 @@ env_logger.workspace = true log.workspace = true # Maturin enables extension-module mode for Python builds. Keeping it out of # Cargo features lets Rust unit tests link against libpython. -pyo3 = { version = "0.28", features = ["abi3-py310", "chrono"] } +pyo3 = { version = "0.28", features = ["abi3-py310", "chrono", "uuid"] } chrono.workspace = true pyo3-async-runtimes = { version = "0.28", features = [ "attributes", @@ -40,6 +40,7 @@ serde.workspace = true serde_json.workspace = true snafu.workspace = true tokio.workspace = true +uuid.workspace = true libc = "0.2" [build-dependencies] diff --git a/python/pyproject.toml b/python/pyproject.toml index 22a41a8a9..ffdfdd948 100644 --- a/python/pyproject.toml +++ b/python/pyproject.toml @@ -139,6 +139,7 @@ include = [ "python/lancedb/exceptions.py", "python/lancedb/background_loop.py", "python/lancedb/schema.py", + "python/lancedb/sql.py", "python/lancedb/remote/__init__.py", "python/lancedb/remote/errors.py", "python/lancedb/embeddings/__init__.py", diff --git a/python/python/lancedb/__init__.py b/python/python/lancedb/__init__.py index 21ffc8860..af325a29b 100644 --- a/python/python/lancedb/__init__.py +++ b/python/python/lancedb/__init__.py @@ -22,6 +22,9 @@ from .remote.db import RemoteDBConnection from .expr import Expr, col, lit, func from .schema import blob, vector from .job import AsyncJob, Job +from .sql import AsyncQuery as AsyncSqlQuery +from .sql import Query as SqlQuery +from .sql import QueryDescription from .functions import ( FunctionArtifactRequest as FunctionArtifactRequest, FunctionApplication as FunctionApplication, @@ -101,6 +104,7 @@ def connect( api_key: Optional[str] = None, region: str = "us-east-1", host_override: Optional[str] = None, + sql_host_override: Optional[str] = None, read_consistency_interval: Optional[timedelta] = None, request_thread_pool: Optional[Union[int, ThreadPoolExecutor]] = None, client_config: Union[ClientConfig, Dict[str, Any], None] = None, @@ -129,6 +133,9 @@ def connect( The region to use for LanceDB Cloud. host_override: str, optional The override url for LanceDB Cloud. + sql_host_override: str, optional + The remote SQL service endpoint override. The client connects lazily when SQL + is first executed and retains that connection. read_consistency_interval: timedelta, default None The interval at which to check for updates to the table from other processes. If None, then consistency is not checked. For performance @@ -270,6 +277,7 @@ def connect( api_key, region, host_override, + sql_host_override=sql_host_override, # TODO: remove this (deprecation warning downstream) request_thread_pool=request_thread_pool, client_config=client_config, @@ -412,6 +420,7 @@ def deserialize_conn( parsed["api_key"], parsed.get("region", "us-east-1"), host_override=parsed.get("host_override"), + sql_host_override=parsed.get("sql_host_override"), client_config=parsed.get("client_config"), storage_options=storage_options, ) @@ -425,6 +434,7 @@ async def connect_async( api_key: Optional[str] = None, region: str = "us-east-1", host_override: Optional[str] = None, + sql_host_override: Optional[str] = None, read_consistency_interval: Optional[timedelta] = None, client_config: Optional[Union[ClientConfig, Dict[str, Any]]] = None, storage_options: Optional[Dict[str, str]] = None, @@ -447,6 +457,9 @@ async def connect_async( The region to use for LanceDB Cloud. host_override: str, optional The override url for LanceDB Cloud. + sql_host_override: str, optional + The remote SQL service endpoint override. The client connects lazily when SQL + is first executed and retains that connection. read_consistency_interval: timedelta, default None The interval at which to check for updates to the table from other processes. If None, then consistency is not checked. For performance @@ -534,6 +547,7 @@ async def connect_async( api_key, region, host_override, + sql_host_override, read_consistency_interval_secs, client_config, storage_options, @@ -556,6 +570,7 @@ __all__ = [ "connect_namespace_async", "AsyncConnection", "AsyncJob", + "AsyncSqlQuery", "AsyncLanceNamespaceDBConnection", "AsyncTable", "FtsToken", @@ -570,6 +585,8 @@ __all__ = [ "vector", "DBConnection", "Job", + "QueryDescription", + "SqlQuery", "LanceDBConnection", "LanceNamespaceDBConnection", "LsmWriteSpec", diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index 78826df92..3ea8d15c7 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -1,6 +1,7 @@ from datetime import date, datetime, timedelta from decimal import Decimal from typing import Dict, List, Optional, Tuple, Any, TypedDict, Union, Literal +from uuid import UUID import pyarrow as pa @@ -158,6 +159,13 @@ class Connection(object): async def job_history( self, job_id: Optional[str] = None ) -> List[pa.RecordBatch]: ... + async def execute_query_async( + self, + query: str, + *, + default_namespace_path: Optional[List[str]] = None, + ) -> SqlQuery: ... + async def describe_query(self, query_id: UUID) -> QueryDescription: ... async def create_table( self, name: str, @@ -274,6 +282,23 @@ class JobDescription: @property def failure(self) -> Optional[JobFailureInfo]: ... +class SqlQuery: + @property + def id(self) -> UUID: ... + async def describe(self) -> QueryDescription: ... + async def reader(self) -> RecordBatchStream: ... + async def cancel(self) -> None: ... + +class QueryDescription: + @property + def id(self) -> UUID: ... + @property + def status(self) -> str: ... + @property + def progress(self) -> Optional[float]: ... + @property + def expires_at(self) -> Optional[datetime]: ... + class Table: def name(self) -> str: ... def __repr__(self) -> str: ... @@ -452,6 +477,7 @@ async def connect( api_key: Optional[str], region: Optional[str], host_override: Optional[str], + sql_host_override: Optional[str], read_consistency_interval: Optional[float], client_config: Optional[Union[ClientConfig, Dict[str, Any]]], storage_options: Optional[Dict[str, str]], diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index 2554e9908..82dfa22f1 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -19,6 +19,7 @@ from typing import ( Optional, Union, ) +from uuid import UUID if sys.version_info >= (3, 12): from typing import override @@ -47,6 +48,9 @@ from . import __version__ from ._lancedb import connect as lancedb_connect # type: ignore from .functions import FunctionVersion, UdfDefinition from .job import AsyncJob, Job, _typed_job +from .sql import AsyncQuery as AsyncSqlQuery +from .sql import Query as SqlQuery +from .sql import QueryDescription from .materialized_view import ( AsyncMaterializedView, MaterializedView, @@ -68,6 +72,7 @@ import deprecation if TYPE_CHECKING: import pyarrow as pa + from .arrow import AsyncRecordBatchReader from .pydantic import LanceModel from ._lancedb import Connection as LanceDbConnection @@ -780,6 +785,39 @@ class DBConnection(EnforceOverrides): "job_history is not supported for this connection type" ) + def execute_query( + self, + query: str, + *, + default_namespace_path: Optional[List[str]] = None, + ) -> pa.RecordBatchReader: + """Execute SQL and return a blocking Arrow reader. + + This submits through :meth:`execute_query_async` and waits until the + initial result stream is readable. It does not wait for the full query + to finish. + """ + return self.execute_query_async( + query, + default_namespace_path=default_namespace_path, + ).reader() + + def execute_query_async( + self, + query: str, + *, + default_namespace_path: Optional[List[str]] = None, + ) -> SqlQuery: + """Start executing SQL and return its query handle. + + Local connections do not support SQL. + """ + raise NotImplementedError("SQL is not supported for this connection type") + + def describe_query(self, query_id: UUID) -> QueryDescription: + """Describe a submitted SQL query by its connection-scoped id.""" + raise NotImplementedError("SQL is not supported for this connection type") + class LanceDBConnection(DBConnection): """ @@ -875,6 +913,7 @@ class LanceDBConnection(DBConnection): None, None, None, + None, read_consistency_interval_secs, None, storage_options, @@ -2321,6 +2360,47 @@ class AsyncConnection(object): """ return await self._inner.job_history(job_id) + async def execute_query( + self, + query: str, + *, + default_namespace_path: Optional[List[str]] = None, + ) -> AsyncRecordBatchReader: + """Execute SQL and return an asynchronous Arrow reader. + + This submits through :meth:`execute_query_async` and waits until the + initial result stream is readable. It does not wait for the full query + to finish. + """ + submitted = await self.execute_query_async( + query, + default_namespace_path=default_namespace_path, + ) + return await submitted.reader() + + async def execute_query_async( + self, + query: str, + *, + default_namespace_path: Optional[List[str]] = None, + ) -> AsyncSqlQuery: + """Start executing SQL and return its query handle. + + The database from ``connect_async`` is used for unqualified database + references. The namespace defaults to ``["public"]``. Local + connections raise ``NotImplementedError``. + """ + return AsyncSqlQuery( + await self._inner.execute_query_async( + query, + default_namespace_path=default_namespace_path, + ) + ) + + async def describe_query(self, query_id: UUID) -> QueryDescription: + """Describe a submitted SQL query by its connection-scoped id.""" + return await self._inner.describe_query(query_id) + async def namespace_client(self) -> LanceNamespace: """Get the equivalent namespace client for this connection. diff --git a/python/python/lancedb/namespace.py b/python/python/lancedb/namespace.py index f2e553321..61d2122f3 100644 --- a/python/python/lancedb/namespace.py +++ b/python/python/lancedb/namespace.py @@ -12,6 +12,7 @@ from __future__ import annotations import sys from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Union +from uuid import UUID if sys.version_info >= (3, 12): from typing import override @@ -48,8 +49,11 @@ from lancedb._lancedb import ( connect_namespace_client as _connect_namespace_client, ) from lancedb.background_loop import LOOP +from lancedb.arrow import AsyncRecordBatchReader from lancedb.db import AsyncConnection, DBConnection from lancedb.job import AsyncJob, Job +from lancedb.sql import AsyncQuery as AsyncSqlQuery +from lancedb.sql import QueryDescription from lance_namespace import ( LanceNamespace, connect as namespace_connect, @@ -1447,6 +1451,37 @@ class AsyncLanceNamespaceDBConnection: namespace_path=namespace_path, page_token=page_token, limit=limit ) + async def execute_query( + self, + query: str, + *, + default_namespace_path: Optional[List[str]] = None, + ) -> AsyncRecordBatchReader: + """Execute SQL when supported by the underlying connection.""" + return await self._inner.execute_query( + query, + default_namespace_path=default_namespace_path, + ) + + async def execute_query_async( + self, + query: str, + *, + default_namespace_path: Optional[List[str]] = None, + ) -> AsyncSqlQuery: + """Start executing SQL when supported by the underlying connection. + + Namespace-backed local connections do not support SQL. + """ + return await self._inner.execute_query_async( + query, + default_namespace_path=default_namespace_path, + ) + + async def describe_query(self, query_id: UUID) -> QueryDescription: + """Describe a submitted SQL query when supported.""" + return await self._inner.describe_query(query_id) + async def namespace_client(self) -> LanceNamespace: """Get the namespace client for this connection. diff --git a/python/python/lancedb/remote/db.py b/python/python/lancedb/remote/db.py index 0e95e035b..1c0dd3afa 100644 --- a/python/python/lancedb/remote/db.py +++ b/python/python/lancedb/remote/db.py @@ -9,6 +9,7 @@ from concurrent.futures import ThreadPoolExecutor import sys from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Union from urllib.parse import urlparse +from uuid import UUID import warnings if sys.version_info >= (3, 12): @@ -25,6 +26,8 @@ from ..common import DATA from ..db import DBConnection, LOOP from ..functions import FunctionVersion, UdfDefinition from ..job import AsyncJob, Job +from ..sql import Query as SqlQuery +from ..sql import QueryDescription from ..materialized_view import MaterializedView, SelectArg if TYPE_CHECKING: @@ -116,6 +119,7 @@ class RemoteDBConnection(DBConnection): read_timeout: Optional[float] = None, storage_options: Optional[Dict[str, str]] = None, read_consistency_interval: Optional[timedelta] = None, + sql_host_override: Optional[str] = None, ): """Connect to a remote LanceDB database.""" if isinstance(client_config, dict): @@ -161,6 +165,7 @@ class RemoteDBConnection(DBConnection): self.api_key = api_key self.region = region self.host_override = host_override + self.sql_host_override = sql_host_override self.storage_options = storage_options self.db_name = parsed.netloc @@ -175,6 +180,7 @@ class RemoteDBConnection(DBConnection): api_key=api_key, region=region, host_override=host_override, + sql_host_override=sql_host_override, client_config=client_config, storage_options=storage_options, read_consistency_interval=read_consistency_interval, @@ -193,6 +199,7 @@ class RemoteDBConnection(DBConnection): "api_key": self.api_key, "region": self.region, "host_override": self.host_override, + "sql_host_override": self.sql_host_override, "client_config": _client_config_to_dict(self.client_config), "storage_options": self.storage_options, } @@ -788,6 +795,37 @@ class RemoteDBConnection(DBConnection): """ return LOOP.run(self._conn.job_history(job_id)) + @override + def execute_query_async( + self, + query: str, + *, + default_namespace_path: Optional[List[str]] = None, + ) -> SqlQuery: + """Start executing SQL through this remote connection. + + Unqualified tables use this connection's database and the + ``["public"]`` namespace by default. Fully qualified table names may + reference other databases available to the same deployment. + """ + return SqlQuery( + LOOP.run( + self._conn.execute_query_async( + query, + default_namespace_path=default_namespace_path, + ) + ) + ) + + @override + def describe_query(self, query_id: UUID) -> QueryDescription: + """Describe a submitted SQL query by its connection-scoped id.""" + return LOOP.run( + self._conn.describe_query( + query_id, + ) + ) + @override def namespace_client(self) -> LanceNamespace: """Get the equivalent namespace client for this connection. diff --git a/python/python/lancedb/remote/header.py b/python/python/lancedb/remote/header.py index 06e3599f5..206b7a7cb 100644 --- a/python/python/lancedb/remote/header.py +++ b/python/python/lancedb/remote/header.py @@ -177,4 +177,7 @@ class OAuthProvider(HeaderProvider): if not self._current_token: raise RuntimeError("Failed to obtain OAuth token") - return {"Authorization": f"Bearer {self._current_token}"} + return { + "Authorization": f"Bearer {self._current_token}", + "x-lancedb-credential-type": "oidc", + } diff --git a/python/python/lancedb/sql.py b/python/python/lancedb/sql.py new file mode 100644 index 000000000..41bbb5328 --- /dev/null +++ b/python/python/lancedb/sql.py @@ -0,0 +1,88 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The LanceDB Authors + +"""Handles to SQL queries running on a remote database.""" + +from uuid import UUID + +import pyarrow as pa + +from lancedb.background_loop import LOOP + +from . import _lancedb +from .arrow import AsyncRecordBatchReader + +QueryDescription = _lancedb.QueryDescription + + +class AsyncQuery: + """A handle to a submitted SQL query on an asynchronous connection.""" + + def __init__(self, inner: "_lancedb.SqlQuery"): + self._inner = inner + + @property + def id(self) -> UUID: + """The stable identifier scoped to the connection that submitted it.""" + return self._inner.id + + async def describe(self) -> QueryDescription: + """Get a point-in-time description of the query.""" + return await self._inner.describe() + + async def reader(self) -> AsyncRecordBatchReader: + """Wait for the initial result stream and return its Arrow reader. + + Results are single-consumer. Calling this method more than once on the + same query raises an error. Later batches are streamed as they become + available without waiting for the full query to finish. + """ + return AsyncRecordBatchReader(await self._inner.reader()) + + async def cancel(self) -> None: + """Request cancellation of the query.""" + await self._inner.cancel() + + +class Query: + """Synchronous counterpart of :class:`AsyncQuery`.""" + + def __init__(self, inner: AsyncQuery): + self._inner = inner + + @property + def id(self) -> UUID: + """The stable identifier scoped to the connection that submitted it.""" + return self._inner.id + + def describe(self) -> QueryDescription: + """Get a point-in-time description of the query.""" + return LOOP.run(self._inner.describe()) + + def reader(self) -> pa.RecordBatchReader: + """Wait for the initial result stream and return a blocking reader. + + Results are single-consumer. Calling this method more than once on the + same query raises an error. Later batches block only until they become + available, without waiting for the full query to finish. + """ + reader = LOOP.run(self._inner.reader()) + + def next_batch(): + try: + return LOOP.run(reader.__anext__()) + except StopAsyncIteration: + return None + + def batches(): + while (batch := next_batch()) is not None: + yield batch + + return pa.RecordBatchReader.from_batches(reader.schema, batches()) + + def cancel(self) -> None: + """Request cancellation of the query.""" + LOOP.run(self._inner.cancel()) + + +__all__ = ["AsyncQuery", "Query", "QueryDescription"] diff --git a/python/python/tests/test_header_provider.py b/python/python/tests/test_header_provider.py index 84c5d7729..187b0a50a 100644 --- a/python/python/tests/test_header_provider.py +++ b/python/python/tests/test_header_provider.py @@ -54,7 +54,10 @@ class TestOAuthProvider: provider = OAuthProvider(fetcher) headers = provider.get_headers() - assert headers == {"Authorization": "Bearer token123"} + assert headers == { + "Authorization": "Bearer token123", + "x-lancedb-credential-type": "oidc", + } assert provider._current_token == "token123" assert provider._token_expires_at is not None @@ -73,14 +76,20 @@ class TestOAuthProvider: # First call headers1 = provider.get_headers() - assert headers1 == {"Authorization": "Bearer token1"} + assert headers1 == { + "Authorization": "Bearer token1", + "x-lancedb-credential-type": "oidc", + } # Wait for token to expire time.sleep(1.1) # Second call should refresh headers2 = provider.get_headers() - assert headers2 == {"Authorization": "Bearer token2"} + assert headers2 == { + "Authorization": "Bearer token2", + "x-lancedb-credential-type": "oidc", + } assert call_count == 2 def test_no_expiry_info(self): @@ -92,12 +101,18 @@ class TestOAuthProvider: provider = OAuthProvider(fetcher) headers = provider.get_headers() - assert headers == {"Authorization": "Bearer permanent_token"} + assert headers == { + "Authorization": "Bearer permanent_token", + "x-lancedb-credential-type": "oidc", + } assert provider._token_expires_at is None # Should not refresh on second call headers2 = provider.get_headers() - assert headers2 == {"Authorization": "Bearer permanent_token"} + assert headers2 == { + "Authorization": "Bearer permanent_token", + "x-lancedb-credential-type": "oidc", + } def test_missing_access_token(self): """Test error handling when access_token is missing.""" @@ -121,7 +136,10 @@ class TestOAuthProvider: provider = OAuthProvider(fetcher) headers = provider.get_headers() - assert headers == {"Authorization": "Bearer sync_token"} + assert headers == { + "Authorization": "Bearer sync_token", + "x-lancedb-credential-type": "oidc", + } class TestClientConfigIntegration: diff --git a/python/python/tests/test_sql.py b/python/python/tests/test_sql.py new file mode 100644 index 000000000..eeadb3a33 --- /dev/null +++ b/python/python/tests/test_sql.py @@ -0,0 +1,162 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright The LanceDB Authors + +from uuid import UUID + +import pytest +import pyarrow as pa + +import lancedb +from lancedb import _lancedb +from lancedb.arrow import AsyncRecordBatchReader +from lancedb.db import AsyncConnection +from lancedb.remote.db import RemoteDBConnection +from lancedb.sql import AsyncQuery, Query + +NIL_QUERY_ID = UUID(int=0) + + +class FakeNativeQuery: + id = UUID("0198f1b2-c3d4-7e5f-8123-456789abcdef") + + async def reader(self): + return pa.table({"value": [1, 2]}) + + +class FakeNativeConnection: + async def execute_query_async(self, query, *, default_namespace_path=None): + return FakeNativeQuery() + + +class FakeAsyncConnection: + async def execute_query_async(self, query, *, default_namespace_path=None): + return AsyncQuery(FakeNativeQuery()) + + +def remote_connection(sql_host_override=None): + return lancedb.connect( + "db://analytics", + api_key="test-key", + host_override="http://localhost:10024", + sql_host_override=sql_host_override, + ) + + +def test_sql_is_connection_scoped(): + assert hasattr(lancedb, "sql") + assert not callable(lancedb.sql) + assert not hasattr(_lancedb, "sql") + assert not hasattr(remote_connection(), "sql") + assert hasattr(remote_connection(), "execute_query") + assert hasattr(remote_connection(), "execute_query_async") + assert hasattr(remote_connection(), "describe_query") + + +def test_query_id_is_uuid(): + query = AsyncQuery(FakeNativeQuery()) + assert isinstance(query.id, UUID) + assert Query(query).id == query.id + + +def test_connection_serializes_sql_host_override(): + endpoint = "grpc+tls://sql.example.com:10026" + restored = lancedb.deserialize_conn( + remote_connection(sql_host_override=endpoint).serialize() + ) + assert restored.sql_host_override == endpoint + + +@pytest.mark.asyncio +async def test_async_sql_reader_is_record_batch_stream(): + reader = await AsyncQuery(FakeNativeQuery()).reader() + assert isinstance(reader, AsyncRecordBatchReader) + assert (await reader.read_all())[0].column(0).to_pylist() == [1, 2] + + +def test_sync_sql_reader_is_record_batch_reader(): + reader = Query(AsyncQuery(FakeNativeQuery())).reader() + assert isinstance(reader, pa.RecordBatchReader) + assert reader.read_all().column(0).to_pylist() == [1, 2] + + +def test_execute_query_returns_blocking_reader(): + connection = RemoteDBConnection.__new__(RemoteDBConnection) + connection._conn = FakeAsyncConnection() + reader = connection.execute_query("SELECT 1") + assert isinstance(reader, pa.RecordBatchReader) + assert reader.read_all().column(0).to_pylist() == [1, 2] + + +@pytest.mark.asyncio +async def test_async_execute_query_returns_async_reader(): + connection = AsyncConnection(FakeNativeConnection()) + reader = await connection.execute_query("SELECT 1") + assert isinstance(reader, AsyncRecordBatchReader) + assert (await reader.read_all())[0].column(0).to_pylist() == [1, 2] + + +def test_local_connection_rejects_sql(tmp_path): + connection = lancedb.connect(tmp_path) + with pytest.raises(NotImplementedError, match="SQL"): + connection.execute_query("SELECT 1") + with pytest.raises(NotImplementedError, match="SQL"): + connection.execute_query_async("SELECT 1") + with pytest.raises(NotImplementedError, match="SQL"): + connection.describe_query(NIL_QUERY_ID) + + +@pytest.mark.asyncio +async def test_local_async_connection_rejects_sql(tmp_path): + connection = await lancedb.connect_async(tmp_path) + with pytest.raises(NotImplementedError, match="SQL"): + await connection.execute_query("SELECT 1") + with pytest.raises(NotImplementedError, match="SQL"): + await connection.execute_query_async("SELECT 1") + with pytest.raises(NotImplementedError, match="SQL"): + await connection.describe_query(NIL_QUERY_ID) + + +@pytest.mark.asyncio +async def test_async_namespace_connection_rejects_sql(tmp_path): + connection = lancedb.connect_namespace_async("dir", {"root": str(tmp_path)}) + with pytest.raises(NotImplementedError, match="SQL"): + await connection.execute_query("SELECT 1") + with pytest.raises(NotImplementedError, match="SQL"): + await connection.execute_query_async("SELECT 1") + with pytest.raises(NotImplementedError, match="SQL"): + await connection.describe_query(NIL_QUERY_ID) + + +def test_describe_query_requires_uuid(): + with pytest.raises(TypeError, match="UUID"): + remote_connection().describe_query(str(NIL_QUERY_ID)) + + +@pytest.mark.parametrize( + "default_namespace_path", + ["public", ("public",), [1]], +) +def test_execute_query_async_requires_namespace_path_list(default_namespace_path): + with pytest.raises(ValueError, match="default_namespace_path"): + remote_connection().execute_query_async( + "SELECT 1", default_namespace_path=default_namespace_path + ) + + +def test_execute_query_async_rejects_invalid_endpoint(): + connection = remote_connection(sql_host_override="invalid://localhost") + with pytest.raises(ValueError, match="sql_host_override"): + connection.execute_query_async("SELECT 1") + + +@pytest.mark.parametrize( + "default_namespace_path", + [[""], ["café"], ["pub\tlic"], ["events$raw"]], +) +def test_execute_query_async_rejects_invalid_namespace_components( + default_namespace_path, +): + with pytest.raises(ValueError, match="default_namespace_path"): + remote_connection().execute_query_async( + "SELECT 1", default_namespace_path=default_namespace_path + ) diff --git a/python/src/connection.rs b/python/src/connection.rs index 5477ab3d2..882fdfd29 100644 --- a/python/src/connection.rs +++ b/python/src/connection.rs @@ -28,7 +28,7 @@ use pyo3::{ Bound, FromPyObject, Py, PyAny, PyRef, PyResult, Python, exceptions::{PyRuntimeError, PyValueError}, pyclass, pyfunction, pymethods, - types::{PyDict, PyDictMethods, PyList, PyListMethods}, + types::{PyAnyMethods, PyDict, PyDictMethods, PyList, PyListMethods}, }; #[pyclass] @@ -86,6 +86,24 @@ impl Connection { } } +fn parse_default_namespace_path(path: Option>) -> PyResult> { + match path { + Some(path) => { + if !path.is_instance_of::() { + return Err(PyValueError::new_err( + "Connection.execute_query_async default_namespace_path must be a list", + )); + } + path.extract::>().map_err(|_| { + PyValueError::new_err( + "Connection.execute_query_async default_namespace_path components must be strings", + ) + }) + } + None => Ok(vec!["public".to_string()]), + } +} + #[pymethods] impl Connection { fn __repr__(&self) -> String { @@ -108,6 +126,40 @@ impl Connection { self.get_inner().map(|inner| inner.uri().to_string()) } + #[pyo3(signature = (query, *, default_namespace_path=None))] + pub fn execute_query_async<'a>( + self_: PyRef<'a, Self>, + query: String, + default_namespace_path: Option>, + ) -> PyResult> { + let inner = self_.get_inner()?.clone(); + let default_namespace_path = parse_default_namespace_path(default_namespace_path)?; + future_into_py(self_.py(), async move { + let operation = inner + .execute_query_async(query) + .default_namespace_path(default_namespace_path); + operation + .execute() + .await + .map(crate::sql::Query::new) + .infer_error() + }) + } + + pub fn describe_query<'a>( + self_: PyRef<'a, Self>, + query_id: uuid::Uuid, + ) -> PyResult> { + let inner = self_.get_inner()?.clone(); + future_into_py(self_.py(), async move { + inner + .describe_query(query_id) + .await + .map(crate::sql::QueryDescription::from) + .infer_error() + }) + } + #[pyo3(signature = ())] pub fn get_read_consistency_interval(self_: PyRef<'_, Self>) -> PyResult> { let inner = self_.get_inner()?.clone(); @@ -699,7 +751,7 @@ impl Connection { } #[pyfunction] -#[pyo3(signature = (uri, api_key=None, region=None, host_override=None, read_consistency_interval=None, client_config=None, storage_options=None, session=None, manifest_enabled=false, namespace_client_properties=None, oauth_config=None))] +#[pyo3(signature = (uri, api_key=None, region=None, host_override=None, sql_host_override=None, read_consistency_interval=None, client_config=None, storage_options=None, session=None, manifest_enabled=false, namespace_client_properties=None, oauth_config=None))] #[allow(clippy::too_many_arguments)] pub fn connect( py: Python<'_>, @@ -707,6 +759,7 @@ pub fn connect( api_key: Option, region: Option, host_override: Option, + sql_host_override: Option, read_consistency_interval: Option, client_config: Option, storage_options: Option>, @@ -726,6 +779,12 @@ pub fn connect( if let Some(host_override) = host_override { builder = builder.host_override(&host_override); } + #[cfg(feature = "remote")] + if let Some(sql_host_override) = sql_host_override { + builder = builder.sql_host_override(&sql_host_override); + } + #[cfg(not(feature = "remote"))] + let _ = sql_host_override; if let Some(read_consistency_interval) = read_consistency_interval { let read_consistency_interval = Duration::from_secs_f64(read_consistency_interval); builder = builder.read_consistency_interval(read_consistency_interval); diff --git a/python/src/lib.rs b/python/src/lib.rs index 8d3eab787..06b31d033 100644 --- a/python/src/lib.rs +++ b/python/src/lib.rs @@ -34,6 +34,7 @@ pub mod permutation; pub mod query; pub mod runtime; pub mod session; +pub mod sql; pub mod table; pub mod util; @@ -50,6 +51,8 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_class::()?; m.add_class::()?; + m.add_class::()?; + m.add_class::()?; m.add_class::()?; m.add_class::()?; m.add_class::()?; diff --git a/python/src/sql.rs b/python/src/sql.rs new file mode 100644 index 000000000..612521207 --- /dev/null +++ b/python/src/sql.rs @@ -0,0 +1,90 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +use std::sync::Arc; + +use chrono::{DateTime, Utc}; +use pyo3::{Bound, PyAny, PyRef, PyResult, pyclass, pymethods}; +use uuid::Uuid; + +use crate::arrow::RecordBatchStream; +use crate::error::PythonErrorExt; +use crate::runtime::future_into_py; + +#[pyclass(name = "SqlQuery")] +pub struct Query { + inner: Arc, +} + +impl Query { + pub(crate) fn new(inner: lancedb::sql::Query) -> Self { + Self { + inner: Arc::new(inner), + } + } +} + +#[pymethods] +impl Query { + #[getter] + pub fn id(&self) -> Uuid { + self.inner.id() + } + + pub fn describe(self_: PyRef<'_, Self>) -> PyResult> { + let inner = self_.inner.clone(); + future_into_py(self_.py(), async move { + inner + .describe() + .await + .map(QueryDescription::from) + .infer_error() + }) + } + + pub fn reader(self_: PyRef<'_, Self>) -> PyResult> { + let inner = self_.inner.clone(); + future_into_py(self_.py(), async move { + let stream = inner.reader().await.infer_error()?; + Ok(RecordBatchStream::new(stream)) + }) + } + + pub fn cancel(self_: PyRef<'_, Self>) -> PyResult> { + let inner = self_.inner.clone(); + future_into_py(self_.py(), async move { + inner.cancel().await.infer_error()?; + Ok(()) + }) + } +} + +#[pyclass(get_all, skip_from_py_object)] +#[derive(Clone)] +pub struct QueryDescription { + id: Uuid, + status: String, + progress: Option, + expires_at: Option>, +} + +#[pymethods] +impl QueryDescription { + fn __repr__(&self) -> String { + format!( + "QueryDescription(id={:?}, status={:?}, progress={:?}, expires_at={:?})", + self.id, self.status, self.progress, self.expires_at + ) + } +} + +impl From for QueryDescription { + fn from(description: lancedb::sql::QueryDescription) -> Self { + Self { + id: description.id, + status: description.status.to_string(), + progress: description.progress, + expires_at: description.expires_at, + } + } +} diff --git a/rust/lancedb/Cargo.toml b/rust/lancedb/Cargo.toml index a5aa12e52..836bffa4f 100644 --- a/rust/lancedb/Cargo.toml +++ b/rust/lancedb/Cargo.toml @@ -21,6 +21,8 @@ arrow-select = { workspace = true } arrow-ord = { workspace = true } arrow-cast = { workspace = true } arrow-ipc.workspace = true +arrow-flight = { workspace = true, optional = true } +prost = { version = "0.14", optional = true } chrono = { workspace = true } datafusion-catalog.workspace = true datafusion-common.workspace = true @@ -77,6 +79,7 @@ reqwest = { version = "0.12.0", default-features = false, features = [ "rustls-tls-native-roots", "stream", ], optional = true } +tonic = { workspace = true, optional = true } http = { version = "1", optional = true } # Matching what is in reqwest urlencoding = { version = "2", optional = true } uuid = { workspace = true, features = ["v5"] } @@ -145,8 +148,11 @@ huggingface = [ ] dynamodb = ["lance/dynamodb", "aws"] remote = [ + "dep:arrow-flight", + "dep:prost", "dep:reqwest", "dep:http", + "dep:tonic", "dep:urlencoding", "lance-namespace-impls/rest", "lance-namespace-impls/rest-adapter", diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs index 35c5c0737..f04f04ce2 100644 --- a/rust/lancedb/src/connection.rs +++ b/rust/lancedb/src/connection.rs @@ -31,7 +31,10 @@ use crate::error::{Error, Result}; #[cfg(feature = "remote")] use crate::remote::{ client::ClientConfig, - db::{OPT_REMOTE_API_KEY, OPT_REMOTE_HOST_OVERRIDE, OPT_REMOTE_REGION}, + db::{ + OPT_REMOTE_API_KEY, OPT_REMOTE_HOST_OVERRIDE, OPT_REMOTE_REGION, + OPT_REMOTE_SQL_HOST_OVERRIDE, + }, }; use lance::io::ObjectStoreParams; pub use lance_file::version::LanceFileVersion; @@ -322,6 +325,43 @@ pub struct CloneTableBuilder { request: CloneTableRequest, } +/// Builder for asynchronously executing a SQL statement on a remote database. +pub struct ExecuteQueryAsyncBuilder { + parent: Arc, + query: String, + default_namespace_path: Vec, +} + +impl ExecuteQueryAsyncBuilder { + fn new(parent: Arc, query: String) -> Self { + Self { + parent, + query, + default_namespace_path: vec!["public".to_string()], + } + } + + /// Set the namespace used for unqualified table names. + /// + /// An empty path is treated as `public`, which is the SQL name for the + /// root Lance namespace. + pub fn default_namespace_path(mut self, path: I) -> Self + where + I: IntoIterator, + S: Into, + { + self.default_namespace_path = path.into_iter().map(Into::into).collect(); + self + } + + /// Start the statement and return its asynchronous query handle. + pub async fn execute(self) -> Result { + self.parent + .execute_query_async(&self.query, &self.default_namespace_path) + .await + } +} + impl CloneTableBuilder { fn new(parent: Arc, target_table_name: String, source_uri: String) -> Self { Self { @@ -405,6 +445,51 @@ impl Connection { &self.internal } + /// Start executing SQL on a remote LanceDB database. + /// + /// The query can reference tables in other databases with SQL dot notation. + /// Use [`ExecuteQueryAsyncBuilder::default_namespace_path`] to avoid qualifying + /// tables in the default namespace. Local connections return + /// [`Error::NotSupported`]. + /// + /// # Example + /// + /// ```no_run + /// # async fn query(db: &lancedb::Connection) -> lancedb::Result<()> { + /// use futures::TryStreamExt; + /// + /// let query = db + /// .execute_query_async("SELECT * FROM events LIMIT 10") + /// .default_namespace_path(["public"]) + /// .execute() + /// .await?; + /// println!("query id: {}", query.id()); + /// let mut batches = query.reader().await?; + /// while let Some(batch) = batches.try_next().await? { + /// println!("received {} rows", batch.num_rows()); + /// } + /// # Ok(()) + /// # } + /// ``` + pub fn execute_query_async(&self, query: impl Into) -> ExecuteQueryAsyncBuilder { + ExecuteQueryAsyncBuilder::new(self.internal.clone(), query.into()) + } + + /// Describe a submitted SQL query by its connection-scoped id. + /// + /// This performs one bounded status poll using state retained by this + /// connection. Running state with a live query handle is not evicted; + /// abandoned state has bounded retention, and server expiration is + /// honored. Terminal state is retained briefly. + /// Query ids are not portable to another connection. Local connections + /// return [`Error::NotSupported`]. + pub async fn describe_query( + &self, + query_id: uuid::Uuid, + ) -> Result { + self.internal.describe_query(query_id).await + } + /// Get the names of all tables in the database /// /// The names will be returned in lexicographical order (ascending) @@ -864,6 +949,19 @@ impl ConnectBuilder { self } + /// Set the SQL service host override for a remote connection. + /// + /// The SQL client is initialized lazily when the connection first executes + /// SQL and is retained for the connection's lifetime. + #[cfg(feature = "remote")] + pub fn sql_host_override(mut self, sql_host_override: &str) -> Self { + self.request.options.insert( + OPT_REMOTE_SQL_HOST_OVERRIDE.to_string(), + sql_host_override.to_string(), + ); + self + } + /// Set the database specific options /// /// See [crate::database::listing::ListingDatabaseOptions] for the options available for @@ -1053,6 +1151,7 @@ impl ConnectBuilder { let mut merged_options = self.request.options.clone(); Self::apply_env_defaults(&ENV_VARS_TO_STORAGE_OPTS, &mut merged_options); + let sql_host_override = merged_options.get(OPT_REMOTE_SQL_HOST_OVERRIDE).cloned(); let options = RemoteDatabaseOptions::parse_from_map(&merged_options)?; let region = options.region.ok_or_else(|| Error::InvalidInput { @@ -1094,11 +1193,15 @@ impl ConnectBuilder { } let storage_options = StorageOptions(options.storage_options.clone()); + let host_overrides = crate::remote::db::RemoteHostOverrides { + rest: options.host_override, + sql: sql_host_override, + }; let internal = Arc::new(crate::remote::db::RemoteDatabase::try_new( &self.request.uri, &api_key, ®ion, - options.host_override, + host_overrides, client_config, storage_options.into(), self.request.read_consistency_interval, @@ -1392,6 +1495,23 @@ mod tests { assert_eq!(tc.connection.uri(), tc.uri); } + #[tokio::test] + async fn test_local_connection_rejects_sql_queries() { + let directory = tempdir().unwrap(); + let connection = connect(directory.path().to_str().unwrap()) + .execute() + .await + .unwrap(); + assert!(matches!( + connection.execute_query_async("SELECT 1").execute().await, + Err(Error::NotSupported { .. }) + )); + assert!(matches!( + connection.describe_query(uuid::Uuid::nil()).await, + Err(Error::NotSupported { .. }) + )); + } + #[cfg(feature = "remote")] #[test] fn test_apply_env_defaults() { diff --git a/rust/lancedb/src/database.rs b/rust/lancedb/src/database.rs index 61424bb05..532ea3658 100644 --- a/rust/lancedb/src/database.rs +++ b/rust/lancedb/src/database.rs @@ -340,6 +340,22 @@ pub trait Database: async fn job_history(&self, _job_id: Option<&str>) -> Result> { job_op_not_supported("job_history") } + /// Start executing a SQL statement on a remote database. + async fn execute_query_async( + &self, + _query: &str, + _default_namespace_path: &[String], + ) -> Result { + Err(crate::error::Error::NotSupported { + message: "SQL is not supported by this database".to_string(), + }) + } + /// Describe a submitted SQL query by its connection-scoped id. + async fn describe_query(&self, _query_id: uuid::Uuid) -> Result { + Err(crate::error::Error::NotSupported { + message: "SQL is not supported by this database".to_string(), + }) + } /// Open a table in the database async fn open_table(&self, request: OpenTableRequest) -> Result>; /// Rename a table in the database diff --git a/rust/lancedb/src/lib.rs b/rust/lancedb/src/lib.rs index 9c3c199ff..44c5dd616 100644 --- a/rust/lancedb/src/lib.rs +++ b/rust/lancedb/src/lib.rs @@ -195,6 +195,7 @@ pub mod query; #[cfg(feature = "remote")] pub mod remote; pub mod rerankers; +pub mod sql; pub mod table; #[cfg(test)] pub mod test_utils; diff --git a/rust/lancedb/src/remote.rs b/rust/lancedb/src/remote.rs index be9d0eef6..6c37ec6a0 100644 --- a/rust/lancedb/src/remote.rs +++ b/rust/lancedb/src/remote.rs @@ -11,6 +11,7 @@ pub(crate) mod db; pub(crate) mod job; pub mod oauth; mod retry; +pub(crate) mod sql; pub(crate) mod table; pub(crate) mod util; diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index ecdadf464..87486a522 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -33,6 +33,7 @@ use crate::table::BaseTable; use super::client::{ ClientConfig, HeaderProvider, HttpSend, RequestResultExt, RestfulLanceDbClient, Sender, }; +use super::sql::SqlClient; use super::table::RemoteTable; use super::util::parse_server_version; use super::{ARROW_STREAM_CONTENT_TYPE, extract_job_id}; @@ -97,6 +98,7 @@ pub const OPT_REMOTE_PREFIX: &str = "remote_database_"; pub const OPT_REMOTE_API_KEY: &str = "remote_database_api_key"; pub const OPT_REMOTE_REGION: &str = "remote_database_region"; pub const OPT_REMOTE_HOST_OVERRIDE: &str = "remote_database_host_override"; +pub const OPT_REMOTE_SQL_HOST_OVERRIDE: &str = "remote_database_sql_host_override"; // TODO: add support for configuring client config via key/value options #[derive(Clone, Debug, Default)] @@ -212,6 +214,7 @@ pub struct RemoteDatabase { namespace_context_provider: Option>, /// TLS configuration for mTLS support tls_config: Option, + sql_client: Option, } #[derive(Clone)] @@ -269,22 +272,35 @@ impl DynamicContextProvider for NamespaceHeaderProviderContext { } } +pub struct RemoteHostOverrides { + pub rest: Option, + pub sql: Option, +} + impl RemoteDatabase { - pub fn try_new( + pub(crate) fn try_new( uri: &str, api_key: &str, region: &str, - host_override: Option, + host_overrides: RemoteHostOverrides, client_config: ClientConfig, options: RemoteOptions, read_consistency_interval: Option, ) -> Result { let parsed = super::client::parse_db_url(uri)?; + let sql_client = SqlClient::new( + parsed.db_name.clone(), + parsed.db_prefix.clone(), + api_key.to_string(), + host_overrides.rest.clone(), + host_overrides.sql, + client_config.clone(), + ); let header_map = RestfulLanceDbClient::::default_headers( api_key, region, &parsed.db_name, - host_override.is_some(), + host_overrides.rest.is_some(), &options, parsed.db_prefix.as_deref(), &client_config, @@ -312,7 +328,7 @@ impl RemoteDatabase { let client = RestfulLanceDbClient::try_new( &parsed, region, - host_override, + host_overrides.rest, header_map, client_config.clone(), read_consistency_interval, @@ -330,6 +346,7 @@ impl RemoteDatabase { namespace_headers, namespace_context_provider, tls_config: client_config.tls_config, + sql_client: Some(sql_client), }) } } @@ -427,6 +444,7 @@ mod test_utils { namespace_headers: HashMap::new(), namespace_context_provider: None, tls_config: None, + sql_client: None, } } @@ -449,6 +467,7 @@ mod test_utils { namespace_headers: config.extra_headers.clone(), namespace_context_provider, tls_config: config.tls_config.clone(), + sql_client: None, } } } @@ -749,6 +768,30 @@ impl Database for RemoteDatabase { .map_err(Into::into) } + async fn execute_query_async( + &self, + query: &str, + default_namespace_path: &[String], + ) -> Result { + let client = self + .sql_client + .as_ref() + .ok_or_else(|| Error::NotSupported { + message: "SQL is unavailable for this remote database client".to_string(), + })?; + client.submit(query, default_namespace_path).await + } + + async fn describe_query(&self, query_id: uuid::Uuid) -> Result { + let client = self + .sql_client + .as_ref() + .ok_or_else(|| Error::NotSupported { + message: "SQL is unavailable for this remote database client".to_string(), + })?; + client.describe(query_id).await + } + async fn table_names(&self, request: TableNamesRequest) -> Result> { let (tables, version) = if request.namespace_path.is_empty() { // The flat route resumes after a table name and orders by name, which is exactly diff --git a/rust/lancedb/src/remote/oauth.rs b/rust/lancedb/src/remote/oauth.rs index fd61db919..3ebe8f86e 100644 --- a/rust/lancedb/src/remote/oauth.rs +++ b/rust/lancedb/src/remote/oauth.rs @@ -466,7 +466,9 @@ impl TokenSource for AzureImdsSource { /// OAuth header provider that manages the full token lifecycle. /// /// Implements [`HeaderProvider`] to inject `Authorization: Bearer ` -/// headers into every LanceDB request, with automatic token refresh. +/// headers into every LanceDB request, with automatic token refresh. It also +/// identifies the bearer credential as OIDC so LanceDB's SQL service selects +/// OIDC validation instead of API-key validation. pub struct OAuthHeaderProvider { token_source: Box, token_state: Arc>, @@ -554,10 +556,10 @@ impl OAuthHeaderProvider { impl HeaderProvider for OAuthHeaderProvider { async fn get_headers(&self) -> Result> { let token = self.get_valid_token().await?; - Ok(HashMap::from([( - "authorization".to_string(), - format!("Bearer {token}"), - )])) + Ok(HashMap::from([ + ("authorization".to_string(), format!("Bearer {token}")), + ("x-lancedb-credential-type".to_string(), "oidc".to_string()), + ])) } } diff --git a/rust/lancedb/src/remote/sql.rs b/rust/lancedb/src/remote/sql.rs new file mode 100644 index 000000000..489158922 --- /dev/null +++ b/rust/lancedb/src/remote/sql.rs @@ -0,0 +1,1471 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +use std::collections::HashMap; +use std::fs; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex as StdMutex, OnceLock}; +use std::time::{Duration, Instant}; + +use arrow_array::RecordBatch; +use arrow_flight::decode::FlightRecordBatchStream; +use arrow_flight::error::FlightError; +use arrow_flight::flight_service_client::FlightServiceClient; +use arrow_flight::sql::{CommandStatementQuery, ProstMessageExt}; +use arrow_flight::{ + Action, CancelFlightInfoRequest, CancelFlightInfoResult, CancelStatus, FlightClient, + FlightDescriptor, FlightEndpoint, FlightInfo, PollInfo, +}; +use arrow_schema::{Schema, SchemaRef}; +use futures::TryStreamExt; +use http::header::{HeaderMap, HeaderName, HeaderValue}; +use prost::Message; +use tokio::sync::{Mutex, Notify, OnceCell, mpsc}; +use tonic::transport::{Certificate, Channel, ClientTlsConfig, Endpoint, Identity}; +use uuid::Uuid; + +use crate::arrow::{SendableRecordBatchStream, SimpleRecordBatchStream}; +use crate::error::{Error, Result}; +use crate::remote::client::{ClientConfig, TlsConfig}; +use crate::remote::retry::ResolvedRetryConfig; +use crate::sql::{Query, QueryDescription, QueryHandle, QueryStatus}; + +const DEFAULT_SQL_PORT: u16 = 10025; +const DEFAULT_SQL_TLS_PORT: u16 = 10026; +const DEFAULT_CONNECT_TIMEOUT: Duration = Duration::from_secs(120); +const DEFAULT_READ_TIMEOUT: Duration = Duration::from_secs(300); +const STATUS_POLL_TIMEOUT: Duration = Duration::from_secs(1); +const MIN_POLL_INTERVAL: Duration = Duration::from_millis(50); +const MAX_SQL_MESSAGE_SIZE: usize = 1024 * 1024 * 1024; +const TERMINAL_QUERY_RETENTION: Duration = Duration::from_secs(300); +const ABANDONED_QUERY_RETENTION: Duration = Duration::from_secs(24 * 60 * 60); + +#[derive(Clone)] +pub(super) struct SqlClient { + inner: Arc, + queries: Arc, +} + +struct SqlClientInner { + database: String, + database_prefix: Option, + api_key: String, + host_override: Option, + sql_host_override: Option, + client_config: ClientConfig, + client: Arc>, +} + +struct SqlConnection { + // FlightClient does not expose its transport. Cancellation retains the channel so it can + // install a per-call interceptor that records whether a request was dispatched. + channel: Channel, + client: FlightServiceClient, +} + +struct ResultEndpointStream { + stream: FlightRecordBatchStream, + request_id: String, + read_timeout: Duration, +} + +struct PreparedSqlResult { + schema: SchemaRef, + next_endpoint: usize, + endpoint_stream: Option, + buffered_batch: Option, +} + +impl ResultEndpointStream { + async fn next_batch(&mut self) -> Result> { + tokio::time::timeout(self.read_timeout, self.stream.try_next()) + .await + .map_err(|_| sql_error(&self.request_id, "SQL result read timed out"))? + .map_err(|err| sql_error(&self.request_id, err)) + } +} + +enum CancelOutcome { + Status(CancelStatus), + NotFound(String), +} + +struct CancelAttempt { + dispatched: Arc, + unresolved: Arc, + resolved: bool, +} + +struct ResultStartGuard<'a> { + started: &'a AtomicBool, + committed: bool, +} + +impl<'a> ResultStartGuard<'a> { + fn new(started: &'a AtomicBool) -> Self { + Self { + started, + committed: false, + } + } + + fn commit(mut self) { + self.committed = true; + } +} + +impl Drop for ResultStartGuard<'_> { + fn drop(&mut self) { + if !self.committed { + self.started.store(false, Ordering::SeqCst); + } + } +} + +impl CancelAttempt { + fn new(dispatched: Arc, unresolved: Arc) -> Self { + Self { + dispatched, + unresolved, + resolved: false, + } + } + + fn resolve(&mut self) { + self.resolved = true; + } +} + +impl Drop for CancelAttempt { + fn drop(&mut self) { + if !self.resolved && self.dispatched.load(Ordering::SeqCst) { + self.unresolved.store(true, Ordering::SeqCst); + } + } +} + +impl std::fmt::Debug for SqlClient { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("SqlClient") + .field("database", &self.inner.database) + .field("database_prefix", &self.inner.database_prefix) + .field("api_key", &"") + .field("host_override", &self.inner.host_override) + .field("sql_host_override", &self.inner.sql_host_override) + .field("client_config", &"") + .field("initialized", &self.inner.client.get().is_some()) + .finish() + } +} + +impl SqlClient { + pub(super) fn new( + database: String, + database_prefix: Option, + api_key: String, + host_override: Option, + sql_host_override: Option, + client_config: ClientConfig, + ) -> Self { + Self { + inner: Arc::new(SqlClientInner { + database, + database_prefix, + api_key, + host_override, + sql_host_override, + client_config, + client: Arc::new(OnceCell::new()), + }), + queries: Arc::new(QueryRegistry::new()), + } + } + + pub(super) async fn submit( + &self, + query: &str, + default_namespace_path: &[String], + ) -> Result { + let timeout = self.inner.overall_timeout()?; + with_overall_timeout(timeout, "SQL query submission", async { + validate_namespace_path(default_namespace_path)?; + let command = CommandStatementQuery { + query: query.to_string(), + transaction_id: None, + }; + let descriptor = FlightDescriptor::new_cmd(command.as_any().encode_to_vec()); + let poll_info = self.inner.poll(descriptor, default_namespace_path).await?; + let query_id = Uuid::now_v7(); + let query = Arc::new(RemoteQuery::new( + query_id, + self.inner.clone(), + default_namespace_path.to_vec(), + poll_info, + )?); + self.queries.insert(query_id, query.clone()); + Ok(Query::new(Arc::new(RemoteQueryHandle::new(query)))) + }) + .await + } + + pub(super) async fn describe(&self, query_id: Uuid) -> Result { + let query = self + .queries + .get(query_id) + .ok_or_else(|| Error::InvalidInput { + message: "Unknown or expired SQL query id for this connection".to_string(), + })?; + query.describe().await + } + + #[cfg(test)] + async fn initialized_client_count(&self) -> usize { + usize::from(self.inner.client.get().is_some()) + } +} + +impl SqlClientInner { + fn overall_timeout(&self) -> Result> { + resolve_timeout( + self.client_config.timeout_config.timeout, + "LANCE_CLIENT_TIMEOUT", + None, + ) + } + + async fn poll( + &self, + descriptor: FlightDescriptor, + default_namespace_path: &[String], + ) -> Result { + let request_id = uuid::Uuid::new_v4().to_string(); + let read_timeout = resolve_timeout( + self.client_config.timeout_config.read_timeout, + "LANCE_CLIENT_READ_TIMEOUT", + Some(DEFAULT_READ_TIMEOUT), + )? + .unwrap(); + let mut client = self + .client_with_headers(default_namespace_path, &request_id) + .await?; + tokio::time::timeout(read_timeout, client.poll_flight_info(descriptor)) + .await + .map_err(|_| sql_error(&request_id, "SQL query poll timed out"))? + .map_err(|err| sql_error(&request_id, err)) + } + + async fn poll_status( + &self, + descriptor: FlightDescriptor, + default_namespace_path: &[String], + ) -> Result> { + let request_id = uuid::Uuid::new_v4().to_string(); + let mut client = self + .client_with_headers(default_namespace_path, &request_id) + .await + .map_err(|err| sql_error(&request_id, err))?; + match tokio::time::timeout(STATUS_POLL_TIMEOUT, client.poll_flight_info(descriptor)).await { + Ok(result) => result.map(Some).map_err(|err| sql_error(&request_id, err)), + Err(_) => Ok(None), + } + } + + async fn poll_continuation( + &self, + descriptor: FlightDescriptor, + default_namespace_path: &[String], + ) -> Result { + let read_timeout = resolve_timeout( + self.client_config.timeout_config.read_timeout, + "LANCE_CLIENT_READ_TIMEOUT", + Some(DEFAULT_READ_TIMEOUT), + )? + .unwrap(); + let retry_config = ResolvedRetryConfig::try_from(self.client_config.retry_config.clone())?; + let mut retry_count = 0_u8; + loop { + let started = Instant::now(); + let request_id = uuid::Uuid::new_v4().to_string(); + let mut client = self + .client_with_headers(default_namespace_path, &request_id) + .await?; + let result = + tokio::time::timeout(read_timeout, client.poll_flight_info(descriptor.clone())) + .await; + let poll_info = match result { + Err(_) if retry_count < retry_config.read_retries => { + retry_count += 1; + tokio::time::sleep(poll_retry_delay(&retry_config, retry_count)).await; + continue; + } + Err(_) => return Err(sql_error(&request_id, "SQL query poll timed out")), + Ok(Err(FlightError::Tonic(status))) + if matches!( + status.code(), + tonic::Code::DeadlineExceeded | tonic::Code::Unavailable + ) && retry_count < retry_config.read_retries => + { + retry_count += 1; + tokio::time::sleep(poll_retry_delay(&retry_config, retry_count)).await; + continue; + } + Ok(Err(error)) => return Err(sql_error(&request_id, error)), + Ok(Ok(poll_info)) => poll_info, + }; + if let Some(delay) = MIN_POLL_INTERVAL.checked_sub(started.elapsed()) { + tokio::time::sleep(delay).await; + } + return Ok(poll_info); + } + } + + async fn open_result_endpoint( + &self, + endpoint: FlightEndpoint, + default_namespace_path: &[String], + ) -> Result { + let request_id = uuid::Uuid::new_v4().to_string(); + let read_timeout = resolve_timeout( + self.client_config.timeout_config.read_timeout, + "LANCE_CLIENT_READ_TIMEOUT", + Some(DEFAULT_READ_TIMEOUT), + )? + .unwrap(); + let ticket = endpoint.ticket.ok_or_else(|| { + sql_error(&request_id, "SQL result endpoint did not include a ticket") + })?; + let mut endpoint_client = self + .client_with_headers(default_namespace_path, &request_id) + .await?; + let stream = tokio::time::timeout(read_timeout, endpoint_client.do_get(ticket)) + .await + .map_err(|_| sql_error(&request_id, "SQL result fetch timed out"))? + .map_err(|err| sql_error(&request_id, err))?; + Ok(ResultEndpointStream { + stream, + request_id, + read_timeout, + }) + } + + async fn cancel( + &self, + info: FlightInfo, + default_namespace_path: &[String], + unresolved_attempt: Arc, + ) -> Result { + let request_id = uuid::Uuid::new_v4().to_string(); + let read_timeout = resolve_timeout( + self.client_config.timeout_config.read_timeout, + "LANCE_CLIENT_READ_TIMEOUT", + Some(DEFAULT_READ_TIMEOUT), + )? + .unwrap(); + let connection = self.connection(&request_id).await?; + let headers = self.headers(default_namespace_path, &request_id).await?; + let metadata = client_with_headers(connection.client.clone(), &headers)? + .metadata() + .clone(); + let dispatched = Arc::new(AtomicBool::new(false)); + let mut attempt = CancelAttempt::new(dispatched.clone(), unresolved_attempt); + let mut client = FlightServiceClient::with_interceptor( + connection.channel.clone(), + move |request: tonic::Request<()>| { + dispatched.store(true, Ordering::SeqCst); + Ok(request) + }, + ) + .max_decoding_message_size(MAX_SQL_MESSAGE_SIZE); + let action = Action::new( + "CancelFlightInfo", + CancelFlightInfoRequest::new(info).encode_to_vec(), + ); + let mut request = tonic::Request::new(action); + *request.metadata_mut() = metadata; + let result = tokio::time::timeout(read_timeout, async { + let response = client + .do_action(request) + .await + .map_err(|status| FlightError::Tonic(Box::new(status)))?; + let response = response + .into_inner() + .message() + .await + .map_err(|status| FlightError::Tonic(Box::new(status)))? + .ok_or_else(|| { + FlightError::protocol("Received no response for cancel_flight_info call") + })?; + CancelFlightInfoResult::decode(response.body) + .map_err(|err| FlightError::DecodeError(err.to_string())) + }) + .await + .map_err(|_| sql_error(&request_id, "SQL query cancellation timed out"))?; + let result = match result { + Ok(result) => result, + Err(FlightError::Tonic(status)) if status.code() == tonic::Code::NotFound => { + attempt.resolve(); + return Ok(CancelOutcome::NotFound(request_id)); + } + Err(FlightError::Tonic(status)) if !cancellation_status_is_ambiguous(status.code()) => { + attempt.resolve(); + return Err(sql_error(&request_id, status)); + } + Err(error) => return Err(sql_error(&request_id, error)), + }; + let status = CancelStatus::try_from(result.status) + .map_err(|_| sql_error(&request_id, "SQL query returned an invalid cancel status"))?; + if status != CancelStatus::Unspecified { + attempt.resolve(); + } + Ok(CancelOutcome::Status(status)) + } + + async fn client_with_headers( + &self, + default_namespace_path: &[String], + request_id: &str, + ) -> Result { + let connection = self.connection(request_id).await?; + let headers = self.headers(default_namespace_path, request_id).await?; + client_with_headers(connection.client.clone(), &headers) + } + + async fn connection(&self, request_id: &str) -> Result<&SqlConnection> { + self.client + .get_or_try_init(|| async { + let target = resolve_sql_host_override( + self.host_override.as_deref(), + self.sql_host_override.as_deref(), + )?; + let channel = connect_channel(&target, &self.client_config, request_id).await?; + let client = FlightServiceClient::new(channel.clone()) + .max_decoding_message_size(MAX_SQL_MESSAGE_SIZE); + Ok::<_, Error>(SqlConnection { channel, client }) + }) + .await + } + + async fn headers( + &self, + default_namespace_path: &[String], + request_id: &str, + ) -> Result { + let mut headers = HeaderMap::new(); + merge_headers(&mut headers, &self.client_config.extra_headers)?; + if let Some(provider) = &self.client_config.header_provider { + merge_headers(&mut headers, &provider.get_headers().await?)?; + } + + let has_authorization = headers.contains_key("authorization"); + let has_api_key = headers.contains_key("x-api-key"); + if has_authorization && has_api_key { + return Err(Error::InvalidInput { + message: "SQL accepts either authorization or x-api-key, not both".to_string(), + }); + } + if !has_authorization && !has_api_key { + if self.api_key.is_empty() { + return Err(Error::InvalidInput { + message: "SQL authentication credentials are required".to_string(), + }); + } + insert_header(&mut headers, "x-api-key", &self.api_key)?; + } + + insert_header(&mut headers, "database", &self.database)?; + if let Some(database_prefix) = &self.database_prefix { + insert_header(&mut headers, "x-lancedb-database-prefix", database_prefix)?; + } + let namespace_path = if default_namespace_path.is_empty() { + "public".to_string() + } else { + default_namespace_path.join("$") + }; + insert_header(&mut headers, "namespace-path", &namespace_path)?; + insert_header(&mut headers, "x-request-id", request_id)?; + if let Some(user_id) = self.client_config.resolve_user_id() { + insert_header(&mut headers, "x-lancedb-user-id", &user_id)?; + } + Ok(headers) + } +} + +struct QueryRegistry { + queries: StdMutex>>, +} + +impl QueryRegistry { + fn new() -> Self { + Self { + queries: StdMutex::new(HashMap::new()), + } + } + + fn insert(&self, id: Uuid, query: Arc) { + self.remove_expired(); + self.queries.lock().unwrap().insert(id, query); + } + + fn get(&self, id: Uuid) -> Option> { + self.remove_expired(); + let query = self.queries.lock().unwrap().get(&id).cloned(); + if let Some(query) = &query { + query.touch(); + } + query + } + + fn remove_expired(&self) { + self.queries + .lock() + .unwrap() + .retain(|_, query| !query.registry_expired(Arc::strong_count(query) == 1)); + } +} + +struct RemoteQuery { + id: Uuid, + client: Arc, + default_namespace_path: Vec, + state: Mutex, + poll_gate: Mutex<()>, + cancel_gate: Mutex<()>, + state_changed: Notify, + cancelled: Notify, + expires_at: StdMutex>>, + terminal_at: OnceLock, + last_accessed: StdMutex, + lifecycle: StdMutex, + cancel_request_uncertain: Arc, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum QueryLifecycle { + Running, + Ready, + Cancelling, + Completed, + Cancelled, +} + +impl RemoteQuery { + fn new( + id: Uuid, + client: Arc, + default_namespace_path: Vec, + poll_info: PollInfo, + ) -> Result { + let expires_at = query_expiration(&poll_info)?; + let terminal_at = OnceLock::new(); + let lifecycle = if poll_info.flight_descriptor.is_none() { + let _ = terminal_at.set(Instant::now()); + QueryLifecycle::Ready + } else { + QueryLifecycle::Running + }; + Ok(Self { + id, + client, + default_namespace_path, + state: Mutex::new(poll_info), + poll_gate: Mutex::new(()), + cancel_gate: Mutex::new(()), + state_changed: Notify::new(), + cancelled: Notify::new(), + expires_at: StdMutex::new(expires_at), + terminal_at, + last_accessed: StdMutex::new(Instant::now()), + lifecycle: StdMutex::new(lifecycle), + cancel_request_uncertain: Arc::new(AtomicBool::new(false)), + }) + } + + fn registry_expired(&self, abandoned: bool) -> bool { + if let Some(finished) = self.terminal_at.get() { + return finished.elapsed() >= TERMINAL_QUERY_RETENTION; + } + self.expires_at + .lock() + .unwrap() + .is_some_and(|expires_at| expires_at <= chrono::Utc::now()) + || (abandoned + && self.last_accessed.lock().unwrap().elapsed() >= ABANDONED_QUERY_RETENTION) + } + + fn mark_terminal(&self) { + let _ = self.terminal_at.set(Instant::now()); + } + + fn mark_ready(&self) { + let mut lifecycle = self.lifecycle.lock().unwrap(); + if *lifecycle == QueryLifecycle::Running { + *lifecycle = QueryLifecycle::Ready; + } + drop(lifecycle); + self.mark_terminal(); + } + + fn mark_cancelled(&self) -> bool { + self.cancel_request_uncertain.store(false, Ordering::SeqCst); + let mut lifecycle = self.lifecycle.lock().unwrap(); + if matches!( + *lifecycle, + QueryLifecycle::Cancelled | QueryLifecycle::Completed + ) { + return false; + } + *lifecycle = QueryLifecycle::Cancelled; + drop(lifecycle); + self.mark_terminal(); + self.cancelled.notify_waiters(); + self.state_changed.notify_waiters(); + true + } + + fn mark_cancelling(&self) { + self.cancel_request_uncertain.store(false, Ordering::SeqCst); + let mut lifecycle = self.lifecycle.lock().unwrap(); + if matches!(*lifecycle, QueryLifecycle::Running | QueryLifecycle::Ready) { + *lifecycle = QueryLifecycle::Cancelling; + drop(lifecycle); + self.cancelled.notify_waiters(); + self.state_changed.notify_waiters(); + } + } + + async fn restore_after_rejected_cancellation(&self) { + self.cancel_request_uncertain.store(false, Ordering::SeqCst); + let running = self.state.lock().await.flight_descriptor.is_some(); + let mut lifecycle = self.lifecycle.lock().unwrap(); + if *lifecycle == QueryLifecycle::Cancelling { + *lifecycle = if running { + QueryLifecycle::Running + } else { + QueryLifecycle::Ready + }; + drop(lifecycle); + self.state_changed.notify_waiters(); + } + } + + fn mark_result_completed(&self) -> Result<()> { + let mut lifecycle = self.lifecycle.lock().unwrap(); + if matches!( + *lifecycle, + QueryLifecycle::Cancelling | QueryLifecycle::Cancelled + ) { + return Err(self.cancelled_error()); + } + *lifecycle = QueryLifecycle::Completed; + Ok(()) + } + + fn lifecycle(&self) -> QueryLifecycle { + *self.lifecycle.lock().unwrap() + } + + fn is_cancellation_requested(&self) -> bool { + matches!( + self.lifecycle(), + QueryLifecycle::Cancelling | QueryLifecycle::Cancelled + ) + } + + fn cancelled_error(&self) -> Error { + Error::JobCancelled { + job_id: Some(self.id.to_string()), + } + } + + async fn wait_for_cancellation(&self) { + loop { + let cancelled = self.cancelled.notified(); + if self.is_cancellation_requested() { + return; + } + cancelled.await; + } + } + + fn touch(&self) { + *self.last_accessed.lock().unwrap() = Instant::now(); + } + + async fn poll_next_state(&self, descriptor: FlightDescriptor) -> Result { + self.touch(); + if self.is_cancellation_requested() { + return Err(self.cancelled_error()); + } + let _poll_guard = tokio::select! { + biased; + _ = self.wait_for_cancellation() => return Err(self.cancelled_error()), + poll_guard = self.poll_gate.lock() => poll_guard, + }; + let latest = self.state.lock().await.clone(); + if latest.flight_descriptor.as_ref() != Some(&descriptor) { + return Ok(latest); + } + let updated = tokio::select! { + biased; + _ = self.wait_for_cancellation() => return Err(self.cancelled_error()), + result = self.client.poll_continuation( + descriptor.clone(), + &self.default_namespace_path, + ) => match result { + Err(_) if self.is_cancellation_requested() => { + return Err(self.cancelled_error()); + } + result => result?, + }, + }; + self.update_state(&descriptor, updated).await + } + + async fn prepare_result(self: &Arc) -> Result { + loop { + if self.is_cancellation_requested() { + return Err(self.cancelled_error()); + } + let state = self.state.lock().await.clone(); + if let Some(info) = state.info { + if let Some(endpoint) = info.endpoint.first().cloned() { + let mut endpoint_stream = tokio::select! { + biased; + _ = self.wait_for_cancellation() => return Err(self.cancelled_error()), + result = self.client.open_result_endpoint( + endpoint, + &self.default_namespace_path, + ) => result?, + }; + let buffered_batch = tokio::select! { + biased; + _ = self.wait_for_cancellation() => return Err(self.cancelled_error()), + result = endpoint_stream.next_batch() => result?, + }; + let schema = buffered_batch + .as_ref() + .map(RecordBatch::schema) + .or_else(|| endpoint_stream.stream.schema().cloned()) + .ok_or_else(|| Error::Runtime { + message: "SQL result endpoint did not include a schema".to_string(), + })?; + return Ok(PreparedSqlResult { + schema, + next_endpoint: 1, + endpoint_stream: buffered_batch.is_some().then_some(endpoint_stream), + buffered_batch, + }); + } + if state.flight_descriptor.is_none() { + let schema = if info.schema.is_empty() { + Arc::new(Schema::empty()) + } else { + let request_id = uuid::Uuid::new_v4().to_string(); + Arc::new( + info.try_decode_schema() + .map_err(|err| sql_error(&request_id, err))?, + ) + }; + return Ok(PreparedSqlResult { + schema, + next_endpoint: 0, + endpoint_stream: None, + buffered_batch: None, + }); + } + } else if state.flight_descriptor.is_none() { + return Err(Error::Runtime { + message: "Completed SQL query did not include result information".to_string(), + }); + } + let descriptor = state.flight_descriptor.ok_or_else(|| Error::Runtime { + message: "Completed SQL query did not include result information".to_string(), + })?; + self.poll_next_state(descriptor).await?; + } + } + + async fn run_result_stream( + self: Arc, + mut prepared: PreparedSqlResult, + sender: mpsc::Sender>, + ) -> Result<()> { + if let Some(batch) = prepared.buffered_batch.take() + && !self + .send_result_batch(&sender, &prepared.schema, batch) + .await? + { + return Ok(()); + } + loop { + if self.is_cancellation_requested() { + return Err(self.cancelled_error()); + } + if let Some(endpoint_stream) = prepared.endpoint_stream.as_mut() { + let batch = tokio::select! { + biased; + _ = sender.closed() => return Ok(()), + _ = self.wait_for_cancellation() => return Err(self.cancelled_error()), + result = endpoint_stream.next_batch() => result?, + }; + if let Some(batch) = batch { + if !self + .send_result_batch(&sender, &prepared.schema, batch) + .await? + { + return Ok(()); + } + } else { + prepared.endpoint_stream = None; + } + continue; + } + + let state = self.state.lock().await.clone(); + let endpoints = state + .info + .as_ref() + .map(|info| info.endpoint.as_slice()) + .unwrap_or_default(); + if prepared.next_endpoint > endpoints.len() { + return Err(Error::Runtime { + message: "SQL service removed a previously advertised result endpoint" + .to_string(), + }); + } + if let Some(endpoint) = endpoints.get(prepared.next_endpoint).cloned() { + prepared.next_endpoint += 1; + prepared.endpoint_stream = Some(tokio::select! { + biased; + _ = sender.closed() => return Ok(()), + _ = self.wait_for_cancellation() => return Err(self.cancelled_error()), + result = self.client.open_result_endpoint( + endpoint, + &self.default_namespace_path, + ) => result?, + }); + continue; + } + if let Some(descriptor) = state.flight_descriptor { + tokio::select! { + biased; + _ = sender.closed() => return Ok(()), + _ = self.wait_for_cancellation() => return Err(self.cancelled_error()), + result = self.poll_next_state(descriptor) => result?, + }; + continue; + } + self.mark_result_completed()?; + return Ok(()); + } + } + + async fn send_result_batch( + &self, + sender: &mpsc::Sender>, + schema: &SchemaRef, + batch: RecordBatch, + ) -> Result { + if batch.schema().as_ref() != schema.as_ref() { + return Err(Error::Runtime { + message: "SQL result endpoint returned a different schema".to_string(), + }); + } + tokio::select! { + biased; + _ = self.wait_for_cancellation() => Err(self.cancelled_error()), + result = sender.send(Ok(batch)) => Ok(result.is_ok()), + } + } + + async fn update_state( + &self, + descriptor: &FlightDescriptor, + updated: PollInfo, + ) -> Result { + self.touch(); + let expires_at = query_expiration(&updated)?; + let mut state = self.state.lock().await; + if state.flight_descriptor.as_ref() == Some(descriptor) { + if updated.flight_descriptor.is_none() { + self.mark_ready(); + } + *self.expires_at.lock().unwrap() = expires_at; + *state = updated; + self.state_changed.notify_waiters(); + } + Ok(state.clone()) + } +} + +impl RemoteQuery { + async fn describe(&self) -> Result { + let timeout = self.client.overall_timeout()?; + with_overall_timeout(timeout, "SQL query description", self.describe_inner()).await + } + + async fn describe_inner(&self) -> Result { + self.touch(); + if self.is_cancellation_requested() { + let state = self.state.lock().await.clone(); + return query_description(self.id, &state, self.lifecycle()); + } + let state = self.state.lock().await.clone(); + let state = if let Some(descriptor) = state.flight_descriptor.clone() { + let poll_guard = tokio::select! { + biased; + _ = self.wait_for_cancellation() => return query_description( + self.id, + &state, + self.lifecycle(), + ), + poll_guard = tokio::time::timeout( + STATUS_POLL_TIMEOUT, + self.poll_gate.lock(), + ) => poll_guard, + }; + let Ok(_poll_guard) = poll_guard else { + return query_description(self.id, &state, self.lifecycle()); + }; + let latest = self.state.lock().await.clone(); + if latest.flight_descriptor.as_ref() != Some(&descriptor) { + latest + } else { + let updated = tokio::select! { + biased; + _ = self.wait_for_cancellation() => return query_description( + self.id, + &latest, + self.lifecycle(), + ), + result = self.client.poll_status( + descriptor.clone(), + &self.default_namespace_path, + ) => match result { + Err(_) if self.is_cancellation_requested() => return query_description( + self.id, + &latest, + self.lifecycle(), + ), + result => result?, + }, + }; + if let Some(updated) = updated { + self.update_state(&descriptor, updated).await? + } else { + latest + } + } + } else { + state + }; + query_description(self.id, &state, self.lifecycle()) + } + + async fn cancel(&self) -> Result<()> { + let timeout = self.client.overall_timeout()?; + with_overall_timeout(timeout, "SQL query cancellation", self.cancel_inner()).await + } + + async fn cancel_inner(&self) -> Result<()> { + self.touch(); + let _cancel_guard = self.cancel_gate.lock().await; + if matches!( + self.lifecycle(), + QueryLifecycle::Cancelled | QueryLifecycle::Completed + ) { + return Ok(()); + } + loop { + let notified = self.state_changed.notified(); + let state = self.state.lock().await.clone(); + if let Some(info) = state.info { + let previously_uncertain = self.cancel_request_uncertain.load(Ordering::SeqCst); + let outcome = match self + .client + .cancel( + info, + &self.default_namespace_path, + self.cancel_request_uncertain.clone(), + ) + .await + { + Ok(outcome) => outcome, + Err(_) + if matches!( + self.lifecycle(), + QueryLifecycle::Cancelled | QueryLifecycle::Completed + ) => + { + return Ok(()); + } + Err(error) => return Err(error), + }; + if matches!( + self.lifecycle(), + QueryLifecycle::Cancelled | QueryLifecycle::Completed + ) { + return Ok(()); + } + let status = match outcome { + CancelOutcome::Status(status) => status, + CancelOutcome::NotFound(_) + if self.lifecycle() == QueryLifecycle::Cancelling => + { + self.mark_cancelled(); + return Ok(()); + } + CancelOutcome::NotFound(request_id) => { + let message = if previously_uncertain { + "SQL query cancellation outcome is unknown because a prior request may have reached the service and the target was not found on retry" + } else { + "SQL query cancellation target was not found" + }; + return Err(sql_error(&request_id, message)); + } + }; + return match status { + CancelStatus::Cancelled => { + self.mark_cancelled(); + Ok(()) + } + CancelStatus::Cancelling => { + self.mark_cancelling(); + Ok(()) + } + CancelStatus::NotCancellable => { + self.restore_after_rejected_cancellation().await; + Err(Error::NotSupported { + message: "The SQL query is not cancellable".to_string(), + }) + } + CancelStatus::Unspecified => Err(Error::Runtime { + message: "The SQL service returned an unspecified cancellation status" + .to_string(), + }), + }; + } + let Some(descriptor) = state.flight_descriptor else { + return Ok(()); + }; + + tokio::select! { + poll_guard = self.poll_gate.lock() => { + let _poll_guard = poll_guard; + if self.state.lock().await.flight_descriptor.as_ref() != Some(&descriptor) { + continue; + } + let updated = self.client.poll_continuation( + descriptor.clone(), + &self.default_namespace_path, + ).await?; + self.update_state(&descriptor, updated).await?; + } + _ = notified => {} + } + } + } +} + +struct RemoteQueryHandle { + query: Arc, + result_started: AtomicBool, +} + +impl RemoteQueryHandle { + fn new(query: Arc) -> Self { + Self { + query, + result_started: AtomicBool::new(false), + } + } +} + +#[async_trait::async_trait] +impl QueryHandle for RemoteQueryHandle { + fn id(&self) -> Uuid { + self.query.touch(); + self.query.id + } + + async fn describe(&self) -> Result { + self.query.describe().await + } + + async fn reader(&self) -> Result { + let timeout = self.query.client.overall_timeout()?; + if self + .result_started + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .is_err() + { + return Err(Error::Runtime { + message: "SQL query results can only be consumed once".to_string(), + }); + } + let result_start = ResultStartGuard::new(&self.result_started); + let started = Instant::now(); + let prepared = + with_overall_timeout(timeout, "SQL query result", self.query.prepare_result()).await?; + let remaining_timeout = timeout.map(|timeout| timeout.saturating_sub(started.elapsed())); + let schema = prepared.schema.clone(); + let (sender, receiver) = mpsc::channel(2); + let error_sender = sender.clone(); + let query = self.query.clone(); + tokio::spawn(async move { + let result = with_overall_timeout( + remaining_timeout, + "SQL query result", + query.run_result_stream(prepared, sender), + ) + .await; + if let Err(error) = result { + let _ = error_sender.send(Err(error)).await; + } + }); + let stream = futures::stream::unfold(receiver, |mut receiver| async move { + receiver.recv().await.map(|item| (item, receiver)) + }); + result_start.commit(); + Ok(Box::pin(SimpleRecordBatchStream::new(stream, schema))) + } + + async fn cancel(&self) -> Result<()> { + self.query.cancel().await + } +} + +fn query_description( + id: Uuid, + poll_info: &PollInfo, + lifecycle: QueryLifecycle, +) -> Result { + let expires_at = query_expiration(poll_info)?; + Ok(QueryDescription { + id, + status: match lifecycle { + QueryLifecycle::Cancelling => QueryStatus::Cancelling, + QueryLifecycle::Cancelled => QueryStatus::Cancelled, + QueryLifecycle::Running if poll_info.flight_descriptor.is_some() => { + QueryStatus::Running + } + QueryLifecycle::Running | QueryLifecycle::Ready | QueryLifecycle::Completed => { + QueryStatus::Finished + } + }, + progress: poll_info.progress, + expires_at, + }) +} + +fn query_expiration(poll_info: &PollInfo) -> Result>> { + poll_info + .expiration_time + .as_ref() + .map(|timestamp| { + u32::try_from(timestamp.nanos) + .ok() + .and_then(|nanos| chrono::DateTime::from_timestamp(timestamp.seconds, nanos)) + .ok_or_else(|| Error::Runtime { + message: "SQL service returned an invalid query expiration time".to_string(), + }) + }) + .transpose() +} + +#[derive(Clone, Debug, PartialEq, Eq)] +struct SqlTarget { + uri: String, + tls: bool, +} + +fn resolve_sql_host_override( + host_override: Option<&str>, + sql_host_override: Option<&str>, +) -> Result { + if let Some(uri) = sql_host_override { + return normalize_sql_host_override(uri); + } + let host_override = host_override.ok_or_else(|| Error::InvalidInput { + message: "sql_host_override is required when the SQL service endpoint cannot be derived from host_override".to_string(), + })?; + let parsed = url::Url::parse(host_override).map_err(|err| Error::InvalidInput { + message: format!("Invalid host_override: {err}"), + })?; + if parsed.scheme() != "http" { + return Err(Error::InvalidInput { + message: "sql_host_override is required for TLS or non-HTTP host overrides".to_string(), + }); + } + validate_endpoint_url(&parsed, "host_override")?; + let port = match parsed.port().or(explicit_port(host_override)) { + Some(u16::MAX) => { + return Err(Error::InvalidInput { + message: "sql_host_override is required when host_override uses port 65535" + .to_string(), + }); + } + Some(port) => port + 1, + None => DEFAULT_SQL_PORT, + }; + Ok(SqlTarget { + uri: endpoint_uri("http", parsed.host_str().unwrap(), port), + tls: false, + }) +} + +fn normalize_sql_host_override(uri: &str) -> Result { + let parsed = url::Url::parse(uri).map_err(|err| Error::InvalidInput { + message: format!("Invalid sql_host_override: {err}"), + })?; + validate_endpoint_url(&parsed, "sql_host_override")?; + let tls = match parsed.scheme().to_ascii_lowercase().as_str() { + "grpc" | "grpc+tcp" | "http" => false, + "grpc+tls" | "grpcs" | "https" => true, + _ => { + return Err(Error::InvalidInput { + message: + "sql_host_override must use grpc, grpc+tcp, grpc+tls, grpcs, http, or https" + .to_string(), + }); + } + }; + let port = parsed.port().or(explicit_port(uri)).unwrap_or(if tls { + DEFAULT_SQL_TLS_PORT + } else { + DEFAULT_SQL_PORT + }); + if port == 0 { + return Err(Error::InvalidInput { + message: "sql_host_override port must be greater than zero".to_string(), + }); + } + Ok(SqlTarget { + uri: endpoint_uri( + if tls { "https" } else { "http" }, + parsed.host_str().unwrap(), + port, + ), + tls, + }) +} + +fn explicit_port(uri: &str) -> Option { + let authority = uri.split_once("://")?.1.split(['/', '?', '#']).next()?; + let suffix = if authority.starts_with('[') { + authority.split_once(']')?.1.strip_prefix(':')? + } else { + authority.rsplit_once(':')?.1 + }; + suffix.parse().ok() +} + +fn validate_endpoint_url(parsed: &url::Url, name: &str) -> Result<()> { + if parsed.host_str().is_none() { + return Err(Error::InvalidInput { + message: format!("{name} must include a hostname"), + }); + } + if !parsed.username().is_empty() || parsed.password().is_some() { + return Err(Error::InvalidInput { + message: format!("{name} must not include user information"), + }); + } + if !matches!(parsed.path(), "" | "/") || parsed.query().is_some() || parsed.fragment().is_some() + { + return Err(Error::InvalidInput { + message: format!("{name} must not include a path, query, or fragment"), + }); + } + Ok(()) +} + +fn endpoint_uri(scheme: &str, host: &str, port: u16) -> String { + if host.contains(':') { + let host = host + .strip_prefix('[') + .and_then(|host| host.strip_suffix(']')) + .unwrap_or(host); + format!("{scheme}://[{host}]:{port}") + } else { + format!("{scheme}://{host}:{port}") + } +} + +async fn connect_channel( + target: &SqlTarget, + config: &ClientConfig, + request_id: &str, +) -> Result { + let connect_timeout = resolve_timeout( + config.timeout_config.connect_timeout, + "LANCE_CLIENT_CONNECT_TIMEOUT", + Some(DEFAULT_CONNECT_TIMEOUT), + )? + .unwrap(); + let mut endpoint = Endpoint::from_shared(target.uri.clone()) + .map_err(|err| sql_error(request_id, err))? + .connect_timeout(connect_timeout); + if target.tls { + endpoint = endpoint + .tls_config(tls_config(config.tls_config.as_ref())?) + .map_err(|err| sql_error(request_id, err))?; + } + tokio::time::timeout(connect_timeout, endpoint.connect()) + .await + .map_err(|_| sql_error(request_id, "SQL connection timed out"))? + .map_err(|err| sql_error(request_id, err)) +} + +fn tls_config(config: Option<&TlsConfig>) -> Result { + let mut tls = ClientTlsConfig::new().with_enabled_roots(); + if let Some(config) = config { + if !config.assert_hostname { + return Err(Error::InvalidInput { + message: "SQL cannot disable TLS hostname verification".to_string(), + }); + } + if let Some(path) = &config.ssl_ca_cert { + let pem = fs::read(path).map_err(|err| Error::InvalidInput { + message: format!("Failed to read SQL CA certificate {path}: {err}"), + })?; + tls = tls.ca_certificate(Certificate::from_pem(pem)); + } + match (&config.cert_file, &config.key_file) { + (Some(cert), Some(key)) => { + let cert_pem = fs::read(cert).map_err(|err| Error::InvalidInput { + message: format!("Failed to read SQL client certificate {cert}: {err}"), + })?; + let key_pem = fs::read(key).map_err(|err| Error::InvalidInput { + message: format!("Failed to read SQL client key {key}: {err}"), + })?; + tls = tls.identity(Identity::from_pem(cert_pem, key_pem)); + } + (None, None) => {} + _ => { + return Err(Error::InvalidInput { + message: "SQL mTLS requires both cert_file and key_file".to_string(), + }); + } + } + } + Ok(tls) +} + +fn client_with_headers( + client: FlightServiceClient, + headers: &HeaderMap, +) -> Result { + let mut client = FlightClient::new_from_inner(client); + for (key, value) in headers { + let value = value.to_str().map_err(|err| Error::InvalidInput { + message: format!("Invalid SQL metadata value for {key:?}: {err}"), + })?; + client + .add_header(key.as_str(), value) + .map_err(|err| Error::InvalidInput { + message: format!("Invalid SQL metadata header {key:?}: {err}"), + })?; + } + Ok(client) +} + +fn merge_headers(destination: &mut HeaderMap, source: &HashMap) -> Result<()> { + for (key, value) in source { + insert_header(destination, key, value)?; + } + Ok(()) +} + +fn insert_header(headers: &mut HeaderMap, key: &str, value: &str) -> Result<()> { + let key = HeaderName::from_bytes(key.as_bytes()).map_err(|err| Error::InvalidInput { + message: format!("Invalid SQL metadata key {key:?}: {err}"), + })?; + let value = HeaderValue::try_from(value).map_err(|err| Error::InvalidInput { + message: format!("Invalid SQL metadata value for {key:?}: {err}"), + })?; + headers.insert(key, value); + Ok(()) +} + +fn validate_namespace_path(path: &[String]) -> Result<()> { + for component in path { + if component.is_empty() + || !component.is_ascii() + || component.contains('$') + || component.bytes().any(|byte| !(0x20..=0x7e).contains(&byte)) + { + return Err(Error::InvalidInput { + message: "default_namespace_path components must be non-empty printable ASCII strings without '$'".to_string(), + }); + } + } + Ok(()) +} + +fn poll_retry_delay(config: &ResolvedRetryConfig, retry_count: u8) -> Duration { + let exponent = i32::from(retry_count.saturating_sub(1).min(16)); + let backoff = config.backoff_factor * 2.0_f32.powi(exponent); + let jitter = rand::random::() * config.backoff_jitter; + Duration::from_secs_f32((backoff + jitter).clamp(MIN_POLL_INTERVAL.as_secs_f32(), 60.0)) +} + +fn cancellation_status_is_ambiguous(code: tonic::Code) -> bool { + matches!( + code, + tonic::Code::Cancelled + | tonic::Code::Unknown + | tonic::Code::DeadlineExceeded + | tonic::Code::Internal + | tonic::Code::Unavailable + | tonic::Code::DataLoss + ) +} + +fn resolve_timeout( + configured: Option, + env_name: &str, + default: Option, +) -> Result> { + if configured.is_some() { + return Ok(configured); + } + match std::env::var(env_name) { + Ok(value) => value + .parse::() + .map(Duration::from_secs) + .map(Some) + .map_err(|_| Error::InvalidInput { + message: format!("Invalid value for {env_name} environment variable: {value:?}"), + }), + Err(_) => Ok(default), + } +} + +async fn with_overall_timeout( + timeout: Option, + operation: &str, + future: impl std::future::Future>, +) -> Result { + match timeout { + Some(timeout) => { + tokio::time::timeout(timeout, future) + .await + .map_err(|_| Error::Runtime { + message: format!("{operation} timed out"), + })? + } + None => future.await, + } +} + +fn sql_error(request_id: &str, error: impl std::fmt::Display) -> Error { + Error::Runtime { + message: format!("SQL error (request_id={request_id}): {error}"), + } +} + +#[cfg(test)] +#[path = "sql_test.rs"] +mod tests; diff --git a/rust/lancedb/src/remote/sql_test.rs b/rust/lancedb/src/remote/sql_test.rs new file mode 100644 index 000000000..a05a11259 --- /dev/null +++ b/rust/lancedb/src/remote/sql_test.rs @@ -0,0 +1,1040 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +use std::sync::atomic::AtomicUsize; + +use arrow_array::builder::StringDictionaryBuilder; +use arrow_array::{Array, Int64Array, StringArray, types::Int32Type}; +use arrow_flight::encode::FlightDataEncoderBuilder; +use arrow_flight::flight_service_server::{FlightService, FlightServiceServer}; +use arrow_flight::sql::{Any, CommandStatementQuery}; +use arrow_flight::{ + Action, ActionType, CancelFlightInfoResult, Criteria, Empty, FlightData, FlightEndpoint, + FlightInfo, HandshakeRequest, HandshakeResponse, PollInfo, PutResult, SchemaResult, Ticket, +}; +use arrow_schema::{DataType, Field, Schema}; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; +use tonic::{Request, Response, Status, Streaming}; + +use super::*; +use crate::remote::client::HeaderProvider; + +#[derive(Debug, Default)] +struct DelayedHeaderProvider { + delay_next: AtomicBool, +} + +#[async_trait::async_trait] +impl HeaderProvider for DelayedHeaderProvider { + async fn get_headers(&self) -> Result> { + if self.delay_next.swap(false, Ordering::SeqCst) { + tokio::time::sleep(Duration::from_millis(1_100)).await; + } + Ok(HashMap::new()) + } +} + +fn assert_overall_timeout(result: Result, operation: &str) { + match result { + Err(Error::Runtime { message }) => { + assert_eq!(message, format!("SQL query {operation} timed out")); + } + _ => panic!("SQL query {operation} did not honor the overall timeout"), + } +} + +async fn collect_result(query: &Query) -> Result> { + query.reader().await?.try_collect().await +} + +#[derive(Debug)] +struct CapturedHeaders { + database: String, + namespace_path: String, + request_id: String, + api_key: String, + database_prefix: String, +} + +#[derive(Clone)] +struct TestSqlService { + query_count: Arc, + do_get_count: Arc, + cancel_count: Arc, + cancel_denied_count: Arc, + cancel_timeout_count: Arc, + cancel_unspecified_count: Arc, + cancelling_response_count: Arc, + incremental_finished: Arc, + first_continuation_count: Arc, + transient_poll_failures: Arc, + headers: Arc>>, + result: RecordBatch, + large_result: RecordBatch, + dictionary_result: RecordBatch, +} + +impl Default for TestSqlService { + fn default() -> Self { + let schema = Arc::new(Schema::new(vec![Field::new( + "value", + DataType::Int64, + false, + )])); + let result = + RecordBatch::try_new(schema, vec![Arc::new(Int64Array::from(vec![42_i64]))]).unwrap(); + let large_schema = Arc::new(Schema::new(vec![Field::new( + "value", + DataType::Utf8, + false, + )])); + let large_result = RecordBatch::try_new( + large_schema, + vec![Arc::new(StringArray::from(vec![ + "x".repeat(5 * 1024 * 1024), + ]))], + ) + .unwrap(); + let mut dictionary_builder = StringDictionaryBuilder::::new(); + dictionary_builder.append("dictionary value").unwrap(); + let dictionary = dictionary_builder.finish(); + let dictionary_schema = Arc::new(Schema::new(vec![Field::new( + "value", + dictionary.data_type().clone(), + false, + )])); + let dictionary_result = + RecordBatch::try_new(dictionary_schema, vec![Arc::new(dictionary)]).unwrap(); + Self { + query_count: Arc::new(AtomicUsize::new(0)), + do_get_count: Arc::new(AtomicUsize::new(0)), + cancel_count: Arc::new(AtomicUsize::new(0)), + cancel_denied_count: Arc::new(AtomicUsize::new(0)), + cancel_timeout_count: Arc::new(AtomicUsize::new(0)), + cancel_unspecified_count: Arc::new(AtomicUsize::new(0)), + cancelling_response_count: Arc::new(AtomicUsize::new(0)), + incremental_finished: Arc::new(AtomicBool::new(false)), + first_continuation_count: Arc::new(AtomicUsize::new(0)), + transient_poll_failures: Arc::new(AtomicUsize::new(0)), + headers: Arc::new(std::sync::Mutex::new(Vec::new())), + result, + large_result, + dictionary_result, + } + } +} + +#[tonic::async_trait] +impl FlightService for TestSqlService { + type HandshakeStream = BoxStream<'static, std::result::Result>; + type ListFlightsStream = BoxStream<'static, std::result::Result>; + type DoGetStream = BoxStream<'static, std::result::Result>; + type DoPutStream = BoxStream<'static, std::result::Result>; + type DoActionStream = BoxStream<'static, std::result::Result>; + type ListActionsStream = BoxStream<'static, std::result::Result>; + type DoExchangeStream = BoxStream<'static, std::result::Result>; + + async fn handshake( + &self, + _request: Request>, + ) -> std::result::Result, Status> { + Err(Status::unimplemented("handshake")) + } + + async fn list_flights( + &self, + _request: Request, + ) -> std::result::Result, Status> { + Err(Status::unimplemented("list_flights")) + } + + async fn get_flight_info( + &self, + _request: Request, + ) -> std::result::Result, Status> { + Err(Status::unimplemented("get_flight_info")) + } + + async fn poll_flight_info( + &self, + request: Request, + ) -> std::result::Result, Status> { + let metadata = request.metadata(); + let header = |name| { + metadata + .get(name) + .and_then(|value| value.to_str().ok()) + .unwrap() + .to_string() + }; + self.headers.lock().unwrap().push(CapturedHeaders { + database: header("database"), + namespace_path: header("namespace-path"), + request_id: header("x-request-id"), + api_key: header("x-api-key"), + database_prefix: header("x-lancedb-database-prefix"), + }); + + let command = Any::decode(request.get_ref().cmd.as_ref()) + .ok() + .and_then(|any| any.unpack::().ok().flatten()); + let (query, stage) = if let Some(command) = command { + self.query_count.fetch_add(1, Ordering::SeqCst); + (command.query, 0_u8) + } else { + let continuation = std::str::from_utf8(request.get_ref().cmd.as_ref()) + .map_err(|_| Status::invalid_argument("invalid continuation"))?; + let mut parts = continuation.splitn(3, ':'); + if parts.next() != Some("poll") { + return Err(Status::invalid_argument("invalid continuation")); + } + let stage = parts + .next() + .and_then(|stage| stage.parse().ok()) + .ok_or_else(|| Status::invalid_argument("invalid continuation"))?; + if stage == 1 { + self.first_continuation_count.fetch_add(1, Ordering::SeqCst); + } + let query = parts + .next() + .ok_or_else(|| Status::invalid_argument("invalid continuation"))?; + (query.to_string(), stage) + }; + if (query == "SELECT slow" || query == "SELECT cancelling") && stage > 0 { + tokio::time::sleep(Duration::from_millis(250)).await; + } + if query == "SELECT no info" && stage == 1 { + tokio::time::sleep(Duration::from_millis(100)).await; + } + if query == "SELECT incremental" && stage == 1 { + tokio::time::sleep(Duration::from_millis(250)).await; + self.incremental_finished.store(true, Ordering::SeqCst); + } + if stage == 1 + && (query == "SELECT fail" + || (query == "SELECT retry" + && self.transient_poll_failures.fetch_add(1, Ordering::SeqCst) == 0)) + { + return Err(Status::unavailable("transient polling failure")); + } + let complete = if query == "SELECT no info" { + stage >= 2 + } else { + stage >= 1 + }; + + let first_ticket = if query == "SELECT incremental" { + format!("{query}:first") + } else { + query.clone() + }; + let mut info = FlightInfo::new().with_endpoint( + FlightEndpoint::new() + .with_ticket(Ticket::new(first_ticket)) + .with_location("grpc://127.0.0.1:1"), + ); + if query == "SELECT incremental" && stage > 0 { + info = info.with_endpoint( + FlightEndpoint::new() + .with_ticket(Ticket::new(format!("{query}:second"))) + .with_location("grpc://127.0.0.1:1"), + ); + } + if query != "SELECT empty" { + let schema = if query == "SELECT large message" { + self.large_result.schema_ref() + } else if query == "SELECT dictionary" { + self.dictionary_result.schema_ref() + } else { + self.result.schema_ref() + }; + info = info.try_with_schema(schema).unwrap(); + } + Ok(Response::new(PollInfo { + info: (query != "SELECT no info" || stage > 0).then_some(info), + flight_descriptor: (!complete) + .then(|| FlightDescriptor::new_cmd(format!("poll:{}:{query}", stage + 1))), + progress: Some(if complete { 1.0 } else { 0.25 }), + expiration_time: None, + })) + } + + async fn get_schema( + &self, + _request: Request, + ) -> std::result::Result, Status> { + Err(Status::unimplemented("get_schema")) + } + + async fn do_get( + &self, + request: Request, + ) -> std::result::Result::DoGetStream>, Status> { + self.do_get_count.fetch_add(1, Ordering::SeqCst); + let ticket = request.get_ref().ticket.as_ref(); + let empty = ticket == b"SELECT empty"; + let slow = ticket == b"SELECT slow get"; + let large = ticket == b"SELECT large message"; + let result = if large { + self.large_result.clone() + } else if ticket == b"SELECT dictionary" { + self.dictionary_result.clone() + } else { + self.result.clone() + }; + let schema = result.schema(); + let input = futures::stream::once(async move { + if slow { + tokio::time::sleep(Duration::from_millis(250)).await; + } + (!empty).then_some(Ok(result)) + }) + .filter_map(futures::future::ready); + let mut encoder = FlightDataEncoderBuilder::new().with_schema(schema); + if large { + encoder = encoder.with_max_flight_data_size(8 * 1024 * 1024); + } + let stream = encoder.build(input).map_err(Status::from); + Ok(Response::new(Box::pin(stream))) + } + + async fn do_put( + &self, + _request: Request>, + ) -> std::result::Result, Status> { + Err(Status::unimplemented("do_put")) + } + + async fn do_action( + &self, + request: Request, + ) -> std::result::Result, Status> { + if request.get_ref().r#type != "CancelFlightInfo" { + return Err(Status::invalid_argument("unexpected action")); + } + self.cancel_count.fetch_add(1, Ordering::SeqCst); + let cancel_request = CancelFlightInfoRequest::decode(request.get_ref().body.clone()) + .map_err(|_| Status::invalid_argument("invalid cancellation request"))?; + let query = cancel_request + .info + .and_then(|info| info.endpoint.into_iter().next()) + .and_then(|endpoint| endpoint.ticket) + .and_then(|ticket| String::from_utf8(ticket.ticket.to_vec()).ok()) + .ok_or_else(|| Status::invalid_argument("cancellation request had no ticket"))?; + if query == "SELECT cancel race" { + tokio::time::sleep(Duration::from_millis(250)).await; + } + if query == "SELECT cancel timeout" { + if self.cancel_timeout_count.fetch_add(1, Ordering::SeqCst) == 0 { + tokio::time::sleep(Duration::from_millis(250)).await; + } else { + return Err(Status::not_found("query cancellation completed")); + } + } + if query == "SELECT cancel missing" { + return Err(Status::not_found("query was not found")); + } + if query == "SELECT cancel denied" { + if self.cancel_denied_count.fetch_add(1, Ordering::SeqCst) == 0 { + return Err(Status::permission_denied("cancellation is not allowed")); + } + return Err(Status::not_found("query was not found")); + } + if query == "SELECT cancel unspecified" + && self.cancel_unspecified_count.fetch_add(1, Ordering::SeqCst) > 0 + { + return Err(Status::not_found("query cancellation completed")); + } + let status = if query == "SELECT cancel unspecified" { + CancelStatus::Unspecified + } else if query == "SELECT cancelling" { + if self + .cancelling_response_count + .fetch_add(1, Ordering::SeqCst) + == 0 + { + CancelStatus::Cancelling + } else { + return Err(Status::not_found("query cancellation completed")); + } + } else if query == "SELECT cancel race" { + CancelStatus::NotCancellable + } else { + CancelStatus::Cancelled + }; + let response = arrow_flight::Result { + body: CancelFlightInfoResult::new(status).encode_to_vec().into(), + }; + Ok(Response::new(Box::pin(futures::stream::iter([Ok( + response, + )])))) + } + + async fn list_actions( + &self, + _request: Request, + ) -> std::result::Result, Status> { + Err(Status::unimplemented("list_actions")) + } + + async fn do_exchange( + &self, + _request: Request>, + ) -> std::result::Result, Status> { + Err(Status::unimplemented("do_exchange")) + } +} + +#[tokio::test] +async fn submits_polls_fetches_cancels_and_reuses_client() { + let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); + let address = listener.local_addr().unwrap(); + drop(listener); + + let service = TestSqlService::default(); + let query_count = service.query_count.clone(); + let do_get_count = service.do_get_count.clone(); + let cancel_count = service.cancel_count.clone(); + let incremental_finished = service.incremental_finished.clone(); + let first_continuation_count = service.first_continuation_count.clone(); + let headers = service.headers.clone(); + let expected = service.result.clone(); + let (shutdown_tx, shutdown_rx) = tokio::sync::oneshot::channel(); + let server = tokio::spawn( + tonic::transport::Server::builder() + .add_service(FlightServiceServer::new(service)) + .serve_with_shutdown(address, async { + let _ = shutdown_rx.await; + }), + ); + let mut ready = false; + for _ in 0..100 { + if tokio::net::TcpStream::connect(address).await.is_ok() { + ready = true; + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + assert!(ready, "SQL test server did not start"); + + let mut client_config = ClientConfig::default(); + client_config.retry_config.read_retries = Some(1); + client_config.retry_config.backoff_factor = Some(0.0); + client_config.retry_config.backoff_jitter = Some(0.0); + client_config + .extra_headers + .insert("x-static-secret".to_string(), "static-secret".to_string()); + let header_provider = Arc::new(DelayedHeaderProvider::default()); + client_config.header_provider = Some(header_provider.clone()); + let client = SqlClient::new( + "analytics".to_string(), + Some("tenant/production".to_string()), + "test-key".to_string(), + None, + Some(format!("grpc://{address}")), + client_config, + ); + assert_eq!(client.initialized_client_count().await, 0); + assert!(!format!("{client:?}").contains("test-key")); + assert!(!format!("{client:?}").contains("static-secret")); + + let mut timeout_client_config = ClientConfig::default(); + timeout_client_config.timeout_config.timeout = Some(Duration::from_millis(50)); + let timeout_header_provider = Arc::new(DelayedHeaderProvider::default()); + timeout_client_config.header_provider = Some(timeout_header_provider.clone()); + let timeout_client = SqlClient::new( + "analytics".to_string(), + Some("tenant/production".to_string()), + "test-key".to_string(), + None, + Some(format!("grpc://{address}")), + timeout_client_config, + ); + timeout_header_provider + .delay_next + .store(true, Ordering::SeqCst); + assert_overall_timeout( + timeout_client + .submit("SELECT overall timeout", &["public".to_string()]) + .await, + "submission", + ); + let timeout_query = timeout_client + .submit("SELECT overall timeout", &["public".to_string()]) + .await + .unwrap(); + timeout_header_provider + .delay_next + .store(true, Ordering::SeqCst); + assert_overall_timeout( + timeout_client.describe(timeout_query.id()).await, + "description", + ); + timeout_header_provider + .delay_next + .store(true, Ordering::SeqCst); + assert_overall_timeout(collect_result(&timeout_query).await, "result"); + timeout_header_provider + .delay_next + .store(true, Ordering::SeqCst); + assert_overall_timeout(timeout_query.cancel().await, "cancellation"); + timeout_query.cancel().await.unwrap(); + + let pre_dispatch_timeout = timeout_client + .submit("SELECT cancel missing", &["public".to_string()]) + .await + .unwrap(); + timeout_header_provider + .delay_next + .store(true, Ordering::SeqCst); + assert_overall_timeout(pre_dispatch_timeout.cancel().await, "cancellation"); + assert!(pre_dispatch_timeout.cancel().await.is_err()); + assert_ne!( + pre_dispatch_timeout.describe().await.unwrap().status, + QueryStatus::Cancelled + ); + + let rejected_cancel = timeout_client + .submit("SELECT cancel denied", &["public".to_string()]) + .await + .unwrap(); + assert!(rejected_cancel.cancel().await.is_err()); + assert!(rejected_cancel.cancel().await.is_err()); + assert_ne!( + rejected_cancel.describe().await.unwrap().status, + QueryStatus::Cancelled + ); + + let unspecified_cancel = timeout_client + .submit("SELECT cancel unspecified", &["public".to_string()]) + .await + .unwrap(); + assert!(unspecified_cancel.cancel().await.is_err()); + assert!(unspecified_cancel.cancel().await.is_err()); + assert_ne!( + unspecified_cancel.describe().await.unwrap().status, + QueryStatus::Cancelled + ); + assert_eq!( + collect_result(&unspecified_cancel).await.unwrap(), + vec![expected.clone()] + ); + + let uncertain_cancel = timeout_client + .submit("SELECT cancel timeout", &["public".to_string()]) + .await + .unwrap(); + assert_overall_timeout(uncertain_cancel.cancel().await, "cancellation"); + assert!(uncertain_cancel.cancel().await.is_err()); + assert_ne!( + uncertain_cancel.describe().await.unwrap().status, + QueryStatus::Cancelled + ); + assert_eq!( + collect_result(&uncertain_cancel).await.unwrap(), + vec![expected.clone()] + ); + + let first = client + .submit("SELECT 'super-secret'", &["public".to_string()]) + .await + .unwrap(); + assert_eq!(first.id().get_version_num(), 7); + assert!(!first.id().to_string().contains("super-secret")); + header_provider.delay_next.store(true, Ordering::SeqCst); + let describe_started = Instant::now(); + let first_description = client.describe(first.id()).await.unwrap(); + assert!(describe_started.elapsed() >= Duration::from_millis(1_100)); + assert_eq!(first_description.status, QueryStatus::Finished); + assert_eq!(first_description.progress, Some(1.0)); + let first_result = collect_result(&first).await.unwrap(); + assert!(first.reader().await.is_err()); + + let incremental = client + .submit("SELECT incremental", &["public".to_string()]) + .await + .unwrap(); + let mut incremental_result = incremental.reader().await.unwrap(); + let first_incremental_batch = + tokio::time::timeout(Duration::from_millis(100), incremental_result.try_next()) + .await + .expect("the first partial result must arrive before query completion") + .unwrap() + .unwrap(); + assert_eq!(first_incremental_batch, expected); + assert!(!incremental_finished.load(Ordering::SeqCst)); + let remaining_incremental_batches = incremental_result.try_collect::>().await.unwrap(); + assert_eq!(remaining_incremental_batches, vec![expected.clone()]); + assert!(incremental_finished.load(Ordering::SeqCst)); + + let interrupted_result = Arc::new( + client + .submit("SELECT no info", &["public".to_string()]) + .await + .unwrap(), + ); + let continuation_count_before = first_continuation_count.load(Ordering::SeqCst); + let interrupted_result_task = { + let interrupted_result = interrupted_result.clone(); + tokio::spawn(async move { interrupted_result.reader().await }) + }; + tokio::time::timeout(Duration::from_millis(100), async { + while first_continuation_count.load(Ordering::SeqCst) == continuation_count_before { + tokio::task::yield_now().await; + } + }) + .await + .expect("result preparation must start polling"); + interrupted_result_task.abort(); + assert!( + interrupted_result_task + .await + .is_err_and(|error| error.is_cancelled()) + ); + assert_eq!( + collect_result(&interrupted_result).await.unwrap(), + vec![expected.clone()], + "cancelling result preparation must release the one-shot result claim", + ); + + let dropped_reader = client + .submit("SELECT slow", &["public".to_string()]) + .await + .unwrap(); + let tracked_dropped_reader = client.queries.get(dropped_reader.id()).unwrap(); + let continuation_count_before = first_continuation_count.load(Ordering::SeqCst); + let dropped_result_stream = dropped_reader.reader().await.unwrap(); + tokio::time::timeout(Duration::from_millis(100), async { + while first_continuation_count.load(Ordering::SeqCst) == continuation_count_before { + tokio::task::yield_now().await; + } + }) + .await + .expect("the result producer must start continuation polling"); + assert!(Arc::strong_count(&tracked_dropped_reader) >= 4); + drop(dropped_result_stream); + tokio::time::timeout(Duration::from_millis(100), async { + while Arc::strong_count(&tracked_dropped_reader) != 3 { + tokio::task::yield_now().await; + } + }) + .await + .expect("dropping a result reader must stop its producer"); + + let staged = client + .submit("SELECT no info", &["public".to_string()]) + .await + .unwrap(); + let staged_running = client.describe(staged.id()).await.unwrap(); + assert_eq!(staged_running.status, QueryStatus::Running); + let staged_finished = client.describe(staged.id()).await.unwrap(); + assert_eq!(staged_finished.status, QueryStatus::Finished); + + let empty = client + .submit("SELECT empty", &["public".to_string()]) + .await + .unwrap(); + let empty_result = empty.reader().await.unwrap(); + assert_eq!(empty_result.schema(), expected.schema()); + let empty_result = empty_result.try_collect::>().await.unwrap(); + + let large = client + .submit("SELECT large message", &["public".to_string()]) + .await + .unwrap(); + let large_result = collect_result(&large).await.unwrap(); + assert_eq!(large_result.len(), 1); + assert_eq!(large_result[0].num_rows(), 1); + assert_eq!( + large_result[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0) + .len(), + 5 * 1024 * 1024, + ); + + let dictionary = client + .submit("SELECT dictionary", &["public".to_string()]) + .await + .unwrap(); + let dictionary_result = collect_result(&dictionary).await.unwrap(); + assert_eq!(dictionary_result.len(), 1); + assert_eq!( + dictionary_result[0].schema().field(0).data_type(), + &DataType::Utf8, + ); + assert_eq!( + dictionary_result[0] + .column(0) + .as_any() + .downcast_ref::() + .unwrap() + .value(0), + "dictionary value", + ); + + let cancelled = client + .submit( + "SELECT cancelled", + &["events".to_string(), "raw".to_string()], + ) + .await + .unwrap(); + cancelled.cancel().await.unwrap(); + assert_eq!( + cancelled.describe().await.unwrap().status, + QueryStatus::Cancelled + ); + assert!(matches!( + cancelled.reader().await, + Err(Error::JobCancelled { .. }) + )); + + let slow = Arc::new( + client + .submit("SELECT slow", &["public".to_string()]) + .await + .unwrap(), + ); + let result_task = { + let slow = slow.clone(); + tokio::spawn(async move { collect_result(&slow).await }) + }; + tokio::time::sleep(Duration::from_millis(25)).await; + tokio::time::timeout(Duration::from_millis(150), slow.cancel()) + .await + .expect("cancellation must not wait for result polling") + .unwrap(); + assert!(matches!( + tokio::time::timeout(Duration::from_millis(150), result_task) + .await + .expect("cancellation must wake result polling") + .unwrap(), + Err(Error::JobCancelled { .. }) + )); + let cancel_count_after_slow = cancel_count.load(Ordering::SeqCst); + slow.cancel().await.unwrap(); + assert_eq!( + cancel_count.load(Ordering::SeqCst), + cancel_count_after_slow, + "a confirmed cancellation must not be sent again", + ); + + let slow_get = Arc::new( + client + .submit("SELECT slow get", &["public".to_string()]) + .await + .unwrap(), + ); + let do_get_count_before_slow = do_get_count.load(Ordering::SeqCst); + let slow_get_result_task = { + let slow_get = slow_get.clone(); + tokio::spawn(async move { collect_result(&slow_get).await }) + }; + while do_get_count.load(Ordering::SeqCst) == do_get_count_before_slow { + tokio::task::yield_now().await; + } + slow_get.cancel().await.unwrap(); + assert_eq!( + slow_get.describe().await.unwrap().status, + QueryStatus::Cancelled + ); + assert!(matches!( + tokio::time::timeout(Duration::from_millis(150), slow_get_result_task) + .await + .expect("cancellation must wake result fetching") + .unwrap(), + Err(Error::JobCancelled { .. }) + )); + assert!(slow_get.reader().await.is_err()); + + let restored = Arc::new( + RemoteQuery::new( + Uuid::now_v7(), + client.inner.clone(), + vec!["public".to_string()], + PollInfo { + flight_descriptor: Some(FlightDescriptor::new_cmd("restored")), + ..Default::default() + }, + ) + .unwrap(), + ); + let mut restored_waiter = { + let restored = restored.clone(); + tokio::spawn(async move { restored.wait_for_cancellation().await }) + }; + tokio::task::yield_now().await; + restored.mark_cancelling(); + restored.restore_after_rejected_cancellation().await; + assert_eq!(restored.lifecycle(), QueryLifecycle::Running); + assert!( + tokio::time::timeout(Duration::from_millis(25), &mut restored_waiter) + .await + .is_err(), + "a stale cancellation notification must not complete the waiter", + ); + restored.mark_cancelling(); + tokio::time::timeout(Duration::from_millis(100), restored_waiter) + .await + .expect("a current cancellation must complete the waiter") + .unwrap(); + + let cancelling = Arc::new( + client + .submit("SELECT cancelling", &["public".to_string()]) + .await + .unwrap(), + ); + let cancelling_result_task = { + let cancelling = cancelling.clone(); + tokio::spawn(async move { collect_result(&cancelling).await }) + }; + tokio::time::sleep(Duration::from_millis(25)).await; + cancelling.cancel().await.unwrap(); + assert_eq!( + cancelling.describe().await.unwrap().status, + QueryStatus::Cancelling + ); + assert!(matches!( + tokio::time::timeout(Duration::from_millis(150), cancelling_result_task) + .await + .expect("an accepted cancellation must wake result polling") + .unwrap(), + Err(Error::JobCancelled { .. }) + )); + cancelling.cancel().await.unwrap(); + assert_eq!( + cancelling.describe().await.unwrap().status, + QueryStatus::Cancelled + ); + + let cancel_race = Arc::new( + client + .submit("SELECT cancel race", &["public".to_string()]) + .await + .unwrap(), + ); + let cancel_count_before_race = cancel_count.load(Ordering::SeqCst); + let cancel_race_task = { + let cancel_race = cancel_race.clone(); + tokio::spawn(async move { cancel_race.cancel().await }) + }; + while cancel_count.load(Ordering::SeqCst) == cancel_count_before_race { + tokio::task::yield_now().await; + } + let cancel_race_result = collect_result(&cancel_race).await.unwrap(); + tokio::time::timeout(Duration::from_millis(500), cancel_race_task) + .await + .expect("completed result must make in-flight cancellation a no-op") + .unwrap() + .unwrap(); + assert_eq!( + cancel_race.describe().await.unwrap().status, + QueryStatus::Finished + ); + assert_eq!(cancel_race_result, vec![expected.clone()]); + assert!(cancel_race.reader().await.is_err()); + + let no_info = Arc::new( + client + .submit("SELECT no info", &["public".to_string()]) + .await + .unwrap(), + ); + let continuation_count_before = first_continuation_count.load(Ordering::SeqCst); + let no_info_result_task = { + let no_info = no_info.clone(); + tokio::spawn(async move { collect_result(&no_info).await }) + }; + tokio::time::sleep(Duration::from_millis(10)).await; + tokio::time::timeout(Duration::from_secs(1), no_info.cancel()) + .await + .expect("cancellation should wait for cancellable query information") + .unwrap(); + assert!(matches!( + no_info_result_task.await.unwrap(), + Err(Error::JobCancelled { .. }) + )); + assert_eq!( + first_continuation_count.load(Ordering::SeqCst), + continuation_count_before + 1, + "result and cancel must share one continuation poll", + ); + + let retried = client + .submit("SELECT retry", &["public".to_string()]) + .await + .unwrap(); + assert_eq!( + collect_result(&retried).await.unwrap(), + vec![expected.clone()] + ); + + let failed = client + .submit("SELECT fail", &["public".to_string()]) + .await + .unwrap(); + assert!(collect_result(&failed).await.is_err()); + + let registry = QueryRegistry::new(); + for descriptor in ["active-one", "active-two"] { + let id = Uuid::now_v7(); + let query = Arc::new( + RemoteQuery::new( + id, + client.inner.clone(), + vec!["public".to_string()], + PollInfo { + flight_descriptor: Some(FlightDescriptor::new_cmd(descriptor)), + ..Default::default() + }, + ) + .unwrap(), + ); + registry.insert(id, query.clone()); + assert!(Arc::ptr_eq(®istry.get(id).unwrap(), &query)); + } + + let expired_id = Uuid::now_v7(); + let expired_query = Arc::new( + RemoteQuery::new( + expired_id, + client.inner.clone(), + vec!["public".to_string()], + PollInfo { + flight_descriptor: Some(FlightDescriptor::new_cmd("expired")), + expiration_time: Some(Default::default()), + ..Default::default() + }, + ) + .unwrap(), + ); + registry.insert(expired_id, expired_query); + assert!(registry.get(expired_id).is_none()); + + let stale_id = Uuid::now_v7(); + let stale_query = Arc::new( + RemoteQuery::new( + stale_id, + client.inner.clone(), + vec!["public".to_string()], + PollInfo { + flight_descriptor: Some(FlightDescriptor::new_cmd("stale")), + ..Default::default() + }, + ) + .unwrap(), + ); + *stale_query.last_accessed.lock().unwrap() = Instant::now() - ABANDONED_QUERY_RETENTION; + registry.insert(stale_id, stale_query.clone()); + drop(stale_query); + assert!(registry.get(stale_id).is_none()); + + assert_eq!(client.initialized_client_count().await, 1); + assert_eq!(query_count.load(Ordering::SeqCst), 21); + assert_eq!(do_get_count.load(Ordering::SeqCst), 17); + assert_eq!(cancel_count.load(Ordering::SeqCst), 15); + assert_eq!(first_result, vec![expected.clone()]); + assert!(empty_result.is_empty()); + assert!(client.describe(Uuid::nil()).await.is_err()); + { + let headers = headers.lock().unwrap(); + assert_eq!(headers[0].database, "analytics"); + assert_eq!(headers[0].namespace_path, "public"); + assert_eq!(headers[0].api_key, "test-key"); + assert_eq!(headers[0].database_prefix, "tenant/production"); + assert!( + headers + .iter() + .any(|header| header.namespace_path == "events$raw") + ); + assert!( + headers + .windows(2) + .all(|headers| headers[0].request_id != headers[1].request_id) + ); + } + let _ = shutdown_tx.send(()); + server.await.unwrap().unwrap(); +} + +#[test] +fn normalizes_supported_uris() { + assert_eq!( + normalize_sql_host_override("grpc://localhost").unwrap(), + SqlTarget { + uri: "http://localhost:10025".to_string(), + tls: false, + } + ); + assert_eq!( + normalize_sql_host_override("grpcs://example.com").unwrap(), + SqlTarget { + uri: "https://example.com:10026".to_string(), + tls: true, + } + ); + assert_eq!( + normalize_sql_host_override("grpc://[::1]:10025").unwrap(), + SqlTarget { + uri: "http://[::1]:10025".to_string(), + tls: false, + } + ); + assert_eq!( + normalize_sql_host_override("https://example.com:443").unwrap(), + SqlTarget { + uri: "https://example.com:443".to_string(), + tls: true, + } + ); +} + +#[test] +fn derives_plaintext_endpoint_from_host_override() { + assert_eq!( + resolve_sql_host_override(Some("http://localhost:10024"), None).unwrap(), + SqlTarget { + uri: "http://localhost:10025".to_string(), + tls: false, + } + ); + assert_eq!( + resolve_sql_host_override(Some("http://localhost:80"), None).unwrap(), + SqlTarget { + uri: "http://localhost:81".to_string(), + tls: false, + } + ); +} + +#[test] +fn rejects_unsafe_or_ambiguous_endpoints() { + assert!(normalize_sql_host_override("ftp://localhost").is_err()); + assert!(normalize_sql_host_override("grpc://user@localhost").is_err()); + assert!(normalize_sql_host_override("grpc://localhost/path").is_err()); + assert!(resolve_sql_host_override(Some("https://localhost"), None).is_err()); +} + +#[test] +fn validates_namespace_components() { + assert!(validate_namespace_path(&[]).is_ok()); + assert!(validate_namespace_path(&["events".into(), "raw".into()]).is_ok()); + assert!(validate_namespace_path(&["events$raw".into()]).is_err()); + assert!(validate_namespace_path(&["".into()]).is_err()); + assert!(validate_namespace_path(&["café".into()]).is_err()); +} + +#[test] +fn validates_metadata_with_header_map() { + let mut headers = HeaderMap::new(); + insert_header(&mut headers, "X-Custom-Header", "value").unwrap(); + assert_eq!(headers.get("x-custom-header").unwrap(), "value"); + assert!(insert_header(&mut headers, "bad header", "value").is_err()); + assert!(insert_header(&mut headers, "valid-header", "bad\nvalue").is_err()); +} diff --git a/rust/lancedb/src/sql.rs b/rust/lancedb/src/sql.rs new file mode 100644 index 000000000..7c040c51b --- /dev/null +++ b/rust/lancedb/src/sql.rs @@ -0,0 +1,124 @@ +// SPDX-License-Identifier: Apache-2.0 +// SPDX-FileCopyrightText: Copyright The LanceDB Authors + +//! Handles to SQL queries running on a remote database. + +use std::{fmt, sync::Arc}; + +use async_trait::async_trait; +use chrono::{DateTime, Utc}; +use uuid::Uuid; + +use crate::{Result, arrow::SendableRecordBatchStream}; + +/// The externally visible lifecycle state of a submitted SQL query. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum QueryStatus { + /// The server is still executing the query. + Running, + /// The server has made the complete result available. + Finished, + /// The server accepted cancellation but has not confirmed it yet. + Cancelling, + /// The server confirmed cancellation. + Cancelled, +} + +impl fmt::Display for QueryStatus { + fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result { + formatter.write_str(match self { + Self::Running => "running", + Self::Finished => "finished", + Self::Cancelling => "cancelling", + Self::Cancelled => "cancelled", + }) + } +} + +/// A point-in-time description of a submitted SQL query. +#[derive(Clone, Debug, PartialEq)] +pub struct QueryDescription { + /// The stable, connection-scoped identifier assigned when the query was submitted. + pub id: Uuid, + /// The server-visible lifecycle state. + pub status: QueryStatus, + /// Server-reported completion progress, when known. Values are in `[0.0, 1.0]`, + /// with `1.0` meaning complete. + pub progress: Option, + /// When the server may stop accepting this query's continuation token. + pub expires_at: Option>, +} + +#[async_trait] +pub(crate) trait QueryHandle: Send + Sync { + fn id(&self) -> Uuid; + async fn describe(&self) -> Result; + async fn reader(&self) -> Result; + async fn cancel(&self) -> Result<()>; +} + +/// A handle to a submitted SQL query. +/// +/// The handle can be inspected, opened as an Arrow reader, or cancelled. +/// Dropping it does not cancel the server-side query. +/// Identifier lookup is scoped to the connection that submitted the query and +/// is not a durable resume mechanism. +pub struct Query { + handle: Arc, +} + +impl std::fmt::Debug for Query { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter + .debug_struct("Query") + .field("id", &self.id()) + .finish() + } +} + +impl Query { + #[cfg(feature = "remote")] + pub(crate) fn new(handle: Arc) -> Self { + Self { handle } + } + + /// Return the stable, connection-scoped identifier for this query. + pub fn id(&self) -> Uuid { + self.handle.id() + } + + /// Get a point-in-time description of the query. + pub async fn describe(&self) -> Result { + self.handle.describe().await + } + + /// Wait for the initial result stream and return its Arrow record batches. + /// + /// The stream can begin yielding partial results before query execution is + /// complete. It continues polling for newly available result endpoints + /// until the query finishes and all endpoints have been consumed. + /// + /// Results are single-consumer. Calling this method more than once on the + /// same handle returns an error. + pub async fn reader(&self) -> Result { + self.handle.reader().await + } + + /// Request cancellation of the query. + pub async fn cancel(&self) -> Result<()> { + self.handle.cancel().await + } +} + +#[cfg(test)] +mod tests { + use super::QueryStatus; + + #[test] + fn query_status_display_is_stable() { + assert_eq!(QueryStatus::Running.to_string(), "running"); + assert_eq!(QueryStatus::Finished.to_string(), "finished"); + assert_eq!(QueryStatus::Cancelling.to_string(), "cancelling"); + assert_eq!(QueryStatus::Cancelled.to_string(), "cancelled"); + } +}