feat: add asynchronous remote SQL queries (#4070)

## Summary Add SQL execution to remote LanceDB connections. On the standard synchronous connection, `execute_query` waits for the initial result stream and returns its Arrow reader. `execute_query_async` is called without Python `await` and immediately returns a query handle for status inspection, streaming, or cancellation. Local databases report that SQL is not supported. The transport and query lifecycle live in Rust. Python exposes native-backed synchronous and asynchronous connection methods and query wrappers; it does not use PyArrow's Flight client. ## User experience The standard synchronous connection supports both direct reads and background query execution: ```python db = lancedb.connect( "db://analytics", api_key="ldb_...", sql_host_override="grpc+tls://sql.example.com:10026", ) # Direct execution waits only until the initial result stream is available. # Later batches continue streaming as the query progresses. reader = db.execute_query( "SELECT * FROM events", default_namespace_path=["production"], ) for batch in reader: print(batch.num_rows) # Background execution returns a query handle immediately. Despite the # `_async` suffix, no Python `await` is needed on a synchronous connection. query = db.execute_query_async("SELECT * FROM events") print(query.id) description = db.describe_query(query.id) print(description.status) print(description.progress) print(description.expires_at) # Start reading as soon as the service advertises partial results. The reader # continues polling and yields newly available record batches until the query # and all result endpoints are complete. reader = query.reader() for batch in reader: print(batch.num_rows) # Or cancel a different still-running query. Its status becomes "cancelling" # while the server is still working, then "cancelled" once confirmed. cancelled_query = db.execute_query_async("SELECT * FROM large_events") cancelled_query.cancel() ``` The less commonly used asynchronous connection exposes the same operations as coroutines: ```python async_db = await lancedb.connect_async( "db://analytics", api_key="ldb_...", sql_host_override="grpc+tls://sql.example.com:10026", ) query = await async_db.execute_query_async("SELECT * FROM events") async for batch in await query.reader(): print(batch.num_rows) ``` The UUIDv7 query id is scoped to the connection that submitted it. The connection retains lightweight shared query state used by `query.describe()` and `db.describe_query(query.id)`; the id does not encode SQL or a Flight continuation token and is not a cross-connection resume token. Abandoned state has bounded retention, and terminal state remains available briefly. Unqualified table names use the connected database and the `public` namespace by default. `default_namespace_path` accepts a list such as `["production", "events"]`. SQL can still use qualified names to reference other databases and namespaces available to the deployment. ## Design - Uses Arrow Flight `PollFlightInfo` for submission and long polling, `DoGet` for results, and `CancelFlightInfo` for cancellation. Each `PollInfo.info` is treated as the cumulative set of currently available endpoints, so advertised tickets are consumed once and batches can be delivered before execution is complete. - Serializes result completion and cancellation into one lifecycle. A server-accepted request reports `cancelling` and wakes blocked status/result work; a later retry can confirm `cancelled`. Result retrieval is rejected after cancellation is accepted, while cancellation after a result was already delivered is a no-op. - Assigns a time-ordered UUIDv7 connection-scoped query id and retains only shared evolving lifecycle state, keeping SQL, Flight continuation tokens, and Arrow result data out of public ids and the registry. - Leaves admission control to the server while honoring server expiration and a local fallback retention window for abandoned entries. - Retains terminal ids for five minutes so they remain available for connection-level description. - Keeps one lazily initialized SQL client on each remote database connection and attaches fresh authentication, routing, namespace, and request metadata to every operation. - Applies the configured overall timeout to each execution, description, reader, and cancellation operation. A result reader carries one absolute deadline from `reader()` through the end of streaming; connect and read timeouts continue to bound their individual phases. - Returns a bounded, backpressured, single-consumer Arrow stream rather than collecting the full result in memory. Dropping the reader stops downloading but does not implicitly cancel the server query. - Preserves typed schemas for empty result sets through the stream schema. - Accepts Flight result messages up to 1 GiB so a valid row containing a large blob, string, or vector is not rejected by tonic's 4 MiB default receive limit. - Supports the Python client first while keeping the authoritative implementation in the Rust core.
2026-09-04 12:38:38 +00:00 · 2026-09-03 14:59:14 -07:00
parent 2779b75d0d
commit e639b1b650
28 changed files with 3620 additions and 26 deletions
@@ -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:
@@ -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",
@@ -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]
@@ -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
@@ -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]
@@ -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",
@@ -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",
@@ -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]],
@@ -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.

@@ -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.

@@ -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.
@@ -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",
+        }
@@ -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"]
@@ -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:
@@ -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
+        )
@@ -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<Bound<'_, PyAny>>) -> PyResult<Vec<String>> {
+    match path {
+        Some(path) => {
+            if !path.is_instance_of::<PyList>() {
+                return Err(PyValueError::new_err(
+                    "Connection.execute_query_async default_namespace_path must be a list",
+                ));
+            }
+            path.extract::<Vec<String>>().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<Bound<'_, PyAny>>,
+    ) -> PyResult<Bound<'a, PyAny>> {
+        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<Bound<'a, PyAny>> {
+        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<Bound<'_, PyAny>> {
        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<String>,
    region: Option<String>,
    host_override: Option<String>,
+    sql_host_override: Option<String>,
    read_consistency_interval: Option<f64>,
    client_config: Option<PyClientConfig>,
    storage_options: Option<HashMap<String, String>>,
@@ -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);
@@ -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::<crate::job::JobInfo>()?;
    m.add_class::<crate::job::JobDescription>()?;
    m.add_class::<crate::job::JobFailureInfo>()?;
+    m.add_class::<crate::sql::Query>()?;
+    m.add_class::<crate::sql::QueryDescription>()?;
    m.add_class::<PyBlobFile>()?;
    m.add_class::<IndexConfig>()?;
    m.add_class::<Query>()?;
@@ -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<lancedb::sql::Query>,
+}
+
+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<Bound<'_, PyAny>> {
+        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<Bound<'_, PyAny>> {
+        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<Bound<'_, PyAny>> {
+        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<f64>,
+    expires_at: Option<DateTime<Utc>>,
+}
+
+#[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<lancedb::sql::QueryDescription> 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,
+        }
+    }
+}
@@ -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",
@@ -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<dyn Database>,
+    query: String,
+    default_namespace_path: Vec<String>,
+}
+
+impl ExecuteQueryAsyncBuilder {
+    fn new(parent: Arc<dyn Database>, 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<I, S>(mut self, path: I) -> Self
+    where
+        I: IntoIterator<Item = S>,
+        S: Into<String>,
+    {
+        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<crate::sql::Query> {
+        self.parent
+            .execute_query_async(&self.query, &self.default_namespace_path)
+            .await
+    }
+}
+
 impl CloneTableBuilder {
    fn new(parent: Arc<dyn Database>, 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<String>) -> 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<crate::sql::QueryDescription> {
+        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,
            &region,
-            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() {
@@ -340,6 +340,22 @@ pub trait Database:
    async fn job_history(&self, _job_id: Option<&str>) -> Result<Vec<RecordBatch>> {
        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<crate::sql::Query> {
+        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<crate::sql::QueryDescription> {
+        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<Arc<dyn BaseTable>>;
    /// Rename a table in the database
@@ -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;
@@ -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;

@@ -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<S: HttpSend = Sender> {
    namespace_context_provider: Option<Arc<dyn DynamicContextProvider>>,
    /// TLS configuration for mTLS support
    tls_config: Option<super::client::TlsConfig>,
+    sql_client: Option<SqlClient>,
 }

 #[derive(Clone)]
@@ -269,22 +272,35 @@ impl DynamicContextProvider for NamespaceHeaderProviderContext {
    }
 }

+pub struct RemoteHostOverrides {
+    pub rest: Option<String>,
+    pub sql: Option<String>,
+}
+
 impl RemoteDatabase {
-    pub fn try_new(
+    pub(crate) fn try_new(
        uri: &str,
        api_key: &str,
        region: &str,
-        host_override: Option<String>,
+        host_overrides: RemoteHostOverrides,
        client_config: ClientConfig,
        options: RemoteOptions,
        read_consistency_interval: Option<std::time::Duration>,
    ) -> Result<Self> {
        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::<Sender>::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<S: HttpSend> Database for RemoteDatabase<S> {
            .map_err(Into::into)
    }

+    async fn execute_query_async(
+        &self,
+        query: &str,
+        default_namespace_path: &[String],
+    ) -> Result<crate::sql::Query> {
+        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<crate::sql::QueryDescription> {
+        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<Vec<String>> {
        let (tables, version) = if request.namespace_path.is_empty() {
            // The flat route resumes after a table name and orders by name, which is exactly
@@ -466,7 +466,9 @@ impl TokenSource for AzureImdsSource {
 /// OAuth header provider that manages the full token lifecycle.
 ///
 /// Implements [`HeaderProvider`] to inject `Authorization: Bearer <token>`
-/// 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<dyn TokenSource>,
    token_state: Arc<RwLock<TokenState>>,
@@ -554,10 +556,10 @@ impl OAuthHeaderProvider {
 impl HeaderProvider for OAuthHeaderProvider {
    async fn get_headers(&self) -> Result<HashMap<String, String>> {
        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()),
+        ]))
    }
 }

@@ -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<f64>,
+    /// When the server may stop accepting this query's continuation token.
+    pub expires_at: Option<DateTime<Utc>>,
+}
+
+#[async_trait]
+pub(crate) trait QueryHandle: Send + Sync {
+    fn id(&self) -> Uuid;
+    async fn describe(&self) -> Result<QueryDescription>;
+    async fn reader(&self) -> Result<SendableRecordBatchStream>;
+    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<dyn QueryHandle>,
+}
+
+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<dyn QueryHandle>) -> 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<QueryDescription> {
+        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<SendableRecordBatchStream> {
+        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");
+    }
+}