mirror of
https://github.com/lancedb/lancedb.git
synced 2026-11-14 14:46:53 +00:00
fix(python): surface Function drop warnings (#4880)
Remote Python `drop_function` currently discards the dependent-view warnings returned by SQL DROP. Both the synchronous and coroutine APIs now emit those warnings through `warnings.warn` while preserving their boolean result and executing DROP once. The server support is already merged in lancedb/sophon#9820. The Rust decoder reads warnings across every result batch, including an empty first batch. Missing warning columns remain compatible with older servers; malformed or null warning lists are rejected. The hidden binding bridge carries the warnings to Python. The separate `drop_function_async` API and its `(bool, Job)` contract are unchanged. Python warning filters can raise after the catalog mutation has completed. Validation: a compiled PyO3/public-server reproduction failed before the fix and passed afterward; the public async path also returned `true` with the expected dependent-view warning. Final source passes two focused Rust tests, two Python wrapper tests, strict targeted Clippy, formatting, and Ruff. Three decoder tests cover batch/null/type and older-server behavior. The public runs preceded a two-file simplification of the completed Job projection; focused final-source checks cover that delta. Current-main integration is conflict-free, and its overlapping changes concern independent retry-policy APIs/tests. Documentation now covers warning filters, the post-removal warnings-as-errors behavior, and silent compatibility with older servers. The follow-up changes only prose and a Python docstring; the runtime validation above remains applicable. Refs ENT-3659. Whole-issue independent acceptance remains pending.
This commit is contained in:
@@ -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.
|
||||
|
||||
@@ -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]: ...
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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}
|
||||
|
||||
@@ -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<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_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>,
|
||||
|
||||
@@ -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<str>,
|
||||
version: impl AsRef<str>,
|
||||
namespace_path: &[String],
|
||||
) -> Result<(bool, Vec<String>)> {
|
||||
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.
|
||||
|
||||
@@ -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<String>)> {
|
||||
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,
|
||||
|
||||
@@ -932,6 +932,23 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
|
||||
.await
|
||||
}
|
||||
|
||||
#[doc(hidden)]
|
||||
async fn drop_function_with_warnings(
|
||||
&self,
|
||||
name: &str,
|
||||
version: &str,
|
||||
namespace_path: &[String],
|
||||
) -> Result<(bool, Vec<String>)> {
|
||||
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,
|
||||
|
||||
@@ -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<String>,
|
||||
}
|
||||
|
||||
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))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<Vec<String>
|
||||
.collect::<Result<Vec<_>>>()
|
||||
.map(|rows| rows.into_iter().flatten().collect())
|
||||
}
|
||||
|
||||
/// Read an optional `List<Utf8>` 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<Vec<String>> {
|
||||
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::<ListArray>()
|
||||
.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::<StringArray>()
|
||||
.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<T: DeserializeOwned>(batches: Vec<RecordBatch>) -> Result<T> {
|
||||
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<Schema> {
|
||||
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::<String>::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());
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<String> {
|
||||
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] {
|
||||
|
||||
Reference in New Issue
Block a user