feat(rust): add conditional function name removal

rust/lancedb/src/connection.rs

@@ -590,6 +590,33 @@ impl Connection {
         self.internal.lookup_function_by_id(function_id).await
     }
 
+    /// Conditionally remove a database-scoped Function catalog name.
+    ///
+    /// This is a direct synchronous catalog compare-and-swap (CAS), not a
+    /// [`crate::job::Job`], not physical [`Function`] deletion, and not
+    /// revocation. The caller supplies an observed immutable [`Function`]
+    /// handle; only [`Function::id`] is authority for the CAS precondition.
+    ///
+    /// Empty names return [`Error::InvalidInput`] before backend dispatch.
+    /// Nonempty names on local/default backends return [`Error::NotSupported`].
+    /// Remote backends complete only when the server reports durable CAS
+    /// success for the `(name, current.id)` pair.
+    pub async fn remove_function_name(
+        &self,
+        name: impl AsRef<str>,
+        current: &Function,
+    ) -> Result<()> {
+        let name = name.as_ref();
+        // Public nonempty invariant: validate before any Database backend sees
+        // the call so local and remote Connections agree on InvalidInput.
+        if name.is_empty() {
+            return Err(Error::InvalidInput {
+                message: "function name removal name must be non-empty".into(),
+            });
+        }
+        self.internal.remove_function_name(name, current).await
+    }
+
     /// Drop a table in the database.
     ///
     /// # Arguments

rust/lancedb/src/database.rs

@@ -350,6 +350,24 @@ pub trait Database:
     async fn lookup_function_by_id(&self, _function_id: &FunctionId) -> Result<Function> {
         job_op_not_supported("lookup_function_by_id")
     }
+    /// Conditionally remove a database-scoped Function catalog name.
+    ///
+    /// Direct synchronous catalog CAS, not a Job and not physical Function
+    /// deletion. Empty names return [`crate::Error::InvalidInput`] before the
+    /// unsupported fallback so local and remote backends agree. Nonempty names
+    /// on databases without enterprise catalog mutation return
+    /// [`crate::Error::NotSupported`].
+    async fn remove_function_name(&self, name: &str, _current: &Function) -> Result<()> {
+        // Public nonempty invariant on the Database trait seam itself:
+        // Connection::database() exposes Arc<dyn Database>, so empty-name
+        // rejection cannot rely solely on Connection prevalidation.
+        if name.is_empty() {
+            return Err(crate::error::Error::InvalidInput {
+                message: "function name removal name must be non-empty".into(),
+            });
+        }
+        job_op_not_supported("remove_function_name")
+    }
     /// Open a table in the database
     async fn open_table(&self, request: OpenTableRequest) -> Result<Arc<dyn BaseTable>>;
     /// Rename a table in the database

rust/lancedb/src/remote/client.rs

@@ -793,10 +793,10 @@ impl<S: HttpSend> RestfulLanceDbClient<S> {
 
     /// Send one attempt with a caller-owned request id.
     ///
-    /// Used by Function lookup so the helper can inspect non-success bodies for
-    /// an explicit `error_code` before deciding whether to retry. Does not log
-    /// request bodies or headers (lookup selectors must stay out of logs) and
-    /// does not interpret HTTP status.
+    /// Used by Function catalog helpers (lookup/remove) so they can inspect
+    /// non-success bodies for an explicit `error_code` before deciding whether
+    /// to retry. Does not log request bodies or headers (selectors must stay
+    /// out of logs) and does not interpret HTTP status.
     pub(crate) async fn send_attempt_with_request_id(
         &self,
         req_builder: RequestBuilder,

rust/lancedb/src/remote/db.rs

@@ -627,6 +627,10 @@ impl<S: HttpSend> Database for RemoteDatabase<S> {
         super::function::lookup_function(&self.client, selector).await
     }
 
+    async fn remove_function_name(&self, name: &str, current: &Function) -> Result<()> {
+        super::function::remove_function_name(&self.client, name, current).await
+    }
+
     async fn table_names(&self, request: TableNamesRequest) -> Result<Vec<String>> {
         let mut req = if !request.namespace_path.is_empty() {
             let namespace_id =
@@ -4178,4 +4182,730 @@ mod tests {
             "nonretryable client/header error must not be repeated"
         );
     }
+
+    // -------------------------------------------------------------------------
+    // Conditional Function name removal
+    //
+    // Direct synchronous catalog CAS via POST /v1/functions/remove. Not a Job
+    // and not physical Function deletion. Caller supplies an observed immutable
+    // Function handle; only current.id is authority on the wire.
+    // -------------------------------------------------------------------------
+
+    const REMOVE_CATALOG_NAME: &str = "text.normalize.remove-name";
+    const REMOVE_FUNCTION_ID: &str = "fn.exact.remove-handle";
+    const REMOVE_SERVER_MESSAGE_MARKER: &str =
+        "SERVER_REMOVE_DIAGNOSTIC_MARKER name=text.normalize.remove-name id=fn.exact.remove-handle";
+
+    fn sample_remove_function() -> Function {
+        let id = FunctionId::try_new(REMOVE_FUNCTION_ID).expect("valid FunctionId");
+        let signature = FunctionSignature::try_new(
+            vec![
+                FunctionParameter::new("text", DataType::Utf8),
+                FunctionParameter::new("limit", DataType::Int32),
+            ],
+            FunctionOutput::new(DataType::Utf8, true),
+        )
+        .expect("valid FunctionSignature");
+        Function::new(id, signature)
+    }
+
+    fn remove_error_chain_text(err: &Error) -> String {
+        let mut text = format!("{err}\n{err:?}");
+        let mut current: Option<&(dyn std::error::Error + 'static)> = Some(err);
+        while let Some(e) = current {
+            text.push('\n');
+            text.push_str(&e.to_string());
+            text.push('\n');
+            text.push_str(&format!("{e:?}"));
+            current = e.source();
+        }
+        text
+    }
+
+    fn assert_remove_payload_free(err: &Error) {
+        let text = remove_error_chain_text(err);
+        assert!(
+            !text.contains(REMOVE_SERVER_MESSAGE_MARKER),
+            "server diagnostic marker must be absent from error/debug/source chain: {text}"
+        );
+        assert!(
+            !text.contains(REMOVE_CATALOG_NAME),
+            "catalog name must be absent from error/debug/source chain: {text}"
+        );
+        assert!(
+            !text.contains(REMOVE_FUNCTION_ID),
+            "FunctionId must be absent from error/debug/source chain: {text}"
+        );
+        assert!(
+            !text.contains("SENSITIVE_REMOVE_BODY_MARKER"),
+            "non-success/malformed body marker must be absent from error/debug/source chain: {text}"
+        );
+    }
+
+    fn assert_remove_request(
+        request: &reqwest::Request,
+        expected_name: &str,
+        expected_function_id: &str,
+    ) {
+        assert_eq!(request.method(), &reqwest::Method::POST);
+        assert_eq!(request.url().path(), "/v1/functions/remove");
+        assert!(
+            request.url().query().is_none(),
+            "remove selectors must stay out of the URL query: {}",
+            request.url()
+        );
+        let body = request
+            .body()
+            .and_then(|b| b.as_bytes())
+            .expect("remove request must carry a JSON body");
+        let actual: Value = serde_json::from_slice(body).expect("remove body must be JSON");
+        assert_eq!(
+            actual,
+            json!({
+                "name": expected_name,
+                "expected_current_function_id": expected_function_id,
+            }),
+            "remove body must be exactly {{\"name\":...,\"expected_current_function_id\":...}}"
+        );
+        let object = actual.as_object().expect("remove body must be an object");
+        assert_eq!(
+            object.len(),
+            2,
+            "remove body must not carry format_version or extra user fields: {actual}"
+        );
+        assert!(
+            object.get("format_version").is_none(),
+            "remove body must not include format_version: {actual}"
+        );
+        assert!(
+            object.get("function_id").is_none(),
+            "remove body must use expected_current_function_id, not function_id: {actual}"
+        );
+        assert!(
+            object.get("current").is_none() && object.get("function").is_none(),
+            "remove must not send a Function record: {actual}"
+        );
+        assert!(
+            object.get("job_id").is_none() && object.get("idempotency_key").is_none(),
+            "remove is not a Job and must not send user idempotency keys: {actual}"
+        );
+    }
+
+    /// Exact path/body and 204 success use only the observed Function id.
+    #[tokio::test]
+    async fn remove_function_name_posts_exact_body_and_succeeds_on_204() {
+        let current = sample_remove_function();
+        let expected_id = current.id().as_str().to_string();
+        let conn = Connection::new_with_handler(move |request| {
+            assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id);
+            // Illegal body on 204 must be ignored; success is status-driven only.
+            http::Response::builder()
+                .status(204)
+                .body(format!(
+                    "{{\"SENSITIVE_REMOVE_BODY_MARKER\":true,\"message\":{REMOVE_SERVER_MESSAGE_MARKER:?}}}"
+                ))
+                .unwrap()
+        });
+
+        conn.remove_function_name(REMOVE_CATALOG_NAME, &current)
+            .await
+            .expect("HTTP 204 must complete the CAS");
+    }
+
+    /// One configured 5xx retry then 204 keeps identical request id/body.
+    #[tokio::test]
+    async fn remove_function_name_retry_preserves_request_id_and_body() {
+        let current = sample_remove_function();
+        let expected_id = current.id().as_str().to_string();
+        let seen_request_id = Arc::new(OnceLock::new());
+        let seen_request_id_ref = seen_request_id.clone();
+        let attempts = Arc::new(AtomicUsize::new(0));
+        let attempts_ref = attempts.clone();
+
+        let conn = Connection::new_with_handler_and_config(
+            move |request| {
+                assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id);
+                let request_id = request.headers()["x-request-id"]
+                    .to_str()
+                    .unwrap()
+                    .to_string();
+                assert!(!request_id.is_empty(), "SDK must generate a request id");
+                let seen = seen_request_id_ref.get_or_init(|| request_id.clone());
+                assert_eq!(
+                    &request_id, seen,
+                    "request id must be identical across retries"
+                );
+
+                let n = attempts_ref.fetch_add(1, Ordering::SeqCst);
+                if n == 0 {
+                    http::Response::builder()
+                        .status(500)
+                        .body(format!(
+                            "{REMOVE_SERVER_MESSAGE_MARKER} SENSITIVE_REMOVE_BODY_MARKER"
+                        ))
+                        .unwrap()
+                } else {
+                    http::Response::builder()
+                        .status(204)
+                        .body(String::new())
+                        .unwrap()
+                }
+            },
+            ClientConfig {
+                retry_config: RetryConfig {
+                    retries: Some(2),
+                    backoff_factor: Some(0.0),
+                    backoff_jitter: Some(0.0),
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+        );
+
+        conn.remove_function_name(REMOVE_CATALOG_NAME, &current)
+            .await
+            .expect("remove must succeed after one retry");
+        assert_eq!(
+            attempts.load(Ordering::SeqCst),
+            2,
+            "one 5xx then 204 must be exactly two attempts"
+        );
+        assert!(seen_request_id.get().is_some());
+    }
+
+    /// Exhausted always-retryable 5xx without explicit error_code surfaces Error::Retry.
+    ///
+    /// retries=2 is max request failures: exactly two identical attempts, then
+    /// Retry with request_failures == max_request_failures == 2, zero connect/read
+    /// failures, the retryable status retained, and a payload-free source chain.
+    #[tokio::test]
+    async fn remove_function_name_exhausted_retryable_5xx_returns_retry_with_request_counters() {
+        let current = sample_remove_function();
+        let expected_id = current.id().as_str().to_string();
+        let seen_request_id = Arc::new(OnceLock::new());
+        let seen_request_id_ref = seen_request_id.clone();
+        let attempts = Arc::new(AtomicUsize::new(0));
+        let attempts_ref = attempts.clone();
+        let body = format!("{REMOVE_SERVER_MESSAGE_MARKER} SENSITIVE_REMOVE_BODY_MARKER");
+
+        let conn = Connection::new_with_handler_and_config(
+            move |request| {
+                assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id);
+                let request_id = request.headers()["x-request-id"]
+                    .to_str()
+                    .unwrap()
+                    .to_string();
+                assert!(!request_id.is_empty(), "SDK must generate a request id");
+                let seen = seen_request_id_ref.get_or_init(|| request_id.clone());
+                assert_eq!(
+                    &request_id, seen,
+                    "request id must be identical across exhausted retries"
+                );
+
+                attempts_ref.fetch_add(1, Ordering::SeqCst);
+                http::Response::builder()
+                    .status(500)
+                    .body(body.clone())
+                    .unwrap()
+            },
+            ClientConfig {
+                retry_config: RetryConfig {
+                    // RetryCounter treats `retries` as max request failures, so
+                    // retries=2 yields exactly two transport attempts before Error::Retry.
+                    retries: Some(2),
+                    backoff_factor: Some(0.0),
+                    backoff_jitter: Some(0.0),
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+        );
+
+        let err = conn
+            .remove_function_name(REMOVE_CATALOG_NAME, &current)
+            .await
+            .expect_err("exhausted retryable 5xx must fail");
+        match &err {
+            Error::Retry {
+                request_failures,
+                max_request_failures,
+                connect_failures,
+                read_failures,
+                status_code,
+                ..
+            } => {
+                assert_eq!(*request_failures, 2);
+                assert_eq!(*max_request_failures, 2);
+                assert_eq!(
+                    request_failures, max_request_failures,
+                    "request budget must be fully exhausted"
+                );
+                assert_eq!(*connect_failures, 0, "5xx must not consume connect budget");
+                assert_eq!(*read_failures, 0, "5xx must not consume read budget");
+                assert_eq!(
+                    status_code.map(|s| s.as_u16()),
+                    Some(500),
+                    "retryable status must be retained on Error::Retry"
+                );
+            }
+            other => panic!("exhausted 5xx must surface as Error::Retry, got {other:?}"),
+        }
+        assert_eq!(
+            attempts.load(Ordering::SeqCst),
+            2,
+            "retries=2 must make exactly two identical attempts"
+        );
+        assert!(seen_request_id.get().is_some());
+        assert_remove_payload_free(&err);
+    }
+
+    /// Explicit name_conflict on a retryable status is terminal Error::Function.
+    #[tokio::test]
+    async fn remove_function_name_explicit_name_conflict_is_terminal_on_retryable_status() {
+        let current = sample_remove_function();
+        let expected_id = current.id().as_str().to_string();
+        let attempts = Arc::new(AtomicUsize::new(0));
+        let attempts_ref = attempts.clone();
+        let body = json!({
+            "error_code": "name_conflict",
+            "message": format!(
+                "{REMOVE_SERVER_MESSAGE_MARKER} looks_like revoked_function"
+            ),
+            "SENSITIVE_REMOVE_BODY_MARKER": true,
+        })
+        .to_string();
+
+        let conn = Connection::new_with_handler_and_config(
+            move |request| {
+                assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id);
+                attempts_ref.fetch_add(1, Ordering::SeqCst);
+                http::Response::builder()
+                    .status(503)
+                    .body(body.clone())
+                    .unwrap()
+            },
+            ClientConfig {
+                retry_config: RetryConfig {
+                    retries: Some(3),
+                    backoff_factor: Some(0.0),
+                    backoff_jitter: Some(0.0),
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+        );
+
+        let err = conn
+            .remove_function_name(REMOVE_CATALOG_NAME, &current)
+            .await
+            .expect_err("explicit name_conflict must fail");
+        match &err {
+            Error::Function { code, message } => {
+                assert_eq!(code.as_str(), "name_conflict");
+                assert!(
+                    matches!(code, FunctionErrorCode::NameConflict),
+                    "expected NameConflict, got {code:?}"
+                );
+                assert_ne!(code.as_str(), "revoked_function");
+                assert!(
+                    !message.contains(REMOVE_SERVER_MESSAGE_MARKER),
+                    "Function error message must be sanitized, got {message}"
+                );
+                assert!(
+                    !message.contains(REMOVE_CATALOG_NAME),
+                    "Function error message must not echo the catalog name, got {message}"
+                );
+                assert!(
+                    !message.contains(REMOVE_FUNCTION_ID),
+                    "Function error message must not echo the FunctionId, got {message}"
+                );
+            }
+            other => panic!("expected Error::Function, got {other:?}"),
+        }
+        assert_remove_payload_free(&err);
+        assert_eq!(
+            attempts.load(Ordering::SeqCst),
+            1,
+            "explicit semantic code must not consume request retries"
+        );
+    }
+
+    /// Unknown nonempty explicit code is preserved; status/message do not override.
+    #[tokio::test]
+    async fn remove_function_name_preserves_unknown_explicit_code_despite_status_and_message() {
+        let current = sample_remove_function();
+        let expected_id = current.id().as_str().to_string();
+        let raw = "enterprise_future_remove_category_xyz";
+        let body = json!({
+            "error_code": raw,
+            "message": format!(
+                "{REMOVE_SERVER_MESSAGE_MARKER} name_conflict revoked_function"
+            ),
+            "SENSITIVE_REMOVE_BODY_MARKER": true,
+        })
+        .to_string();
+        let conn = Connection::new_with_handler(move |request| {
+            assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id);
+            http::Response::builder()
+                .status(409)
+                .body(body.clone())
+                .unwrap()
+        });
+
+        let err = conn
+            .remove_function_name(REMOVE_CATALOG_NAME, &current)
+            .await
+            .expect_err("unknown explicit code must surface");
+        match &err {
+            Error::Function { code, message } => {
+                assert_eq!(code.as_str(), raw);
+                assert!(
+                    matches!(code, FunctionErrorCode::Unrecognized(_)),
+                    "unknown code must stay Unrecognized, got {code:?}"
+                );
+                assert_ne!(code.as_str(), "name_conflict");
+                assert_ne!(code.as_str(), "revoked_function");
+                assert!(
+                    !message.contains(REMOVE_SERVER_MESSAGE_MARKER),
+                    "diagnostic message must be sanitized"
+                );
+            }
+            other => panic!("expected Error::Function, got {other:?}"),
+        }
+        assert_remove_payload_free(&err);
+    }
+
+    /// Missing/empty/null/wrong-type/malformed error_code stays payload-free Http.
+    #[tokio::test]
+    async fn remove_function_name_missing_or_invalid_error_code_is_payload_free_http() {
+        let cases: Vec<(&str, u16, String)> = vec![
+            (
+                "missing_code_404",
+                404,
+                json!({
+                    "message": REMOVE_SERVER_MESSAGE_MARKER,
+                    "SENSITIVE_REMOVE_BODY_MARKER": true,
+                })
+                .to_string(),
+            ),
+            (
+                "empty_code",
+                400,
+                json!({
+                    "error_code": "",
+                    "message": REMOVE_SERVER_MESSAGE_MARKER,
+                    "SENSITIVE_REMOVE_BODY_MARKER": true,
+                })
+                .to_string(),
+            ),
+            (
+                "wrong_type_code",
+                400,
+                json!({
+                    "error_code": 123,
+                    "message": REMOVE_SERVER_MESSAGE_MARKER,
+                    "SENSITIVE_REMOVE_BODY_MARKER": true,
+                })
+                .to_string(),
+            ),
+            (
+                // Non-retryable status: invalid/null error_code must stay immediate Http.
+                // Exhausted retryable 5xx is covered by
+                // remove_function_name_exhausted_retryable_5xx_returns_retry_with_request_counters.
+                "null_code",
+                400,
+                json!({
+                    "error_code": null,
+                    "message": REMOVE_SERVER_MESSAGE_MARKER,
+                    "SENSITIVE_REMOVE_BODY_MARKER": true,
+                })
+                .to_string(),
+            ),
+            (
+                // Non-retryable status: this case proves invalid-code -> Http only.
+                // Exhausted retryable 5xx without error_code is covered separately
+                // by remove_function_name_exhausted_retryable_5xx_returns_retry_with_request_counters.
+                "non_json",
+                400,
+                format!("not-json {REMOVE_SERVER_MESSAGE_MARKER} SENSITIVE_REMOVE_BODY_MARKER"),
+            ),
+        ];
+
+        let mut unexpected = Vec::new();
+        for (label, status, response_body) in cases {
+            let current = sample_remove_function();
+            let expected_id = current.id().as_str().to_string();
+            let body_for_handler = response_body.clone();
+            let conn = Connection::new_with_handler(move |request| {
+                assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id);
+                http::Response::builder()
+                    .status(status)
+                    .body(body_for_handler.clone())
+                    .unwrap()
+            });
+            match conn
+                .remove_function_name(REMOVE_CATALOG_NAME, &current)
+                .await
+            {
+                Err(err @ Error::Http { .. }) => assert_remove_payload_free(&err),
+                Err(Error::Function { .. }) => unexpected.push(format!(
+                    "{label}: must not invent Error::Function without explicit code"
+                )),
+                other => unexpected.push(format!("{label}: {other:?}")),
+            }
+        }
+        assert!(
+            unexpected.is_empty(),
+            "invalid/missing error_code must stay payload-free Http: {unexpected:?}"
+        );
+    }
+
+    /// 200/202 are payload-free protocol Http failures, never CAS success.
+    #[tokio::test]
+    async fn remove_function_name_other_2xx_are_payload_free_http_failures() {
+        let cases: Vec<(&str, u16, String)> = vec![
+            (
+                "200_with_body",
+                200,
+                json!({
+                    "ok": true,
+                    "message": REMOVE_SERVER_MESSAGE_MARKER,
+                    "SENSITIVE_REMOVE_BODY_MARKER": true,
+                    "job_id": "must-not-infer-job",
+                })
+                .to_string(),
+            ),
+            (
+                "202_empty",
+                202,
+                format!("{REMOVE_SERVER_MESSAGE_MARKER} SENSITIVE_REMOVE_BODY_MARKER"),
+            ),
+            ("200_empty", 200, String::new()),
+        ];
+
+        let mut unexpected = Vec::new();
+        for (label, status, response_body) in cases {
+            let current = sample_remove_function();
+            let expected_id = current.id().as_str().to_string();
+            let body_for_handler = response_body.clone();
+            let conn = Connection::new_with_handler(move |request| {
+                assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id);
+                http::Response::builder()
+                    .status(status)
+                    .body(body_for_handler.clone())
+                    .unwrap()
+            });
+            match conn
+                .remove_function_name(REMOVE_CATALOG_NAME, &current)
+                .await
+            {
+                Ok(()) => {
+                    unexpected.push(format!("{label}: must not treat non-204 2xx as success"))
+                }
+                Err(err @ Error::Http { .. }) => assert_remove_payload_free(&err),
+                Err(Error::Function { .. }) => unexpected.push(format!(
+                    "{label}: must not invent Error::Function from 2xx body"
+                )),
+                other => unexpected.push(format!("{label}: {other:?}")),
+            }
+        }
+        assert!(
+            unexpected.is_empty(),
+            "200/202 must be payload-free Http failures: {unexpected:?}"
+        );
+    }
+
+    /// Non-204 success must fail from status alone without reading the body.
+    ///
+    /// A protocol-invalid HTTP 200 whose body stream fails if read must return
+    /// payload-free Error::Http with status 200 on exactly one attempt, even when
+    /// read/request retry budgets are configured above one. Body must not be
+    /// read and no retry budget may be consumed.
+    #[tokio::test]
+    async fn remove_function_name_non_204_success_does_not_read_failing_body() {
+        let current = sample_remove_function();
+        let expected_id = current.id().as_str().to_string();
+        let attempts = Arc::new(AtomicUsize::new(0));
+        let attempts_ref = attempts.clone();
+
+        let conn = Connection::new_with_handler_and_config(
+            move |request| {
+                assert_remove_request(&request, REMOVE_CATALOG_NAME, &expected_id);
+                attempts_ref.fetch_add(1, Ordering::SeqCst);
+                let stream = futures::stream::once(async {
+                    Err::<bytes::Bytes, _>(std::io::Error::other(
+                        "simulated remove response body read failure SENSITIVE_REMOVE_BODY_MARKER",
+                    ))
+                });
+                http::Response::builder()
+                    .status(200)
+                    .body(reqwest::Body::wrap_stream(stream))
+                    .unwrap()
+            },
+            ClientConfig {
+                // Budgets above one must not be consumed: status 200 is terminal
+                // protocol Http before any body read/retry classification.
+                retry_config: RetryConfig {
+                    retries: Some(3),
+                    read_retries: Some(3),
+                    connect_retries: Some(3),
+                    backoff_factor: Some(0.0),
+                    backoff_jitter: Some(0.0),
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+        );
+
+        let err = conn
+            .remove_function_name(REMOVE_CATALOG_NAME, &current)
+            .await
+            .expect_err("non-204 success must be protocol Http");
+        assert_eq!(
+            attempts.load(Ordering::SeqCst),
+            1,
+            "invalid 2xx must not read the body or consume retry budget"
+        );
+        match &err {
+            Error::Http { status_code, .. } => {
+                assert_eq!(
+                    status_code.map(|s| s.as_u16()),
+                    Some(200),
+                    "protocol Http must retain status 200 from the response alone"
+                );
+            }
+            other => panic!("expected payload-free Error::Http, got {other:?}"),
+        }
+        assert_remove_payload_free(&err);
+    }
+
+    /// Empty name is InvalidInput before backend through Connection.
+    #[tokio::test]
+    async fn remove_function_name_empty_name_fails_before_transport() {
+        let current = sample_remove_function();
+        let attempts = Arc::new(AtomicUsize::new(0));
+        let attempts_ref = attempts.clone();
+        let conn = Connection::new_with_handler(move |_request| -> http::Response<String> {
+            attempts_ref.fetch_add(1, Ordering::SeqCst);
+            panic!("empty name must not issue an HTTP request");
+        });
+
+        let err = conn
+            .remove_function_name("", &current)
+            .await
+            .expect_err("empty name must fail before transport");
+        assert!(
+            matches!(err, Error::InvalidInput { .. }),
+            "empty name must be InvalidInput, got {err:?}"
+        );
+        assert_eq!(
+            attempts.load(Ordering::SeqCst),
+            0,
+            "empty name must not touch transport"
+        );
+        assert_remove_payload_free(&err);
+    }
+
+    /// Database trait seam must reject empty names as InvalidInput (not NotSupported).
+    #[tokio::test]
+    async fn remove_function_name_database_trait_empty_name_is_invalid_input_without_mutation() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let conn = ConnectBuilder::new(dir.path().to_str().unwrap())
+            .execute()
+            .await
+            .expect("local connect");
+        let before = conn
+            .table_names()
+            .execute()
+            .await
+            .expect("table_names before");
+        assert!(before.is_empty());
+
+        let current = sample_remove_function();
+        let err = conn
+            .database()
+            .remove_function_name("", &current)
+            .await
+            .expect_err("empty name must fail on Database trait default");
+        assert!(
+            matches!(err, Error::InvalidInput { .. }),
+            "empty name via Database trait must be InvalidInput, got {err:?}"
+        );
+
+        let after = conn
+            .table_names()
+            .execute()
+            .await
+            .expect("table_names after");
+        assert_eq!(
+            before, after,
+            "Database-trait empty-name rejection must not mutate tables"
+        );
+    }
+
+    /// Valid local removal is NotSupported and does not mutate tables.
+    #[tokio::test]
+    async fn remove_function_name_local_database_returns_not_supported_without_mutation() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let conn = ConnectBuilder::new(dir.path().to_str().unwrap())
+            .execute()
+            .await
+            .expect("local connect");
+        let before = conn
+            .table_names()
+            .execute()
+            .await
+            .expect("table_names before");
+        assert!(before.is_empty());
+
+        let current = sample_remove_function();
+        let err = conn
+            .remove_function_name(REMOVE_CATALOG_NAME, &current)
+            .await
+            .expect_err("local remove_function_name must be unsupported");
+        assert!(
+            matches!(err, Error::NotSupported { .. }),
+            "expected NotSupported for local remove, got {err:?}"
+        );
+
+        let after = conn
+            .table_names()
+            .execute()
+            .await
+            .expect("table_names after");
+        assert_eq!(before, after, "unsupported remove must not mutate tables");
+    }
+
+    /// Empty name on local Connection is InvalidInput before NotSupported.
+    #[tokio::test]
+    async fn remove_function_name_local_empty_name_is_invalid_input_without_mutation() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let conn = ConnectBuilder::new(dir.path().to_str().unwrap())
+            .execute()
+            .await
+            .expect("local connect");
+        let before = conn
+            .table_names()
+            .execute()
+            .await
+            .expect("table_names before");
+        assert!(before.is_empty());
+
+        let current = sample_remove_function();
+        let err = conn
+            .remove_function_name("", &current)
+            .await
+            .expect_err("empty name must fail before backend dispatch");
+        assert!(
+            matches!(err, Error::InvalidInput { .. }),
+            "empty name must be InvalidInput on local Connection, got {err:?}"
+        );
+
+        let after = conn
+            .table_names()
+            .execute()
+            .await
+            .expect("table_names after");
+        assert_eq!(before, after, "empty-name rejection must not mutate tables");
+    }
 }

rust/lancedb/src/remote/function.rs

@@ -1,12 +1,17 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
-//! Remote first-class Function catalog lookup wire helper.
+//! Remote first-class Function catalog wire helpers.
 //!
 //! POST `/v1/functions/lookup` resolves a database-scoped name or exact
 //! [`FunctionId`] to an immutable [`Function`] value. Name is lookup
 //! indirection only and never becomes part of the returned handle.
+//!
+//! POST `/v1/functions/remove` performs a direct synchronous catalog CAS that
+//! unbinds a name when the caller's observed [`Function`] id still matches.
+//! This is not a Job, not physical Function deletion, and not revocation.
 
+use reqwest::{RequestBuilder, StatusCode};
 use serde::Deserialize;
 use serde_json::Value;
 
@@ -17,6 +22,7 @@ use super::client::{HttpSend, RestfulLanceDbClient};
 use super::retry::RetryCounter;
 
 const LOOKUP_PATH: &str = "/v1/functions/lookup";
+const REMOVE_PATH: &str = "/v1/functions/remove";
 
 /// Fixed client diagnostic for [`Error::Function`]. Never carry server text,
 /// selector values, or response payload bytes.
@@ -29,6 +35,13 @@ const LOOKUP_HTTP_ERROR_MESSAGE: &str = "function lookup request failed";
 /// Fixed client diagnostic for malformed success payloads.
 const LOOKUP_INVALID_SUCCESS_MESSAGE: &str = "function lookup response missing or invalid function";
 
+/// Fixed client diagnostic for remove [`Error::Function`]. Never carry server
+/// text, catalog name, Function id, or response payload bytes.
+const REMOVE_FUNCTION_ERROR_MESSAGE: &str = "function name removal failed";
+
+/// Fixed client diagnostic for remove protocol / HTTP failures.
+const REMOVE_HTTP_ERROR_MESSAGE: &str = "function name removal request failed";
+
 /// One exact lookup selector. Exactly one variant is serialized on the wire.
 pub enum FunctionLookupSelector {
     Name(String),
@@ -65,6 +78,23 @@ struct LookupSuccessResponse {
     function: Function,
 }
 
+/// Decision before reading response bytes.
+enum BeforeBody<T> {
+    /// Finish without reading or interpreting any body (HTTP 204 remove CAS).
+    Done(Result<T>),
+    /// Read bytes and continue classification.
+    ReadBody,
+}
+
+/// How a catalog POST treats a successfully read response body.
+enum CatalogBodyAction<T> {
+    /// Terminal success or failure for this attempt.
+    Done(Result<T>),
+    /// Configured retryable status without an explicit `error_code`: consume
+    /// the request budget and retry with the same request id and body.
+    RetryRequest,
+}
+
 /// Resolve a Function via POST `/v1/functions/lookup`.
 ///
 /// Transport classification matches [`RestfulLanceDbClient::send_with_retry`]:
@@ -78,16 +108,125 @@ pub async fn lookup_function<S: HttpSend>(
     selector: FunctionLookupSelector,
 ) -> Result<Function> {
     let req_builder = client.post(LOOKUP_PATH).json(&selector.to_wire());
+    catalog_post_with_retry(
+        client,
+        req_builder,
+        LOOKUP_HTTP_ERROR_MESSAGE,
+        |_status, _request_id| BeforeBody::ReadBody,
+        |status, bytes, request_id| {
+            if status.is_success() {
+                return CatalogBodyAction::Done(decode_lookup_success(bytes, request_id));
+            }
+            if let Some(code) = explicit_error_code(bytes) {
+                return CatalogBodyAction::Done(Err(Error::Function {
+                    code,
+                    message: LOOKUP_FUNCTION_ERROR_MESSAGE.to_string(),
+                }));
+            }
+            if client.retry_config.statuses.contains(&status) {
+                return CatalogBodyAction::RetryRequest;
+            }
+            CatalogBodyAction::Done(Err(Error::Http {
+                source: LOOKUP_HTTP_ERROR_MESSAGE.into(),
+                request_id,
+                status_code: Some(status),
+            }))
+        },
+    )
+    .await
+}
 
-    let tmp_req = req_builder.try_clone().ok_or_else(|| Error::Runtime {
-        message: "Attempted to retry a request that cannot be cloned".to_string(),
-    })?;
-    let (_, built) = tmp_req.build_split();
-    let mut built = built.map_err(|e| Error::Runtime {
-        message: format!("Failed to build request: {}", e),
-    })?;
-    let request_id = client.extract_request_id(&mut built);
-    let mut retry_counter = RetryCounter::new(&client.retry_config, request_id.clone());
+/// Conditionally remove a database-scoped Function name via POST
+/// `/v1/functions/remove`.
+///
+/// Direct synchronous catalog CAS: the wire body is exactly
+/// `{"name","expected_current_function_id"}` using only `current.id`. Only
+/// HTTP 204 means the CAS completed; other 2xx are payload-free protocol
+/// [`Error::Http`]. Empty names are [`Error::InvalidInput`] before transport.
+///
+/// Retry budgets match lookup: stable internal request id and exact cloned
+/// body across attempts; response-byte failures consume read budget; configured
+/// retryable status without explicit `error_code` consumes request budget;
+/// header/client errors are immediate. Sophon deduplicates the internal request
+/// id; it is not a user-facing idempotency key.
+pub async fn remove_function_name<S: HttpSend>(
+    client: &RestfulLanceDbClient<S>,
+    name: &str,
+    current: &Function,
+) -> Result<()> {
+    if name.is_empty() {
+        return Err(Error::InvalidInput {
+            message: "function name removal name must be non-empty".into(),
+        });
+    }
+
+    // Authority is the observed immutable Function id only; never send
+    // signature, raw Function objects, Job fields, or user idempotency keys.
+    let body = serde_json::json!({
+        "name": name,
+        "expected_current_function_id": current.id().as_str(),
+    });
+    let req_builder = client.post(REMOVE_PATH).json(&body);
+
+    catalog_post_with_retry(
+        client,
+        req_builder,
+        REMOVE_HTTP_ERROR_MESSAGE,
+        |status, request_id| {
+            // Exact HTTP 204 completes the CAS; do not read or interpret any body.
+            if status == StatusCode::NO_CONTENT {
+                BeforeBody::Done(Ok(()))
+            } else if status.is_success() {
+                // Other 2xx are payload-free protocol failures from status alone.
+                BeforeBody::Done(Err(Error::Http {
+                    source: REMOVE_HTTP_ERROR_MESSAGE.into(),
+                    request_id: request_id.to_string(),
+                    status_code: Some(status),
+                }))
+            } else {
+                BeforeBody::ReadBody
+            }
+        },
+        |status, bytes, request_id| {
+            // Explicit nonempty error_code wins over HTTP status and precludes retry.
+            if let Some(code) = explicit_error_code(bytes) {
+                return CatalogBodyAction::Done(Err(Error::Function {
+                    code,
+                    message: REMOVE_FUNCTION_ERROR_MESSAGE.to_string(),
+                }));
+            }
+
+            if client.retry_config.statuses.contains(&status) {
+                return CatalogBodyAction::RetryRequest;
+            }
+
+            CatalogBodyAction::Done(Err(Error::Http {
+                source: REMOVE_HTTP_ERROR_MESSAGE.into(),
+                request_id,
+                status_code: Some(status),
+            }))
+        },
+    )
+    .await
+}
+
+/// Shared Function-catalog POST retry loop used by lookup and remove.
+///
+/// Sensitive attempt sending logs no body/header selectors. One SDK-generated
+/// request id and the exact cloned JSON body are reused across attempts.
+async fn catalog_post_with_retry<S, Before, After, T>(
+    client: &RestfulLanceDbClient<S>,
+    req_builder: RequestBuilder,
+    http_error_message: &'static str,
+    mut before_body: Before,
+    mut after_body: After,
+) -> Result<T>
+where
+    S: HttpSend,
+    Before: FnMut(StatusCode, &str) -> BeforeBody<T>,
+    After: FnMut(StatusCode, &[u8], String) -> CatalogBodyAction<T>,
+{
+    let mut retry_counter = prepare_catalog_retry(client, &req_builder)?;
 
     loop {
         let attempt = req_builder.try_clone().ok_or_else(|| Error::Runtime {
@@ -100,13 +239,18 @@ pub async fn lookup_function<S: HttpSend>(
         {
             Ok(rsp) => rsp,
             Err(err) => {
-                classify_lookup_send_error(&mut retry_counter, err)?;
+                classify_catalog_send_error(&mut retry_counter, err)?;
                 tokio::time::sleep(retry_counter.next_sleep_time()).await;
                 continue;
             }
         };
 
         let status = rsp.status();
+        match before_body(status, &retry_counter.request_id) {
+            BeforeBody::Done(result) => return result,
+            BeforeBody::ReadBody => {}
+        }
+
         let bytes = match rsp.bytes().await {
             Ok(bytes) => bytes,
             Err(err) => {
@@ -118,42 +262,41 @@ pub async fn lookup_function<S: HttpSend>(
             }
         };
 
-        if status.is_success() {
-            return decode_lookup_success(&bytes, retry_counter.request_id);
+        match after_body(status, &bytes, retry_counter.request_id.clone()) {
+            CatalogBodyAction::Done(result) => return result,
+            CatalogBodyAction::RetryRequest => {
+                let source = Error::Http {
+                    source: http_error_message.into(),
+                    request_id: retry_counter.request_id.clone(),
+                    status_code: Some(status),
+                };
+                retry_counter.increment_request_failures(source)?;
+                tokio::time::sleep(retry_counter.next_sleep_time()).await;
+            }
         }
-
-        // Explicit nonempty error_code wins over HTTP status and precludes retry.
-        if let Some(code) = explicit_error_code(&bytes) {
-            return Err(Error::Function {
-                code,
-                message: LOOKUP_FUNCTION_ERROR_MESSAGE.to_string(),
-            });
-        }
-
-        if client.retry_config.statuses.contains(&status) {
-            let source = Error::Http {
-                source: LOOKUP_HTTP_ERROR_MESSAGE.into(),
-                request_id: retry_counter.request_id.clone(),
-                status_code: Some(status),
-            };
-            retry_counter.increment_request_failures(source)?;
-            tokio::time::sleep(retry_counter.next_sleep_time()).await;
-            continue;
-        }
-
-        return Err(Error::Http {
-            source: LOOKUP_HTTP_ERROR_MESSAGE.into(),
-            request_id: retry_counter.request_id,
-            status_code: Some(status),
-        });
     }
 }
 
+fn prepare_catalog_retry<'a, S: HttpSend>(
+    client: &'a RestfulLanceDbClient<S>,
+    req_builder: &RequestBuilder,
+) -> Result<RetryCounter<'a>> {
+    let tmp_req = req_builder.try_clone().ok_or_else(|| Error::Runtime {
+        message: "Attempted to retry a request that cannot be cloned".to_string(),
+    })?;
+    let (_, built) = tmp_req.build_split();
+    let mut built = built.map_err(|e| Error::Runtime {
+        message: format!("Failed to build request: {}", e),
+    })?;
+    let request_id = client.extract_request_id(&mut built);
+    Ok(RetryCounter::new(&client.retry_config, request_id))
+}
+
 /// Classify a send-attempt error using the same buckets as `send_with_retry`.
 ///
 /// Returns `Ok(())` when the caller should sleep and retry. Returns `Err` for
 /// nonretryable failures or when a retry budget is exhausted (no extra attempt).
-fn classify_lookup_send_error(retry_counter: &mut RetryCounter<'_>, err: Error) -> Result<()> {
+fn classify_catalog_send_error(retry_counter: &mut RetryCounter<'_>, err: Error) -> Result<()> {
     match err {
         Error::Http {
             source,