Merge branch 'main' into refactor/wal-naming

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

python/src/catalog.rs

@@ -0,0 +1,155 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use futures::{StreamExt, future::poll_fn};
+
+use lancedb::catalog::{CatalogConnection, CreateDatabaseRequest, DropDatabaseRequest};
+use pyo3::exceptions::{PyStopAsyncIteration, 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 = (*, page_token=None, page_limit=None))]
+    fn list_databases(&self, page_token: Option<String>, page_limit: Option<u32>) -> DatabaseNames {
+        DatabaseNames {
+            inner: Arc::new(Mutex::new(
+                self.inner.list_databases(page_token, page_limit),
+            )),
+            next_lock: Arc::new(tokio::sync::Mutex::new(())),
+        }
+    }
+}
+
+#[pyclass]
+pub struct DatabaseNames {
+    inner: Arc<Mutex<lancedb::catalog::DatabaseNames>>,
+    next_lock: Arc<tokio::sync::Mutex<()>>,
+}
+
+#[pymethods]
+impl DatabaseNames {
+    fn num_page_results(&self, py: Python<'_>) -> usize {
+        py.detach(|| self.inner.lock().unwrap().num_page_results())
+    }
+
+    fn page_token(&self, py: Python<'_>) -> Option<String> {
+        py.detach(|| self.inner.lock().unwrap().page_token().map(str::to_owned))
+    }
+
+    fn __aiter__(self_: PyRef<'_, Self>) -> PyRef<'_, Self> {
+        self_
+    }
+
+    fn __anext__(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner.clone();
+        let next_lock = self_.next_lock.clone();
+        future_into_py(self_.py(), async move {
+            // Serialize advances while allowing synchronous state inspection
+            // between polls, including while a REST request is pending.
+            let _guard = next_lock.lock().await;
+            poll_fn(|cx| inner.lock().unwrap().poll_next_unpin(cx))
+                .await
+                .ok_or_else(|| PyStopAsyncIteration::new_err(""))?
+                .infer_error()
+        })
+    }
+}
+
+#[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

@@ -14,9 +14,10 @@ use crate::{
     table::Table,
 };
 use arrow::{
+    array::RecordBatch,
     datatypes::Schema,
     ffi_stream::ArrowArrayStreamReader,
-    pyarrow::{FromPyArrow, ToPyArrow},
+    pyarrow::{FromPyArrow, PyArrowType, ToPyArrow},
 };
 use lancedb::{
     connection::Connection as LanceConnection,
@@ -28,7 +29,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 +87,46 @@ 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()]),
+    }
+}
+
+/// A view description on its way to Python: name, namespace, query, default
+/// database, and the schema as pyarrow renders it.
+type PyViewDescription = (String, Vec<String>, String, String, Vec<String>, Py<PyAny>);
+
+/// A view description as a plain tuple, with the schema converted to the
+/// pyarrow schema the caller would get from any other lancedb API. The Python
+/// layer names the fields; this keeps the binding free of a class that would
+/// have to be kept in step with the Rust struct.
+fn view_description_to_py(view: lancedb::view::ViewDescription) -> PyResult<PyViewDescription> {
+    Python::attach(|py| {
+        let schema = view.schema.to_pyarrow(py)?.unbind();
+        Ok((
+            view.name,
+            view.namespace_path,
+            view.query,
+            view.default_database,
+            view.default_namespace_path,
+            schema,
+        ))
+    })
+}
+
 #[pymethods]
 impl Connection {
     fn __repr__(&self) -> String {
@@ -108,6 +149,44 @@ impl Connection {
         self.get_inner().map(|inner| inner.uri().to_string())
     }
 
+    #[pyo3(signature = (query, *, default_namespace_path=None, parameters=None))]
+    pub fn execute_query_async<'a>(
+        self_: PyRef<'a, Self>,
+        query: String,
+        default_namespace_path: Option<Bound<'_, PyAny>>,
+        parameters: Option<PyArrowType<RecordBatch>>,
+    ) -> 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 mut operation = inner
+                .execute_query_async(query)
+                .default_namespace_path(default_namespace_path);
+            if let Some(PyArrowType(parameters)) = parameters {
+                operation = operation.parameters(parameters);
+            }
+            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 +412,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 +420,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 +434,83 @@ 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()))
         })
     }
 
-    pub fn list_materialized_views(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+    #[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 views = inner.list_materialized_views().await.infer_error()?;
-            Ok(views.into_iter().map(|view| view.name).collect::<Vec<_>>())
+            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))
+        })
+    }
+
+    #[pyo3(signature = (*, page_token=None, page_limit=None))]
+    pub fn list_materialized_views(
+        &self,
+        page_token: Option<String>,
+        page_limit: Option<u32>,
+    ) -> PyResult<crate::listing::NameListing> {
+        Ok(crate::listing::NameListing::new(
+            self.get_inner()?
+                .list_materialized_views(crate::listing::options(page_token, page_limit)),
+        ))
+    }
+
+    #[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,36 +739,45 @@ 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))
+        })
     }
 
+    #[pyo3(signature = (request_json, namespace_path=None))]
     pub fn create_function_async(
         self_: PyRef<'_, Self>,
         request_json: String,
+        namespace_path: Option<Vec<String>>,
     ) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.get_inner()?.clone();
         let request = lancedb::function::FunctionRegistrationRequest::from_json(&request_json)
             .infer_error()?;
+        let namespace_path = namespace_path.unwrap_or_default();
         future_into_py(self_.py(), async move {
             inner
-                .create_function_async(request)
+                .create_function_async(request, &namespace_path)
                 .await
                 .infer_error()
                 .map(crate::job::Job::new_typed)
         })
     }
 
+    #[pyo3(signature = (name, version, namespace_path=None))]
     pub fn get_function(
         self_: PyRef<'_, Self>,
         name: String,
         version: 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
-                .get_function(name, version)
+                .get_function(name, version, &namespace_path)
                 .await
                 .infer_error()?
                 .to_canonical_json()
@@ -629,25 +785,230 @@ impl Connection {
         })
     }
 
-    pub fn list_jobs(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+    #[pyo3(signature = (namespace_path=None, *, page_token=None, page_limit=None))]
+    pub fn list_functions(
+        &self,
+        namespace_path: Option<Vec<String>>,
+        page_token: Option<String>,
+        page_limit: Option<u32>,
+    ) -> PyResult<crate::listing::FunctionListing> {
+        Ok(crate::listing::FunctionListing::new(
+            self.get_inner()?.list_functions(
+                &namespace_path.unwrap_or_default(),
+                crate::listing::options(page_token, page_limit),
+            ),
+        ))
+    }
+
+    #[pyo3(signature = (name, version, namespace_path=None))]
+    pub fn drop_function(
+        self_: PyRef<'_, Self>,
+        name: String,
+        version: 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 jobs = inner.list_jobs().await.infer_error()?;
-            Ok(jobs
-                .into_iter()
-                .map(crate::job::JobInfo::from)
-                .collect::<Vec<_>>())
+            inner
+                .drop_function(name, version, &namespace_path)
+                .await
+                .infer_error()
         })
     }
 
-    pub fn get_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
+    #[pyo3(signature = (name, version, namespace_path=None))]
+    pub fn drop_function_async(
+        self_: PyRef<'_, Self>,
+        name: String,
+        version: 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 description = inner.get_job(&job_id).await.infer_error()?;
-            Ok(description.map(crate::job::JobDescription::from))
+            inner
+                .drop_function_async(name, version, &namespace_path)
+                .await
+                .infer_error()
+                .map(|(dropped, job)| (dropped, crate::job::Job::new(job)))
         })
     }
 
+    #[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, *, page_token=None, page_limit=None))]
+    pub fn list_secrets(
+        &self,
+        namespace_path: Option<Vec<String>>,
+        page_token: Option<String>,
+        page_limit: Option<u32>,
+    ) -> PyResult<crate::listing::NameListing> {
+        Ok(crate::listing::NameListing::new(
+            self.get_inner()?.list_secrets(
+                &namespace_path.unwrap_or_default(),
+                crate::listing::options(page_token, page_limit),
+            ),
+        ))
+    }
+
+    #[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))
+        })
+    }
+
+    #[pyo3(signature = (name, query, namespace_path=None))]
+    pub fn create_view(
+        self_: PyRef<'_, Self>,
+        name: String,
+        query: 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 view = inner
+                .create_view(name, query, &namespace_path)
+                .await
+                .infer_error()?;
+            view_description_to_py(view)
+        })
+    }
+
+    #[pyo3(signature = (name, namespace_path=None))]
+    pub fn describe_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 {
+            let view = inner
+                .describe_view(name, &namespace_path)
+                .await
+                .infer_error()?;
+            view_description_to_py(view)
+        })
+    }
+
+    #[pyo3(signature = (name, namespace_path=None))]
+    pub fn drop_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_view(name, &namespace_path).await.infer_error()
+        })
+    }
+
+    #[pyo3(signature = (name, namespace_path=None))]
+    pub fn drop_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_view_async(name, &namespace_path)
+                .await
+                .infer_error()
+                .map(crate::job::Job::new)
+        })
+    }
+
+    #[pyo3(signature = (namespace_path=None, *, page_token=None, page_limit=None))]
+    pub fn list_views(
+        &self,
+        namespace_path: Option<Vec<String>>,
+        page_token: Option<String>,
+        page_limit: Option<u32>,
+    ) -> PyResult<crate::listing::NameListing> {
+        Ok(crate::listing::NameListing::new(
+            self.get_inner()?.list_views(
+                &namespace_path.unwrap_or_default(),
+                crate::listing::options(page_token, page_limit),
+            ),
+        ))
+    }
+
+    #[pyo3(signature = (*, page_token=None, page_limit=None))]
+    pub fn list_jobs(
+        &self,
+        page_token: Option<String>,
+        page_limit: Option<u32>,
+    ) -> PyResult<crate::listing::JobListing> {
+        Ok(crate::listing::JobListing::new(
+            self.get_inner()?
+                .list_jobs(crate::listing::options(page_token, page_limit)),
+        ))
+    }
+
     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 {
@@ -655,27 +1016,33 @@ impl Connection {
         })
     }
 
-    #[pyo3(signature = (job_id=None))]
-    pub fn job_history(
-        self_: PyRef<'_, Self>,
-        job_id: Option<String>,
-    ) -> PyResult<Bound<'_, PyAny>> {
+    pub fn pause_job(self_: PyRef<'_, Self>, job_id: 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())
+            let status = inner.pause_job(&job_id).await.infer_error()?;
+            Ok(match status {
+                lancedb::database::PauseJobStatus::Pausing => "pausing",
+                lancedb::database::PauseJobStatus::AlreadyPaused => "already_paused",
+                lancedb::database::PauseJobStatus::Committing => "committing",
+            })
+        })
+    }
+
+    pub fn resume_job(self_: PyRef<'_, Self>, job_id: String) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.get_inner()?.clone();
+        future_into_py(self_.py(), async move {
+            let status = inner.resume_job(&job_id).await.infer_error()?;
+            Ok(match status {
+                lancedb::database::ResumeJobStatus::Resumed => "resumed",
+                lancedb::database::ResumeJobStatus::StillPausing => "still_pausing",
+                lancedb::database::ResumeJobStatus::NotPaused => "not_paused",
             })
         })
     }
 }
 
 #[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 +1050,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 +1070,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

@@ -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/index.rs

@@ -54,8 +54,6 @@ pub fn extract_index_params(source: &Option<Bound<'_, PyAny>>) -> PyResult<Lance
                     .with_position(params.with_position)
                     .lower_case(params.lower_case)
                     .max_token_length(params.max_token_length)
-                    .remove_stop_words(params.remove_stop_words)
-                    .stem(params.stem)
                     .ascii_folding(params.ascii_folding)
                     .ngram_min_length(params.ngram_min_length)
                     .ngram_max_length(params.ngram_max_length)
@@ -65,6 +63,12 @@ pub fn extract_index_params(source: &Option<Bound<'_, PyAny>>) -> PyResult<Lance
                         DocumentGranularity::try_from(params.document_granularity.as_str())
                             .map_err(|err| PyValueError::new_err(err.to_string()))?,
                     );
+                if let Some(stem) = params.stem {
+                    inner_opts = inner_opts.stem(stem);
+                }
+                if let Some(remove_stop_words) = params.remove_stop_words {
+                    inner_opts = inner_opts.remove_stop_words(remove_stop_words);
+                }
                 if let Some(memory_limit) = params.memory_limit {
                     inner_opts = inner_opts.memory_limit_mb(memory_limit);
                 }
@@ -215,8 +219,8 @@ struct FtsParams {
     language: String,
     max_token_length: Option<usize>,
     lower_case: bool,
-    stem: bool,
-    remove_stop_words: bool,
+    stem: Option<bool>,
+    remove_stop_words: Option<bool>,
     custom_stop_words: Option<Vec<String>>,
     ascii_folding: bool,
     ngram_min_length: u32,

python/src/job.rs

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

@@ -8,25 +8,28 @@ 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};
 use session::Session;
 use table::{
     AddColumnsResult, AddResult, AlterColumnsResult, DeleteResult, DropColumnsResult, FtsToken,
-    LsmWriteSpec, MergeResult, PyBlobFile, RefreshColumnResult, RefreshMaterializedViewResult,
-    Table, UpdateFieldMetadataResult, UpdateResult,
+    FunctionErrorFragment, FunctionErrorRecord, FunctionErrors, LsmWriteSpec, MergeResult,
+    PyBlobFile, RefreshColumnResult, RefreshMaterializedViewResult, Table,
+    UpdateFieldMetadataResult, UpdateResult,
 };
 
 pub mod arrow;
+pub mod catalog;
 pub mod connection;
 pub mod error;
 pub mod expr;
 pub mod header;
 pub mod index;
 pub mod job;
+mod listing;
 pub mod namespace;
 pub mod oauth;
 pub mod otel;
@@ -34,22 +37,51 @@ 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_class::<catalog::DatabaseNames>()?;
+    m.add_class::<listing::NameListing>()?;
+    m.add_class::<listing::FunctionListing>()?;
+    m.add_class::<listing::JobListing>()?;
+
+    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>()?;
@@ -59,6 +91,9 @@ pub fn _lancedb(_py: Python, m: &Bound<'_, PyModule>) -> PyResult<()> {
     m.add_class::<RecordBatchStream>()?;
     m.add_class::<AddColumnsResult>()?;
     m.add_class::<RefreshColumnResult>()?;
+    m.add_class::<FunctionErrors>()?;
+    m.add_class::<FunctionErrorRecord>()?;
+    m.add_class::<FunctionErrorFragment>()?;
     m.add_class::<RefreshMaterializedViewResult>()?;
     m.add_class::<AlterColumnsResult>()?;
     m.add_class::<UpdateFieldMetadataResult>()?;
@@ -92,5 +127,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/listing.rs

@@ -0,0 +1,131 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: Copyright The LanceDB Authors
+
+use crate::error::PythonErrorExt;
+use crate::runtime::future_into_py;
+use futures::{StreamExt, future::poll_fn};
+use lancedb::listing::{Listing, ListingOptions};
+use pyo3::exceptions::PyStopAsyncIteration;
+use pyo3::{Bound, PyAny, PyRef, PyResult, Python, pyclass, pymethods};
+use std::sync::{Arc, Mutex};
+
+pub fn options(page_token: Option<String>, page_limit: Option<u32>) -> ListingOptions {
+    let mut options = ListingOptions::default();
+    options.page_token = page_token;
+    options.page_limit = page_limit;
+    options
+}
+
+#[pyclass]
+pub struct NameListing {
+    inner: Arc<Mutex<Listing<String>>>,
+    next_lock: Arc<tokio::sync::Mutex<()>>,
+}
+impl NameListing {
+    pub fn new(inner: Listing<String>) -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(inner)),
+            next_lock: Arc::new(tokio::sync::Mutex::new(())),
+        }
+    }
+}
+#[pymethods]
+impl NameListing {
+    fn num_page_results(&self, py: Python<'_>) -> usize {
+        py.detach(|| self.inner.lock().unwrap().num_page_results())
+    }
+    fn page_token(&self, py: Python<'_>) -> Option<String> {
+        py.detach(|| self.inner.lock().unwrap().page_token().map(str::to_owned))
+    }
+    fn __aiter__(self_: PyRef<'_, Self>) -> PyRef<'_, Self> {
+        self_
+    }
+    fn __anext__(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner.clone();
+        let next_lock = self_.next_lock.clone();
+        future_into_py(self_.py(), async move {
+            let _guard = next_lock.lock().await;
+            let value = poll_fn(|cx| inner.lock().unwrap().poll_next_unpin(cx))
+                .await
+                .ok_or_else(|| PyStopAsyncIteration::new_err(""))?
+                .infer_error()?;
+            Ok(value)
+        })
+    }
+}
+
+#[pyclass]
+pub struct FunctionListing {
+    inner: Arc<Mutex<Listing<lancedb::function::FunctionVersion>>>,
+    next_lock: Arc<tokio::sync::Mutex<()>>,
+}
+impl FunctionListing {
+    pub fn new(inner: Listing<lancedb::function::FunctionVersion>) -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(inner)),
+            next_lock: Arc::new(tokio::sync::Mutex::new(())),
+        }
+    }
+}
+#[pymethods]
+impl FunctionListing {
+    fn num_page_results(&self, py: Python<'_>) -> usize {
+        py.detach(|| self.inner.lock().unwrap().num_page_results())
+    }
+    fn page_token(&self, py: Python<'_>) -> Option<String> {
+        py.detach(|| self.inner.lock().unwrap().page_token().map(str::to_owned))
+    }
+    fn __aiter__(self_: PyRef<'_, Self>) -> PyRef<'_, Self> {
+        self_
+    }
+    fn __anext__(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner.clone();
+        let next_lock = self_.next_lock.clone();
+        future_into_py(self_.py(), async move {
+            let _guard = next_lock.lock().await;
+            let value = poll_fn(|cx| inner.lock().unwrap().poll_next_unpin(cx))
+                .await
+                .ok_or_else(|| PyStopAsyncIteration::new_err(""))?
+                .infer_error()?;
+            value.to_canonical_json().infer_error()
+        })
+    }
+}
+
+#[pyclass]
+pub struct JobListing {
+    inner: Arc<Mutex<Listing<lancedb::database::JobInfo>>>,
+    next_lock: Arc<tokio::sync::Mutex<()>>,
+}
+impl JobListing {
+    pub fn new(inner: Listing<lancedb::database::JobInfo>) -> Self {
+        Self {
+            inner: Arc::new(Mutex::new(inner)),
+            next_lock: Arc::new(tokio::sync::Mutex::new(())),
+        }
+    }
+}
+#[pymethods]
+impl JobListing {
+    fn num_page_results(&self, py: Python<'_>) -> usize {
+        py.detach(|| self.inner.lock().unwrap().num_page_results())
+    }
+    fn page_token(&self, py: Python<'_>) -> Option<String> {
+        py.detach(|| self.inner.lock().unwrap().page_token().map(str::to_owned))
+    }
+    fn __aiter__(self_: PyRef<'_, Self>) -> PyRef<'_, Self> {
+        self_
+    }
+    fn __anext__(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner.clone();
+        let next_lock = self_.next_lock.clone();
+        future_into_py(self_.py(), async move {
+            let _guard = next_lock.lock().await;
+            let value = poll_fn(|cx| inner.lock().unwrap().poll_next_unpin(cx))
+                .await
+                .ok_or_else(|| PyStopAsyncIteration::new_err(""))?
+                .infer_error()?;
+            Ok(crate::job::JobInfo::from(value))
+        })
+    }
+}

python/src/oauth.rs

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

@@ -71,7 +71,7 @@ impl<'a, 'py> FromPyObject<'a, 'py> for PyLanceDB<FtsQuery> {
         match class_name(&ob)?.as_str() {
             "MatchQuery" => {
                 let query = ob.getattr("query")?.extract()?;
-                let column = ob.getattr("column")?.extract()?;
+                let column = ob.getattr("column")?.extract::<Option<String>>()?;
                 let boost = ob.getattr("boost")?.extract()?;
                 let fuzziness = ob.getattr("fuzziness")?.extract()?;
                 let max_expansions = ob.getattr("max_expansions")?.extract()?;
@@ -88,7 +88,7 @@ impl<'a, 'py> FromPyObject<'a, 'py> for PyLanceDB<FtsQuery> {
 
                 let mut query =
                     MatchQuery::new(query)
-                        .with_column(Some(column))
+                        .with_column(column)
                         .with_boost(boost)
                         .with_fuzziness(fuzziness)
                         .with_max_expansions(max_expansions)
@@ -180,6 +180,9 @@ impl<'py> IntoPyObject<'py> for PyLanceDB<FtsQuery> {
             .expect("Failed to import namespace");
 
         match self.0 {
+            FtsQuery::CombinedFields(_) => Err(PyValueError::new_err(
+                "Combined-fields FTS queries are not yet supported in Python",
+            )),
             FtsQuery::Match(query) => {
                 let kwargs = PyDict::new(py);
                 kwargs.set_item("boost", query.boost)?;
@@ -196,7 +199,7 @@ impl<'py> IntoPyObject<'py> for PyLanceDB<FtsQuery> {
                 }
                 namespace
                     .getattr(intern!(py, "MatchQuery"))?
-                    .call((query.terms, query.column.unwrap()), Some(&kwargs))
+                    .call((query.terms, query.column), Some(&kwargs))
             }
             FtsQuery::Phrase(query) => {
                 let kwargs = PyDict::new(py);
@@ -323,6 +326,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,
@@ -332,10 +336,10 @@ pub struct PyQueryRequest {
     pub use_lsm: Option<bool>,
     pub column: Option<String>,
     pub query_vector: Option<PyQueryVectors>,
+    pub nprobes: Option<usize>,
     pub minimum_nprobes: Option<usize>,
-    // None means user did not set it and default shoud be used (currenty 20)
-    // Some(0) means user set it to None and there is no limit
     pub maximum_nprobes: Option<usize>,
+    pub search_effort: Option<f64>,
     pub lower_bound: Option<f32>,
     pub upper_bound: Option<f32>,
     pub ef: Option<usize>,
@@ -353,6 +357,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
@@ -364,8 +369,10 @@ impl From<AnyQuery> for PyQueryRequest {
                 use_lsm: query_request.use_lsm,
                 column: None,
                 query_vector: None,
+                nprobes: None,
                 minimum_nprobes: None,
                 maximum_nprobes: None,
+                search_effort: None,
                 lower_bound: None,
                 upper_bound: None,
                 ef: None,
@@ -381,6 +388,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),
@@ -390,11 +398,10 @@ impl From<AnyQuery> for PyQueryRequest {
                 use_lsm: vector_query.base.use_lsm,
                 column: vector_query.column,
                 query_vector: Some(PyQueryVectors(vector_query.query_vector)),
-                minimum_nprobes: Some(vector_query.minimum_nprobes),
-                maximum_nprobes: match vector_query.maximum_nprobes {
-                    None => Some(0),
-                    Some(value) => Some(value),
-                },
+                nprobes: vector_query.nprobes,
+                minimum_nprobes: vector_query.minimum_nprobes,
+                maximum_nprobes: vector_query.maximum_nprobes,
+                search_effort: vector_query.search_effort,
                 lower_bound: vector_query.lower_bound,
                 upper_bound: vector_query.upper_bound,
                 ef: vector_query.ef,
@@ -989,8 +996,18 @@ impl VectorQuery {
         self.inner = self.inner.clone().refine_factor(refine_factor);
     }
 
-    pub fn nprobes(&mut self, nprobe: u32) {
-        self.inner = self.inner.clone().nprobes(nprobe as usize);
+    pub fn nprobes(&mut self, nprobe: u32) -> PyResult<()> {
+        self.inner = self.inner.clone().nprobes(nprobe as usize).infer_error()?;
+        Ok(())
+    }
+
+    pub fn search_effort(&mut self, search_effort: Option<f64>) -> PyResult<()> {
+        self.inner = self
+            .inner
+            .clone()
+            .search_effort(search_effort)
+            .infer_error()?;
+        Ok(())
     }
 
     pub fn minimum_nprobes(&mut self, minimum_nprobes: u32) -> PyResult<()> {
@@ -1182,8 +1199,20 @@ impl HybridQuery {
         self.inner_vec.refine_factor(refine_factor);
     }
 
-    pub fn nprobes(&mut self, nprobe: u32) {
-        self.inner_vec.nprobes(nprobe);
+    pub fn nprobes(&mut self, nprobe: u32) -> PyResult<()> {
+        self.inner_vec.nprobes(nprobe)
+    }
+
+    pub fn search_effort(&mut self, search_effort: Option<f64>) -> PyResult<()> {
+        self.inner_vec.search_effort(search_effort)
+    }
+
+    pub fn minimum_nprobes(&mut self, minimum_nprobes: u32) -> PyResult<()> {
+        self.inner_vec.minimum_nprobes(minimum_nprobes)
+    }
+
+    pub fn maximum_nprobes(&mut self, maximum_nprobes: u32) -> PyResult<()> {
+        self.inner_vec.maximum_nprobes(maximum_nprobes)
     }
 
     pub fn ef(&mut self, ef: u32) {
@@ -1230,6 +1259,10 @@ impl HybridQuery {
         req.query_vector = vec_req.query_vector;
         req.column = vec_req.column;
         req.distance_type = vec_req.distance_type;
+        req.nprobes = vec_req.nprobes;
+        req.minimum_nprobes = vec_req.minimum_nprobes;
+        req.maximum_nprobes = vec_req.maximum_nprobes;
+        req.search_effort = vec_req.search_effort;
         req.ef = vec_req.ef;
         req.refine_factor = vec_req.refine_factor;
         req.lower_bound = vec_req.lower_bound;

python/src/runtime.rs

@@ -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,98 @@ 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.
+/// Run a detached task, including when the caller is already on a runtime worker.
+/// Keep it visible to [`shutdown`] until it finishes.
+pub fn spawn_background<F>(fut: F)
+where
+    F: Future<Output = ()> + Send + 'static,
+{
+    let guard = OutstandingGuard::new();
+    let task = async move {
+        let _guard = guard;
+        fut.await;
+    };
+    // Detaching is the point: dropping the `JoinHandle` leaves the task running, and the
+    // `OutstandingGuard` it carries is what keeps it visible to `shutdown`. Written as `drop`
+    // rather than `let _ =`, which reads as discarding an unpolled future.
+    match runtime::Handle::try_current() {
+        Ok(handle) => drop(handle.spawn(task)),
+        Err(_) => drop(get_runtime().spawn(task)),
+    }
+}
+
+/// 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 +315,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 +342,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 +386,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

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

@@ -1,8 +1,14 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::HashMap,
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
+};
 
-use crate::runtime::{block_on, future_into_py};
+use crate::runtime::{block_on, future_into_py, spawn_background};
 use crate::{
     connection::Connection,
     error::PythonErrorExt,
@@ -261,7 +267,7 @@ fn fmt_maintained(maintained: &Option<Vec<String>>) -> String {
 /// Constructed via the `bucket(...)`, `identity(...)`, or `unsharded()`
 /// classmethods, then optionally chain `with_maintained_indexes(...)` and
 /// `with_writer_config_defaults(...)`. A fresh spec maintains every index the
-/// MemWAL supports, resolved on install.
+/// table has, including ones created later.
 #[pyclass(module = "lancedb._lancedb", from_py_object)]
 #[derive(Clone, Debug)]
 pub struct LsmWriteSpec {
@@ -301,9 +307,9 @@ impl LsmWriteSpec {
         }
     }
 
-    /// Set which indexes the MemWAL maintains. `None` (the default)
-    /// resolves every supported index on install; a list is verbatim,
-    /// and an empty list maintains nothing.
+    /// Set which indexes the MemWAL maintains. `None` (the default) is
+    /// every index the table has, including ones created later; a list is
+    /// verbatim, and an empty list maintains nothing.
     #[pyo3(signature = (indexes))]
     pub fn with_maintained_indexes(&self, indexes: Option<Vec<String>>) -> Self {
         Self {
@@ -441,6 +447,117 @@ impl From<lancedb::table::RefreshColumnResult> for RefreshColumnResult {
     }
 }
 
+/// One row a Function refresh skipped, as the server recorded it.
+#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)]
+#[derive(Clone, Debug)]
+pub struct FunctionErrorRecord {
+    pub job_id: String,
+    pub fragment_id: u64,
+    pub row_offset: Option<u32>,
+    pub column: String,
+    pub function: String,
+    pub function_version: String,
+    pub table_version: u64,
+    pub error_type: String,
+    pub error_message: String,
+    pub created_at_millis: i64,
+}
+
+#[pymethods]
+impl FunctionErrorRecord {
+    pub fn __repr__(&self) -> String {
+        format!(
+            "FunctionErrorRecord(job_id={:?}, fragment_id={}, row_offset={:?}, column={:?}, \
+             error_type={:?}, error_message={:?})",
+            self.job_id,
+            self.fragment_id,
+            self.row_offset,
+            self.column,
+            self.error_type,
+            self.error_message
+        )
+    }
+}
+
+impl From<lancedb::function::FunctionErrorRecord> for FunctionErrorRecord {
+    fn from(record: lancedb::function::FunctionErrorRecord) -> Self {
+        Self {
+            job_id: record.job_id,
+            fragment_id: record.fragment_id,
+            row_offset: record.row_offset,
+            column: record.column,
+            function: record.function,
+            function_version: record.function_version,
+            table_version: record.table_version,
+            error_type: record.error_type,
+            error_message: record.error_message,
+            created_at_millis: record.created_at_millis,
+        }
+    }
+}
+
+/// A fragment whose per-row error detail was capped.
+#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)]
+#[derive(Clone, Debug)]
+pub struct FunctionErrorFragment {
+    pub job_id: String,
+    pub fragment_id: u64,
+    pub rows_skipped: u64,
+    pub rows_recorded: u64,
+}
+
+#[pymethods]
+impl FunctionErrorFragment {
+    pub fn __repr__(&self) -> String {
+        format!(
+            "FunctionErrorFragment(job_id={:?}, fragment_id={}, rows_skipped={}, rows_recorded={})",
+            self.job_id, self.fragment_id, self.rows_skipped, self.rows_recorded
+        )
+    }
+}
+
+impl From<lancedb::function::FunctionErrorFragment> for FunctionErrorFragment {
+    fn from(fragment: lancedb::function::FunctionErrorFragment) -> Self {
+        Self {
+            job_id: fragment.job_id,
+            fragment_id: fragment.fragment_id,
+            rows_skipped: fragment.rows_skipped,
+            rows_recorded: fragment.rows_recorded,
+        }
+    }
+}
+
+/// A table's per-row Function errors.
+#[pyclass(module = "lancedb._lancedb", get_all, from_py_object)]
+#[derive(Clone, Debug)]
+pub struct FunctionErrors {
+    pub records: Vec<FunctionErrorRecord>,
+    pub fragments: Vec<FunctionErrorFragment>,
+    pub truncated: bool,
+}
+
+#[pymethods]
+impl FunctionErrors {
+    pub fn __repr__(&self) -> String {
+        format!(
+            "FunctionErrors(records={}, fragments={}, truncated={})",
+            self.records.len(),
+            self.fragments.len(),
+            self.truncated
+        )
+    }
+}
+
+impl From<lancedb::function::FunctionErrors> for FunctionErrors {
+    fn from(errors: lancedb::function::FunctionErrors) -> Self {
+        Self {
+            records: errors.records.into_iter().map(Into::into).collect(),
+            fragments: errors.fragments.into_iter().map(Into::into).collect(),
+            truncated: errors.truncated,
+        }
+    }
+}
+
 #[pyclass(get_all, from_py_object)]
 #[derive(Clone, Debug)]
 pub struct RefreshMaterializedViewResult {
@@ -452,6 +569,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={})",
@@ -464,8 +592,6 @@ impl From<lancedb::RefreshMaterializedViewResult> for RefreshMaterializedViewRes
     fn from(result: lancedb::RefreshMaterializedViewResult) -> Self {
         let mode = match result.mode {
             lancedb::RefreshMode::Rebuild => "rebuild",
-            lancedb::RefreshMode::Incremental => "incremental",
-            lancedb::RefreshMode::NoOp => "no_op",
         };
         Self {
             mode: mode.to_string(),
@@ -558,11 +684,23 @@ impl From<lancedb::table::DropColumnsResult> for DropColumnsResult {
 #[pyclass(name = "BlobFile")]
 pub struct PyBlobFile {
     inner: Arc<BlobFile>,
+    closed: AtomicBool,
+}
+
+impl PyBlobFile {
+    fn ensure_open(&self) -> PyResult<()> {
+        if self.closed.load(Ordering::Acquire) {
+            Err(PyRuntimeError::new_err("blob file is already closed"))
+        } else {
+            Ok(())
+        }
+    }
 }
 
 #[pymethods]
 impl PyBlobFile {
     fn read_bytes(self_: PyRef<'_, Self>) -> PyResult<Py<PyBytes>> {
+        self_.ensure_open()?;
         let inner = self_.inner.clone();
         let py = self_.py();
         let bytes = py
@@ -572,6 +710,7 @@ impl PyBlobFile {
     }
 
     pub fn read(self_: PyRef<'_, Self>) -> PyResult<Bound<'_, PyAny>> {
+        self_.ensure_open()?;
         let inner = self_.inner.clone();
         future_into_py(self_.py(), async move {
             let bytes = inner
@@ -583,7 +722,20 @@ impl PyBlobFile {
     }
 
     fn close(self_: PyRef<'_, Self>) -> PyResult<()> {
+        if self_.closed.swap(true, Ordering::AcqRel) {
+            return Ok(());
+        }
         let inner = self_.inner.clone();
+        if tokio::runtime::Handle::try_current().is_ok() {
+            // IOBase.__del__ can call close while cyclic GC runs on a worker.
+            // The status changes immediately; release Lance's async state there.
+            spawn_background(async move {
+                if let Err(error) = inner.close().await {
+                    log::warn!("blob close failed: {error}");
+                }
+            });
+            return Ok(());
+        }
         self_
             .py()
             .detach(move || block_on(async move { inner.close().await }))
@@ -591,13 +743,11 @@ impl PyBlobFile {
     }
 
     fn is_closed(self_: PyRef<'_, Self>) -> bool {
-        let inner = self_.inner.clone();
-        self_
-            .py()
-            .detach(move || block_on(async move { inner.is_closed().await }))
+        self_.closed.load(Ordering::Acquire)
     }
 
     fn seek(self_: PyRef<'_, Self>, position: u64) -> PyResult<()> {
+        self_.ensure_open()?;
         let inner = self_.inner.clone();
         self_
             .py()
@@ -606,6 +756,7 @@ impl PyBlobFile {
     }
 
     fn tell(self_: PyRef<'_, Self>) -> PyResult<u64> {
+        self_.ensure_open()?;
         let inner = self_.inner.clone();
         self_
             .py()
@@ -619,6 +770,7 @@ impl PyBlobFile {
 
     /// Read a blob-local byte range without moving the cursor.
     fn read_range(self_: PyRef<'_, Self>, offset: u64, length: usize) -> PyResult<Py<PyBytes>> {
+        self_.ensure_open()?;
         let end = offset
             .checked_add(length as u64)
             .ok_or_else(|| PyValueError::new_err("offset + length overflowed"))?;
@@ -631,6 +783,7 @@ impl PyBlobFile {
     }
 
     fn read_up_to(self_: PyRef<'_, Self>, length: usize) -> PyResult<Py<PyBytes>> {
+        self_.ensure_open()?;
         let inner = self_.inner.clone();
         let py = self_.py();
         let bytes = py
@@ -670,8 +823,8 @@ impl From<LanceDbFtsToken> for FtsToken {
     language = "English".to_string(),
     max_token_length = Some(40),
     lower_case = true,
-    stem = true,
-    remove_stop_words = true,
+    stem = None,
+    remove_stop_words = None,
     custom_stop_words = None,
     ascii_folding = true,
     ngram_min_length = 3,
@@ -685,15 +838,15 @@ pub fn tokenize(
     language: String,
     max_token_length: Option<u32>,
     lower_case: bool,
-    stem: bool,
-    remove_stop_words: bool,
+    stem: Option<bool>,
+    remove_stop_words: Option<bool>,
     custom_stop_words: Option<Vec<String>>,
     ascii_folding: bool,
     ngram_min_length: u32,
     ngram_max_length: u32,
     prefix_only: bool,
 ) -> PyResult<Vec<FtsToken>> {
-    let params = FtsIndexBuilder::default()
+    let mut params = FtsIndexBuilder::default()
         .base_tokenizer(base_tokenizer)
         .language(&language)
         .map_err(|_| {
@@ -704,13 +857,17 @@ pub fn tokenize(
         })?
         .max_token_length(max_token_length.map(|value| value as usize))
         .lower_case(lower_case)
-        .stem(stem)
-        .remove_stop_words(remove_stop_words)
         .ascii_folding(ascii_folding)
         .ngram_min_length(ngram_min_length)
         .ngram_max_length(ngram_max_length)
         .ngram_prefix_only(prefix_only)
         .custom_stop_words(custom_stop_words);
+    if let Some(stem) = stem {
+        params = params.stem(stem);
+    }
+    if let Some(remove_stop_words) = remove_stop_words {
+        params = params.remove_stop_words(remove_stop_words);
+    }
     let tokens = lancedb_tokenize(&query, &params).infer_error()?;
     Ok(tokens.into_iter().map(FtsToken::from).collect())
 }
@@ -760,6 +917,11 @@ impl Table {
         self.inner.is_some()
     }
 
+    /// Whether this table has a local Lance dataset that the client may open.
+    pub fn _is_native(&self) -> PyResult<bool> {
+        Ok(self.inner_ref()?.as_native().is_some())
+    }
+
     /// Closes the table, releasing any resources associated with it.
     pub fn close(&mut self) {
         self.inner.take();
@@ -1113,6 +1275,7 @@ impl Table {
                 let dict = PyDict::new(py);
                 dict.set_item("total_bytes", stats.total_bytes)?;
                 dict.set_item("num_rows", stats.num_rows)?;
+                dict.set_item("num_deleted_rows", stats.num_deleted_rows)?;
                 dict.set_item("num_indices", stats.num_indices)?;
 
                 let fragment_stats = PyDict::new(py);
@@ -1333,6 +1496,7 @@ impl Table {
                 .map(|handle| {
                     handle.map(|file| PyBlobFile {
                         inner: Arc::new(file),
+                        closed: AtomicBool::new(false),
                     })
                 })
                 .collect::<Vec<_>>())
@@ -1481,6 +1645,21 @@ impl Table {
         })
     }
 
+    /// Whether a hybrid query on this table has already been refused `_rowid`.
+    /// Learned from a previous refusal, never probed, so this is free.
+    ///
+    /// WAL-PK-FUSION: delete this and `note_hybrid_pk_fusion`.
+    pub fn hybrid_pk_fusion_learned(self_: PyRef<'_, Self>) -> PyResult<bool> {
+        Ok(self_.inner_ref()?.base_table().hybrid_pk_fusion_learned())
+    }
+
+    /// Remember that this table refused `_rowid`, so later hybrid queries skip
+    /// straight to the primary-key fusion.
+    pub fn note_hybrid_pk_fusion(self_: PyRef<'_, Self>) -> PyResult<()> {
+        self_.inner_ref()?.base_table().note_hybrid_pk_fusion();
+        Ok(())
+    }
+
     /// Converge the table's LSM write path into its base table.
     ///
     /// Best-effort: with writes flowing, new rows may land after the last
@@ -1538,7 +1717,10 @@ impl Table {
         future_into_py(self_.py(), async move {
             inner
                 .as_native()
-                .ok_or_else(|| PyValueError::new_err("This cannot be run on a remote table"))?
+                .ok_or_else(|| lancedb::Error::NotSupported {
+                    message: "uses_v2_manifest_paths is not supported for remote tables.".into(),
+                })
+                .infer_error()?
                 .uses_v2_manifest_paths()
                 .await
                 .infer_error()
@@ -1550,7 +1732,10 @@ impl Table {
         future_into_py(self_.py(), async move {
             inner
                 .as_native()
-                .ok_or_else(|| PyValueError::new_err("This cannot be run on a remote table"))?
+                .ok_or_else(|| lancedb::Error::NotSupported {
+                    message: "migrate_manifest_paths_v2 is not supported for remote tables.".into(),
+                })
+                .infer_error()?
                 .migrate_manifest_paths_v2()
                 .await
                 .infer_error()
@@ -1627,10 +1812,28 @@ impl Table {
         })
     }
 
-    #[pyo3(signature = (full=false, source_version=None))]
+    #[pyo3(signature = (job_id=None, column=None, limit=None))]
+    pub fn function_errors(
+        self_: PyRef<'_, Self>,
+        job_id: Option<String>,
+        column: Option<String>,
+        limit: Option<usize>,
+    ) -> PyResult<Bound<'_, PyAny>> {
+        let inner = self_.inner_ref()?.clone();
+        let request = lancedb::function::FunctionErrorsRequest {
+            job_id,
+            column,
+            limit,
+        };
+        future_into_py(self_.py(), async move {
+            let errors = inner.function_errors(request).await.infer_error()?;
+            Ok(FunctionErrors::from(errors))
+        })
+    }
+
+    #[pyo3(signature = (source_version=None))]
     pub fn refresh_materialized_view(
         self_: PyRef<'_, Self>,
-        full: bool,
         source_version: Option<u64>,
     ) -> PyResult<Bound<'_, PyAny>> {
         let inner = self_.inner_ref()?.clone();
@@ -1638,7 +1841,7 @@ impl Table {
             let view = lancedb::MaterializedView::from_table(inner)
                 .await
                 .infer_error()?;
-            let mut builder = view.refresh().full(full);
+            let mut builder = view.refresh();
             if let Some(version) = source_version {
                 builder = builder.source_version(version);
             }
@@ -1647,6 +1850,37 @@ impl Table {
         })
     }
 
+    #[pyo3(signature = (source_version=None))]
+    pub fn refresh_materialized_view_async(
+        self_: PyRef<'_, Self>,
+        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();
+            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()?;
+            Ok(lancedb::materialized_view::definition_metadata_from_sql(
+                view.definition_sql(),
+            ))
+        })
+    }
+
     pub fn add_columns_with_schema(
         self_: PyRef<'_, Self>,
         schema: PyArrowType<Schema>,
@@ -1673,6 +1907,17 @@ impl Table {
         let alterations = alterations
             .iter()
             .map(|alteration| {
+                for key in alteration.keys().iter() {
+                    let key = key.extract::<String>()?;
+                    if !matches!(
+                        key.as_str(),
+                        "path" | "rename" | "name" | "nullable" | "data_type"
+                    ) {
+                        return Err(PyValueError::new_err(format!(
+                            "Unknown column alteration key '{key}'"
+                        )));
+                    }
+                }
                 let path = alteration
                     .get_item("path")?
                     .ok_or_else(|| PyValueError::new_err("Missing path"))?