fix: align Secret APIs with native SQL syntax (#4784)

Secret APIs now submit the native SQL statements with the credential
bound as an Arrow string parameter:

```sql
CREATE SECRET "api-key" TYPE = GENERIC_STRING SECRET_STRING = $1;
ALTER SECRET "api-key" SET SECRET_STRING = $1;
DESCRIBE SECRET "api-key";
```

`create_secret` and `alter_secret` execute SQL through the existing
query-backed job adapter and wait for that query to complete. The job's
`id` is the query ID and its type is `query`; credentials stay out of
SQL text. `describe_secret` reads the native `name`,
`created_at_millis`, and `updated_at_millis` columns. This replaces the
unreleased `AS $1` syntax and JSON-envelope response without a
compatibility fallback.

Python methods and namespace handling remain the same:

```python
db.create_secret("api-key", "credential", namespace_path=["team"])
db.alter_secret("api-key", "rotated", namespace_path=["team"])
info = db.describe_secret("api-key", namespace_path=["team"])
```

docs/web/basics/jobs.mdx

@@ -127,8 +127,9 @@ physical cleanup continues separately.
 
 Remote Secret create, alter, list, describe, and drop use the SQL query backend.
 The synchronous APIs wait on query completion internally. Create and alter bind
-the credential as a parameter to `CREATE SECRET name AS $1` or
-`ALTER SECRET name AS $1`, keeping it out of SQL text. Results contain only
+the credential as a parameter to
+`CREATE SECRET name TYPE = GENERIC_STRING SECRET_STRING = $1` or
+`ALTER SECRET name SET SECRET_STRING = $1`, keeping it out of SQL text. Results contain only
 completion status or Secret names and timestamps; they never return credential
 values. SQL `DESCRIBE SECRET` additionally exposes rotation history.
 

rust/lancedb/src/remote/sql/secret.rs

@@ -1,11 +1,7 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The LanceDB Authors
 
-use super::{
-    SqlClient,
-    materialized_view::quote_identifier,
-    objects::{definition, strings},
-};
+use super::{SqlClient, materialized_view::quote_identifier, objects::strings};
 use crate::{
     database::ExecuteQueryRequest,
     error::{Error, Result},
@@ -13,10 +9,38 @@ use crate::{
     secrets::SecretInfo,
     utils::{validate_namespace, validate_secret_name},
 };
-use arrow_array::{RecordBatch, StringArray};
+use arrow_array::{Array, Int64Array, RecordBatch, StringArray};
 use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
 use futures::TryStreamExt;
 use std::sync::Arc;
+fn secret_info(batches: Vec<RecordBatch>) -> Result<SecretInfo> {
+    if batches.iter().map(RecordBatch::num_rows).sum::<usize>() != 1 {
+        return Err(Error::Runtime {
+            message: "DESCRIBE SECRET must return exactly one row".into(),
+        });
+    }
+    let batch = batches.iter().find(|batch| batch.num_rows() == 1).unwrap();
+    let timestamp = |name: &str| -> Result<i64> {
+        let values = batch
+            .column_by_name(name)
+            .and_then(|column| column.as_any().downcast_ref::<Int64Array>())
+            .ok_or_else(|| Error::Runtime {
+                message: format!("DESCRIBE SECRET is missing INT64 column '{name}'"),
+            })?;
+        if values.is_null(0) {
+            return Err(Error::Runtime {
+                message: format!("DESCRIBE SECRET contains null '{name}'"),
+            });
+        }
+        Ok(values.value(0))
+    };
+    Ok(SecretInfo {
+        name: strings(&batches, "name")?.remove(0),
+        created_at_millis: timestamp("created_at_millis")?,
+        updated_at_millis: timestamp("updated_at_millis")?,
+    })
+}
+
 impl SqlClient {
     pub(in crate::remote) async fn write_secret(
         &self,
@@ -37,9 +61,14 @@ impl SqlClient {
         self.object_job(
             ExecuteQueryRequest {
                 query: format!(
-                    "{} SECRET {} AS $1",
+                    "{} SECRET {} {}SECRET_STRING = $1",
                     if create { "CREATE" } else { "ALTER" },
-                    quote_identifier(name)
+                    quote_identifier(name),
+                    if create {
+                        "TYPE = GENERIC_STRING "
+                    } else {
+                        "SET "
+                    }
                 ),
                 default_namespace_path: namespace.to_vec(),
                 parameters: Some(parameters),
@@ -63,7 +92,7 @@ impl SqlClient {
                 default_namespace_path: namespace.to_vec(),
                 parameters: None,
             },
-            definition,
+            secret_info,
         )
         .await?
         .wait()
@@ -154,3 +183,40 @@ impl SqlClient {
         Ok(ListingPage { items, page_token })
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use arrow_array::ArrayRef;
+
+    #[test]
+    fn decodes_native_secret_metadata() {
+        let batch = RecordBatch::try_from_iter(vec![
+            (
+                "name",
+                Arc::new(StringArray::from(vec!["api-key"])) as ArrayRef,
+            ),
+            (
+                "created_at_millis",
+                Arc::new(Int64Array::from(vec![10])) as ArrayRef,
+            ),
+            (
+                "updated_at_millis",
+                Arc::new(Int64Array::from(vec![20])) as ArrayRef,
+            ),
+        ])
+        .unwrap();
+        let info = secret_info(vec![batch.clone()]).unwrap();
+        assert_eq!(info.name, "api-key");
+        assert_eq!(info.created_at_millis, 10);
+        assert_eq!(info.updated_at_millis, 20);
+        assert!(secret_info(vec![]).is_err());
+        assert!(secret_info(vec![batch.clone(), batch]).is_err());
+        let missing = RecordBatch::try_from_iter(vec![(
+            "name",
+            Arc::new(StringArray::from(vec!["api-key"])) as ArrayRef,
+        )])
+        .unwrap();
+        assert!(secret_info(vec![missing]).is_err());
+    }
+}

rust/lancedb/src/remote/sql_test.rs

@@ -2419,8 +2419,11 @@ async fn secret_writes_bind_credentials_and_never_put_them_in_sql() {
         );
         let seen = exchanges.lock().unwrap();
         assert_eq!(seen.len(), 2);
-        assert_eq!(seen[0].0, "CREATE SECRET \"api-key\" AS $1");
-        assert_eq!(seen[1].0, "ALTER SECRET \"api-key\" AS $1");
+        assert_eq!(
+            seen[0].0,
+            "CREATE SECRET \"api-key\" TYPE = GENERIC_STRING SECRET_STRING = $1"
+        );
+        assert_eq!(seen[1].0, "ALTER SECRET \"api-key\" SET SECRET_STRING = $1");
         assert_eq!(
             seen[0]
                 .1
@@ -2450,11 +2453,20 @@ async fn secret_writes_bind_credentials_and_never_put_them_in_sql() {
 }
 #[tokio::test]
 async fn secret_description_uses_the_sql_metadata_projection() {
-    let definition = r#"{"name":"api-key","created_at_millis":1,"updated_at_millis":2,"history":[{"generation":2,"at_millis":2}]}"#;
-    let result = RecordBatch::try_from_iter(vec![(
-        "definition",
-        Arc::new(StringArray::from(vec![definition])) as ArrayRef,
-    )])
+    let result = RecordBatch::try_from_iter(vec![
+        (
+            "name",
+            Arc::new(StringArray::from(vec!["api-key"])) as ArrayRef,
+        ),
+        (
+            "created_at_millis",
+            Arc::new(arrow_array::Int64Array::from(vec![1])) as ArrayRef,
+        ),
+        (
+            "updated_at_millis",
+            Arc::new(arrow_array::Int64Array::from(vec![2])) as ArrayRef,
+        ),
+    ])
     .unwrap();
     let service = TestSqlService {
         adaptive: true,