rust/lancedb
1
0
Fork 0
mirror of https://github.com/lancedb/lancedb.git synced 2026-10-23 04:02:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into gatekeeper/fix-2325-1

Browse Source
This commit is contained in:
Gatefixer
2026-09-17 06:49:46 +00:00
parent 9bfd475702 60a1b4c219
commit 64e4d42488
221 changed files with 34375 additions and 14194 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/src/catalog.rs
+125
View File
@@ -0,0 +1,125 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
use std::time::Duration;
use lancedb::catalog::{
CatalogConnection, CreateDatabaseRequest, DropDatabaseRequest, ListDatabasesRequest,
};
use pyo3::exceptions::PyValueError;
use pyo3::{Bound, PyAny, PyRef, PyResult, Python, pyclass, pyfunction, pymethods};
use crate::connection::{Connection, PyClientConfig};
use crate::error::PythonErrorExt;
use crate::runtime::future_into_py;
#[pyclass]
pub struct Catalog {
inner: CatalogConnection,
}
#[pymethods]
impl Catalog {
#[getter]
fn uri(&self) -> &str {
self.inner.uri()
}
#[pyo3(signature = (name, *, exist_ok=false))]
fn create_database<'py>(
self_: PyRef<'py, Self>,
name: String,
exist_ok: bool,
) -> PyResult<Bound<'py, PyAny>> {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
inner
.create_database(CreateDatabaseRequest::new(name).exist_ok(exist_ok))
.await
.map(Connection::new)
.infer_error()
})
}
fn connect_database<'py>(self_: PyRef<'py, Self>, name: String) -> PyResult<Bound<'py, PyAny>> {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
inner
.connect_database(name)
.await
.map(Connection::new)
.infer_error()
})
}
#[pyo3(signature = (name, *, ignore_missing=false))]
fn drop_database<'py>(
self_: PyRef<'py, Self>,
name: String,
ignore_missing: bool,
) -> PyResult<Bound<'py, PyAny>> {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
inner
.drop_database(DropDatabaseRequest::new(name).ignore_missing(ignore_missing))
.await
.infer_error()
})
}
#[pyo3(signature = (*, limit=None, page_token=None))]
fn list_databases<'py>(
self_: PyRef<'py, Self>,
limit: Option<u32>,
page_token: Option<String>,
) -> PyResult<Bound<'py, PyAny>> {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
let mut request = ListDatabasesRequest::default();
request.limit = limit;
request.page_token = page_token;
let response = inner.list_databases(request).await.infer_error()?;
Ok((response.databases, response.page_token))
})
}
}
#[pyfunction]
#[pyo3(signature = (endpoint, *, api_key=None, client_config=None, sql_host_override=None, read_consistency_interval=None, oauth_config=None))]
pub fn connect_catalog(
py: Python<'_>,
endpoint: String,
api_key: Option<String>,
client_config: Option<PyClientConfig>,
sql_host_override: Option<String>,
read_consistency_interval: Option<f64>,
oauth_config: Option<crate::oauth::PyOAuthConfig>,
) -> PyResult<Bound<'_, PyAny>> {
let interval = read_consistency_interval
.map(Duration::try_from_secs_f64)
.transpose()
.map_err(|err| {
PyValueError::new_err(format!("Invalid read consistency interval: {err}"))
})?;
future_into_py(py, async move {
let mut builder = lancedb::connect_catalog(endpoint);
if let Some(api_key) = api_key {
builder = builder.api_key(api_key);
}
if let Some(config) = client_config {
builder = builder.client_config(config.into());
}
if let Some(endpoint) = sql_host_override {
builder = builder.sql_host_override(endpoint);
}
if let Some(interval) = interval {
builder = builder.read_consistency_interval(interval);
}
if let Some(config) = oauth_config {
builder = builder.oauth_config(config.try_into().infer_error()?);
}
Ok(Catalog {
inner: builder.execute().await.infer_error()?,
})
})
}
python/src/connection.rs
+238 -39
View File
@@ -13,11 +13,7 @@ use crate::{
runtime::future_into_py,
table::Table,
};
use arrow::{
datatypes::Schema,
ffi_stream::ArrowArrayStreamReader,
pyarrow::{FromPyArrow, ToPyArrow},
};
use arrow::{datatypes::Schema, ffi_stream::ArrowArrayStreamReader, pyarrow::FromPyArrow};
use lancedb::{
connection::Connection as LanceConnection,
connection::NamespaceClientPushdownOperation,
@@ -28,7 +24,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},
};
#[pyclass]
@@ -86,6 +82,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 +122,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();
@@ -333,7 +381,7 @@ impl Connection {
})
}
#[pyo3(signature = (name, source, projections=None, filter=None, limit=None))]
#[pyo3(signature = (name, source, projections=None, filter=None, limit=None, with_no_data=false))]
pub fn create_materialized_view(
self_: PyRef<'_, Self>,
name: String,
@@ -341,6 +389,7 @@ impl Connection {
projections: Option<Vec<(String, String)>>,
filter: Option<String>,
limit: Option<u64>,
with_no_data: bool,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
@@ -354,16 +403,79 @@ impl Connection {
if let Some(limit) = limit {
builder = builder.limit(limit);
}
let view = builder.execute().await.infer_error()?;
builder = builder.with_no_data(with_no_data);
let view = Box::pin(builder.execute()).await.infer_error()?;
Ok(Table::new(view.table().clone()))
})
}
#[pyo3(signature = (name, source, projections=None, filter=None, limit=None, with_no_data=false))]
pub fn create_materialized_view_async(
self_: PyRef<'_, Self>,
name: String,
source: String,
projections: Option<Vec<(String, String)>>,
filter: Option<String>,
limit: Option<u64>,
with_no_data: bool,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
let mut builder = inner.create_materialized_view(name, source);
if let Some(projections) = projections {
builder = builder.select(projections);
}
if let Some(filter) = filter {
builder = builder.only_if(filter);
}
if let Some(limit) = limit {
builder = builder.limit(limit);
}
let job = Box::pin(builder.with_no_data(with_no_data).execute_async())
.await
.infer_error()?;
Ok(crate::job::Job::new(job))
})
}
pub fn list_materialized_views(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
let views = inner.list_materialized_views().await.infer_error()?;
Ok(views.into_iter().map(|view| view.name).collect::<Vec<_>>())
Ok(views)
})
}
#[pyo3(signature = (name, namespace_path=None))]
pub fn drop_materialized_view(
self_: PyRef<'_, Self>,
name: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner
.drop_materialized_view(name, &namespace_path)
.await
.infer_error()
})
}
#[pyo3(signature = (name, namespace_path=None))]
pub fn drop_materialized_view_async(
self_: PyRef<'_, Self>,
name: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner
.drop_materialized_view_async(name, &namespace_path)
.await
.infer_error()
.map(crate::job::Job::new)
})
}
@@ -592,9 +704,12 @@ impl Connection {
})
}
pub fn job(&self, job_id: String) -> PyResult<crate::job::Job> {
let inner = self.get_inner()?.clone();
Ok(crate::job::Job::new(inner.job(job_id).infer_error()?))
pub fn open_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
let job = inner.open_job(&job_id).await.infer_error()?;
Ok(crate::job::Job::new(job))
})
}
pub fn create_function_async(
@@ -629,6 +744,109 @@ impl Connection {
})
}
pub fn list_functions(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
inner
.list_functions()
.await
.infer_error()?
.into_iter()
.map(|function| function.to_canonical_json().infer_error())
.collect::<PyResult<Vec<_>>>()
})
}
pub fn drop_function(
self_: PyRef<'_, Self>,
name: String,
version: String,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
inner.drop_function(name, version).await.infer_error()
})
}
#[pyo3(signature = (name, value, namespace_path=None))]
pub fn create_secret(
self_: PyRef<'_, Self>,
name: String,
value: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner
.create_secret(name, value, &namespace_path)
.await
.infer_error()
})
}
#[pyo3(signature = (name, value, namespace_path=None))]
pub fn alter_secret(
self_: PyRef<'_, Self>,
name: String,
value: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner
.alter_secret(name, value, &namespace_path)
.await
.infer_error()
})
}
#[pyo3(signature = (namespace_path=None))]
pub fn list_secrets(
self_: PyRef<'_, Self>,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner.list_secrets(&namespace_path).await.infer_error()
})
}
#[pyo3(signature = (name, namespace_path=None))]
pub fn drop_secret(
self_: PyRef<'_, Self>,
name: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
inner.drop_secret(name, &namespace_path).await.infer_error()
})
}
/// Name and timestamps as a plain tuple. `SecretInfo` carries no value, so
/// there is none to filter out here. Timestamps stay integers rather than
/// going through a string, so the caller can compare two without parsing.
#[pyo3(signature = (name, namespace_path=None))]
pub fn describe_secret(
self_: PyRef<'_, Self>,
name: String,
namespace_path: Option<Vec<String>>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
let namespace_path = namespace_path.unwrap_or_default();
future_into_py(self_.py(), async move {
let info = inner
.describe_secret(name, &namespace_path)
.await
.infer_error()?;
Ok((info.name, info.created_at_millis, info.updated_at_millis))
})
}
pub fn list_jobs(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
@@ -640,42 +858,16 @@ impl Connection {
})
}
pub fn get_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
let description = inner.get_job(&job_id).await.infer_error()?;
Ok(description.map(crate::job::JobDescription::from))
})
}
pub fn cancel_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
inner.cancel_job(&job_id).await.infer_error()
})
}
#[pyo3(signature = (job_id=None))]
pub fn job_history(
self_: PyRef<'_, Self>,
job_id: Option<String>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.get_inner()?.clone();
future_into_py(self_.py(), async move {
let batches = inner.job_history(job_id.as_deref()).await.infer_error()?;
Python::attach(|py| {
let list = PyList::empty(py);
for batch in batches {
list.append(batch.to_pyarrow(py)?)?;
}
Ok(list.unbind())
})
})
}
}
#[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<'_>,
@@ -683,6 +875,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>>,
@@ -702,6 +895,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);
python/src/error.rs
+9
View File
@@ -29,7 +29,10 @@ impl<T> PythonErrorExt<T> for std::result::Result<T, LanceError> {
LanceError::InvalidInput { .. }
| LanceError::InvalidTableName { .. }
| LanceError::TableNotFound { .. }
| LanceError::NotAMaterializedView { .. }
| LanceError::Schema { .. }
| LanceError::DatabaseNotFound { .. }
| LanceError::DatabaseAlreadyExists { .. }
| LanceError::TableAlreadyExists { .. } => self.value_error(),
LanceError::CreateDir { .. } => self.os_error(),
LanceError::ObjectStore { .. } => Err(PyIOError::new_err(err.to_string())),
@@ -114,6 +117,12 @@ impl<T> PythonErrorExt<T> for std::result::Result<T, LanceError> {
.getattr(intern!(py, "JobCancelledError"))?;
Err(PyErr::from_value(cls.call1((err.to_string(),))?))
}),
LanceError::JobNotFound { .. } => Python::attach(|py| {
let cls = py
.import(intern!(py, "lancedb.exceptions"))?
.getattr(intern!(py, "JobNotFoundError"))?;
Err(PyErr::from_value(cls.call1((err.to_string(),))?))
}),
_ => self.runtime_error(),
},
}
python/src/job.rs
+129 -12
View File
@@ -4,11 +4,50 @@
use std::sync::Arc;
use crate::runtime::future_into_py;
use pyo3::{Bound, PyAny, PyRef, PyResult, pyclass, pymethods};
use arrow::{
datatypes::Schema,
pyarrow::{IntoPyArrow, Table as PyArrowTable},
};
use lancedb::job::JobEventsRequest;
use pyo3::{
Bound, PyAny, PyRef, PyResult, Python,
exceptions::PyValueError,
pyclass, pymethods,
types::{PyAnyMethods, PyDict, PyDictMethods},
};
use serde::Serialize;
use crate::error::PythonErrorExt;
const REPR_INDENT: &str = " ";
/// Parse a stored JSON payload into Python data. The bindings carry these as
/// strings because that is what crosses the boundary cheaply; the public
/// Python surface is the parsed form.
fn parse_json_payload<'py>(
py: Python<'py>,
raw: Option<&str>,
) -> PyResult<Option<Bound<'py, PyAny>>> {
match raw {
None => Ok(None),
Some(raw) => Ok(Some(py.import("json")?.call_method1("loads", (raw,))?)),
}
}
/// A payload rendered as indented JSON, aligned under the field that holds it.
fn pretty_json_payload(py: Python<'_>, raw: Option<&str>) -> PyResult<Option<String>> {
let Some(parsed) = parse_json_payload(py, raw)? else {
return Ok(None);
};
let kwargs = PyDict::new(py);
kwargs.set_item("indent", 4)?;
let rendered: String = py
.import("json")?
.call_method("dumps", (parsed,), Some(&kwargs))?
.extract()?;
Ok(Some(rendered.replace('\n', &format!("\n{REPR_INDENT}"))))
}
#[pyclass]
pub struct Job {
inner: Arc<lancedb::Job<std::result::Result<Option<String>, String>>>,
@@ -67,10 +106,52 @@ impl Job {
Ok(())
})
}
pub fn refresh(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner.clone();
future_into_py(self_.py(), async move {
inner.refresh().await.infer_error()?;
Ok(())
})
}
/// The last observed lifecycle state, without contacting the backend.
#[getter]
pub fn _state(&self) -> Option<String> {
self.inner.state()
}
/// The last observed server-side record. `None` for an in-process job.
#[getter]
pub fn _description(&self) -> Option<JobDescription> {
self.inner.description().map(JobDescription::from)
}
#[pyo3(signature = (*, limit=None, filter=None))]
pub fn events(
self_: PyRef<'_, Self>,
limit: Option<u32>,
filter: Option<String>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner.clone();
let request = JobEventsRequest { limit, filter };
future_into_py(self_.py(), async move {
let batches = inner.events(request).await.infer_error()?;
Python::attach(|py| {
let schema = batches
.first()
.map(|batch| batch.schema())
.unwrap_or_else(|| Arc::new(Schema::empty()));
let table = PyArrowTable::try_new(batches, schema)
.map_err(|err| PyValueError::new_err(err.to_string()))?;
table.into_pyarrow(py).map(|table| table.unbind())
})
})
}
}
/// A row from `Connection.list_jobs`: one server-side job.
#[pyclass(get_all, skip_from_py_object)]
#[pyclass(module = "lancedb._lancedb", get_all, skip_from_py_object)]
#[derive(Clone)]
pub struct JobInfo {
job_id: String,
@@ -103,7 +184,7 @@ impl From<lancedb::database::JobInfo> for JobInfo {
}
/// The server's account of why a job failed.
#[pyclass(get_all, skip_from_py_object)]
#[pyclass(module = "lancedb._lancedb", get_all, skip_from_py_object)]
#[derive(Clone)]
pub struct JobFailureInfo {
phase: Option<String>,
@@ -121,25 +202,57 @@ impl JobFailureInfo {
}
}
/// A described job from `Connection.get_job`.
#[pyclass(get_all, skip_from_py_object)]
/// The server-side record behind a `Job` handle.
#[pyclass(module = "lancedb._lancedb", get_all, skip_from_py_object)]
#[derive(Clone)]
pub struct JobDescription {
job_id: String,
job_type: String,
state: String,
creation_ms: i64,
spec_json: Option<String>,
/// Internal: the wire form behind the `spec` property.
_spec_json: Option<String>,
/// Internal: the wire form behind the `result` property.
_result_json: Option<String>,
failure: Option<JobFailureInfo>,
}
#[pymethods]
impl JobDescription {
fn __repr__(&self) -> String {
format!(
"JobDescription(job_id={:?}, job_type={:?}, state={:?}, creation_ms={})",
self.job_id, self.job_type, self.state, self.creation_ms
)
/// The job-type-specific specification it was submitted with.
#[getter]
fn spec<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyAny>>> {
parse_json_payload(py, self._spec_json.as_deref())
}
/// The job-type-specific terminal result. `None` until the job succeeds.
#[getter]
fn result<'py>(&self, py: Python<'py>) -> PyResult<Option<Bound<'py, PyAny>>> {
parse_json_payload(py, self._result_json.as_deref())
}
fn __repr__(&self, py: Python<'_>) -> PyResult<String> {
let mut fields = vec![
format!("job_id={:?}", self.job_id),
format!("job_type={:?}", self.job_type),
format!("state={:?}", self.state),
format!("creation_ms={}", self.creation_ms),
];
// Lay the payloads out as indented JSON, the same way the `Job` repr
// does, so the two agree on how the same data looks.
for (name, payload) in [("spec", &self._spec_json), ("result", &self._result_json)] {
if let Some(rendered) = pretty_json_payload(py, payload.as_deref())? {
fields.push(format!("{name}={rendered}"));
}
}
if let Some(failure) = &self.failure {
fields.push(format!("failure={}", failure.__repr__()));
}
let body = fields
.iter()
.map(|field| format!("\n{REPR_INDENT}{field},"))
.collect::<String>();
Ok(format!("JobDescription({body}\n)"))
}
}
@@ -150,7 +263,11 @@ impl From<lancedb::database::JobDescription> for JobDescription {
job_type: description.job_type,
state: description.state,
creation_ms: description.creation_ms,
spec_json: (!description.spec.is_null()).then(|| description.spec.to_string()),
_spec_json: (!description.spec.is_null()).then(|| description.spec.to_string()),
_result_json: description
.result
.filter(|result| !result.is_null())
.map(|result| result.to_string()),
failure: description.failure.map(|failure| JobFailureInfo {
phase: failure.phase,
message: failure.message,
python/src/lib.rs
+32 -3
View File
@@ -8,8 +8,8 @@ use expr::{PyExpr, expr_col, expr_func, expr_lit};
use index::IndexConfig;
use permutation::{PyAsyncPermutationBuilder, PyPermutationReader};
use pyo3::{
Bound, PyResult, Python, pymodule,
types::{PyModule, PyModuleMethods},
Bound, PyResult, Python, pyfunction, pymodule,
types::{PyAnyMethods, PyModule, PyModuleMethods},
wrap_pyfunction,
};
use query::{FTSQuery, HybridQuery, Query, VectorQuery};
@@ -21,6 +21,7 @@ use table::{
};
pub mod arrow;
pub mod catalog;
pub mod connection;
pub mod error;
pub mod expr;
@@ -34,22 +35,46 @@ pub mod permutation;
pub mod query;
pub mod runtime;
pub mod session;
pub mod sql;
pub mod table;
pub mod util;
/// Shut down the shared Tokio runtime (see `runtime::shutdown`).
///
/// Registered below as a Python `atexit` callback rather than called
/// directly: `atexit` runs while the interpreter is still fully valid,
/// which is the coordinated, bounded exit the runtime otherwise never gets.
///
/// Runs the actual wait with the GIL released (`Python::detach`): shutdown
/// blocks the calling thread waiting on the runtime's own worker threads,
/// and if any in-flight task needs the GIL to finish (e.g. one that calls
/// back into Python), holding it here while waiting on that same task would
/// deadlock rather than time out.
#[pyfunction]
fn shutdown_runtime(py: Python<'_>) {
py.detach(|| runtime::shutdown(std::time::Duration::from_secs(5)));
}
#[pymodule]
pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
pub fn _lancedb(py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
let env = Env::new()
.filter_or("LANCEDB_LOG", "warn")
.write_style("LANCEDB_LOG_STYLE");
env_logger::init_from_env(env);
m.add_class::<Connection>()?;
m.add_class::<catalog::Catalog>()?;
m.add_function(wrap_pyfunction!(catalog::connect_catalog, m)?)?;
m.add_class::<Session>()?;
m.add_class::<Table>()?;
m.add_class::<crate::oauth::PyOAuthSession>()?;
m.add_class::<crate::oauth::PySessionStatus>()?;
m.add_class::<crate::oauth::PySessionLogout>()?;
m.add_class::<crate::job::Job>()?;
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>()?;
@@ -92,5 +117,9 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
m.add_function(wrap_pyfunction!(expr_lit, m)?)?;
m.add_function(wrap_pyfunction!(expr_func, m)?)?;
m.add("__version__", env!("CARGO_PKG_VERSION"))?;
// Give the shared runtime a coordinated, bounded shutdown at normal
// process exit -- see `shutdown_runtime` and `runtime::shutdown` for why.
py.import("atexit")?
.call_method1("register", (wrap_pyfunction!(shutdown_runtime, m)?,))?;
Ok(())
}
python/src/oauth.rs
+319 -6
View File
@@ -1,10 +1,33 @@
// SPDX-License-Identifier: Apache-2.0
// SPDX-FileCopyrightText: Copyright The LanceDB Authors
use pyo3::FromPyObject;
use std::path::PathBuf;
use std::sync::Arc;
use pyo3::{FromPyObject, PyResult, Python, pyclass, pymethods};
use crate::error::PythonErrorExt;
use crate::runtime::future_into_py;
use lancedb::error::Error;
use lancedb::remote::oauth::{OAuthConfig, OAuthFlow};
use lancedb::remote::oauth::{AuthorizationCodeOptions, ClientAuthMethod, OAuthConfig, OAuthFlow};
use lancedb::remote::{OAuthSession, SessionLogout, SessionStatus, TokenCacheOptions};
/// Python-side persistent token cache options, extracted via FromPyObject.
/// Maps to `lancedb.remote.oauth.TokenCacheOptions` Python dataclass.
#[derive(FromPyObject, Default)]
pub struct PyTokenCacheOptions {
pub cache_dir: Option<String>,
pub lock_timeout_secs: Option<u64>,
}
impl From<PyTokenCacheOptions> for TokenCacheOptions {
fn from(py: PyTokenCacheOptions) -> Self {
Self {
cache_dir: py.cache_dir.map(PathBuf::from),
lock_timeout_secs: py.lock_timeout_secs,
}
}
}
/// Python-side OAuth configuration, extracted via FromPyObject.
/// Maps to `lancedb.remote.oauth.OAuthConfig` Python dataclass.
@@ -13,10 +36,19 @@ pub struct PyOAuthConfig {
pub issuer_url: String,
pub client_id: String,
pub scopes: Vec<String>,
/// Optional resource indicator for authorization and token requests.
pub resource: Option<String>,
/// Optional provider-specific audience for authorization and token requests.
pub audience: Option<String>,
pub flow: String,
pub client_secret: Option<String>,
pub client_auth_method: Option<String>,
pub redirect_uri: Option<String>,
pub callback_port: Option<u16>,
pub use_pkce: bool,
pub managed_identity_client_id: Option<String>,
pub refresh_buffer_secs: Option<u64>,
pub token_cache: Option<PyTokenCacheOptions>,
}
impl TryFrom<PyOAuthConfig> for OAuthConfig {
@@ -25,6 +57,17 @@ impl TryFrom<PyOAuthConfig> for OAuthConfig {
fn try_from(py: PyOAuthConfig) -> Result<Self, Self::Error> {
let flow = match py.flow.as_str() {
"client_credentials" => OAuthFlow::ClientCredentials,
"authorization_code" => {
let mut options = AuthorizationCodeOptions::new().use_pkce(py.use_pkce);
if let Some(redirect_uri) = py.redirect_uri {
options = options.redirect_uri(redirect_uri);
}
if let Some(callback_port) = py.callback_port {
options = options.callback_port(callback_port);
}
OAuthFlow::AuthorizationCode(options)
}
"device_code" => OAuthFlow::DeviceCode,
"azure_managed_identity" => OAuthFlow::AzureManagedIdentity {
client_id: py.managed_identity_client_id,
},
@@ -35,13 +78,182 @@ impl TryFrom<PyOAuthConfig> for OAuthConfig {
}
};
let client_auth_method = match py.client_auth_method.as_deref() {
Some("none") => Some(ClientAuthMethod::None),
Some("client_secret_basic") => Some(ClientAuthMethod::ClientSecretBasic),
Some("client_secret_post") => Some(ClientAuthMethod::ClientSecretPost),
None => None,
Some(other) => {
return Err(Error::InvalidInput {
message: format!("Unknown OAuth client auth method: {other}"),
});
}
};
Ok(Self {
issuer_url: py.issuer_url,
client_id: py.client_id,
client_secret: py.client_secret,
client_auth_method,
scopes: py.scopes,
resource: py.resource,
audience: py.audience,
flow,
refresh_buffer_secs: py.refresh_buffer_secs,
token_cache: py.token_cache.map(TokenCacheOptions::from),
})
}
}
/// Wrapper around [`lancedb::remote::SessionStatus`] exposing safe metadata.
#[pyclass(name = "SessionStatus", skip_from_py_object)]
#[derive(Clone)]
pub struct PySessionStatus {
inner: SessionStatus,
}
#[pymethods]
impl PySessionStatus {
/// Whether a cached session exists that can obtain tokens without
/// interactive authentication.
#[getter]
pub fn refreshable(&self) -> bool {
self.inner.refreshable
}
/// Canonical issuer URL of the cached session.
#[getter]
pub fn issuer_url(&self) -> String {
self.inner.issuer_url.clone()
}
/// Client ID of the cached session.
#[getter]
pub fn client_id(&self) -> String {
self.inner.client_id.clone()
}
/// Canonical (sorted, de-duplicated) scopes of the cached session.
#[getter]
pub fn scopes(&self) -> Vec<String> {
self.inner.scopes.clone()
}
/// Resource indicator used to obtain the cached session.
#[getter]
pub fn resource(&self) -> Option<String> {
self.inner.resource.clone()
}
/// Provider-specific audience used to obtain the cached session.
#[getter]
pub fn audience(&self) -> Option<String> {
self.inner.audience.clone()
}
/// Flow that produced the cached session.
#[getter]
pub fn flow(&self) -> String {
self.inner.flow.clone()
}
/// When the cached session was obtained, as Unix seconds.
#[getter]
pub fn obtained_at(&self) -> Option<u64> {
self.inner.obtained_at
}
pub fn __repr__(&self) -> String {
format!(
"SessionStatus(refreshable={}, issuer_url='{}', client_id='{}', flow='{}')",
self.inner.refreshable, self.inner.issuer_url, self.inner.client_id, self.inner.flow
)
}
}
impl From<SessionStatus> for PySessionStatus {
fn from(inner: SessionStatus) -> Self {
Self { inner }
}
}
/// Wrapper around [`lancedb::remote::SessionLogout`].
#[pyclass(name = "SessionLogout", skip_from_py_object)]
#[derive(Clone)]
pub struct PySessionLogout {
inner: SessionLogout,
}
#[pymethods]
impl PySessionLogout {
/// Whether a cached credential was removed.
#[getter]
pub fn removed(&self) -> bool {
self.inner.removed
}
pub fn __repr__(&self) -> String {
format!("SessionLogout(removed={})", self.inner.removed)
}
}
impl From<SessionLogout> for PySessionLogout {
fn from(inner: SessionLogout) -> Self {
Self { inner }
}
}
/// Wrapper around [`lancedb::remote::OAuthSession`].
#[pyclass(name = "OAuthSession", skip_from_py_object)]
#[derive(Clone)]
pub struct PyOAuthSession {
inner: Arc<OAuthSession>,
}
#[pymethods]
impl PyOAuthSession {
/// Create a session manager for the given OAuth configuration.
///
/// The configuration must set ``token_cache`` options and use a flow that
/// supports persistent sessions (authorization code or device code).
#[new]
pub fn new(config: PyOAuthConfig) -> PyResult<Self> {
let config: OAuthConfig = config.try_into().infer_error()?;
let inner = OAuthSession::new(config).infer_error()?;
Ok(Self {
inner: Arc::new(inner),
})
}
/// Eagerly run the configured authentication flow and store the session.
pub fn login<'py>(&self, py: Python<'py>) -> PyResult<pyo3::Bound<'py, pyo3::PyAny>> {
let inner = Arc::clone(&self.inner);
future_into_py(py, async move {
inner.login().await.map(PySessionStatus::from).infer_error()
})
}
/// Report whether a matching cached session exists, with safe metadata.
pub fn status<'py>(&self, py: Python<'py>) -> PyResult<pyo3::Bound<'py, pyo3::PyAny>> {
let inner = Arc::clone(&self.inner);
future_into_py(py, async move {
inner
.status()
.await
.map(PySessionStatus::from)
.infer_error()
})
}
/// Remove the matching local cached credential.
pub fn logout<'py>(&self, py: Python<'py>) -> PyResult<pyo3::Bound<'py, pyo3::PyAny>> {
let inner = Arc::clone(&self.inner);
future_into_py(py, async move {
inner
.logout()
.await
.map(PySessionLogout::from)
.infer_error()
})
}
}
@@ -50,16 +262,30 @@ impl TryFrom<PyOAuthConfig> for OAuthConfig {
mod tests {
use super::*;
#[test]
fn test_unknown_oauth_flow_returns_invalid_input() {
let config = PyOAuthConfig {
fn base_config() -> PyOAuthConfig {
PyOAuthConfig {
issuer_url: "https://issuer.example.com".to_string(),
client_id: "client-id".to_string(),
scopes: vec!["scope".to_string()],
flow: "typo".to_string(),
flow: "device_code".to_string(),
client_secret: None,
client_auth_method: None,
redirect_uri: None,
callback_port: None,
use_pkce: true,
managed_identity_client_id: None,
refresh_buffer_secs: None,
resource: None,
audience: None,
token_cache: None,
}
}
#[test]
fn test_unknown_oauth_flow_returns_invalid_input() {
let config = PyOAuthConfig {
flow: "typo".to_string(),
..base_config()
};
let err = OAuthConfig::try_from(config).unwrap_err();
@@ -69,4 +295,91 @@ mod tests {
if message == "Unknown OAuth flow type: typo"
));
}
#[test]
fn test_authorization_code_conversion_preserves_options() {
let config = PyOAuthConfig {
flow: "authorization_code".to_string(),
client_secret: Some("secret".to_string()),
redirect_uri: Some("http://127.0.0.1:9000/callback".to_string()),
callback_port: Some(9000),
use_pkce: false,
resource: Some("urn:resource".into()),
audience: Some("audience".into()),
..base_config()
};
let converted = OAuthConfig::try_from(config).unwrap();
let OAuthFlow::AuthorizationCode(options) = converted.flow else {
panic!("expected authorization code flow");
};
assert_eq!(
options.redirect_uri.as_deref(),
Some("http://127.0.0.1:9000/callback")
);
assert_eq!(options.callback_port, Some(9000));
assert!(!options.use_pkce);
assert_eq!(converted.resource.as_deref(), Some("urn:resource"));
assert_eq!(converted.audience.as_deref(), Some("audience"));
}
#[test]
fn test_device_code_conversion() {
let config = base_config();
let converted = OAuthConfig::try_from(config).unwrap();
assert!(matches!(converted.flow, OAuthFlow::DeviceCode));
}
#[test]
fn test_client_auth_method_conversion() {
for (value, expected) in [
("none", ClientAuthMethod::None),
("client_secret_basic", ClientAuthMethod::ClientSecretBasic),
("client_secret_post", ClientAuthMethod::ClientSecretPost),
] {
let config = PyOAuthConfig {
client_auth_method: Some(value.to_string()),
..base_config()
};
let converted = OAuthConfig::try_from(config).unwrap();
assert_eq!(converted.client_auth_method, Some(expected));
}
}
#[test]
fn test_unknown_client_auth_method_returns_invalid_input() {
let config = PyOAuthConfig {
client_auth_method: Some("typo".to_string()),
..base_config()
};
let err = OAuthConfig::try_from(config).unwrap_err();
assert!(matches!(
err,
Error::InvalidInput { message }
if message == "Unknown OAuth client auth method: typo"
));
}
#[test]
fn test_token_cache_conversion() {
let config = PyOAuthConfig {
resource: None,
audience: None,
token_cache: Some(PyTokenCacheOptions {
cache_dir: Some("/tmp/oauth-cache".to_string()),
lock_timeout_secs: Some(5),
}),
..base_config()
};
let converted = OAuthConfig::try_from(config).unwrap();
let cache = converted.token_cache.expect("token cache options");
assert_eq!(
cache.cache_dir.as_deref(),
Some(std::path::Path::new("/tmp/oauth-cache"))
);
assert_eq!(cache.lock_timeout_secs, Some(5));
}
}
python/src/query.rs
+4 -1
View File
@@ -323,6 +323,7 @@ impl<'py> IntoPyObject<'py> for PyQueryVectors {
pub struct PyQueryRequest {
pub limit: Option<usize>,
pub offset: Option<usize>,
pub take_offsets: Option<Vec<u64>>,
pub filter: Option<PyQueryFilter>,
pub full_text_search: Option<PyLanceDB<FtsQuery>>,
pub select: PySelect,
@@ -333,7 +334,7 @@ pub struct PyQueryRequest {
pub column: Option<String>,
pub query_vector: Option<PyQueryVectors>,
pub minimum_nprobes: Option<usize>,
// None means user did not set it and default shoud be used (currenty 20)
// None means user did not set it and default should be used (currently 20)
// Some(0) means user set it to None and there is no limit
pub maximum_nprobes: Option<usize>,
pub lower_bound: Option<f32>,
@@ -353,6 +354,7 @@ impl From<AnyQuery> for PyQueryRequest {
AnyQuery::Query(query_request) => Self {
limit: query_request.limit,
offset: query_request.offset,
take_offsets: query_request.take_offsets,
filter: query_request.filter.map(PyQueryFilter),
full_text_search: query_request
.full_text_search
@@ -381,6 +383,7 @@ impl From<AnyQuery> for PyQueryRequest {
AnyQuery::VectorQuery(vector_query) => Self {
limit: vector_query.base.limit,
offset: vector_query.base.offset,
take_offsets: vector_query.base.take_offsets,
filter: vector_query.base.filter.map(PyQueryFilter),
full_text_search: None,
select_source_columns: PySelect::source_columns(&vector_query.base.select),
python/src/runtime.rs
+460 -30
View File
@@ -4,20 +4,84 @@
//! Fork-safe wrapper around tokio + pyo3-async-runtimes.
//!
//! `pyo3_async_runtimes::tokio` keeps its multi-threaded runtime in a
//! `OnceLock` that can never be replaced. Tokio's worker threads do not
//! `OnceLock` that can never be replaced. Tokio's worker threads do not
//! survive `fork()`, so once a child inherits a "frozen" runtime, every
//! `future_into_py` call hangs forever.
//! `future_into_py` call hangs forever. Normal (non-fork) process exit has
//! its own gap: nothing tells the runtime to shut down, so its worker
//! threads keep running, uncoordinated with the interpreter, right up until
//! the process ends. If one of them is mid-task exactly as `Py_Finalize`
//! starts tearing down interpreter state, it can panic on state that's
//! already gone -- and since that happens on a background thread with no
//! PyO3-wrapped call frame to catch it, Rust aborts the whole process
//! rather than failing that one call. [`shutdown`], registered as a Python
//! `atexit` callback, closes that gap by giving the runtime a coordinated,
//! bounded exit while the interpreter is still fully valid.
//!
//! We sidestep the global by routing every future through our own
//! [`LanceRuntime`] (a [`pyo3_async_runtimes::generic::Runtime`] impl) backed
//! by an [`AtomicPtr`] to a tokio runtime that we own. A `pthread_atfork`
//! child handler nulls the pointer; the next `spawn` rebuilds the runtime in
//! the child. This mirrors the pattern used in the Lance Python bindings.
//! Getting both of these right at once took a few tries; the design here
//! rests on three separate mechanisms, each solving one problem the others
//! cannot:
//!
//! **`OUTSTANDING`, not `Arc::strong_count`, decides when the runtime is
//! idle.** Early versions tried to infer "is anything still using this
//! runtime" from how many `Arc<Runtime>` clones existed. That signal is
//! wrong in both directions: a clone taken only for the instant a task is
//! *submitted* says nothing about whether that task has actually finished
//! running (`Runtime::shutdown_timeout` gives spawned, non-blocking tasks
//! no grace period at all -- a task "keeps running until it yields, then is
//! dropped" -- so a reclaim landing right after submission would silently
//! abandon it before it ever got to run); and a clone held for a task's
//! *whole* lifetime can end up making that task the final owner of the
//! `Runtime`, so completing it drops the `Runtime` from inside one of its
//! own worker threads, which tokio itself forbids ("cannot drop a runtime
//! in a context where blocking is not allowed") and panics. `OUTSTANDING`
//! is an explicit courtesy counter instead: every top-level `spawn`,
//! `spawn_blocking`, or `block_on` call increments it before it starts and
//! decrements it (via [`OutstandingGuard`]) when it is truly done, entirely
//! decoupled from how many `Arc` clones exist at any instant. `shutdown`
//! waits for it to reach zero before ever touching the runtime, which also
//! closes a narrower race: because the counter is incremented *before*
//! `get_runtime()` is even called, an install already in progress when
//! `shutdown` runs is never invisible to it the way an empty slot would be.
//! If the counter never reaches zero within the bound, `shutdown` stops
//! waiting and forces the retirement attempt anyway -- silently returning
//! with the runtime and its workers still fully alive would just recreate
//! the exact race this function exists to close, for any call slower than
//! the grace period.
//!
//! **Tasks never hold an `Arc<Runtime>`.** Because `OUTSTANDING` (not
//! reference counting) is what `shutdown` waits on, a top-level task only
//! needs to carry an `OutstandingGuard` -- a token whose `Drop` is a plain
//! atomic decrement -- not a clone of the runtime itself. That is what
//! makes it impossible for a task's completion to become the final,
//! worker-thread-side drop of the `Runtime`: nothing a task holds ever
//! *is* the `Runtime`.
//!
//! **`atfork_child` touches nothing but a plain counter.** `future_into_py`
//! spawns a task that, once running, spawns a second one to do the real
//! work and awaits its `JoinHandle`; if a nested call re-resolved "the
//! current runtime" independently, a reclaim landing between the two calls
//! could bind them to different instances. `spawn`/`spawn_blocking` close
//! that with `Handle::try_current`: a call already running on one of our
//! worker threads stays pinned to that instance, so only the first,
//! outermost call of a chain ever consults [`get_runtime`]. That leaves
//! fork as the other place identity can change, and it has to be handled
//! without ever calling into `ArcSwapOption` from the child handler itself
//! -- `swap`/`compare_and_swap` reconcile reader "debts" internally (via
//! thread-local state, and potentially an allocation), none of which is
//! safe to run in a forked child that may have inherited another thread's
//! lock mid-acquisition. `atfork_child` therefore does nothing but bump a
//! bare `GENERATION` counter; [`get_runtime`] compares the generation its
//! installed runtime was built in against the live counter on every call,
//! from ordinary (non-signal) context, and treats a mismatch as "stale,
//! rebuild" -- exactly the check `atfork_child` used to perform directly.
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::{AtomicBool, AtomicPtr, Ordering};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::{Duration, Instant};
use arc_swap::ArcSwapOption;
use pyo3::{Bound, PyAny, PyResult, Python, conversion::IntoPyObject};
use pyo3_async_runtimes::{
TaskLocals,
@@ -25,8 +89,31 @@ use pyo3_async_runtimes::{
};
use tokio::{runtime, task};
static RUNTIME: AtomicPtr<runtime::Runtime> = AtomicPtr::new(std::ptr::null_mut());
static RUNTIME_INSTALLING: AtomicBool = AtomicBool::new(false);
/// A runtime tagged with the fork generation it was built in, so a stale
/// (post-fork, dead-worker-threads) instance can be told apart from a live
/// one without `atfork_child` ever having to touch it directly.
struct Tagged {
runtime: runtime::Runtime,
generation: u64,
}
impl std::ops::Deref for Tagged {
type Target = runtime::Runtime;
fn deref(&self) -> &runtime::Runtime {
&self.runtime
}
}
static RUNTIME: ArcSwapOption<Tagged> = ArcSwapOption::const_empty();
/// Bumped only by `atfork_child`, and only ever read elsewhere. This is the
/// entire fork-safety mechanism: no lock, no allocation, no thread-local
/// access -- just one atomic add, which is all a `pthread_atfork` child
/// handler is ever safe to do.
static GENERATION: AtomicU64 = AtomicU64::new(0);
/// Count of top-level `spawn`/`spawn_blocking`/`block_on` calls that have
/// started but not yet finished. See the module docs for why this, and not
/// `Arc::strong_count`, is what `shutdown` waits on.
static OUTSTANDING: AtomicU64 = AtomicU64::new(0);
static ATFORK_INSTALLED: AtomicBool = AtomicBool::new(false);
fn create_runtime() -> runtime::Runtime {
@@ -37,23 +124,67 @@ fn create_runtime() -> runtime::Runtime {
.expect("Failed to build tokio runtime")
}
fn get_runtime() -> &'static runtime::Runtime {
/// Get a live, owned handle to the shared runtime, rebuilding it if the
/// installed one predates the most recent `fork()`.
fn get_runtime() -> Arc<Tagged> {
let current_gen = GENERATION.load(Ordering::SeqCst);
loop {
let ptr = RUNTIME.load(Ordering::SeqCst);
if !ptr.is_null() {
return unsafe { &*ptr };
let existing = RUNTIME.load_full();
if let Some(existing) = &existing
&& existing.generation == current_gen
{
return Arc::clone(existing);
}
if !RUNTIME_INSTALLING.fetch_or(true, Ordering::SeqCst) {
break;
if !ATFORK_INSTALLED.fetch_or(true, Ordering::SeqCst) {
install_atfork();
}
std::thread::yield_now();
// Built optimistically, outside any lock: on the rare race where two
// threads both find the slot empty (or stale), one candidate wins
// the compare-and-swap below and the other is simply dropped here,
// tearing down its own (never shared, never used) worker pool the
// ordinary way.
let candidate = Arc::new(Tagged {
runtime: create_runtime(),
generation: current_gen,
});
let previous = RUNTIME.compare_and_swap(&existing, Some(Arc::clone(&candidate)));
let won = match (&*previous, &existing) {
(None, None) => true,
(Some(prev), Some(exist)) => Arc::ptr_eq(prev, exist),
_ => false,
};
if won {
if let Some(stale) = existing {
// A prior generation's runtime: its worker threads are dead
// in this process (they do not survive fork), so dropping it
// normally would try to join them and hang. Leak it instead.
std::mem::forget(stale);
}
return candidate;
}
// Someone else's candidate (or a concurrent shutdown) won; go around
// and reload.
}
if !ATFORK_INSTALLED.fetch_or(true, Ordering::SeqCst) {
install_atfork();
}
/// RAII token tracked by [`OUTSTANDING`]. Held for the duration of a
/// top-level `block_on` call, or moved into a top-level `spawn`/
/// `spawn_blocking` task so it decrements only once that task's *entire*
/// body -- including anything it nested-spawns and awaits -- has run to
/// completion or been dropped without completing.
struct OutstandingGuard;
impl OutstandingGuard {
fn new() -> Self {
OUTSTANDING.fetch_add(1, Ordering::SeqCst);
Self
}
}
impl Drop for OutstandingGuard {
fn drop(&mut self) {
OUTSTANDING.fetch_sub(1, Ordering::SeqCst);
}
let new_ptr = Box::into_raw(Box::new(create_runtime()));
RUNTIME.store(new_ptr, Ordering::SeqCst);
unsafe { &*new_ptr }
}
/// Block the current thread on a future using the shared runtime.
@@ -62,16 +193,78 @@ fn get_runtime() -> &'static runtime::Runtime {
/// building a namespace client). Must not be called from within the runtime's
/// own worker threads.
pub fn block_on<F: std::future::Future>(fut: F) -> F::Output {
let _guard = OutstandingGuard::new();
get_runtime().block_on(fut)
}
/// Runs in async-signal context after `fork()` in the child. We can only
/// touch atomics here; we deliberately leak the previous runtime because
/// dropping a tokio `Runtime` would try to join its (now-dead) worker
/// threads and hang.
/// Gracefully quiesce the shared runtime, meant to run at normal process exit.
///
/// Waits (bounded by `timeout`) for [`OUTSTANDING`] to reach zero -- i.e.
/// for every top-level call already under way to actually finish, not just
/// for `Arc::strong_count` to look low -- before ever touching the runtime.
/// If the bound elapses first, it stops waiting and attempts retirement
/// anyway: leaving the runtime and its worker threads untouched would just
/// recreate the exact race this function exists to close, for any call
/// slower than the grace period.
///
/// Retirement itself removes the runtime from the slot and calls
/// `shutdown_timeout` rather than a bare `drop`: dropping a tokio `Runtime`
/// waits (in the worst case indefinitely) for its worker threads to join,
/// whereas `shutdown_timeout` gives real in-flight work -- a connection
/// pool's keep-alive, a graceful close -- a bounded chance to finish first,
/// then forcibly ends whatever has not. Reclaiming can only proceed once
/// `Arc::try_unwrap` proves no other reference remains; if some transient
/// `get_runtime()` caller is, at that exact instant, still between loading
/// the slot and finishing its own call, this abandons the runtime instead
/// of forcing the issue -- the same trade `atfork_child` already makes.
///
/// Neither of the two ways this can fail to cleanly retire the runtime --
/// the wait timing out, or `try_unwrap` losing that race -- has any other
/// signal to report through (`shutdown_timeout` itself returns nothing),
/// so both log a warning instead of failing silently.
pub fn shutdown(timeout: Duration) {
let deadline = Instant::now() + timeout;
loop {
let outstanding = OUTSTANDING.load(Ordering::SeqCst);
if outstanding == 0 {
break;
}
if Instant::now() >= deadline {
log::warn!(
"lancedb: runtime shutdown timed out with {outstanding} call(s) still in flight; forcing shutdown anyway, some in-flight work may be abandoned"
);
break;
}
std::thread::sleep(Duration::from_millis(1));
}
let Some(current) = RUNTIME.load_full() else {
return;
};
RUNTIME.compare_and_swap(&Some(Arc::clone(&current)), None);
match Arc::try_unwrap(current) {
Ok(tagged) => {
tagged
.runtime
.shutdown_timeout(deadline.saturating_duration_since(Instant::now()));
}
Err(_) => {
// Some transient `get_runtime()` caller is, at this exact
// instant, still between loading the slot and finishing its own
// call: we have no owned handle to call `shutdown_timeout` on,
// and no way to force one (tokio has no shutdown API over a
// shared reference). Nothing more to do but say so.
log::warn!(
"lancedb: runtime shutdown could not obtain exclusive ownership; the shared runtime was left running"
);
}
}
}
/// Runs in async-signal context after `fork()` in the child. Touches
/// nothing but a plain atomic add -- see the module docs for why even
/// `ArcSwapOption::swap` is not safe to call here.
extern "C" fn atfork_child() {
RUNTIME.store(std::ptr::null_mut(), Ordering::SeqCst);
RUNTIME_INSTALLING.store(false, Ordering::SeqCst);
GENERATION.fetch_add(1, Ordering::SeqCst);
}
#[cfg(not(windows))]
@@ -102,11 +295,26 @@ impl Runtime for LanceRuntime {
type JoinError = LanceJoinError;
type JoinHandle = Pin<Box<dyn Future<Output = Result<(), Self::JoinError>> + Send>>;
/// `pyo3_async_runtimes::generic::future_into_py` spawns a task that,
/// once it starts running, spawns a second one to do the real work and
/// awaits its `JoinHandle`. `Handle::try_current` pins that nested call
/// to whatever runtime is already executing it, so only the first,
/// outermost call of a chain -- one running on a thread outside any
/// runtime -- ever consults [`get_runtime`] or [`OUTSTANDING`].
fn spawn<F>(fut: F) -> Self::JoinHandle
where
F: Future<Output = ()> + Send + 'static,
{
let handle = get_runtime().spawn(fut);
let handle = match tokio::runtime::Handle::try_current() {
Ok(handle) => handle.spawn(fut),
Err(_) => {
let guard = OutstandingGuard::new();
get_runtime().spawn(async move {
let _guard = guard;
fut.await;
})
}
};
Box::pin(async move { handle.await.map_err(LanceJoinError) })
}
@@ -114,7 +322,16 @@ impl Runtime for LanceRuntime {
where
F: FnOnce() + Send + 'static,
{
let handle = get_runtime().spawn_blocking(f);
let handle = match tokio::runtime::Handle::try_current() {
Ok(handle) => handle.spawn_blocking(f),
Err(_) => {
let guard = OutstandingGuard::new();
get_runtime().spawn_blocking(move || {
let _guard = guard;
f();
})
}
};
Box::pin(async move { handle.await.map_err(LanceJoinError) })
}
}
@@ -149,3 +366,216 @@ where
{
pyo3_async_runtimes::generic::future_into_py::<LanceRuntime, _, T>(py, fut)
}
#[cfg(test)]
mod tests {
use super::*;
// RUNTIME, GENERATION, OUTSTANDING, and ATFORK_INSTALLED are process-wide
// statics, and Rust's test harness runs tests in parallel by default,
// so separate test functions below would otherwise race each other
// through this shared state (this reproduced in CI: one test's
// in-flight task got reclaimed by a *different* test's concurrent
// `shutdown()` call, and another observed `OUTSTANDING` left non-zero
// by a still-running sibling). Every test takes this lock first so
// only one of them touches the shared runtime state at a time; a
// poisoned lock (a previous test's genuine failure) is still honored
// rather than cascading into every later test as an unrelated panic.
static TEST_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
fn lock_runtime_state_for_test() -> std::sync::MutexGuard<'static, ()> {
TEST_LOCK
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
}
// A task's own completion must never be the final drop of the shared
// `Runtime`: tasks carry only an `OutstandingGuard` (a plain counter
// token), never an `Arc<Runtime>`, specifically so this can't happen.
// Getting this wrong panics ("cannot drop a runtime in a context where
// blocking is not allowed") -- this reproduced unprompted, twice, in a
// single run of `test_nested_spawn_survives_concurrent_shutdown` under
// an earlier design that held the `Arc` for a task's whole lifetime.
#[test]
#[allow(unused_must_use)] // fire-and-forget spawn, same as future_into_py itself
fn test_top_level_task_survives_concurrent_shutdown_reclaim() {
use std::sync::mpsc;
let _lock = lock_runtime_state_for_test();
for _ in 0..50 {
let (tx, rx) = mpsc::channel::<()>();
LanceRuntime::spawn(async move {
for _ in 0..5 {
task::yield_now().await;
}
let _ = tx.send(());
});
shutdown(Duration::from_secs(2));
rx.recv_timeout(Duration::from_secs(5))
.expect("top-level task was abandoned by a concurrent shutdown reclaim");
}
}
#[test]
fn test_shutdown_stops_and_the_runtime_rebuilds_lazily_after() {
let _lock = lock_runtime_state_for_test();
// No runtime created yet in this process: shutdown must be a no-op,
// not a null-pointer dereference.
shutdown(Duration::from_secs(1));
// Force the runtime into existence, then shut it down. Bounded by
// the timeout, so a hang here means shutdown itself is broken, not
// that the test is slow.
assert_eq!(block_on(async { 1 + 1 }), 2);
shutdown(Duration::from_secs(5));
// A caller after shutdown -- e.g. a stray call racing with the
// atexit callback -- must get a fresh, working runtime rather than
// a dangling reference into the one just torn down.
assert_eq!(block_on(async { 2 + 2 }), 4);
// Shutting down twice in a row (e.g. atexit firing more than once)
// must not panic or double-free.
shutdown(Duration::from_secs(5));
shutdown(Duration::from_secs(5));
}
// Adapted from the concurrent reproducer that found the original bug:
// many threads hammering the runtime while shutdown races them, not
// just the sequential rebuild path above. Before every reader held its
// own `Arc`, this could dereference a runtime `shutdown` had already
// freed, or hang forever. Repeated, since a race is not guaranteed to
// show up on any single attempt.
#[test]
fn test_shutdown_is_safe_concurrently_with_live_callers() {
use std::sync::Barrier;
use std::sync::atomic::AtomicBool as StopFlag;
let _lock = lock_runtime_state_for_test();
for _ in 0..50 {
let barrier = Arc::new(Barrier::new(9));
let stop = Arc::new(StopFlag::new(false));
let workers: Vec<_> = (0..8)
.map(|_| {
let barrier = Arc::clone(&barrier);
let stop = Arc::clone(&stop);
std::thread::spawn(move || {
barrier.wait();
while !stop.load(Ordering::Relaxed) {
assert_eq!(block_on(async { 1 + 1 }), 2);
}
})
})
.collect();
barrier.wait();
shutdown(Duration::from_millis(50));
stop.store(true, Ordering::Relaxed);
for worker in workers {
worker.join().unwrap();
}
}
}
// Reproduces the actual bug mechanism, not just concurrent `block_on`
// traffic: `future_into_py` spawns an outer task that, once running,
// spawns a second (inner) one for the real work and awaits its
// `JoinHandle` (see `pyo3_async_runtimes::generic::future_into_py_with_locals`).
// Before `spawn`/`spawn_blocking` pinned nested calls to whatever
// runtime is already executing them, a reclaim landing between the
// outer and inner spawn could bind them to two different runtime
// instances -- and if the outer task's own runtime was the one torn
// down while it awaited the inner task, it never resumed. Every wait
// here is bounded so a reintroduced bug fails this test instead of
// hanging the suite.
#[test]
#[allow(unused_must_use)] // fire-and-forget outer spawn, same as future_into_py itself
fn test_nested_spawn_survives_concurrent_shutdown() {
use std::sync::mpsc;
let _lock = lock_runtime_state_for_test();
for _ in 0..50 {
let stop = Arc::new(AtomicBool::new(false));
let (done_tx, done_rx) = mpsc::channel::<()>();
let workers: Vec<_> = (0..8)
.map(|_| {
let stop = Arc::clone(&stop);
let done_tx = done_tx.clone();
std::thread::spawn(move || {
while !stop.load(Ordering::Relaxed) {
let (tx, rx) = mpsc::sync_channel::<()>(1);
LanceRuntime::spawn(async move {
let inner = LanceRuntime::spawn(async {
let _ = 1 + 1;
});
let _ = inner.await;
let _ = tx.send(());
});
// A hang here across a concurrent shutdown is
// exactly the bug this test exists to catch.
let _ = rx.recv_timeout(Duration::from_secs(2));
}
let _ = done_tx.send(());
})
})
.collect();
std::thread::sleep(Duration::from_millis(10));
shutdown(Duration::from_millis(50));
stop.store(true, Ordering::Relaxed);
for _ in 0..8 {
done_rx
.recv_timeout(Duration::from_secs(10))
.expect("worker hung after shutdown raced a nested spawn");
}
for worker in workers {
worker.join().unwrap();
}
}
}
// Forces the exact interleaving the install-race finding described: an
// installer registers as outstanding (as `spawn`'s outermost branch
// does, before it ever calls `get_runtime()`) and then pauses, while a
// concurrent `shutdown()` must not decide "nothing here" and return
// before that install actually completes and is retired in turn.
#[test]
fn test_shutdown_waits_for_a_racing_install() {
use std::sync::mpsc;
let _lock = lock_runtime_state_for_test();
// Clean slate: no runtime installed, OUTSTANDING at zero.
shutdown(Duration::from_secs(5));
let (installer_ready_tx, installer_ready_rx) = mpsc::channel::<()>();
let (proceed_tx, proceed_rx) = mpsc::channel::<()>();
let installer = std::thread::spawn(move || {
let _guard = OutstandingGuard::new();
installer_ready_tx.send(()).unwrap();
proceed_rx.recv().unwrap();
assert_eq!(get_runtime().block_on(async { 1 + 1 }), 2);
});
installer_ready_rx.recv().unwrap();
let shutdown_thread = std::thread::spawn(|| shutdown(Duration::from_secs(5)));
// Give shutdown's polling loop several chances to (wrongly) observe
// an idle runtime before the installer is allowed to proceed.
std::thread::sleep(Duration::from_millis(50));
proceed_tx.send(()).unwrap();
installer.join().unwrap();
shutdown_thread.join().unwrap();
// shutdown must have waited for the install to finish and then
// retired it, not returned early and left it stranded.
assert!(RUNTIME.load_full().is_none());
assert_eq!(OUTSTANDING.load(Ordering::SeqCst), 0);
}
}
python/src/sql.rs
+90
View File
@@ -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,
}
}
}
python/src/table.rs
+45
View File
@@ -575,6 +575,17 @@ pub struct RefreshMaterializedViewResult {
#[pymethods]
impl RefreshMaterializedViewResult {
#[staticmethod]
pub fn from_json(value: &str) -> PyResult<Self> {
let result: lancedb::RefreshMaterializedViewResult =
serde_json::from_str(value).map_err(|err| {
PyValueError::new_err(format!(
"failed to decode materialized-view refresh result: {err}"
))
})?;
Ok(Self::from(result))
}
pub fn __repr__(&self) -> String {
format!(
"RefreshMaterializedViewResult(mode={}, rows_written={}, source_version={}, version={})",
@@ -1772,6 +1783,40 @@ impl Table {
})
}
#[pyo3(signature = (full=false, source_version=None))]
pub fn refresh_materialized_view_async(
self_: PyRef<'_, Self>,
full: bool,
source_version: Option<u64>,
) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let view = lancedb::MaterializedView::from_table(inner)
.await
.infer_error()?;
let mut builder = view.refresh().full(full);
if let Some(version) = source_version {
builder = builder.source_version(version);
}
let job = builder.execute_async().await.infer_error()?;
Ok(crate::job::Job::new_typed(job))
})
}
pub fn materialized_view_definition(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
let inner = self_.inner_ref()?.clone();
future_into_py(self_.py(), async move {
let view = lancedb::MaterializedView::from_table(inner)
.await
.infer_error()?;
serde_json::to_string(view.definition()).map_err(|err| {
PyRuntimeError::new_err(format!(
"failed to serialize materialized-view definition: {err}"
))
})
})
}
pub fn add_columns_with_schema(
self_: PyRef<'_, Self>,
schema: PyArrowType<Schema>,