diff --git a/docs/web/management/function.mdx b/docs/web/management/function.mdx index 4709264c7..c8fa0626c 100644 --- a/docs/web/management/function.mdx +++ b/docs/web/management/function.mdx @@ -530,6 +530,14 @@ returns as a [background job](/basics/jobs#teardown). handle. Waiting confirms the revision's catalog removal; physical cleanup can continue afterward. +When a deployment reports dependency warnings for a boolean drop, +`db.drop_function` emits each warning as a `UserWarning`. This also applies to +`await db.drop_function(...)` on an async connection. The warnings are emitted +after catalog removal, and standard Python warnings filters apply. If warnings +are configured as errors, the call can raise after the revision has been +removed; the drop is not rolled back. Servers whose DROP response omits the +warning field remain silent. + A deployment refuses the drop unless it also allows changes to a function's lifecycle, which is a separate setting, off by default even where Functions are enabled. diff --git a/python/python/lancedb/_lancedb.pyi b/python/python/lancedb/_lancedb.pyi index d1cee5091..f0f6027ad 100644 --- a/python/python/lancedb/_lancedb.pyi +++ b/python/python/lancedb/_lancedb.pyi @@ -201,6 +201,9 @@ class Connection(object): async def drop_function( self, name: str, version: str, namespace_path: Optional[List[str]] = None ) -> bool: ... + async def _drop_function_with_warnings( + self, name: str, version: str, namespace_path: Optional[List[str]] = None + ) -> Tuple[bool, list[str]]: ... async def drop_function_async( self, name: str, version: str, namespace_path: Optional[List[str]] = None ) -> Tuple[bool, Job]: ... diff --git a/python/python/lancedb/db.py b/python/python/lancedb/db.py index 9df75a794..3ff657a6c 100644 --- a/python/python/lancedb/db.py +++ b/python/python/lancedb/db.py @@ -11,6 +11,7 @@ from pathlib import Path from dataclasses import asdict import json import sys +import warnings from typing import ( TYPE_CHECKING, Any, @@ -1085,7 +1086,9 @@ class DBConnection(EnforceOverrides): Returns True when the revision was removed and False for a repeated drop or a missing name. A revision never published under an existing name is an error. Dropping the last visible revision releases the name and - schedules physical cleanup in the background. Use + schedules physical cleanup in the background. When the remote server + reports dependent warnings, they are emitted as ``UserWarning`` before + this method returns. Use :meth:`drop_function_async` to access the query job, whose completion confirms catalog removal of the revision. Local connections raise NotImplementedError. @@ -3647,9 +3650,29 @@ class AsyncConnection(object): Returns True when the requested revision was removed and False for a repeated drop or a missing name. A revision never published under an existing name is an error. Only the last visible revision's removal - releases the name and schedules background object cleanup. + releases the name and schedules background object cleanup. When the + remote server reports a warning, it is emitted as ``UserWarning`` + before this method returns, subject to the standard ``warnings`` + filters. A warnings-as-errors setting can therefore raise after the + catalog removal; it does not roll back the drop. Older servers that + omit the warning field remain silent. """ - return await self._inner.drop_function(name, version, namespace_path) + dropped, drop_warnings = await self._drop_function_with_warnings( + name, version, namespace_path + ) + for warning in drop_warnings: + warnings.warn(warning, UserWarning, stacklevel=2) + return dropped + + async def _drop_function_with_warnings( + self, + name: str, + version: str, + namespace_path: Optional[List[str]] = None, + ) -> Tuple[bool, List[str]]: + return await self._inner._drop_function_with_warnings( + name, version, namespace_path + ) async def drop_function_async( self, diff --git a/python/python/lancedb/remote/db.py b/python/python/lancedb/remote/db.py index 655391b1f..2cc6bf056 100644 --- a/python/python/lancedb/remote/db.py +++ b/python/python/lancedb/remote/db.py @@ -1024,11 +1024,14 @@ class RemoteDBConnection(DBConnection): version: str, namespace_path: Optional[List[str]] = None, ) -> bool: - return LOOP.run( - self._conn.drop_function( + dropped, drop_warnings = LOOP.run( + self._conn._drop_function_with_warnings( name, version=version, namespace_path=namespace_path ) ) + for warning in drop_warnings: + warnings.warn(warning, UserWarning, stacklevel=2) + return dropped @override def drop_function_async( diff --git a/python/python/tests/test_first_class_function_slice2.py b/python/python/tests/test_first_class_function_slice2.py index 9b435b256..53b0189f5 100644 --- a/python/python/tests/test_first_class_function_slice2.py +++ b/python/python/tests/test_first_class_function_slice2.py @@ -1972,6 +1972,45 @@ def test_sync_catalog_path_preserves_execution_controls(): assert _assert_execution_control_encoding(version) == definition +def test_sync_drop_function_emits_server_warnings_without_changing_bool_api(): + class _DropConnection: + async def drop_function(self, name, *, version, namespace_path=None): + return True + + async def _drop_function_with_warnings( + self, name, *, version, namespace_path=None + ): + return True, [ + 'materialized view "pinned" references Function "embed" version "1"' + ] + + connection = object.__new__(RemoteDBConnection) + connection._conn = _DropConnection() + + with pytest.warns(UserWarning, match="references Function"): + assert connection.drop_function("embed", version="1") is True + + +@pytest.mark.asyncio +async def test_async_drop_function_emits_server_warnings_without_changing_bool_api(): + class _Inner: + async def _drop_function_with_warnings( + self, name, version, namespace_path=None + ): + return True, [ + 'materialized view "pinned" references Function "embed" version "1"' + ] + + async def drop_function(self, name, version, namespace_path=None): + return True + + connection = object.__new__(AsyncConnection) + connection._inner = _Inner() + + with pytest.warns(UserWarning, match="references Function"): + assert await connection.drop_function("embed", version="1") is True + + @contextlib.contextmanager def _mock_remote_function_catalog(): state = {"requests": [], "version": None} diff --git a/python/src/connection.rs b/python/src/connection.rs index d88991235..bffa167d8 100644 --- a/python/src/connection.rs +++ b/python/src/connection.rs @@ -928,6 +928,23 @@ impl Connection { }) } + #[pyo3(signature = (name, version, namespace_path=None))] + pub fn _drop_function_with_warnings( + self_: PyRef<'_, Self>, + name: String, + version: String, + namespace_path: Option>, + ) -> PyResult> { + let inner = self_.get_inner()?.clone(); + let namespace_path = namespace_path.unwrap_or_default(); + future_into_py(self_.py(), async move { + inner + .drop_function_with_warnings(name, version, &namespace_path) + .await + .infer_error() + }) + } + #[pyo3(signature = (name, version, namespace_path=None))] pub fn drop_function_async( self_: PyRef<'_, Self>, diff --git a/rust/lancedb/src/connection.rs b/rust/lancedb/src/connection.rs index 7d751bdb7..dec5e1359 100644 --- a/rust/lancedb/src/connection.rs +++ b/rust/lancedb/src/connection.rs @@ -738,6 +738,21 @@ impl Connection { .await } + /// Internal bridge for the Python binding to preserve the public bool API + /// while forwarding advisory SQL warnings to Python's existing warning + /// mechanism. + #[doc(hidden)] + pub async fn drop_function_with_warnings( + &self, + name: impl AsRef, + version: impl AsRef, + namespace_path: &[String], + ) -> Result<(bool, Vec)> { + self.internal + .drop_function_with_warnings(name.as_ref(), version.as_ref(), namespace_path) + .await + } + /// Drop one Function revision and return its SQL query handle. /// /// The boolean reports removal of the requested revision from the catalog. diff --git a/rust/lancedb/src/database.rs b/rust/lancedb/src/database.rs index daa7f3f29..44d038521 100644 --- a/rust/lancedb/src/database.rs +++ b/rust/lancedb/src/database.rs @@ -581,6 +581,18 @@ pub trait Database: let dropped = self.drop_function(name, version, namespace_path).await?; Ok((dropped, crate::job::Job::new_done())) } + /// Internal bridge for the Python binding to preserve the public bool API + /// while making advisory SQL warnings available to the wrapper. + #[doc(hidden)] + async fn drop_function_with_warnings( + &self, + name: &str, + version: &str, + namespace_path: &[String], + ) -> Result<(bool, Vec)> { + let dropped = self.drop_function(name, version, namespace_path).await?; + Ok((dropped, Vec::new())) + } /// Create a database-owned storage integration in a namespace. async fn create_storage_integration( &self, diff --git a/rust/lancedb/src/remote/db.rs b/rust/lancedb/src/remote/db.rs index 9af99dced..c33c7267d 100644 --- a/rust/lancedb/src/remote/db.rs +++ b/rust/lancedb/src/remote/db.rs @@ -932,6 +932,23 @@ impl Database for RemoteDatabase { .await } + #[doc(hidden)] + async fn drop_function_with_warnings( + &self, + name: &str, + version: &str, + namespace_path: &[String], + ) -> Result<(bool, Vec)> { + validate_function_name(name)?; + validate_namespace(namespace_path)?; + let (outcome, job) = self + .sql_client()? + .drop_function_outcome(name, version, namespace_path) + .await?; + job.wait().await?; + Ok((outcome.dropped, outcome.warnings)) + } + async fn create_storage_integration( &self, name: &str, diff --git a/rust/lancedb/src/remote/sql/function.rs b/rust/lancedb/src/remote/sql/function.rs index 7682b11c1..e90792a8a 100644 --- a/rust/lancedb/src/remote/sql/function.rs +++ b/rust/lancedb/src/remote/sql/function.rs @@ -4,7 +4,7 @@ use super::{ SqlClient, materialized_view::quote_identifier, - objects::{definition, literal, strings}, + objects::{definition, literal, string_lists, strings}, }; use crate::{ database::ExecuteQueryRequest, @@ -16,6 +16,14 @@ use crate::{ }; use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD}; +/// The SQL result of one Function drop, before the public boolean wrapper +/// discards the advisory details. +#[derive(Clone, serde::Deserialize, serde::Serialize)] +pub(in crate::remote) struct DropFunctionOutcome { + pub(in crate::remote) dropped: bool, + pub(in crate::remote) warnings: Vec, +} + impl SqlClient { pub(in crate::remote) async fn register_function( &self, @@ -146,12 +154,12 @@ impl SqlClient { }; Ok(ListingPage { items, page_token }) } - pub(in crate::remote) async fn drop_function( + pub(in crate::remote) async fn drop_function_outcome( &self, name: &str, version: &str, namespace: &[String], - ) -> Result<(bool, Job)> { + ) -> Result<(DropFunctionOutcome, Job)> { validate_function_name(name)?; validate_namespace(namespace)?; let job = self @@ -167,23 +175,37 @@ impl SqlClient { }, |batches| { let rows = strings(&batches, "status")?; - match rows.as_slice() { - [status] if status == "dropped" => Ok(true), - [status] if status == "not_found" => Ok(false), - _ => Err(Error::Runtime { - message: "invalid DROP FUNCTION result".into(), - }), - } + let dropped = match rows.as_slice() { + [status] if status == "dropped" => true, + [status] if status == "not_found" => false, + _ => { + return Err(Error::Runtime { + message: "invalid DROP FUNCTION result".into(), + }); + } + }; + Ok(DropFunctionOutcome { + dropped, + warnings: string_lists(&batches, "warnings")?, + }) }, ) .await?; let id = job.id().unwrap().parse().map_err(|_| Error::Runtime { message: "invalid SQL query job id".into(), })?; - let dropped = job.wait().await?; - Ok(( - dropped, - Job::from_query_task(id, tokio::spawn(async { Ok(()) }), || async { Ok(()) }), - )) + let outcome = job.wait().await?; + let job = Job::from_query_task(id, tokio::spawn(async { Ok(()) }), || async { Ok(()) }); + Ok((outcome, job)) + } + + pub(in crate::remote) async fn drop_function( + &self, + name: &str, + version: &str, + namespace: &[String], + ) -> Result<(bool, Job)> { + let (outcome, job) = self.drop_function_outcome(name, version, namespace).await?; + Ok((outcome.dropped, job)) } } diff --git a/rust/lancedb/src/remote/sql/objects.rs b/rust/lancedb/src/remote/sql/objects.rs index 2c4d33614..ff87e0000 100644 --- a/rust/lancedb/src/remote/sql/objects.rs +++ b/rust/lancedb/src/remote/sql/objects.rs @@ -7,7 +7,7 @@ use crate::{ error::{Error, Result}, job::Job, }; -use arrow_array::{Array, RecordBatch, StringArray}; +use arrow_array::{Array, ListArray, RecordBatch, StringArray}; use futures::TryStreamExt; use serde::{Serialize, de::DeserializeOwned}; use std::sync::Arc; @@ -38,6 +38,45 @@ pub(super) fn strings(batches: &[RecordBatch], name: &str) -> Result .collect::>>() .map(|rows| rows.into_iter().flatten().collect()) } + +/// Read an optional `List` column. Older servers do not return the +/// column, so a missing column means no warnings rather than a bad result. +pub(super) fn string_lists(batches: &[RecordBatch], name: &str) -> Result> { + let mut values = Vec::new(); + for batch in batches { + let Some(column) = batch.column_by_name(name) else { + continue; + }; + let lists = column + .as_any() + .downcast_ref::() + .ok_or_else(|| Error::Runtime { + message: format!("SQL result column '{name}' is not a string list"), + })?; + for index in 0..lists.len() { + if lists.is_null(index) { + return Err(Error::Runtime { + message: format!("SQL result contains null '{name}'"), + }); + } + let list = lists.value(index); + let strings = + list.as_any() + .downcast_ref::() + .ok_or_else(|| Error::Runtime { + message: format!("SQL result column '{name}' contains a non-string value"), + })?; + if strings.null_count() != 0 { + return Err(Error::Runtime { + message: format!("SQL result column '{name}' contains null values"), + }); + } + values.extend(strings.iter().map(|value| value.unwrap().to_owned())); + } + } + Ok(values) +} + pub(super) fn definition(batches: Vec) -> Result { let values = strings(&batches, "definition")?; let [value]: [String; 1] = values.try_into().map_err(|_| Error::Runtime { @@ -70,3 +109,74 @@ impl SqlClient { })) } } + +#[cfg(test)] +mod tests { + use super::string_lists; + use arrow_array::builder::{ListBuilder, StringBuilder}; + use arrow_array::{ArrayRef, RecordBatch}; + use arrow_schema::{DataType, Field, Schema}; + use std::sync::Arc; + + fn warning_schema() -> Arc { + Arc::new(Schema::new(vec![Field::new( + "warnings", + DataType::List(Arc::new(Field::new("item", DataType::Utf8, true))), + true, + )])) + } + + #[test] + fn reads_warnings_after_an_empty_first_batch() { + let schema = warning_schema(); + let empty = RecordBatch::new_empty(schema.clone()); + let mut builder = ListBuilder::new(StringBuilder::new()); + builder.values().append_value( + "materialized view \"pinned\" references Function \"embed\" version \"1\"", + ); + builder.append(true); + let non_empty = + RecordBatch::try_new(schema.clone(), vec![Arc::new(builder.finish()) as ArrayRef]) + .unwrap(); + + assert_eq!( + string_lists(&[empty, non_empty], "warnings").unwrap(), + ["materialized view \"pinned\" references Function \"embed\" version \"1\""] + ); + } + + #[test] + fn missing_warning_column_remains_an_empty_result() { + let schema = Arc::new(Schema::new(vec![Field::new( + "status", + DataType::Utf8, + false, + )])); + let batch = RecordBatch::new_empty(schema); + + assert_eq!( + string_lists(&[batch], "warnings").unwrap(), + Vec::::new() + ); + } + + #[test] + fn null_or_non_list_warning_values_are_rejected() { + let mut builder = ListBuilder::new(StringBuilder::new()); + builder.append(false); + let null_list = RecordBatch::try_new( + warning_schema(), + vec![Arc::new(builder.finish()) as ArrayRef], + ) + .unwrap(); + assert!(string_lists(&[null_list], "warnings").is_err()); + + let non_list_schema = Arc::new(Schema::new(vec![Field::new( + "warnings", + DataType::Utf8, + false, + )])); + let non_list = RecordBatch::new_empty(non_list_schema); + assert!(string_lists(&[non_list], "warnings").is_err()); + } +} diff --git a/rust/lancedb/src/remote/sql_test.rs b/rust/lancedb/src/remote/sql_test.rs index 37125c1fd..1211263d8 100644 --- a/rust/lancedb/src/remote/sql_test.rs +++ b/rust/lancedb/src/remote/sql_test.rs @@ -3,7 +3,7 @@ use std::sync::atomic::AtomicUsize; -use arrow_array::builder::StringDictionaryBuilder; +use arrow_array::builder::{ListBuilder, StringBuilder, StringDictionaryBuilder}; use arrow_array::types::{Float32Type, Int32Type}; use arrow_array::{Array, ArrayRef, FixedSizeListArray, Float32Array, Int64Array, StringArray}; use arrow_flight::encode::FlightDataEncoderBuilder; @@ -2944,6 +2944,45 @@ async fn function_drop_keeps_the_query_id_and_reports_catalog_removal() { } } +#[tokio::test] +async fn function_drop_decodes_server_warnings_for_the_binding_bridge() { + let mut warning_values = ListBuilder::new(StringBuilder::new()); + warning_values + .values() + .append_value("materialized view \"pinned\" references Function \"embed\" version \"1\""); + warning_values.append(true); + let result = RecordBatch::try_from_iter(vec![ + ( + "status", + Arc::new(StringArray::from(vec!["dropped"])) as ArrayRef, + ), + ("warnings", Arc::new(warning_values.finish()) as ArrayRef), + ]) + .unwrap(); + let service = TestSqlService { + adaptive: true, + defer: true, + result, + ..Default::default() + }; + let (client, shutdown) = start_sql_service(service, ClientConfig::default()).await; + let db = RemoteDatabase::new_mock(|_| -> http::Response { + panic!("unexpected REST request"); + }) + .with_sql_client(client); + + let (dropped, warnings) = db + .drop_function_with_warnings("embed", "1", &[]) + .await + .unwrap(); + assert!(dropped); + assert_eq!( + warnings, + ["materialized view \"pinned\" references Function \"embed\" version \"1\""] + ); + shutdown.send(()).unwrap(); +} + #[tokio::test] async fn secret_writes_bind_credentials_and_never_put_them_in_sql() { for defer in [false, true] {