mirror of
https://github.com/lancedb/lancedb.git
synced 2026-09-11 15:52:17 +00:00
Jack Ye
e639b1b650
## 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.
91 lines
2.3 KiB
Rust
91 lines
2.3 KiB
Rust
// 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,
|
|
}
|
|
}
|
|
}
|