Merge remote-tracking branch 'upstream/main' into codex/update-lance-13-0-0-beta-9

Resolve conflicts with the beta.8 bump (#4242) and the 0.40.0-beta.5
version bump; Cargo.lock regenerated against main with
ci/set_lance_version.py 13.0.0-beta.9.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>

.bumpversion.toml

@@ -1,5 +1,5 @@
 [tool.bumpversion]
-current_version = "0.40.0-beta.4"
+current_version = "0.40.0-beta.5"
 parse = """(?x)
     (?P<major>0|[1-9]\\d*)\\.
     (?P<minor>0|[1-9]\\d*)\\.

Cargo.lock

@@ -5756,7 +5756,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb"
-version = "0.40.0-beta.4"
+version = "0.40.0-beta.5"
 dependencies = [
  "ahash",
  "anyhow",
@@ -5853,7 +5853,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb-nodejs"
-version = "0.40.0-beta.4"
+version = "0.40.0-beta.5"
 dependencies = [
  "arrow-array",
  "arrow-buffer",
@@ -5878,7 +5878,7 @@ dependencies = [
 
 [[package]]
 name = "lancedb-python"
-version = "0.40.0-beta.4"
+version = "0.40.0-beta.5"
 dependencies = [
  "arc-swap",
  "arrow",

docs/src/java/java.md

@@ -14,7 +14,7 @@ Add the following dependency to your `pom.xml`:
 <dependency>
     <groupId>com.lancedb</groupId>
     <artifactId>lancedb-core</artifactId>
-    <version>0.40.0-beta.4</version>
+    <version>0.40.0-beta.5</version>
 </dependency>
 ```
 

java/lancedb-core/pom.xml

@@ -8,7 +8,7 @@
     <parent>
       <groupId>com.lancedb</groupId>
       <artifactId>lancedb-parent</artifactId>
-      <version>0.40.0-beta.4</version>
+      <version>0.40.0-beta.5</version>
       <relativePath>../pom.xml</relativePath>
     </parent>
 

java/pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.lancedb</groupId>
     <artifactId>lancedb-parent</artifactId>
-    <version>0.40.0-beta.4</version>
+    <version>0.40.0-beta.5</version>
     <packaging>pom</packaging>
     <name>${project.artifactId}</name>
     <description>LanceDB Java SDK Parent POM</description>

nodejs/Cargo.toml

@@ -1,7 +1,7 @@
 [package]
 name = "lancedb-nodejs"
 edition.workspace = true
-version = "0.40.0-beta.4"
+version = "0.40.0-beta.5"
 publish = false
 license.workspace = true
 description.workspace = true

nodejs/npm/darwin-arm64/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-darwin-arm64",
-	"version": "0.40.0-beta.4",
+	"version": "0.40.0-beta.5",
 	"os": ["darwin"],
 	"cpu": ["arm64"],
 	"main": "lancedb.darwin-arm64.node",

nodejs/npm/linux-arm64-gnu/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-gnu",
-	"version": "0.40.0-beta.4",
+	"version": "0.40.0-beta.5",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-gnu.node",

nodejs/npm/linux-arm64-musl/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-arm64-musl",
-	"version": "0.40.0-beta.4",
+	"version": "0.40.0-beta.5",
 	"os": ["linux"],
 	"cpu": ["arm64"],
 	"main": "lancedb.linux-arm64-musl.node",

nodejs/npm/linux-x64-gnu/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-gnu",
-	"version": "0.40.0-beta.4",
+	"version": "0.40.0-beta.5",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-gnu.node",

nodejs/npm/linux-x64-musl/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-linux-x64-musl",
-	"version": "0.40.0-beta.4",
+	"version": "0.40.0-beta.5",
 	"os": ["linux"],
 	"cpu": ["x64"],
 	"main": "lancedb.linux-x64-musl.node",

nodejs/npm/win32-arm64-msvc/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@lancedb/lancedb-win32-arm64-msvc",
-  "version": "0.40.0-beta.4",
+  "version": "0.40.0-beta.5",
   "os": [
     "win32"
   ],

nodejs/npm/win32-x64-msvc/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "@lancedb/lancedb-win32-x64-msvc",
-	"version": "0.40.0-beta.4",
+	"version": "0.40.0-beta.5",
 	"os": ["win32"],
 	"cpu": ["x64"],
 	"main": "lancedb.win32-x64-msvc.node",

nodejs/package.json

@@ -11,7 +11,7 @@
     "ann"
   ],
   "private": false,
-  "version": "0.40.0-beta.4",
+  "version": "0.40.0-beta.5",
   "main": "dist/index.js",
   "exports": {
     ".": "./dist/index.js",

python/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb-python"
-version = "0.40.0-beta.4"
+version = "0.40.0-beta.5"
 publish = false
 edition.workspace = true
 description = "Python bindings for LanceDB"

rust/lancedb/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "lancedb"
-version = "0.40.0-beta.4"
+version = "0.40.0-beta.5"
 edition.workspace = true
 description = "LanceDB: A serverless, low-latency vector database for AI applications"
 license.workspace = true

rust/lancedb/src/table/computed_columns.rs

@@ -418,23 +418,18 @@ pub(crate) fn ensure_not_function_bound<S: AsRef<str>>(
     touched: impl IntoIterator<Item = S>,
 ) -> Result<()> {
     ensure_supported_function_metadata(schema)?;
-    let mut protected = BTreeSet::new();
-    for binding in function_bindings(schema)? {
-        for input in binding.inputs() {
-            protected.insert(field_root(&input.field_path)?);
-        }
-        protected.extend(
-            binding
-                .outputs()
-                .iter()
-                .map(|output| output.output_name.clone()),
-        );
-        protected.extend(
-            binding
-                .assignment()
-                .map(|assignment| assignment.output_name.clone()),
-        );
-    }
+    ensure_not_bound_by(&function_bindings(schema)?, operation, touched)
+}
+
+/// [`ensure_not_function_bound`] against an explicit set of bindings, for a
+/// caller that is retiring some of the schema's own in the same operation and
+/// must be checked against what survives it.
+pub(crate) fn ensure_not_bound_by<S: AsRef<str>>(
+    bindings: &[FunctionBinding],
+    operation: &str,
+    touched: impl IntoIterator<Item = S>,
+) -> Result<()> {
+    let protected = protected_roots(bindings)?;
     if protected.is_empty() {
         return Ok(());
     }
@@ -452,6 +447,129 @@ pub(crate) fn ensure_not_function_bound<S: AsRef<str>>(
     Ok(())
 }
 
+/// Every column `bindings` depend on, inputs taken at their root.
+fn protected_roots(bindings: &[FunctionBinding]) -> Result<BTreeSet<String>> {
+    let mut protected = BTreeSet::new();
+    for binding in bindings {
+        for input in binding.inputs() {
+            protected.insert(field_root(&input.field_path)?);
+        }
+        protected.extend(
+            binding
+                .outputs()
+                .iter()
+                .map(|output| output.output_name.clone()),
+        );
+        protected.extend(
+            binding
+                .assignment()
+                .map(|assignment| assignment.output_name.clone()),
+        );
+    }
+    Ok(protected)
+}
+
+/// What a drop must do besides removing columns, so that the Function
+/// bindings it covers are retired rather than stranded.
+#[derive(Debug, Default)]
+pub struct FunctionUnbinding {
+    /// The bindings the drop leaves in place. Later columns in the same
+    /// operation are checked against these, not against the schema's.
+    pub retained: Vec<FunctionBinding>,
+    /// Assignment columns of the retired bindings that the caller did not
+    /// name. Internal bookkeeping columns, so the drop adds them itself.
+    pub assignment_columns: Vec<String>,
+    /// Columns whose `computed_column.*` declaration metadata the unbind
+    /// commit clears.
+    pub cleared_columns: Vec<String>,
+    /// The new value of [`FUNCTION_BINDINGS_META_KEY`]; `None` deletes it.
+    pub bindings_metadata: Option<String>,
+}
+
+impl FunctionUnbinding {
+    /// True when the drop retires nothing and is an ordinary column drop.
+    pub fn is_noop(&self) -> bool {
+        self.cleared_columns.is_empty()
+    }
+
+    /// Refuse `operation` on a path that a binding surviving this drop still
+    /// reads or writes. The retired ones no longer protect anything.
+    pub fn ensure_retained_unaffected<S: AsRef<str>>(
+        &self,
+        operation: &str,
+        touched: impl IntoIterator<Item = S>,
+    ) -> Result<()> {
+        ensure_not_bound_by(&self.retained, operation, touched)
+    }
+}
+
+/// Plan the retirement of every Function binding whose outputs `columns`
+/// covers.
+///
+/// Naming one output of a binding means naming all of them: the outputs of
+/// one Function are written by one refresh in one commit, so a binding that
+/// kept some of them would have no coherent shape to write. A partial drop is
+/// refused and names the siblings that are missing.
+pub fn plan_function_unbinding(
+    schema: &ArrowSchema,
+    columns: &[&str],
+) -> Result<FunctionUnbinding> {
+    ensure_supported_function_metadata(schema)?;
+    let named = columns
+        .iter()
+        .map(|column| whole_column(column))
+        .collect::<Result<Vec<Option<String>>>>()?
+        .into_iter()
+        .flatten()
+        .collect::<BTreeSet<String>>();
+
+    let mut plan = FunctionUnbinding::default();
+    for binding in function_bindings(schema)? {
+        let outputs = binding
+            .outputs()
+            .iter()
+            .map(|output| output.output_name.clone())
+            .collect::<Vec<_>>();
+        let Some(dropped) = outputs.iter().find(|name| named.contains(name.as_str())) else {
+            plan.retained.push(binding);
+            continue;
+        };
+        let missing = outputs
+            .iter()
+            .filter(|name| !named.contains(name.as_str()))
+            .cloned()
+            .collect::<Vec<_>>();
+        if !missing.is_empty() {
+            return Err(Error::InvalidInput {
+                message: format!(
+                    "dropping Function output '{dropped}' must drop every output of its \
+                     binding; the same request must also name {}",
+                    missing.join(", ")
+                ),
+            });
+        }
+        plan.cleared_columns.extend(outputs);
+        if let Some(assignment) = binding.assignment() {
+            plan.cleared_columns.push(assignment.output_name.clone());
+            if !named.contains(assignment.output_name.as_str()) {
+                plan.assignment_columns.push(assignment.output_name.clone());
+            }
+        }
+    }
+
+    // Re-encoding an untouched set would rewrite bytes a stricter reader
+    // compares against its own encoding, so leave the key alone when the drop
+    // retires nothing.
+    if !plan.is_noop() {
+        plan.bindings_metadata = if plan.retained.is_empty() {
+            None
+        } else {
+            Some(function_bindings_metadata(&plan.retained)?)
+        };
+    }
+    Ok(plan)
+}
+
 /// Read a field's computed-column declaration, if it carries one.
 ///
 /// A field flagged computed but carrying no kind, or a SQL one missing its
@@ -1320,7 +1438,30 @@ pub(crate) fn plan_function_application(
 /// Paths are compared at their root: a declaration reading `metadata` is
 /// invalidated by a change to `metadata.age` just as surely.
 pub(crate) fn ensure_not_an_input(schema: &SchemaRef, paths: &[&str]) -> Result<()> {
+    ensure_not_an_input_of(schema, paths, &[])
+}
+
+/// [`ensure_not_an_input`] where `removed` names columns whose own
+/// declarations the same operation deletes. A reader that is going away
+/// cannot be stranded by dropping what it reads, so a group drops together.
+pub(crate) fn ensure_not_an_input_of(
+    schema: &SchemaRef,
+    paths: &[&str],
+    removed: &[&str],
+) -> Result<()> {
+    // By identity, not spelling: a quoted path names the same column, while
+    // a path *into* one removes no declaration at all.
+    let removed = removed
+        .iter()
+        .map(|path| whole_column(path))
+        .collect::<Result<Vec<_>>>()?
+        .into_iter()
+        .flatten()
+        .collect::<BTreeSet<String>>();
     for declaration in computed_columns(schema) {
+        if removed.contains(&declaration.name) {
+            continue;
+        }
         // The expression, not stored inputs, is the source of truth; an
         // expression that no longer parses proves nothing, so refuse.
         let inputs = match &declaration.kind {
@@ -1583,6 +1724,21 @@ pub(crate) fn field_root(path: &str) -> Result<String> {
         })
 }
 
+/// The column a path names when it names the whole column, and `None` when it
+/// addresses a field inside one. Retiring a binding takes the whole output;
+/// a path into one is left to the ordinary guard to refuse.
+fn whole_column(path: &str) -> Result<Option<String>> {
+    let mut segments = parse_field_path(path)
+        .map_err(|e| Error::InvalidInput {
+            message: format!("invalid column path '{path}': {e}"),
+        })?
+        .into_iter();
+    let column = segments.next().ok_or_else(|| Error::InvalidInput {
+        message: format!("column path '{path}' is empty"),
+    })?;
+    Ok(segments.next().is_none().then_some(column))
+}
+
 /// [`field_root`], falling back to the text before the first dot for a
 /// spelling lance would not resolve.
 pub(crate) fn root(path: &str) -> String {

rust/lancedb/src/table/schema_evolution.rs

@@ -9,7 +9,8 @@
 //! - [`drop_columns`](execute_drop_columns): Remove columns from the table
 
 use arrow_schema::Schema as ArrowSchema;
-use lance::dataset::{ColumnAlteration, NewColumnTransform};
+use lance::dataset::transaction::{Operation, Transaction, UpdateMap, UpdateMapEntry};
+use lance::dataset::{ColumnAlteration, CommitBuilder, Dataset, NewColumnTransform};
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 
@@ -240,14 +241,96 @@ pub(crate) async fn execute_drop_columns(
     table.dataset.ensure_mutable()?;
     let mut dataset = (*table.dataset.get().await?).clone();
     let schema = std::sync::Arc::new(ArrowSchema::from(dataset.schema()));
-    computed_columns::ensure_not_function_bound(schema.as_ref(), "schema evolution", columns)?;
-    computed_columns::ensure_not_an_input(&schema, columns)?;
-    dataset.drop_columns(columns).await?;
+    let unbinding = computed_columns::plan_function_unbinding(schema.as_ref(), columns)?;
+
+    let mut names = columns.iter().map(|c| c.to_string()).collect::<Vec<_>>();
+    names.extend(unbinding.assignment_columns.iter().cloned());
+    let dropped = names.iter().map(String::as_str).collect::<Vec<_>>();
+    computed_columns::ensure_not_bound_by(&unbinding.retained, "schema evolution", &dropped)?;
+    computed_columns::ensure_not_an_input_of(&schema, &dropped, &dropped)?;
+
+    if !unbinding.is_noop() {
+        // The retirement commits before the projection, so the whole
+        // projection has to be known valid against this revision first --
+        // otherwise the binding is gone and the columns are not. Lance plans
+        // it rather than this repeating the rules: dropping a struct's last
+        // child removes the struct, and data files that disagree on metadata
+        // semantics are refused, neither of which an approximation here would
+        // get right.
+        dataset.plan_drop_columns(&dropped)?;
+        dataset = commit_function_unbinding(dataset, &unbinding).await?;
+    }
+    dataset.drop_columns(&dropped).await?;
     let version = dataset.version().version;
     table.dataset.update(dataset);
     Ok(DropColumnsResult { version })
 }
 
+/// Retire the bindings a drop covers, in the commit before it.
+///
+/// A binding lives in three places -- the schema-metadata envelope, each
+/// output's `computed_column.*` field metadata, and the columns themselves --
+/// and readers cross-check the first two, so a state with one of them removed
+/// is a table that refuses every write. Clearing both metadata halves in one
+/// `UpdateConfig` leaves the outputs as ordinary columns holding their last
+/// values: valid on its own, and the drop that follows is then an ordinary
+/// drop. Interrupted in between, the columns are still there to be dropped
+/// again.
+///
+/// Not folded into the drop's own `Operation::Project`, though it carries a
+/// schema: the transaction proto records `Project` as fields alone, so the
+/// metadata edit would survive only in this writer's memory.
+async fn commit_function_unbinding(
+    dataset: Dataset,
+    unbinding: &computed_columns::FunctionUnbinding,
+) -> Result<Dataset> {
+    let bindings = UpdateMap {
+        update_entries: vec![UpdateMapEntry {
+            key: computed_columns::FUNCTION_BINDINGS_META_KEY.to_string(),
+            value: unbinding.bindings_metadata.clone(),
+        }],
+        replace: false,
+    };
+    let mut field_metadata_updates = HashMap::new();
+    for column in &unbinding.cleared_columns {
+        let field = dataset
+            .schema()
+            .field(column)
+            .ok_or_else(|| Error::InvalidInput {
+                message: format!("Function output '{column}' does not exist in the table"),
+            })?;
+        let cleared = field
+            .metadata
+            .keys()
+            .filter(|key| computed_columns::is_declaration_key(key))
+            .map(|key| UpdateMapEntry {
+                key: key.clone(),
+                value: None,
+            })
+            .collect::<Vec<_>>();
+        field_metadata_updates.insert(
+            field.id,
+            UpdateMap {
+                update_entries: cleared,
+                replace: false,
+            },
+        );
+    }
+    let transaction = Transaction::new(
+        dataset.manifest.version,
+        Operation::UpdateConfig {
+            config_updates: None,
+            table_metadata_updates: None,
+            schema_metadata_updates: Some(bindings),
+            field_metadata_updates,
+        },
+        None,
+    );
+    Ok(CommitBuilder::new(std::sync::Arc::new(dataset))
+        .execute(transaction)
+        .await?)
+}
+
 /// Internal implementation of the update field metadata logic.
 ///
 /// Merges or replaces per-field metadata, addressing fields by dot-path.
@@ -330,8 +413,9 @@ mod tests {
     use crate::query::{ExecutableQuery, QueryBase, Select};
     use crate::table::NewColumnTransform;
     use crate::table::computed_columns::{
-        FUNCTION_BINDINGS_META_KEY, ensure_supported_function_metadata, function_bindings,
-        function_bindings_metadata, function_computed_column_metadata,
+        FUNCTION_ASSIGNMENT_OUTPUT_ORDINAL, FUNCTION_BINDINGS_META_KEY,
+        ensure_supported_function_metadata, function_bindings, function_bindings_metadata,
+        function_computed_column_metadata, is_declaration_key,
     };
     use crate::{Error, Table};
     use std::collections::HashMap;
@@ -351,38 +435,71 @@ mod tests {
         )
         .unwrap();
         let table = conn.create_table("bound", batch).execute().await.unwrap();
-        let binding = FunctionBinding::from_json(include_str!(
+        stamp_bindings(&table, &[text_features_binding("")]).await;
+        table
+    }
+
+    /// The fixture binding, its id and output names suffixed so a table can
+    /// carry more than one.
+    fn text_features_binding(suffix: &str) -> FunctionBinding {
+        let raw = include_str!(
             "../../tests/fixtures/first_class_functions/v1/remote_function_binding.json"
-        ))
-        .unwrap();
+        );
+        if suffix.is_empty() {
+            return FunctionBinding::from_json(raw).unwrap();
+        }
+        FunctionBinding::from_json(
+            &raw.replace("fb_01K3TEXT", &format!("fb_01K3TEXT{suffix}"))
+                .replace("search_text", &format!("search_text{suffix}"))
+                .replace("search_token_count", &format!("search_token_count{suffix}")),
+        )
+        .unwrap()
+    }
+
+    /// Stamp bindings the way the server does it, since no local path
+    /// declares one: the envelope on the schema, a declaration marker on
+    /// every output field.
+    async fn stamp_bindings(table: &Table, bindings: &[FunctionBinding]) {
         let native = table.as_native().unwrap();
         let mut dataset = native.dataset.get().await.unwrap().as_ref().clone();
         dataset
             .update_schema_metadata(vec![(
                 FUNCTION_BINDINGS_META_KEY.to_string(),
-                Some(function_bindings_metadata(std::slice::from_ref(&binding)).unwrap()),
+                Some(function_bindings_metadata(bindings).unwrap()),
             )])
             .await
             .unwrap();
         let inputs = ["title".to_string(), "body".to_string()];
-        let outputs = binding
-            .outputs()
+        let outputs = bindings
             .iter()
-            .map(|output| {
-                (
-                    dataset.schema().field(&output.output_name).unwrap().id as u32,
-                    function_computed_column_metadata(
-                        binding.binding_id(),
-                        output.output_ordinal,
-                        &inputs,
-                    ),
-                )
+            .flat_map(|binding| {
+                let assignment = binding.assignment().map(|assignment| {
+                    (
+                        assignment.output_name.clone(),
+                        FUNCTION_ASSIGNMENT_OUTPUT_ORDINAL,
+                    )
+                });
+                binding
+                    .outputs()
+                    .iter()
+                    .map(|output| (output.output_name.clone(), output.output_ordinal))
+                    .chain(assignment)
+                    .map(|(name, ordinal)| {
+                        (
+                            dataset.schema().field(&name).unwrap().id as u32,
+                            function_computed_column_metadata(
+                                binding.binding_id(),
+                                ordinal,
+                                &inputs,
+                            ),
+                        )
+                    })
+                    .collect::<Vec<_>>()
             })
             .collect::<Vec<_>>();
         dataset.replace_field_metadata(outputs).await.unwrap();
         native.dataset.update(dataset);
         ensure_supported_function_metadata(&table.schema().await.unwrap()).unwrap();
-        table
     }
 
     fn metadata_update(path: &str) -> FieldMetadataUpdate {
@@ -438,6 +555,18 @@ mod tests {
         );
     }
 
+    fn incomplete_group(err: Error) {
+        assert!(
+            matches!(&err, Error::InvalidInput { message }
+                if message.contains("must drop every output of its binding")),
+            "{err:?}"
+        );
+    }
+
+    async fn bindings_of(table: &Table) -> Vec<FunctionBinding> {
+        function_bindings(table.schema().await.unwrap().as_ref()).unwrap()
+    }
+
     /// Every schema-evolution door refuses a column a binding reads or
     /// writes, including a rename onto one.
     #[tokio::test]
@@ -445,7 +574,14 @@ mod tests {
         let table = bound_table().await;
         let version = table.version().await.unwrap();
         for column in ["title", "body", "search_text", "search_token_count"] {
-            bound(table.drop_columns(&[column]).await.unwrap_err());
+            // A lone drop of an output is refused too, but for its siblings
+            // rather than for the binding; see the retirement tests below.
+            let err = table.drop_columns(&[column]).await.unwrap_err();
+            if column.starts_with("search_") {
+                incomplete_group(err);
+            } else {
+                bound(err);
+            }
             bound(
                 table
                     .alter_columns(&[ColumnAlteration::new(column.into()).rename("moved".into())])
@@ -500,7 +636,12 @@ mod tests {
         let table = bound_table().await;
         let version = table.version().await.unwrap();
         for path in ["`title`", "`search_text`", "`title`.nested"] {
-            bound(table.drop_columns(&[path]).await.unwrap_err());
+            let err = table.drop_columns(&[path]).await.unwrap_err();
+            if path == "`search_text`" {
+                incomplete_group(err);
+            } else {
+                bound(err);
+            }
             bound(
                 table
                     .alter_columns(&[ColumnAlteration::new(path.into()).set_nullable(false)])
@@ -1038,6 +1179,264 @@ mod tests {
         assert_eq!(drop_result.version, v4);
     }
 
+    // Retiring a Function binding by dropping its outputs (ENT-2669).
+
+    /// The whole output set goes, and the binding with it: no envelope, no
+    /// declaration markers, and the inputs it held are ordinary again.
+    #[tokio::test]
+    async fn dropping_every_output_retires_the_binding() {
+        let table = bound_table().await;
+        table
+            .drop_columns(&["search_text", "search_token_count"])
+            .await
+            .unwrap();
+
+        let schema = table.schema().await.unwrap();
+        assert!(schema.field_with_name("search_text").is_err());
+        assert!(schema.field_with_name("search_token_count").is_err());
+        assert!(!schema.metadata().contains_key(FUNCTION_BINDINGS_META_KEY));
+        assert!(bindings_of(&table).await.is_empty());
+
+        // What the binding used to protect is now ordinary schema.
+        table.drop_columns(&["title"]).await.unwrap();
+        table
+            .update_field_metadata(&[metadata_update("body")])
+            .await
+            .unwrap();
+        ensure_supported_function_metadata(&table.schema().await.unwrap()).unwrap();
+    }
+
+    /// One output of a multi-output binding cannot go alone: its siblings are
+    /// written by the same refresh in the same commit.
+    #[tokio::test]
+    async fn dropping_part_of_an_output_group_names_the_missing_siblings() {
+        let table = bound_table().await;
+        let version = table.version().await.unwrap();
+        for (named, missing) in [
+            ("search_text", "search_token_count"),
+            ("search_token_count", "search_text"),
+        ] {
+            let err = table.drop_columns(&[named]).await.unwrap_err();
+            let Error::InvalidInput { message } = &err else {
+                panic!("{err:?}");
+            };
+            assert!(
+                message.contains("must drop every output of its binding")
+                    && message.contains(missing),
+                "{message}"
+            );
+        }
+        assert_eq!(table.version().await.unwrap(), version);
+        assert_eq!(bindings_of(&table).await.len(), 1);
+    }
+
+    /// An input is protected by the binding, not by its own name, so it drops
+    /// in the request that retires the binding reading it.
+    #[tokio::test]
+    async fn an_input_drops_with_the_binding_that_reads_it() {
+        let table = bound_table().await;
+        bound(table.drop_columns(&["title"]).await.unwrap_err());
+
+        table
+            .drop_columns(&["title", "body", "search_text", "search_token_count"])
+            .await
+            .unwrap();
+        let schema = table.schema().await.unwrap();
+        assert_eq!(schema.fields().len(), 1);
+        assert!(schema.field_with_name("spare").is_ok());
+        assert!(bindings_of(&table).await.is_empty());
+    }
+
+    /// Lance resolves a quoted spelling to the same field, so it retires the
+    /// same binding rather than slipping past the group rule.
+    #[tokio::test]
+    async fn a_quoted_output_spelling_retires_the_same_binding() {
+        let table = bound_table().await;
+        incomplete_group(table.drop_columns(&["`search_text`"]).await.unwrap_err());
+        table
+            .drop_columns(&["`search_text`", "`search_token_count`"])
+            .await
+            .unwrap();
+        assert!(bindings_of(&table).await.is_empty());
+    }
+
+    /// The unbind lands before the drop, and on its own: at that version the
+    /// outputs are still there as plain columns, with no binding and no
+    /// declaration metadata left pointing at one. A crash in between leaves
+    /// that state, which every later write must accept.
+    #[tokio::test]
+    async fn the_unbind_commit_leaves_a_valid_table_before_the_drop() {
+        let table = bound_table().await;
+        let before = table.version().await.unwrap();
+        table
+            .drop_columns(&["search_text", "search_token_count"])
+            .await
+            .unwrap();
+        assert_eq!(table.version().await.unwrap(), before + 2);
+
+        table.checkout(before + 1).await.unwrap();
+        let schema = table.schema().await.unwrap();
+        assert!(!schema.metadata().contains_key(FUNCTION_BINDINGS_META_KEY));
+        for column in ["search_text", "search_token_count"] {
+            let field = schema.field_with_name(column).unwrap();
+            assert!(
+                !field.metadata().keys().any(|key| is_declaration_key(key)),
+                "{column}: {:?}",
+                field.metadata()
+            );
+        }
+        ensure_supported_function_metadata(&schema).unwrap();
+        assert!(function_bindings(&schema).unwrap().is_empty());
+    }
+
+    /// A table carrying two bindings retires only the one whose outputs the
+    /// drop names; the other keeps its envelope entry and its protection.
+    #[tokio::test]
+    async fn retiring_one_binding_leaves_the_others_standing() {
+        let conn = connect("memory://").execute().await.unwrap();
+        let batch = record_batch!(
+            ("title", Utf8, ["a"]),
+            ("body", Utf8, ["b"]),
+            ("search_text", Utf8, ["a b"]),
+            ("search_token_count", Int64, [2]),
+            ("search_text_2", Utf8, ["a b"]),
+            ("search_token_count_2", Int64, [2])
+        )
+        .unwrap();
+        let table = conn.create_table("two", batch).execute().await.unwrap();
+        stamp_bindings(
+            &table,
+            &[text_features_binding(""), text_features_binding("_2")],
+        )
+        .await;
+
+        table
+            .drop_columns(&["search_text", "search_token_count"])
+            .await
+            .unwrap();
+        let remaining = bindings_of(&table).await;
+        assert_eq!(remaining.len(), 1);
+        assert_eq!(remaining[0].binding_id(), "fb_01K3TEXT_2");
+        // The survivor still protects the inputs it shared with the retired one.
+        bound(table.drop_columns(&["title"]).await.unwrap_err());
+        // And its own outputs are still a group.
+        incomplete_group(table.drop_columns(&["search_text_2"]).await.unwrap_err());
+    }
+
+    /// A multi-output binding also owns a hidden `__function_assignment_*`
+    /// column, which the caller never names. Retiring the binding has to take
+    /// it: leaving it behind orphans a column nothing can fill or explain.
+    #[tokio::test]
+    async fn retiring_a_binding_drops_its_assignment_column() {
+        const ASSIGNMENT: &str = "__function_assignment_fb_01K3TEXT";
+
+        let conn = connect("memory://").execute().await.unwrap();
+        let batch = record_batch!(
+            ("title", Utf8, ["a"]),
+            ("body", Utf8, ["b"]),
+            ("search_text", Utf8, ["a b"]),
+            ("search_token_count", Int64, [2]),
+            (ASSIGNMENT, Boolean, [Some(true)]),
+            ("spare", Int32, [1])
+        )
+        .unwrap();
+        let table = conn
+            .create_table("assigned", batch)
+            .execute()
+            .await
+            .unwrap();
+
+        let raw = include_str!(
+            "../../tests/fixtures/first_class_functions/v1/remote_function_binding.json"
+        )
+        .replace(
+            r#""future_binding""#,
+            &format!(
+                r#""assignment": {{"output_name": "{ASSIGNMENT}", "output_field_id": -1}}, "future_binding""#
+            ),
+        )
+        // The assignment column is a physical sibling, so it belongs to the
+        // binding's output schema too -- the server writes it that way.
+        .replace(
+            r#"{"name": "search_token_count", "nullable": true, "type": {"type": "int64"}}"#,
+            &format!(
+                r#"{{"name": "search_token_count", "nullable": true, "type": {{"type": "int64"}}}}, {{"name": "{ASSIGNMENT}", "nullable": true, "type": {{"type": "bool"}}}}"#
+            ),
+        );
+        let binding = FunctionBinding::from_json(&raw).unwrap();
+        assert!(binding.assignment().is_some(), "fixture must carry one");
+        stamp_bindings(&table, std::slice::from_ref(&binding)).await;
+
+        // The assignment column is the binding's, so it is protected too...
+        bound(table.drop_columns(&[ASSIGNMENT]).await.unwrap_err());
+
+        // ...and goes with the outputs without the caller naming it.
+        table
+            .drop_columns(&["search_text", "search_token_count"])
+            .await
+            .unwrap();
+        let schema = table.schema().await.unwrap();
+        assert!(
+            schema.field_with_name(ASSIGNMENT).is_err(),
+            "the assignment column outlived its binding: {:?}",
+            schema.fields().iter().map(|f| f.name()).collect::<Vec<_>>()
+        );
+        assert!(bindings_of(&table).await.is_empty());
+        ensure_supported_function_metadata(&schema).unwrap();
+    }
+
+    /// An input error must not cost the binding. The retirement commits
+    /// first, so a projection that could never succeed is refused before it,
+    /// leaving the table exactly as it was.
+    #[rstest::rstest]
+    #[case::unknown_column(
+        &["search_text", "search_token_count", "nope"],
+        "does not exist"
+    )]
+    #[case::every_column(
+        &["title", "body", "search_text", "search_token_count", "spare"],
+        "Cannot drop all columns"
+    )]
+    #[case::every_column_quoted(
+        &["`title`", "`body`", "`search_text`", "`search_token_count`", "`spare`"],
+        "Cannot drop all columns"
+    )]
+    #[tokio::test]
+    async fn an_invalid_drop_leaves_the_binding_in_place(
+        #[case] columns: &[&str],
+        #[case] expected: &str,
+    ) {
+        let table = bound_table().await;
+        let version = table.version().await.unwrap();
+
+        let err = table.drop_columns(columns).await.unwrap_err();
+        let Error::InvalidInput { message } = &err else {
+            panic!("{err:?}");
+        };
+        assert!(message.contains(expected), "{message}");
+
+        // Read durable state, not this handle: the error path never calls
+        // `dataset.update`, so the cached handle would report the old version
+        // even if the unbind had committed.
+        table.checkout_latest().await.unwrap();
+        assert_eq!(table.version().await.unwrap(), version);
+        let schema = table.schema().await.unwrap();
+        assert!(schema.metadata().contains_key(FUNCTION_BINDINGS_META_KEY));
+        assert_eq!(bindings_of(&table).await.len(), 1);
+        for column in ["search_text", "search_token_count"] {
+            assert!(
+                schema
+                    .field_with_name(column)
+                    .unwrap()
+                    .metadata()
+                    .keys()
+                    .any(|key| is_declaration_key(key)),
+                "{column} lost its declaration metadata"
+            );
+        }
+        ensure_supported_function_metadata(&schema).unwrap();
+    }
+
     #[tokio::test]
     async fn test_update_field_metadata() {
         let conn = connect("memory://").execute().await.unwrap();