diff --git a/rust/error.rs b/rust/error.rs
index 70e41de..5eb4c21 100644
--- a/rust/error.rs
+++ b/rust/error.rs
@@ -67,6 +67,9 @@ pub enum ErrorCode {
     Exec = 52,
     DatasetMerge = 53,
     NamespaceQueryTable = 54,
+    NamespaceListNamespaces = 55,
+    NamespaceCreateNamespace = 56,
+    NamespaceDropNamespace = 57,
 }
 
 struct LastError {
diff --git a/rust/ffi/dir_namespace.rs b/rust/ffi/dir_namespace.rs
index 0b1cb41..ef9cd6c 100644
--- a/rust/ffi/dir_namespace.rs
+++ b/rust/ffi/dir_namespace.rs
@@ -15,7 +15,8 @@ use crate::runtime;
 use super::session::record_dataset_open;
 use super::types::DatasetHandle;
 use super::util::{
-    cstr_to_str, optional_session_handle, slice_from_ptr, to_c_string, FfiError, FfiResult,
+    cstr_to_str, export_string_list, optional_session_handle, slice_from_ptr, to_c_string,
+    FfiError, FfiResult, LanceStringList,
 };
 
 fn parse_storage_options(
@@ -102,18 +103,10 @@ pub unsafe extern "C" fn lance_dir_namespace_list_tables(
     option_keys: *const *const c_char,
     option_values: *const *const c_char,
     options_len: usize,
-) -> *const c_char {
-    match dir_namespace_list_tables_inner(root, option_keys, option_values, options_len) {
-        Ok(tables) => {
-            clear_last_error();
-            let joined = tables.join("\n");
-            to_c_string(joined).into_raw() as *const c_char
-        }
-        Err(err) => {
-            set_last_error(err.code, err.message);
-            ptr::null()
-        }
-    }
+    out: *mut LanceStringList,
+) -> i32 {
+    let result = dir_namespace_list_tables_inner(root, option_keys, option_values, options_len);
+    unsafe { export_string_list(result, out) }
 }
 
 fn open_dataset_in_dir_namespace_inner(
diff --git a/rust/ffi/index.rs b/rust/ffi/index.rs
index 3533f69..499c680 100644
--- a/rust/ffi/index.rs
+++ b/rust/ffi/index.rs
@@ -20,7 +20,8 @@ use crate::runtime;
 
 use super::types::{SchemaHandle, StreamHandle};
 use super::util::{
-    canonicalize_lance_field_path, cstr_to_str, dataset_handle, to_c_string, FfiError, FfiResult,
+    canonicalize_lance_field_path, cstr_to_str, dataset_handle, export_string_list, to_c_string,
+    FfiError, FfiResult, LanceStringList,
 };
 
 #[derive(Debug, Default, Deserialize)]
@@ -156,50 +157,15 @@ fn create_index_list_stream_inner(dataset: *mut c_void) -> FfiResult<StreamHandl
     Ok(StreamHandle::Batches(vec![batch].into_iter()))
 }
 
-/// Returns an array of column names that have scalar indices.
-/// The caller must free the returned array with `lance_free_scalar_indexed_columns`.
+/// Returns column names that have scalar indices.
+/// The caller must free the result with `lance_free_string_list`.
 #[no_mangle]
 pub unsafe extern "C" fn lance_dataset_list_scalar_indexed_columns(
     dataset: *mut c_void,
-    out_len: *mut usize,
-) -> *mut *mut c_char {
-    match list_scalar_indexed_columns_inner(dataset) {
-        Ok(cols) => {
-            clear_last_error();
-            unsafe { *out_len = cols.len() };
-            if cols.is_empty() {
-                return std::ptr::null_mut();
-            }
-            let ptrs: Vec<*mut c_char> = cols
-                .into_iter()
-                .map(|s| to_c_string(s).into_raw())
-                .collect();
-            let mut boxed = ptrs.into_boxed_slice();
-            let ptr = boxed.as_mut_ptr();
-            std::mem::forget(boxed);
-            ptr
-        }
-        Err(err) => {
-            set_last_error(err.code, err.message);
-            unsafe { *out_len = 0 };
-            std::ptr::null_mut()
-        }
-    }
-}
-
-#[no_mangle]
-pub unsafe extern "C" fn lance_free_scalar_indexed_columns(ptr: *mut *mut c_char, len: usize) {
-    if ptr.is_null() {
-        return;
-    }
-    unsafe {
-        let slice = Box::from_raw(std::ptr::slice_from_raw_parts_mut(ptr, len));
-        for &p in slice.iter() {
-            if !p.is_null() {
-                drop(std::ffi::CString::from_raw(p));
-            }
-        }
-    }
+    out: *mut LanceStringList,
+) -> i32 {
+    let result = list_scalar_indexed_columns_inner(dataset);
+    unsafe { export_string_list(result, out) }
 }
 
 fn list_scalar_indexed_columns_inner(dataset: *mut c_void) -> FfiResult<Vec<String>> {
diff --git a/rust/ffi/namespace.rs b/rust/ffi/namespace.rs
index cfba44e..2f90b90 100644
--- a/rust/ffi/namespace.rs
+++ b/rust/ffi/namespace.rs
@@ -7,10 +7,11 @@ use lance::dataset::builder::DatasetBuilder;
 use lance_core::Error as LanceError;
 
 use lance_namespace::models::{
-    DeclareTableRequest, DescribeTableRequest, DropTableRequest, ListTablesRequest,
+    CreateNamespaceRequest, DeclareTableRequest, DescribeTableRequest, DropNamespaceRequest,
+    DropTableRequest, ListNamespacesRequest, ListTablesRequest,
 };
 use lance_namespace::schema::convert_json_arrow_schema;
-use lance_namespace::LanceNamespace;
+use lance_namespace::{ErrorCode as NamespaceErrorCode, LanceNamespace, NamespaceError};
 use lance_namespace_impls::RestNamespaceBuilder;
 
 use crate::error::{clear_last_error, set_last_error, ErrorCode};
@@ -19,8 +20,8 @@ use crate::runtime;
 use super::session::{record_dataset_open, record_namespace_describe};
 use super::types::DatasetHandle;
 use super::util::{
-    cstr_to_str, optional_session_handle, schema_to_ffi_arrow_schema, to_c_string, FfiError,
-    FfiResult,
+    cstr_to_str, export_string_list, optional_session_handle, schema_to_ffi_arrow_schema,
+    to_c_string, FfiError, FfiResult, LanceStringList,
 };
 
 unsafe fn optional_cstr_to_string(
@@ -86,6 +87,223 @@ fn storage_options_to_tsv(storage_options: std::collections::HashMap<String, Str
         .join("\n")
 }
 
+fn split_id(id: &str, delimiter: &str) -> Vec<String> {
+    if id.is_empty() {
+        Vec::new()
+    } else {
+        id.split(delimiter).map(ToString::to_string).collect()
+    }
+}
+
+fn normalize_listed_tables(
+    tables: Vec<String>,
+    namespace_id: &str,
+    delimiter: &str,
+) -> Vec<String> {
+    let namespace_prefix = if namespace_id.is_empty() {
+        None
+    } else {
+        Some(format!("{namespace_id}{delimiter}"))
+    };
+    tables
+        .into_iter()
+        .filter_map(|table| {
+            let relative = match namespace_prefix.as_deref() {
+                Some(prefix) if table.starts_with(prefix) => table[prefix.len()..].to_string(),
+                _ if !table.contains(delimiter) => table,
+                _ => return None,
+            };
+            if relative.is_empty() || relative.contains(delimiter) {
+                None
+            } else {
+                Some(relative)
+            }
+        })
+        .collect()
+}
+
+fn namespace_operation_config(
+    endpoint: *const c_char,
+    namespace_id: *const c_char,
+    bearer_token: *const c_char,
+    api_key: *const c_char,
+    delimiter: *const c_char,
+    headers_tsv: *const c_char,
+) -> FfiResult<(impl LanceNamespace, Vec<String>)> {
+    let endpoint = unsafe { cstr_to_str(endpoint, "endpoint")? };
+    let namespace_id = unsafe { cstr_to_str(namespace_id, "namespace_id")? };
+    let delimiter = unsafe { optional_cstr_to_string(delimiter, "delimiter")? }
+        .unwrap_or_else(|| "$".to_string());
+    let bearer_token = unsafe { optional_cstr_to_string(bearer_token, "bearer_token")? };
+    let api_key = unsafe { optional_cstr_to_string(api_key, "api_key")? };
+    let headers_tsv = unsafe { optional_cstr_to_string(headers_tsv, "headers_tsv")? };
+    let id = split_id(namespace_id, &delimiter);
+    let namespace = build_config(
+        endpoint,
+        bearer_token.as_deref(),
+        api_key.as_deref(),
+        headers_tsv.as_deref(),
+    )
+    .delimiter(delimiter)
+    .build();
+    Ok((namespace, id))
+}
+
+fn is_unsupported_namespace_operation(error: &LanceError) -> bool {
+    matches!(error, LanceError::NotSupported { .. })
+        || match error {
+            LanceError::Namespace { source, .. } => source
+                .downcast_ref::<NamespaceError>()
+                .is_some_and(|error| error.code() == NamespaceErrorCode::Unsupported),
+            _ => false,
+        }
+        // lance-namespace 9.0.1 maps an empty HTTP 501 response to Internal
+        // because there is no structured error code to preserve.
+        || error.to_string().contains("status=501 Not Implemented")
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn lance_namespace_list_namespaces(
+    endpoint: *const c_char,
+    namespace_id: *const c_char,
+    bearer_token: *const c_char,
+    api_key: *const c_char,
+    delimiter: *const c_char,
+    headers_tsv: *const c_char,
+    out: *mut LanceStringList,
+) -> i32 {
+    let result = (|| {
+        let (namespace, id) = namespace_operation_config(
+            endpoint,
+            namespace_id,
+            bearer_token,
+            api_key,
+            delimiter,
+            headers_tsv,
+        )?;
+        runtime::block_on(async move {
+            let mut out = Vec::new();
+            let mut page_token = None;
+            loop {
+                let mut request = ListNamespacesRequest::new();
+                request.id = Some(id.clone());
+                request.page_token = page_token.clone();
+                request.limit = Some(1000);
+                let response = match namespace.list_namespaces(request).await {
+                    Ok(response) => response,
+                    Err(err) if is_unsupported_namespace_operation(&err) => break,
+                    Err(err) => {
+                        return Err(FfiError::new(
+                            ErrorCode::NamespaceListNamespaces,
+                            format!("namespace list_namespaces: {err}"),
+                        ));
+                    }
+                };
+                out.extend(response.namespaces);
+                match response.page_token {
+                    Some(token) if !token.is_empty() => page_token = Some(token),
+                    _ => break,
+                }
+            }
+            Ok::<_, FfiError>(out)
+        })
+        .map_err(|err| FfiError::new(ErrorCode::Runtime, format!("runtime: {err}")))?
+    })();
+    unsafe { export_string_list(result, out) }
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn lance_namespace_create_namespace(
+    endpoint: *const c_char,
+    namespace_id: *const c_char,
+    bearer_token: *const c_char,
+    api_key: *const c_char,
+    delimiter: *const c_char,
+    headers_tsv: *const c_char,
+    mode: *const c_char,
+) -> i32 {
+    let result = (|| {
+        let (namespace, id) = namespace_operation_config(
+            endpoint,
+            namespace_id,
+            bearer_token,
+            api_key,
+            delimiter,
+            headers_tsv,
+        )?;
+        let mode = unsafe { optional_cstr_to_string(mode, "mode")? };
+        runtime::block_on(async move {
+            let mut request = CreateNamespaceRequest::new();
+            request.id = Some(id);
+            request.mode = mode;
+            namespace.create_namespace(request).await.map_err(|err| {
+                FfiError::new(
+                    ErrorCode::NamespaceCreateNamespace,
+                    format!("namespace create_namespace: {err}"),
+                )
+            })?;
+            Ok::<_, FfiError>(())
+        })
+        .map_err(|err| FfiError::new(ErrorCode::Runtime, format!("runtime: {err}")))?
+    })();
+    match result {
+        Ok(()) => {
+            clear_last_error();
+            0
+        }
+        Err(err) => {
+            set_last_error(err.code, err.message);
+            -1
+        }
+    }
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn lance_namespace_drop_namespace(
+    endpoint: *const c_char,
+    namespace_id: *const c_char,
+    bearer_token: *const c_char,
+    api_key: *const c_char,
+    delimiter: *const c_char,
+    headers_tsv: *const c_char,
+    behavior: *const c_char,
+) -> i32 {
+    let result = (|| {
+        let (namespace, id) = namespace_operation_config(
+            endpoint,
+            namespace_id,
+            bearer_token,
+            api_key,
+            delimiter,
+            headers_tsv,
+        )?;
+        let behavior = unsafe { optional_cstr_to_string(behavior, "behavior")? };
+        runtime::block_on(async move {
+            let mut request = DropNamespaceRequest::new();
+            request.id = Some(id);
+            request.behavior = behavior;
+            namespace.drop_namespace(request).await.map_err(|err| {
+                FfiError::new(
+                    ErrorCode::NamespaceDropNamespace,
+                    format!("namespace drop_namespace: {err}"),
+                )
+            })?;
+            Ok::<_, FfiError>(())
+        })
+        .map_err(|err| FfiError::new(ErrorCode::Runtime, format!("runtime: {err}")))?
+    })();
+    match result {
+        Ok(()) => {
+            clear_last_error();
+            0
+        }
+        Err(err) => {
+            set_last_error(err.code, err.message);
+            -1
+        }
+    }
+}
+
 fn list_tables_inner(
     endpoint: *const c_char,
     namespace_id: *const c_char,
@@ -102,13 +320,14 @@ fn list_tables_inner(
     let headers_tsv = unsafe { optional_cstr_to_string(headers_tsv, "headers_tsv")? };
 
     let delimiter = delimiter.unwrap_or_else(|| "$".to_string());
+    let namespace_parts = split_id(&namespace_id, &delimiter);
     let namespace = build_config(
         endpoint,
         bearer_token.as_deref(),
         api_key.as_deref(),
         headers_tsv.as_deref(),
     )
-    .delimiter(delimiter)
+    .delimiter(delimiter.clone())
     .build();
 
     let tables = runtime::block_on(async move {
@@ -116,11 +335,7 @@ fn list_tables_inner(
         let mut page_token: Option<String> = None;
         loop {
             let mut req = ListTablesRequest::new();
-            req.id = Some(if namespace_id.is_empty() {
-                Vec::new()
-            } else {
-                vec![namespace_id.to_string()]
-            });
+            req.id = Some(namespace_parts.clone());
             req.page_token = page_token.clone();
             req.limit = Some(1000);
             let resp = namespace.list_tables(req).await.map_err(|err| {
@@ -139,7 +354,7 @@ fn list_tables_inner(
     })
     .map_err(|err| FfiError::new(ErrorCode::Runtime, format!("runtime: {err}")))??;
 
-    Ok(tables)
+    Ok(normalize_listed_tables(tables, &namespace_id, &delimiter))
 }
 
 #[no_mangle]
@@ -150,25 +365,17 @@ pub unsafe extern "C" fn lance_namespace_list_tables(
     api_key: *const c_char,
     delimiter: *const c_char,
     headers_tsv: *const c_char,
-) -> *const c_char {
-    match list_tables_inner(
+    out: *mut LanceStringList,
+) -> i32 {
+    let result = list_tables_inner(
         endpoint,
         namespace_id,
         bearer_token,
         api_key,
         delimiter,
         headers_tsv,
-    ) {
-        Ok(tables) => {
-            clear_last_error();
-            let joined = tables.join("\n");
-            to_c_string(joined).into_raw() as *const c_char
-        }
-        Err(err) => {
-            set_last_error(err.code, err.message);
-            ptr::null()
-        }
-    }
+    );
+    unsafe { export_string_list(result, out) }
 }
 
 fn describe_table_info_inner(
@@ -187,27 +394,20 @@ fn describe_table_info_inner(
     let headers_tsv = unsafe { optional_cstr_to_string(headers_tsv, "headers_tsv")? };
 
     let delimiter = delimiter.unwrap_or_else(|| "$".to_string());
+    let table_id_segments = split_id(table_id, &delimiter);
     let namespace = build_config(
         endpoint,
         bearer_token.as_deref(),
         api_key.as_deref(),
         headers_tsv.as_deref(),
     )
-    .delimiter(delimiter.clone())
+    .delimiter(delimiter)
     .build();
 
     let (location, storage_options_tsv) = runtime::block_on(async move {
         record_namespace_describe();
         let mut req = DescribeTableRequest::new();
-        // FIX: a qualified table id (e.g. "catalog.schema.table") must be sent as
-        // its multi-segment namespace path, not a single segment. Split on the
-        // delimiter so the server sees the full 3-level id instead of "got: 1".
-        req.id = Some(
-            table_id
-                .split(delimiter.as_str())
-                .map(|s| s.to_string())
-                .collect(),
-        );
+        req.id = Some(table_id_segments);
         req.with_table_uri = Some(true);
         let resp = namespace.describe_table(req).await.map_err(|err| {
             FfiError::new(
@@ -302,19 +502,16 @@ fn create_empty_table_inner(
     let headers_tsv = unsafe { optional_cstr_to_string(headers_tsv, "headers_tsv")? };
 
     let delimiter = delimiter.unwrap_or_else(|| "$".to_string());
+    let table_id_segments = split_id(table_id, &delimiter);
     let namespace = build_config(
         endpoint,
         bearer_token.as_deref(),
         api_key.as_deref(),
         headers_tsv.as_deref(),
     )
-    .delimiter(delimiter.clone())
+    .delimiter(delimiter)
     .build();
 
-    let table_id_segments: Vec<String> = table_id
-        .split(delimiter.as_str())
-        .map(|s| s.to_string())
-        .collect();
     let (location, storage_options_tsv) = runtime::block_on(async move {
         let mut req = DeclareTableRequest::new();
         req.id = Some(table_id_segments);
@@ -411,19 +608,16 @@ fn drop_table_inner(
     let headers_tsv = unsafe { optional_cstr_to_string(headers_tsv, "headers_tsv")? };
 
     let delimiter = delimiter.unwrap_or_else(|| "$".to_string());
+    let table_id_segments = split_id(table_id, &delimiter);
     let namespace = build_config(
         endpoint,
         bearer_token.as_deref(),
         api_key.as_deref(),
         headers_tsv.as_deref(),
     )
-    .delimiter(delimiter.clone())
+    .delimiter(delimiter)
     .build();
 
-    let table_id_segments: Vec<String> = table_id
-        .split(delimiter.as_str())
-        .map(|s| s.to_string())
-        .collect();
     runtime::block_on(async move {
         let mut req = DropTableRequest::new();
         req.id = Some(table_id_segments);
@@ -485,24 +679,19 @@ fn describe_table_with_schema_inner(
     let headers_tsv = unsafe { optional_cstr_to_string(headers_tsv, "headers_tsv")? };
 
     let delimiter = delimiter.unwrap_or_else(|| "$".to_string());
+    let table_id_segments = split_id(table_id, &delimiter);
     let namespace = build_config(
         endpoint,
         bearer_token.as_deref(),
         api_key.as_deref(),
         headers_tsv.as_deref(),
     )
-    .delimiter(delimiter.clone())
+    .delimiter(delimiter)
     .build();
 
     let schema_json = runtime::block_on(async move {
         let mut req = DescribeTableRequest::new();
-        // FIX: split the qualified id into its namespace segments (see describe_table_info_inner).
-        req.id = Some(
-            table_id
-                .split(delimiter.as_str())
-                .map(|s| s.to_string())
-                .collect(),
-        );
+        req.id = Some(table_id_segments);
         req.with_table_uri = Some(true);
         req.load_detailed_metadata = Some(true);
         let resp = namespace.describe_table(req).await.map_err(|err| {
@@ -590,21 +779,16 @@ fn open_dataset_in_namespace_inner(
     let headers_tsv = unsafe { optional_cstr_to_string(headers_tsv, "headers_tsv")? };
 
     let delimiter = delimiter.unwrap_or_else(|| "$".to_string());
+    let table_id_segments = split_id(table_id, &delimiter);
     let namespace = build_config(
         endpoint,
         bearer_token.as_deref(),
         api_key.as_deref(),
         headers_tsv.as_deref(),
     )
-    .delimiter(delimiter.clone())
+    .delimiter(delimiter)
     .build();
     let session = unsafe { optional_session_handle(session)? };
-    // FIX: split the qualified id into namespace segments so the crate's internal
-    // describe (DatasetBuilder::from_namespace) gets the full 3-level id, not 1.
-    let table_id_segments: Vec<String> = table_id
-        .split(delimiter.as_str())
-        .map(|s| s.to_string())
-        .collect();
 
     let (dataset, table_uri) = runtime::block_on(async move {
         record_namespace_describe();
@@ -758,3 +942,38 @@ pub unsafe extern "C" fn lance_json_arrow_schema_to_c(
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::normalize_listed_tables;
+
+    #[test]
+    fn listed_tables_are_relative_to_the_requested_namespace() {
+        let tables = vec![
+            "default$child$t".to_string(),
+            "default$child$a\nb".to_string(),
+            "compat".to_string(),
+            "default$child$grand$t".to_string(),
+            "default$other$t".to_string(),
+        ];
+        assert_eq!(
+            normalize_listed_tables(tables, "default$child", "$"),
+            vec!["t", "a\nb", "compat"]
+        );
+    }
+
+    #[test]
+    fn root_table_listing_excludes_descendants() {
+        let tables = vec!["t".to_string(), "child$t".to_string()];
+        assert_eq!(normalize_listed_tables(tables, "", "$"), vec!["t"]);
+    }
+
+    #[test]
+    fn listed_tables_use_the_configured_delimiter() {
+        let tables = vec!["default/child/t".to_string()];
+        assert_eq!(
+            normalize_listed_tables(tables, "default/child", "/"),
+            vec!["t"]
+        );
+    }
+}
diff --git a/rust/ffi/util.rs b/rust/ffi/util.rs
index 96fa4f5..01d5c0f 100644
--- a/rust/ffi/util.rs
+++ b/rust/ffi/util.rs
@@ -8,7 +8,7 @@ use datafusion_expr::Expr;
 use lance::session::Session;
 use lance_core::datatypes::Schema as LanceSchema;
 
-use crate::error::ErrorCode;
+use crate::error::{clear_last_error, set_last_error, ErrorCode};
 
 use super::types::{DatasetHandle, SchemaHandle, SessionHandle, StreamHandle};
 
@@ -29,6 +29,75 @@ impl FfiError {
 
 pub(crate) type FfiResult<T> = Result<T, FfiError>;
 
+#[repr(C)]
+pub struct LanceStringList {
+    pub items: *mut *mut c_char,
+    pub count: usize,
+}
+
+pub(crate) unsafe fn write_string_list(
+    values: Vec<String>,
+    out: *mut LanceStringList,
+) -> FfiResult<()> {
+    if out.is_null() {
+        return Err(FfiError::new(
+            ErrorCode::InvalidArgument,
+            "string list output is null",
+        ));
+    }
+
+    let count = values.len();
+    let items = if values.is_empty() {
+        std::ptr::null_mut()
+    } else {
+        let pointers = values
+            .into_iter()
+            .map(|value| to_c_string(value).into_raw())
+            .collect::<Vec<_>>();
+        let mut boxed = pointers.into_boxed_slice();
+        let items = boxed.as_mut_ptr();
+        std::mem::forget(boxed);
+        items
+    };
+    unsafe { std::ptr::write_unaligned(out, LanceStringList { items, count }) };
+    Ok(())
+}
+
+pub(crate) unsafe fn export_string_list(
+    result: FfiResult<Vec<String>>,
+    out: *mut LanceStringList,
+) -> i32 {
+    match result.and_then(|values| unsafe { write_string_list(values, out) }) {
+        Ok(()) => {
+            clear_last_error();
+            0
+        }
+        Err(err) => {
+            set_last_error(err.code, err.message);
+            -1
+        }
+    }
+}
+
+#[no_mangle]
+pub unsafe extern "C" fn lance_free_string_list(list: *mut LanceStringList) {
+    if list.is_null() {
+        return;
+    }
+    let list = unsafe { &mut *list };
+    if !list.items.is_null() {
+        let items =
+            unsafe { Box::from_raw(std::ptr::slice_from_raw_parts_mut(list.items, list.count)) };
+        for item in items.iter().copied() {
+            if !item.is_null() {
+                unsafe { drop(CString::from_raw(item)) };
+            }
+        }
+    }
+    list.items = std::ptr::null_mut();
+    list.count = 0;
+}
+
 pub(crate) fn to_c_string(s: impl AsRef<str>) -> CString {
     match CString::new(s.as_ref()) {
         Ok(v) => v,
@@ -227,3 +296,35 @@ pub(crate) fn schema_to_ffi_arrow_schema(
     arrow::ffi::FFI_ArrowSchema::try_from(&data_type)
         .map_err(|err| FfiError::new(ErrorCode::SchemaExport, format!("schema export: {err}")))
 }
+
+#[cfg(test)]
+mod tests {
+    use std::ffi::CStr;
+
+    use super::{lance_free_string_list, write_string_list, LanceStringList};
+
+    #[test]
+    fn string_list_preserves_element_boundaries() {
+        let mut list = LanceStringList {
+            items: std::ptr::null_mut(),
+            count: 0,
+        };
+        unsafe {
+            write_string_list(vec!["a\nb".to_string(), "销售".to_string()], &mut list).unwrap();
+            assert_eq!(list.count, 2);
+            assert_eq!(CStr::from_ptr(*list.items).to_str().unwrap(), "a\nb");
+            assert_eq!(CStr::from_ptr(*list.items.add(1)).to_str().unwrap(), "销售");
+            lance_free_string_list(&mut list);
+        }
+        assert!(list.items.is_null());
+        assert_eq!(list.count, 0);
+
+        unsafe {
+            write_string_list(Vec::new(), &mut list).unwrap();
+            assert!(list.items.is_null());
+            assert_eq!(list.count, 0);
+            lance_free_string_list(&mut list);
+            assert!(write_string_list(Vec::new(), std::ptr::null_mut()).is_err());
+        }
+    }
+}
diff --git a/src/include/lance_common.hpp b/src/include/lance_common.hpp
index 1d19d9f..14b0f43 100644
--- a/src/include/lance_common.hpp
+++ b/src/include/lance_common.hpp
@@ -3,6 +3,7 @@
 #include "duckdb.hpp"
 
 struct LanceNamespaceQueryConfig;
+struct LanceStringList;
 
 namespace duckdb {
 
@@ -46,12 +47,27 @@ void ResolveLanceNamespaceAuthOverrides(
     const unordered_map<string, Value> &options, string &out_bearer_token,
     string &out_api_key);
 
+vector<string> LanceConsumeStringList(LanceStringList &list);
+
 bool TryLanceNamespaceListTables(ClientContext &context, const string &endpoint,
                                  const string &namespace_id,
                                  const string &bearer_token,
                                  const string &api_key, const string &delimiter,
                                  const string &headers_tsv,
                                  vector<string> &out_tables, string &out_error);
+bool TryLanceNamespaceListNamespaces(
+    ClientContext &context, const string &endpoint, const string &namespace_id,
+    const string &bearer_token, const string &api_key, const string &delimiter,
+    const string &headers_tsv, vector<string> &out_namespaces,
+    string &out_error);
+bool TryLanceNamespaceCreateNamespace(
+    ClientContext &context, const string &endpoint, const string &namespace_id,
+    const string &bearer_token, const string &api_key, const string &delimiter,
+    const string &headers_tsv, const string &mode, string &out_error);
+bool TryLanceNamespaceDropNamespace(
+    ClientContext &context, const string &endpoint, const string &namespace_id,
+    const string &bearer_token, const string &api_key, const string &delimiter,
+    const string &headers_tsv, bool cascade, string &out_error);
 
 bool TryLanceDirNamespaceListTables(ClientContext &context, const string &root,
                                     vector<string> &out_tables,
diff --git a/src/include/lance_ffi.hpp b/src/include/lance_ffi.hpp
index e6c0c1b..95002ed 100644
--- a/src/include/lance_ffi.hpp
+++ b/src/include/lance_ffi.hpp
@@ -17,6 +17,12 @@ typedef struct LanceDebugCounters {
   uint64_t commit_count;
 } LanceDebugCounters;
 
+typedef struct LanceStringList {
+  // Rust-owned strings; release the complete list with lance_free_string_list.
+  char **items;
+  size_t count;
+} LanceStringList;
+
 void *lance_create_session(uint64_t index_cache_size_bytes,
                            uint64_t metadata_cache_size_bytes);
 void lance_close_session(void *session);
@@ -33,10 +39,11 @@ void *lance_open_dataset_with_storage_options(const char *path,
 void *lance_open_dataset_with_storage_options_and_session(
     const char *path, const char **option_keys, const char **option_values,
     size_t options_len, void *session);
-const char *lance_dir_namespace_list_tables(const char *root,
-                                            const char **option_keys,
-                                            const char **option_values,
-                                            size_t options_len);
+int32_t lance_dir_namespace_list_tables(const char *root,
+                                        const char **option_keys,
+                                        const char **option_values,
+                                        size_t options_len,
+                                        LanceStringList *out);
 int32_t lance_dir_namespace_drop_table(const char *root, const char *table_name,
                                        const char **option_keys,
                                        const char **option_values,
@@ -48,10 +55,27 @@ void *lance_open_dataset_in_dir_namespace_with_session(
     const char *root, const char *table_name, const char **option_keys,
     const char **option_values, size_t options_len, void *session,
     const char **out_table_uri);
-const char *
-lance_namespace_list_tables(const char *endpoint, const char *namespace_id,
-                            const char *bearer_token, const char *api_key,
-                            const char *delimiter, const char *headers_tsv);
+int32_t lance_namespace_list_tables(const char *endpoint,
+                                    const char *namespace_id,
+                                    const char *bearer_token,
+                                    const char *api_key, const char *delimiter,
+                                    const char *headers_tsv,
+                                    LanceStringList *out);
+int32_t
+lance_namespace_list_namespaces(const char *endpoint, const char *namespace_id,
+                                const char *bearer_token, const char *api_key,
+                                const char *delimiter, const char *headers_tsv,
+                                LanceStringList *out);
+int32_t
+lance_namespace_create_namespace(const char *endpoint, const char *namespace_id,
+                                 const char *bearer_token, const char *api_key,
+                                 const char *delimiter, const char *headers_tsv,
+                                 const char *mode);
+int32_t
+lance_namespace_drop_namespace(const char *endpoint, const char *namespace_id,
+                               const char *bearer_token, const char *api_key,
+                               const char *delimiter, const char *headers_tsv,
+                               const char *behavior);
 int32_t lance_json_arrow_schema_to_c(const char *json_schema,
                                      ArrowSchema *out_schema);
 int32_t lance_namespace_describe_table_with_schema(
@@ -97,6 +121,7 @@ void *lance_create_dataset_exec_stream_ir(void *dataset, const uint8_t *exec_ir,
 int32_t lance_last_error_code();
 const char *lance_last_error_message();
 void lance_free_string(const char *s);
+void lance_free_string_list(LanceStringList *list);
 
 int64_t lance_dataset_count_rows(void *dataset);
 int32_t lance_dataset_delete(void *dataset, const uint8_t *filter_ir,
@@ -338,9 +363,8 @@ lance_dataset_optimize_index_with_options(void *dataset, const char *index_name,
                                           const char **out_metrics_json);
 void *lance_get_index_list_schema(void *dataset);
 void *lance_create_index_list_stream(void *dataset);
-char **lance_dataset_list_scalar_indexed_columns(void *dataset,
-                                                 size_t *out_len);
-void lance_free_scalar_indexed_columns(char **ptr, size_t len);
+int32_t lance_dataset_list_scalar_indexed_columns(void *dataset,
+                                                  LanceStringList *out);
 
 void lance_free_batch(void *batch);
 int32_t lance_batch_to_arrow(void *batch, ArrowArray *out_array,
diff --git a/src/lance_common.cpp b/src/lance_common.cpp
index a3d0174..6b53ab7 100644
--- a/src/lance_common.cpp
+++ b/src/lance_common.cpp
@@ -293,6 +293,24 @@ void BuildStorageOptionPointerArrays(const vector<string> &option_keys,
   }
 }
 
+vector<string> LanceConsumeStringList(LanceStringList &list) {
+  vector<string> values;
+  try {
+    values.reserve(list.count);
+    for (idx_t i = 0; i < list.count; i++) {
+      if (!list.items || !list.items[i]) {
+        throw IOException("Invalid Lance string list");
+      }
+      values.emplace_back(list.items[i]);
+    }
+  } catch (...) {
+    lance_free_string_list(&list);
+    throw;
+  }
+  lance_free_string_list(&list);
+  return values;
+}
+
 bool TryLanceNamespaceListTables(
     ClientContext &context, const string &endpoint, const string &namespace_id,
     const string &bearer_token, const string &api_key, const string &delimiter,
@@ -306,28 +324,89 @@ bool TryLanceNamespaceListTables(
   const char *delimiter_ptr = delimiter.empty() ? nullptr : delimiter.c_str();
   const char *headers_ptr = headers_tsv.empty() ? nullptr : headers_tsv.c_str();
 
-  auto *ptr = lance_namespace_list_tables(
-      endpoint.c_str(), namespace_id.c_str(), bearer_ptr, api_key_ptr,
-      delimiter_ptr, headers_ptr);
-  if (!ptr) {
+  LanceStringList list{nullptr, 0};
+  auto rc = lance_namespace_list_tables(endpoint.c_str(), namespace_id.c_str(),
+                                        bearer_ptr, api_key_ptr, delimiter_ptr,
+                                        headers_ptr, &list);
+  if (rc != 0) {
     out_error = LanceConsumeLastError();
     if (out_error.empty()) {
       out_error = "unknown error";
     }
     return false;
   }
-  string joined = ptr;
-  lance_free_string(ptr);
+  out_tables = LanceConsumeStringList(list);
+  return true;
+}
 
-  vector<string> parts = StringUtil::Split(joined, '\n');
-  for (auto &p : parts) {
-    if (!p.empty()) {
-      out_tables.push_back(std::move(p));
+bool TryLanceNamespaceListNamespaces(
+    ClientContext &context, const string &endpoint, const string &namespace_id,
+    const string &bearer_token, const string &api_key, const string &delimiter,
+    const string &headers_tsv, vector<string> &out_namespaces,
+    string &out_error) {
+  (void)context;
+  out_namespaces.clear();
+  out_error.clear();
+  LanceStringList list{nullptr, 0};
+  auto rc = lance_namespace_list_namespaces(
+      endpoint.c_str(), namespace_id.c_str(),
+      bearer_token.empty() ? nullptr : bearer_token.c_str(),
+      api_key.empty() ? nullptr : api_key.c_str(),
+      delimiter.empty() ? nullptr : delimiter.c_str(),
+      headers_tsv.empty() ? nullptr : headers_tsv.c_str(), &list);
+  if (rc != 0) {
+    out_error = LanceConsumeLastError();
+    return false;
+  }
+  auto effective_delimiter = delimiter.empty() ? "$" : delimiter;
+  for (auto &namespace_name : LanceConsumeStringList(list)) {
+    if (!namespace_name.empty() &&
+        namespace_name.find(effective_delimiter) == string::npos) {
+      out_namespaces.push_back(std::move(namespace_name));
     }
   }
   return true;
 }
 
+bool TryLanceNamespaceCreateNamespace(
+    ClientContext &context, const string &endpoint, const string &namespace_id,
+    const string &bearer_token, const string &api_key, const string &delimiter,
+    const string &headers_tsv, const string &mode, string &out_error) {
+  (void)context;
+  out_error.clear();
+  auto rc = lance_namespace_create_namespace(
+      endpoint.c_str(), namespace_id.c_str(),
+      bearer_token.empty() ? nullptr : bearer_token.c_str(),
+      api_key.empty() ? nullptr : api_key.c_str(),
+      delimiter.empty() ? nullptr : delimiter.c_str(),
+      headers_tsv.empty() ? nullptr : headers_tsv.c_str(), mode.c_str());
+  if (rc != 0) {
+    out_error = LanceConsumeLastError();
+    return false;
+  }
+  return true;
+}
+
+bool TryLanceNamespaceDropNamespace(
+    ClientContext &context, const string &endpoint, const string &namespace_id,
+    const string &bearer_token, const string &api_key, const string &delimiter,
+    const string &headers_tsv, bool cascade, string &out_error) {
+  (void)context;
+  out_error.clear();
+  auto behavior = cascade ? "Cascade" : "Restrict";
+  auto rc = lance_namespace_drop_namespace(
+      endpoint.c_str(), namespace_id.c_str(),
+      bearer_token.empty() ? nullptr : bearer_token.c_str(),
+      api_key.empty() ? nullptr : api_key.c_str(),
+      delimiter.empty() ? nullptr : delimiter.c_str(),
+      headers_tsv.empty() ? nullptr : headers_tsv.c_str(), behavior);
+  if (rc != 0) {
+    out_error = LanceConsumeLastError();
+    return false;
+  }
+  return true;
+}
+
 static void ParseStorageOptionsTsv(const char *ptr, vector<string> &out_keys,
                                    vector<string> &out_values) {
   out_keys.clear();
@@ -471,10 +550,12 @@ bool TryLanceDirNamespaceListTables(ClientContext &context, const string &root,
   BuildStorageOptionPointerArrays(option_keys, option_values, key_ptrs,
                                   value_ptrs);
 
-  auto *ptr = lance_dir_namespace_list_tables(
+  LanceStringList list{nullptr, 0};
+  auto rc = lance_dir_namespace_list_tables(
       open_root.c_str(), key_ptrs.empty() ? nullptr : key_ptrs.data(),
-      value_ptrs.empty() ? nullptr : value_ptrs.data(), option_keys.size());
-  if (!ptr) {
+      value_ptrs.empty() ? nullptr : value_ptrs.data(), option_keys.size(),
+      &list);
+  if (rc != 0) {
     out_error = LanceConsumeLastError();
     if (out_error.empty()) {
       out_error = "unknown error";
@@ -482,11 +563,7 @@ bool TryLanceDirNamespaceListTables(ClientContext &context, const string &root,
     return false;
   }
 
-  string joined = ptr;
-  lance_free_string(ptr);
-
-  vector<string> parts = StringUtil::Split(joined, '\n');
-  for (auto &p : parts) {
+  for (auto &p : LanceConsumeStringList(list)) {
     if (!p.empty()) {
       out_tables.push_back(std::move(p));
     }
diff --git a/src/lance_scan.cpp b/src/lance_scan.cpp
index 8058a91..46d3c83 100644
--- a/src/lance_scan.cpp
+++ b/src/lance_scan.cpp
@@ -1480,18 +1480,22 @@ LanceScanInitGlobal(ClientContext &context, TableFunctionInitInput &input) {
     }
 
     if (!filtered_columns.empty()) {
-      size_t indexed_cols_len = 0;
-      auto indexed_cols_ptr = lance_dataset_list_scalar_indexed_columns(
-          bind_data.dataset, &indexed_cols_len);
+      LanceStringList indexed_columns{nullptr, 0};
+      auto rc = lance_dataset_list_scalar_indexed_columns(bind_data.dataset,
+                                                          &indexed_columns);
       bool has_indexed_filter = false;
-      for (size_t i = 0; i < indexed_cols_len; i++) {
-        if (indexed_cols_ptr[i] &&
-            filtered_columns.count(indexed_cols_ptr[i])) {
-          has_indexed_filter = true;
-          break;
+      if (rc == 0) {
+        for (auto &column : LanceConsumeStringList(indexed_columns)) {
+          if (filtered_columns.count(column)) {
+            has_indexed_filter = true;
+            break;
+          }
         }
+      } else {
+        // Index discovery is an optimization; preserve the existing fallback
+        // to the regular scanner when it is unavailable.
+        (void)LanceConsumeLastError();
       }
-      lance_free_scalar_indexed_columns(indexed_cols_ptr, indexed_cols_len);
       if (has_indexed_filter) {
         scan_state.use_dataset_scanner = true;
         scan_state.max_threads = 1;
diff --git a/src/lance_storage.cpp b/src/lance_storage.cpp
index 4c52802..6724a0c 100644
--- a/src/lance_storage.cpp
+++ b/src/lance_storage.cpp
@@ -68,6 +68,26 @@ struct LanceRestNamespaceConfig {
   string headers_tsv; // Tab-separated key\tvalue pairs for custom headers
 };
 
+static string EffectiveNamespaceDelimiter(const string &delimiter) {
+  return delimiter.empty() ? "$" : delimiter;
+}
+
+static string AppendRestIdentifier(const string &identifier,
+                                   const string &delimiter,
+                                   const string &segment) {
+  auto effective_delimiter = EffectiveNamespaceDelimiter(delimiter);
+  if (segment.find(effective_delimiter) != string::npos) {
+    throw InvalidInputException(
+        "Lance REST identifier segment '%s' contains the configured delimiter "
+        "'%s'",
+        segment, effective_delimiter);
+  }
+  if (identifier.empty()) {
+    return segment;
+  }
+  return identifier + effective_delimiter + segment;
+}
+
 static string GetLanceNamespaceEndpoint(const AttachInfo &info) {
   for (auto &kv : info.options) {
     if (!StringUtil::CIEquals(kv.first, "endpoint") || kv.second.IsNull()) {
@@ -210,23 +230,16 @@ ListDirectoryNamespaceTables(const LanceDirectoryNamespaceConfig &ns) {
   BuildStorageOptionPointerArrays(ns.option_keys, ns.option_values, key_ptrs,
                                   value_ptrs);
 
-  auto *ptr = lance_dir_namespace_list_tables(
+  LanceStringList list{nullptr, 0};
+  auto rc = lance_dir_namespace_list_tables(
       ns.root.c_str(), key_ptrs.empty() ? nullptr : key_ptrs.data(),
-      value_ptrs.empty() ? nullptr : value_ptrs.data(), ns.option_keys.size());
-  if (!ptr) {
+      value_ptrs.empty() ? nullptr : value_ptrs.data(), ns.option_keys.size(),
+      &list);
+  if (rc != 0) {
     throw IOException("Failed to list tables from Lance directory namespace: " +
                       ns.root + LanceFormatErrorSuffix());
   }
-  string joined = ptr;
-  lance_free_string(ptr);
-
-  vector<string> out;
-  for (auto &p : StringUtil::Split(joined, '\n')) {
-    if (!p.empty()) {
-      out.push_back(std::move(p));
-    }
-  }
-  return out;
+  return LanceConsumeStringList(list);
 }
 
 static vector<string>
@@ -239,23 +252,15 @@ ListRestNamespaceTables(const string &endpoint, const string &namespace_id,
   const char *delimiter_ptr = delimiter.empty() ? nullptr : delimiter.c_str();
   const char *headers_ptr = headers_tsv.empty() ? nullptr : headers_tsv.c_str();
 
-  auto *ptr = lance_namespace_list_tables(
-      endpoint.c_str(), namespace_id.c_str(), bearer_ptr, api_key_ptr,
-      delimiter_ptr, headers_ptr);
-  if (!ptr) {
+  LanceStringList list{nullptr, 0};
+  auto rc = lance_namespace_list_tables(endpoint.c_str(), namespace_id.c_str(),
+                                        bearer_ptr, api_key_ptr, delimiter_ptr,
+                                        headers_ptr, &list);
+  if (rc != 0) {
     throw IOException("Failed to list tables from Lance namespace: " +
                       endpoint + "/" + namespace_id + LanceFormatErrorSuffix());
   }
-  string joined = ptr;
-  lance_free_string(ptr);
-
-  vector<string> out;
-  for (auto &p : StringUtil::Split(joined, '\n')) {
-    if (!p.empty()) {
-      out.push_back(std::move(p));
-    }
-  }
-  return out;
+  return LanceConsumeStringList(list);
 }
 
 static bool
@@ -449,15 +454,10 @@ class LanceRestNamespaceDefaultGenerator : public DefaultGenerator {
       resolved_api_key = api_key;
     }
 
-    // Build candidate table IDs (bare name + optional namespace-prefixed).
-    vector<string> candidates = {entry_name};
-    if (!namespace_id.empty()) {
-      auto delim = delimiter.empty() ? "$" : delimiter;
-      auto prefix = namespace_id + delim;
-      if (!StringUtil::StartsWith(entry_name, prefix)) {
-        candidates.push_back(prefix + entry_name);
-      }
-    }
+    // A qualified lookup must never retry in a different namespace after it
+    // fails.
+    vector<string> candidates = {
+        AppendRestIdentifier(namespace_id, delimiter, entry_name)};
 
     // Fast path: describe_table with schema from REST API (skips S3 open).
     for (auto &table_id : candidates) {
@@ -523,19 +523,8 @@ class LanceRestNamespaceDefaultGenerator : public DefaultGenerator {
   }
 
   vector<string> GetDefaultEntries() override {
-    auto tables = ListRestNamespaceTables(endpoint, namespace_id, bearer_token,
-                                          api_key, delimiter, headers_tsv);
-    if (namespace_id.empty()) {
-      return tables;
-    }
-    auto delim = delimiter.empty() ? "$" : delimiter;
-    auto prefix = namespace_id + delim;
-    for (auto &t : tables) {
-      if (StringUtil::StartsWith(t, prefix)) {
-        t = t.substr(prefix.size());
-      }
-    }
-    return tables;
+    return ListRestNamespaceTables(endpoint, namespace_id, bearer_token,
+                                   api_key, delimiter, headers_tsv);
   }
 
 private:
@@ -659,10 +648,20 @@ class LanceSchemaEntry final : public DuckSchemaEntry {
       : DuckSchemaEntry(catalog, info), directory_ns(std::move(directory_ns)),
         rest_ns(std::move(rest_ns)) {}
 
+  const shared_ptr<LanceRestNamespaceConfig> &GetRestNamespace() const {
+    return rest_ns;
+  }
+
   void SetTableDefaultGenerator(DefaultGenerator *generator) {
     table_default_generator = generator;
   }
 
+  void InvalidateTableDefaults() {
+    if (table_default_generator) {
+      table_default_generator->created_all_entries = false;
+    }
+  }
+
   void Alter(CatalogTransaction transaction, AlterInfo &info) override {
     auto &set = GetCatalogSet(info.GetCatalogType());
     auto entry = set.GetEntry(transaction, info.name);
@@ -752,6 +751,10 @@ class LanceSchemaEntry final : public DuckSchemaEntry {
     auto existing_type = existing_entry->type;
 
     if (rest_ns) {
+      if (!context.transaction.IsAutoCommit()) {
+        throw NotImplementedException(
+            "Lance table DDL does not support explicit transactions");
+      }
       unordered_map<string, Value> overrides;
       if (!rest_ns->bearer_token_override.empty()) {
         overrides["bearer_token"] = Value(rest_ns->bearer_token_override);
@@ -766,14 +769,9 @@ class LanceSchemaEntry final : public DuckSchemaEntry {
                                 bearer_token, api_key);
 
       auto leaf_id = info.name;
-      string prefixed_id;
-      if (!rest_ns->namespace_id.empty()) {
-        auto delim = rest_ns->delimiter.empty() ? "$" : rest_ns->delimiter;
-        auto prefix = rest_ns->namespace_id + delim;
-        if (!StringUtil::StartsWith(leaf_id, prefix)) {
-          prefixed_id = prefix + leaf_id;
-        }
-      }
+      auto qualified_id = AppendRestIdentifier(rest_ns->namespace_id,
+                                               rest_ns->delimiter, leaf_id);
+      auto qualified_display = qualified_id;
 
       vector<string> discovered;
       string list_error;
@@ -784,14 +782,12 @@ class LanceSchemaEntry final : public DuckSchemaEntry {
         throw IOException("Failed to list tables from Lance namespace: " +
                           (list_error.empty() ? "unknown error" : list_error));
       }
-      string table_id_for_ops = prefixed_id.empty() ? leaf_id : prefixed_id;
+      string table_id_for_ops = qualified_id;
       for (auto &t : discovered) {
-        if (!prefixed_id.empty() && StringUtil::CIEquals(t, prefixed_id)) {
-          table_id_for_ops = prefixed_id;
+        if (StringUtil::CIEquals(t, qualified_display)) {
           break;
         }
         if (StringUtil::CIEquals(t, leaf_id)) {
-          table_id_for_ops = leaf_id;
           break;
         }
       }
@@ -887,6 +883,10 @@ class LanceSchemaEntry final : public DuckSchemaEntry {
     vector<string> option_values;
 
     if (rest_ns) {
+      if (!context.transaction.IsAutoCommit()) {
+        throw NotImplementedException(
+            "Lance table DDL does not support explicit transactions");
+      }
       unordered_map<string, Value> overrides;
       if (!rest_ns->bearer_token_override.empty()) {
         overrides["bearer_token"] = Value(rest_ns->bearer_token_override);
@@ -901,14 +901,9 @@ class LanceSchemaEntry final : public DuckSchemaEntry {
                                 bearer_token, api_key);
 
       auto leaf_id = create_info.table;
-      string prefixed_id;
-      if (!rest_ns->namespace_id.empty()) {
-        auto delim = rest_ns->delimiter.empty() ? "$" : rest_ns->delimiter;
-        auto prefix = rest_ns->namespace_id + delim;
-        if (!StringUtil::StartsWith(leaf_id, prefix)) {
-          prefixed_id = prefix + leaf_id;
-        }
-      }
+      auto qualified_id = AppendRestIdentifier(rest_ns->namespace_id,
+                                               rest_ns->delimiter, leaf_id);
+      auto qualified_display = qualified_id;
 
       vector<string> discovered;
       string list_error;
@@ -922,9 +917,9 @@ class LanceSchemaEntry final : public DuckSchemaEntry {
       bool exists = false;
       string existing_id;
       for (auto &t : discovered) {
-        if (!prefixed_id.empty() && StringUtil::CIEquals(t, prefixed_id)) {
+        if (StringUtil::CIEquals(t, qualified_display)) {
           exists = true;
-          existing_id = prefixed_id;
+          existing_id = t;
           break;
         }
         if (StringUtil::CIEquals(t, leaf_id)) {
@@ -933,8 +928,7 @@ class LanceSchemaEntry final : public DuckSchemaEntry {
           break;
         }
       }
-      auto table_id_for_ops =
-          exists ? existing_id : (prefixed_id.empty() ? leaf_id : prefixed_id);
+      auto table_id_for_ops = qualified_id;
       if (create_info.on_conflict == OnCreateConflict::IGNORE_ON_CONFLICT &&
           exists) {
         InvalidateTableDefaults();
@@ -962,27 +956,9 @@ class LanceSchemaEntry final : public DuckSchemaEntry {
               context, rest_ns->endpoint, table_id_for_ops, bearer_token,
               api_key, rest_ns->delimiter, rest_ns->headers_tsv, dataset_path,
               option_keys, option_values, create_error)) {
-        // Best-effort fallback for namespace implementations that do not use
-        // a qualified object identifier for tables in ListTables.
-        if (!prefixed_id.empty() && table_id_for_ops == prefixed_id) {
-          option_keys.clear();
-          option_values.clear();
-          dataset_path.clear();
-          create_error.clear();
-          if (!TryLanceNamespaceCreateEmptyTable(
-                  context, rest_ns->endpoint, leaf_id, bearer_token, api_key,
-                  rest_ns->delimiter, rest_ns->headers_tsv, dataset_path,
-                  option_keys, option_values, create_error)) {
-            throw IOException(
-                "Failed to create Lance table via namespace: " +
-                (create_error.empty() ? "unknown error" : create_error));
-          }
-          table_id_for_ops = leaf_id;
-        } else {
-          throw IOException(
-              "Failed to create Lance table via namespace: " +
-              (create_error.empty() ? "unknown error" : create_error));
-        }
+        throw IOException(
+            "Failed to create Lance table via namespace: " +
+            (create_error.empty() ? "unknown error" : create_error));
       }
       if (dataset_path.empty()) {
         throw IOException(
@@ -1090,18 +1066,75 @@ class LanceSchemaEntry final : public DuckSchemaEntry {
   }
 
 private:
-  void InvalidateTableDefaults() {
-    if (!table_default_generator) {
-      return;
-    }
-    table_default_generator->created_all_entries = false;
-  }
-
   shared_ptr<LanceDirectoryNamespaceConfig> directory_ns;
   shared_ptr<LanceRestNamespaceConfig> rest_ns;
   DefaultGenerator *table_default_generator = nullptr;
 };
 
+static void InvalidateLanceSchema(ClientContext &context,
+                                  const string &catalog_name,
+                                  const string &schema_name) {
+  auto schema = Catalog::GetSchema(context, catalog_name, schema_name,
+                                   OnEntryNotFound::RETURN_NULL);
+  auto *lance_schema =
+      schema ? dynamic_cast<LanceSchemaEntry *>(schema.get()) : nullptr;
+  if (lance_schema) {
+    lance_schema->InvalidateTableDefaults();
+  }
+}
+
+class PhysicalLanceCopyToFile final : public PhysicalCopyToFile {
+public:
+  PhysicalLanceCopyToFile(PhysicalPlan &physical_plan,
+                          vector<LogicalType> types, CopyFunction function,
+                          unique_ptr<FunctionData> bind_data,
+                          idx_t estimated_cardinality, string catalog_name,
+                          string schema_name)
+      : PhysicalCopyToFile(physical_plan, std::move(types), std::move(function),
+                           std::move(bind_data), estimated_cardinality),
+        catalog_name(std::move(catalog_name)),
+        schema_name(std::move(schema_name)) {}
+
+  SinkFinalizeType Finalize(Pipeline &pipeline, Event &event,
+                            ClientContext &context,
+                            OperatorSinkFinalizeInput &input) const override {
+    auto result = PhysicalCopyToFile::Finalize(pipeline, event, context, input);
+    InvalidateLanceSchema(context, catalog_name, schema_name);
+    return result;
+  }
+
+private:
+  string catalog_name;
+  string schema_name;
+};
+
+class PhysicalLanceBatchCopyToFile final : public PhysicalBatchCopyToFile {
+public:
+  PhysicalLanceBatchCopyToFile(PhysicalPlan &physical_plan,
+                               vector<LogicalType> types, CopyFunction function,
+                               unique_ptr<FunctionData> bind_data,
+                               idx_t estimated_cardinality, string catalog_name,
+                               string schema_name)
+      : PhysicalBatchCopyToFile(physical_plan, std::move(types),
+                                std::move(function), std::move(bind_data),
+                                estimated_cardinality),
+        catalog_name(std::move(catalog_name)),
+        schema_name(std::move(schema_name)) {}
+
+  SinkFinalizeType Finalize(Pipeline &pipeline, Event &event,
+                            ClientContext &context,
+                            OperatorSinkFinalizeInput &input) const override {
+    auto result =
+        PhysicalBatchCopyToFile::Finalize(pipeline, event, context, input);
+    InvalidateLanceSchema(context, catalog_name, schema_name);
+    return result;
+  }
+
+private:
+  string catalog_name;
+  string schema_name;
+};
+
 class LanceDuckCatalog final : public DuckCatalog {
 public:
   using DuckCatalog::PlanDelete;
@@ -1115,6 +1148,111 @@ class LanceDuckCatalog final : public DuckCatalog {
 
   using DuckCatalog::PlanUpdate;
 
+  optional_ptr<CatalogEntry> CreateSchema(CatalogTransaction transaction,
+                                          CreateSchemaInfo &info) override {
+    if (directory_ns && !info.internal && info.schema != DEFAULT_SCHEMA &&
+        !DefaultSchemaGenerator::IsDefaultSchema(info.schema)) {
+      throw NotImplementedException(
+          "CREATE SCHEMA is not supported for legacy Lance directory "
+          "namespaces because manifest mode is disabled");
+    }
+    if (rest_ns && !info.internal && info.schema != DEFAULT_SCHEMA &&
+        !DefaultSchemaGenerator::IsDefaultSchema(info.schema)) {
+      auto &context = transaction.GetContext();
+      if (!context.transaction.IsAutoCommit()) {
+        throw NotImplementedException(
+            "Lance schema DDL does not support explicit transactions");
+      }
+      if (info.on_conflict == OnCreateConflict::REPLACE_ON_CONFLICT) {
+        throw NotImplementedException(
+            "CREATE OR REPLACE SCHEMA is not supported for Lance namespaces");
+      }
+      string bearer_token;
+      string api_key;
+      ResolveRestAuth(context, bearer_token, api_key);
+      auto child_ns = MakeRestChildNamespace(info.schema);
+      auto result = CreateRestSchemaEntry(transaction, info, child_ns,
+                                          bearer_token, api_key);
+      if (!result) {
+        if (info.on_conflict == OnCreateConflict::ERROR_ON_CONFLICT) {
+          throw CatalogException::EntryAlreadyExists(CatalogType::SCHEMA_ENTRY,
+                                                     info.schema);
+        }
+        return nullptr;
+      }
+      string error;
+      if (!TryLanceNamespaceCreateNamespace(
+              context, rest_ns->endpoint, child_ns->namespace_id, bearer_token,
+              api_key, rest_ns->delimiter, rest_ns->headers_tsv,
+              CreateNamespaceMode(info.on_conflict), error)) {
+        throw IOException("Failed to create Lance schema '%s': %s", info.schema,
+                          error);
+      }
+      return result;
+    }
+    return DuckCatalog::CreateSchema(transaction, info);
+  }
+
+  void LoadRestSchemas(ClientContext &context, CatalogTransaction transaction,
+                       const vector<string> &schema_names) {
+    string bearer_token;
+    string api_key;
+    ResolveRestAuth(context, bearer_token, api_key);
+    for (auto &schema_name : schema_names) {
+      CreateSchemaInfo info;
+      info.schema = schema_name;
+      info.internal = false;
+      info.on_conflict = OnCreateConflict::IGNORE_ON_CONFLICT;
+      (void)CreateRestSchemaEntry(transaction, info,
+                                  MakeRestChildNamespace(schema_name),
+                                  bearer_token, api_key);
+    }
+  }
+
+  void DropSchema(ClientContext &context, DropInfo &info) override {
+    if (!rest_ns || info.name == DEFAULT_SCHEMA ||
+        DefaultSchemaGenerator::IsDefaultSchema(info.name)) {
+      auto transaction = GetCatalogTransaction(context);
+      if (!GetSchemaCatalogSet().DropEntry(transaction, info.name,
+                                           info.cascade) &&
+          info.if_not_found == OnEntryNotFound::THROW_EXCEPTION) {
+        throw CatalogException::MissingEntry(CatalogType::SCHEMA_ENTRY,
+                                             info.name, string());
+      }
+      return;
+    }
+    auto transaction = GetCatalogTransaction(context);
+    if (!context.transaction.IsAutoCommit()) {
+      throw NotImplementedException(
+          "Lance schema DDL does not support explicit transactions");
+    }
+    auto existing = GetSchemaCatalogSet().GetEntry(transaction, info.name);
+    if (!existing) {
+      if (info.if_not_found == OnEntryNotFound::THROW_EXCEPTION) {
+        throw CatalogException::MissingEntry(CatalogType::SCHEMA_ENTRY,
+                                             info.name, string());
+      }
+      return;
+    }
+    if (!GetSchemaCatalogSet().DropEntry(transaction, existing->name,
+                                         info.cascade)) {
+      throw InternalException("Failed to drop Lance schema entry: " +
+                              existing->name);
+    }
+    string bearer_token;
+    string api_key;
+    ResolveRestAuth(context, bearer_token, api_key);
+    auto child_ns = MakeRestChildNamespace(existing->name);
+    string error;
+    if (!TryLanceNamespaceDropNamespace(
+            context, rest_ns->endpoint, child_ns->namespace_id, bearer_token,
+            api_key, rest_ns->delimiter, rest_ns->headers_tsv, info.cascade,
+            error)) {
+      throw IOException("Failed to drop Lance schema '%s': %s", info.name,
+                        error);
+    }
+  }
+
   ErrorData SupportsCreateTable(BoundCreateTableInfo &info) override {
     auto &base = info.Base().Cast<CreateTableInfo>();
     if (!base.partition_keys.empty()) {
@@ -1186,15 +1324,27 @@ class LanceDuckCatalog final : public DuckCatalog {
           "Lance ATTACH TYPE LANCE does not support TEMPORARY tables");
     }
     if (rest_ns) {
+      if (!context.transaction.IsAutoCommit()) {
+        throw NotImplementedException(
+            "Lance table DDL does not support explicit transactions");
+      }
+      auto *lance_schema = dynamic_cast<LanceSchemaEntry *>(&op.schema);
+      if (!lance_schema || !lance_schema->GetRestNamespace()) {
+        throw InternalException(
+            "REST Lance CTAS requires a namespace-backed schema");
+      }
+      auto schema_rest_ns = lance_schema->GetRestNamespace();
+
       class PhysicalLanceCreateTableAs final : public PhysicalOperator {
       public:
         PhysicalLanceCreateTableAs(
             PhysicalPlan &physical_plan, vector<LogicalType> types_p,
             string endpoint, string namespace_id, string delimiter,
             string bearer_token_override, string api_key_override,
-            string headers_tsv, string table_name, string writer_mode,
-            string data_storage_version, vector<string> column_names_p,
-            vector<LogicalType> column_types_p, idx_t estimated_cardinality)
+            string headers_tsv, string catalog_name, string schema_name,
+            string table_name, string writer_mode, string data_storage_version,
+            vector<string> column_names_p, vector<LogicalType> column_types_p,
+            idx_t estimated_cardinality)
             : PhysicalOperator(physical_plan, PhysicalOperatorType::EXTENSION,
                                std::move(types_p), estimated_cardinality),
               endpoint(std::move(endpoint)),
@@ -1203,6 +1353,8 @@ class LanceDuckCatalog final : public DuckCatalog {
               bearer_token_override(std::move(bearer_token_override)),
               api_key_override(std::move(api_key_override)),
               headers_tsv(std::move(headers_tsv)),
+              catalog_name(std::move(catalog_name)),
+              schema_name(std::move(schema_name)),
               table_name(std::move(table_name)),
               writer_mode(std::move(writer_mode)),
               data_storage_version(std::move(data_storage_version)),
@@ -1293,15 +1445,10 @@ class LanceDuckCatalog final : public DuckCatalog {
           ResolveLanceNamespaceAuth(context, state->endpoint, overrides,
                                     bearer_token, api_key);
 
-          auto delim = state->delimiter.empty() ? "$" : state->delimiter;
-          auto prefix = state->namespace_id.empty()
-                            ? string()
-                            : (state->namespace_id + delim);
           auto leaf_id = state->table_name;
-          string prefixed_id;
-          if (!prefix.empty() && !StringUtil::StartsWith(leaf_id, prefix)) {
-            prefixed_id = prefix + leaf_id;
-          }
+          auto qualified_id = AppendRestIdentifier(state->namespace_id,
+                                                   state->delimiter, leaf_id);
+          auto qualified_display = qualified_id;
 
           vector<string> discovered;
           string list_error;
@@ -1314,14 +1461,12 @@ class LanceDuckCatalog final : public DuckCatalog {
                 (list_error.empty() ? "unknown error" : list_error));
           }
 
-          state->table_id = prefixed_id.empty() ? leaf_id : prefixed_id;
+          state->table_id = qualified_id;
           for (auto &t : discovered) {
-            if (!prefixed_id.empty() && StringUtil::CIEquals(t, prefixed_id)) {
-              state->table_id = prefixed_id;
+            if (StringUtil::CIEquals(t, qualified_display)) {
               break;
             }
             if (StringUtil::CIEquals(t, leaf_id)) {
-              state->table_id = leaf_id;
               break;
             }
           }
@@ -1345,26 +1490,9 @@ class LanceDuckCatalog final : public DuckCatalog {
                   api_key, state->delimiter, state->headers_tsv,
                   state->open_path, state->option_keys, state->option_values,
                   create_error)) {
-            if (!prefixed_id.empty() && state->table_id == prefixed_id) {
-              state->table_id = leaf_id;
-              state->open_path.clear();
-              state->option_keys.clear();
-              state->option_values.clear();
-              create_error.clear();
-              if (!TryLanceNamespaceCreateEmptyTable(
-                      context, state->endpoint, state->table_id, bearer_token,
-                      api_key, state->delimiter, state->headers_tsv,
-                      state->open_path, state->option_keys,
-                      state->option_values, create_error)) {
-                throw IOException(
-                    "Failed to create Lance table via namespace: " +
-                    (create_error.empty() ? "unknown error" : create_error));
-              }
-            } else {
-              throw IOException(
-                  "Failed to create Lance table via namespace: " +
-                  (create_error.empty() ? "unknown error" : create_error));
-            }
+            throw IOException(
+                "Failed to create Lance table via namespace: " +
+                (create_error.empty() ? "unknown error" : create_error));
           }
           if (state->open_path.empty()) {
             throw IOException(
@@ -1441,7 +1569,6 @@ class LanceDuckCatalog final : public DuckCatalog {
         SinkFinalizeType
         Finalize(Pipeline &, Event &, ClientContext &context,
                  OperatorSinkFinalizeInput &input) const override {
-          (void)context;
           auto &gstate = input.global_state.Cast<GlobalState>();
 
           {
@@ -1455,6 +1582,8 @@ class LanceDuckCatalog final : public DuckCatalog {
             }
           }
 
+          InvalidateLanceSchema(context, catalog_name, schema_name);
+
           return SinkFinalizeType::READY;
         }
 
@@ -1493,6 +1622,8 @@ class LanceDuckCatalog final : public DuckCatalog {
         string bearer_token_override;
         string api_key_override;
         string headers_tsv;
+        string catalog_name;
+        string schema_name;
         string table_name;
         string writer_mode;
         string data_storage_version;
@@ -1503,43 +1634,38 @@ class LanceDuckCatalog final : public DuckCatalog {
       // Use LIST TABLES to implement conflict behavior in a side-effect-free
       // way.
       unordered_map<string, Value> overrides;
-      if (!rest_ns->bearer_token_override.empty()) {
-        overrides["bearer_token"] = Value(rest_ns->bearer_token_override);
+      if (!schema_rest_ns->bearer_token_override.empty()) {
+        overrides["bearer_token"] =
+            Value(schema_rest_ns->bearer_token_override);
       }
-      if (!rest_ns->api_key_override.empty()) {
-        overrides["api_key"] = Value(rest_ns->api_key_override);
+      if (!schema_rest_ns->api_key_override.empty()) {
+        overrides["api_key"] = Value(schema_rest_ns->api_key_override);
       }
       string bearer_token;
       string api_key;
-      ResolveLanceNamespaceAuth(context, rest_ns->endpoint, overrides,
+      ResolveLanceNamespaceAuth(context, schema_rest_ns->endpoint, overrides,
                                 bearer_token, api_key);
 
       vector<string> discovered;
       string list_error;
       if (!TryLanceNamespaceListTables(
-              context, rest_ns->endpoint, rest_ns->namespace_id, bearer_token,
-              api_key, rest_ns->delimiter, rest_ns->headers_tsv, discovered,
-              list_error)) {
+              context, schema_rest_ns->endpoint, schema_rest_ns->namespace_id,
+              bearer_token, api_key, schema_rest_ns->delimiter,
+              schema_rest_ns->headers_tsv, discovered, list_error)) {
         throw IOException("Failed to list tables from Lance namespace: " +
                           (list_error.empty() ? "unknown error" : list_error));
       }
 
-      auto delim = rest_ns->delimiter.empty() ? "$" : rest_ns->delimiter;
-      auto prefix = rest_ns->namespace_id.empty()
-                        ? string()
-                        : (rest_ns->namespace_id + delim);
       auto leaf_id = create_info.table;
-      string prefixed_id;
-      if (!prefix.empty() && !StringUtil::StartsWith(leaf_id, prefix)) {
-        prefixed_id = prefix + leaf_id;
-      }
+      auto qualified_display = AppendRestIdentifier(
+          schema_rest_ns->namespace_id, schema_rest_ns->delimiter, leaf_id);
 
       bool exists = false;
       string existing_id;
       for (auto &t : discovered) {
-        if (!prefixed_id.empty() && StringUtil::CIEquals(t, prefixed_id)) {
+        if (StringUtil::CIEquals(t, qualified_display)) {
           exists = true;
-          existing_id = prefixed_id;
+          existing_id = t;
           break;
         }
         if (StringUtil::CIEquals(t, leaf_id)) {
@@ -1563,10 +1689,11 @@ class LanceDuckCatalog final : public DuckCatalog {
       auto types = create_info.columns.GetColumnTypes();
       string mode = CreateTableModeFromConflict(create_info.on_conflict);
       auto &create_as = planner.Make<PhysicalLanceCreateTableAs>(
-          op.types, rest_ns->endpoint, rest_ns->namespace_id,
-          rest_ns->delimiter, rest_ns->bearer_token_override,
-          rest_ns->api_key_override, rest_ns->headers_tsv, create_info.table,
-          mode, data_storage_version, std::move(names), std::move(types),
+          op.types, schema_rest_ns->endpoint, schema_rest_ns->namespace_id,
+          schema_rest_ns->delimiter, schema_rest_ns->bearer_token_override,
+          schema_rest_ns->api_key_override, schema_rest_ns->headers_tsv,
+          op.schema.catalog.GetName(), op.schema.name, create_info.table, mode,
+          data_storage_version, std::move(names), std::move(types),
           op.estimated_cardinality);
       create_as.children.push_back(plan);
       return create_as;
@@ -1635,10 +1762,11 @@ class LanceDuckCatalog final : public DuckCatalog {
     }
 
     if (execution_mode == CopyFunctionExecutionMode::BATCH_COPY_TO_FILE) {
-      auto &copy = planner.Make<PhysicalBatchCopyToFile>(
+      auto &copy = planner.Make<PhysicalLanceBatchCopyToFile>(
           op.types, copy_function, std::move(bind_data),
-          op.estimated_cardinality);
-      auto &cast_copy = copy.Cast<PhysicalBatchCopyToFile>();
+          op.estimated_cardinality, op.schema.catalog.GetName(),
+          op.schema.name);
+      auto &cast_copy = copy.Cast<PhysicalLanceBatchCopyToFile>();
       cast_copy.file_path = dataset_path;
       cast_copy.use_tmp_file = false;
       cast_copy.return_type = CopyFunctionReturnType::CHANGED_ROWS;
@@ -1647,10 +1775,10 @@ class LanceDuckCatalog final : public DuckCatalog {
       return copy;
     }
 
-    auto &copy = planner.Make<PhysicalCopyToFile>(op.types, copy_function,
-                                                  std::move(bind_data),
-                                                  op.estimated_cardinality);
-    auto &cast_copy = copy.Cast<PhysicalCopyToFile>();
+    auto &copy = planner.Make<PhysicalLanceCopyToFile>(
+        op.types, copy_function, std::move(bind_data), op.estimated_cardinality,
+        op.schema.catalog.GetName(), op.schema.name);
+    auto &cast_copy = copy.Cast<PhysicalLanceCopyToFile>();
     cast_copy.file_path = dataset_path;
     cast_copy.use_tmp_file = false;
     cast_copy.filename_pattern = FilenamePattern();
@@ -1701,6 +1829,64 @@ class LanceDuckCatalog final : public DuckCatalog {
   }
 
 private:
+  string CreateNamespaceMode(OnCreateConflict conflict) const {
+    switch (conflict) {
+    case OnCreateConflict::ERROR_ON_CONFLICT:
+      return "Create";
+    case OnCreateConflict::IGNORE_ON_CONFLICT:
+      return "ExistOk";
+    case OnCreateConflict::REPLACE_ON_CONFLICT:
+      return "Overwrite";
+    default:
+      throw InternalException("Unsupported CREATE SCHEMA conflict mode");
+    }
+  }
+
+  shared_ptr<LanceRestNamespaceConfig>
+  MakeRestChildNamespace(const string &schema_name) const {
+    auto child = make_shared_ptr<LanceRestNamespaceConfig>(*rest_ns);
+    child->namespace_id = AppendRestIdentifier(rest_ns->namespace_id,
+                                               rest_ns->delimiter, schema_name);
+    return child;
+  }
+
+  void ResolveRestAuth(ClientContext &context, string &bearer_token,
+                       string &api_key) const {
+    unordered_map<string, Value> overrides;
+    if (!rest_ns->bearer_token_override.empty()) {
+      overrides["bearer_token"] = Value(rest_ns->bearer_token_override);
+    }
+    if (!rest_ns->api_key_override.empty()) {
+      overrides["api_key"] = Value(rest_ns->api_key_override);
+    }
+    ResolveLanceNamespaceAuth(context, rest_ns->endpoint, overrides,
+                              bearer_token, api_key);
+  }
+
+  optional_ptr<CatalogEntry>
+  CreateRestSchemaEntry(CatalogTransaction transaction, CreateSchemaInfo &info,
+                        shared_ptr<LanceRestNamespaceConfig> schema_ns,
+                        const string &bearer_token, const string &api_key) {
+    auto &schemas = GetSchemaCatalogSet();
+    LogicalDependencyList dependencies;
+    auto entry = make_uniq<LanceSchemaEntry>(*this, info, nullptr, schema_ns);
+    auto result = entry.get();
+    if (!schemas.CreateEntry(transaction, info.schema, std::move(entry),
+                             dependencies)) {
+      return nullptr;
+    }
+    auto &table_set = result->GetCatalogSet(CatalogType::TABLE_ENTRY);
+    auto generator = make_uniq<LanceRestNamespaceDefaultGenerator>(
+        *this, *result, schema_ns->endpoint, schema_ns->namespace_id,
+        bearer_token, api_key, schema_ns->delimiter,
+        schema_ns->bearer_token_override, schema_ns->api_key_override,
+        schema_ns->headers_tsv);
+    auto *generator_ptr = generator.get();
+    table_set.SetDefaultGenerator(std::move(generator));
+    result->SetTableDefaultGenerator(generator_ptr);
+    return result;
+  }
+
   shared_ptr<LanceDirectoryNamespaceConfig> directory_ns;
   shared_ptr<LanceRestNamespaceConfig> rest_ns;
 };
@@ -1740,6 +1926,7 @@ LanceStorageAttach(optional_ptr<StorageExtensionInfo>, ClientContext &context,
 
   auto is_rest_namespace = !endpoint.empty();
   string namespace_id;
+  vector<string> discovered_namespaces;
   string bearer_token;
   string api_key;
   string bearer_token_override;
@@ -1767,11 +1954,11 @@ LanceStorageAttach(optional_ptr<StorageExtensionInfo>, ClientContext &context,
     directory_ns->option_keys = std::move(option_keys);
     directory_ns->option_values = std::move(option_values);
   } else {
-    namespace_id = attach_path;
-    if (namespace_id.empty()) {
+    if (attach_path.empty()) {
       throw InvalidInputException(
           "ATTACH TYPE LANCE with ENDPOINT requires a non-empty namespace id");
     }
+    namespace_id = attach_path;
     ResolveLanceNamespaceAuth(context, endpoint, info.options, bearer_token,
                               api_key);
     ResolveLanceNamespaceAuthOverrides(info.options, bearer_token_override,
@@ -1793,6 +1980,13 @@ LanceStorageAttach(optional_ptr<StorageExtensionInfo>, ClientContext &context,
     rest_ns->bearer_token_override = bearer_token_override;
     rest_ns->api_key_override = api_key_override;
     rest_ns->headers_tsv = headers_tsv;
+
+    if (!TryLanceNamespaceListNamespaces(
+            context, endpoint, namespace_id, bearer_token, api_key, delimiter,
+            headers_tsv, discovered_namespaces, list_error)) {
+      throw IOException("Failed to list schemas from Lance namespace: " +
+                        list_error);
+    }
   }
 
   // Back the attached catalog by an in-memory DuckCatalog that lazily
@@ -1824,6 +2018,11 @@ LanceStorageAttach(optional_ptr<StorageExtensionInfo>, ClientContext &context,
   catalog_set.SetDefaultGenerator(std::move(generator));
   lance_schema.SetTableDefaultGenerator(generator_ptr);
 
+  if (rest_ns && !discovered_namespaces.empty()) {
+    catalog->LoadRestSchemas(context, system_transaction,
+                             discovered_namespaces);
+  }
+
   (void)name;
   return std::move(catalog);
 }
diff --git a/test/sql/namespace_ctas_catalog_visibility.test b/test/sql/namespace_ctas_catalog_visibility.test
new file mode 100644
index 0000000..7bca665
--- /dev/null
+++ b/test/sql/namespace_ctas_catalog_visibility.test
@@ -0,0 +1,33 @@
+# name: test/sql/namespace_ctas_catalog_visibility.test
+# description: CTAS refreshes the shared directory schema catalog after commit
+# group: [sql]
+
+require lance
+
+statement ok con1
+ATTACH '__TEST_DIR__/nsroot_ctas_visibility' AS ns (TYPE LANCE);
+
+statement ok con1
+CREATE TABLE ns.main.seed AS SELECT 1::BIGINT AS id;
+
+query T con1
+SHOW TABLES FROM ns.main
+----
+seed
+
+statement ok con2
+CREATE TABLE ns.main.ctas_new AS SELECT 42::BIGINT AS id;
+
+query T con1
+SHOW TABLES FROM ns.main
+----
+ctas_new
+seed
+
+query I con1
+SELECT sum(id) FROM ns.main.ctas_new
+----
+42
+
+statement ok con1
+DETACH ns;
diff --git a/test/sql/namespace_directory_schema_unsupported.test b/test/sql/namespace_directory_schema_unsupported.test
new file mode 100644
index 0000000..2aaf7b7
--- /dev/null
+++ b/test/sql/namespace_directory_schema_unsupported.test
@@ -0,0 +1,39 @@
+# name: test/sql/namespace_directory_schema_unsupported.test
+# description: Legacy directory namespaces reject child schemas
+# group: [sql]
+
+require lance
+
+statement ok
+ATTACH '__TEST_DIR__/nsroot_directory_schema_unsupported' AS ns (TYPE LANCE);
+
+statement error
+CREATE SCHEMA ns.s1;
+----
+Not implemented Error: CREATE SCHEMA is not supported for legacy Lance directory namespaces because manifest mode is disabled
+
+statement error
+CREATE TABLE ns.s1.regular_t (id BIGINT);
+----
+Catalog Error: Schema with name s1 does not exist!
+
+statement error
+CREATE TABLE ns.s1.ctas_t AS SELECT 1::BIGINT AS id;
+----
+Catalog Error: Schema with name s1 does not exist!
+
+statement ok
+CREATE TABLE ns.main.main_t AS SELECT 1::BIGINT AS id;
+
+query I
+SELECT sum(id) FROM '__TEST_DIR__/nsroot_directory_schema_unsupported/main_t.lance'
+----
+1
+
+statement error
+SELECT count(*) FROM '__TEST_DIR__/nsroot_directory_schema_unsupported/main/main_t.lance';
+----
+IO Error: Failed to open Lance dataset:
+
+statement ok
+DETACH ns;
diff --git a/test/sql/namespace_rest_ctas_catalog_visibility.test b/test/sql/namespace_rest_ctas_catalog_visibility.test
new file mode 100644
index 0000000..89c2819
--- /dev/null
+++ b/test/sql/namespace_rest_ctas_catalog_visibility.test
@@ -0,0 +1,35 @@
+# name: test/sql/namespace_rest_ctas_catalog_visibility.test
+# description: REST CTAS refreshes the shared schema catalog after commit
+# group: [sql]
+
+require-env LANCE_TEST_NAMESPACE 1
+
+test-env LANCE_NAMESPACE_ENDPOINT http://127.0.0.1:2333
+
+test-env LANCE_NAMESPACE_ID default
+
+test-env LANCE_TEST_CTAS_TABLE rest_ctas_visibility_356ebea
+
+require lance
+
+statement ok con1
+ATTACH '${LANCE_NAMESPACE_ID}' AS ns (TYPE LANCE, ENDPOINT '${LANCE_NAMESPACE_ENDPOINT}');
+
+statement ok con1
+SHOW TABLES FROM ns.main;
+
+statement ok con2
+CREATE TABLE ns.main."${LANCE_TEST_CTAS_TABLE}" AS SELECT 4242::BIGINT AS id;
+
+query I con1
+SELECT count(*)
+FROM (SHOW TABLES FROM ns.main)
+WHERE name = '${LANCE_TEST_CTAS_TABLE}'
+----
+1
+
+statement ok con1
+DROP TABLE ns.main."${LANCE_TEST_CTAS_TABLE}";
+
+statement ok con1
+DETACH ns;
diff --git a/test/sql/namespace_rest_ctas_schema.test b/test/sql/namespace_rest_ctas_schema.test
new file mode 100644
index 0000000..6e6521b
--- /dev/null
+++ b/test/sql/namespace_rest_ctas_schema.test
@@ -0,0 +1,49 @@
+# name: test/sql/namespace_rest_ctas_schema.test
+# description: REST CTAS writes to the namespace bound to its schema
+# group: [sql]
+
+require-env LANCE_TEST_NAMESPACE 1
+
+test-env LANCE_NAMESPACE_ENDPOINT http://127.0.0.1:2333
+
+test-env LANCE_NAMESPACE_ID default
+
+test-env LANCE_TEST_SCHEMA rest_ctas_schema_routing
+
+require lance
+
+statement ok
+ATTACH '${LANCE_NAMESPACE_ID}' AS ns (TYPE LANCE, ENDPOINT '${LANCE_NAMESPACE_ENDPOINT}');
+
+statement ok
+CREATE SCHEMA ns."${LANCE_TEST_SCHEMA}";
+
+statement ok
+CREATE TABLE ns."${LANCE_TEST_SCHEMA}".routed_table AS SELECT 42 AS id;
+
+statement ok
+DETACH ns;
+
+statement ok
+ATTACH '${LANCE_NAMESPACE_ID}' AS ns (TYPE LANCE, ENDPOINT '${LANCE_NAMESPACE_ENDPOINT}');
+
+query T
+SHOW TABLES FROM ns."${LANCE_TEST_SCHEMA}"
+----
+routed_table
+
+query I
+SELECT count(*) FROM duckdb_tables()
+WHERE database_name = 'ns' AND schema_name = 'main'
+  AND table_name = 'routed_table'
+----
+0
+
+statement ok
+DROP TABLE ns."${LANCE_TEST_SCHEMA}".routed_table;
+
+statement ok
+DROP SCHEMA ns."${LANCE_TEST_SCHEMA}";
+
+statement ok
+DETACH ns;
diff --git a/test/sql/namespace_rest_schema.test b/test/sql/namespace_rest_schema.test
new file mode 100644
index 0000000..b1f6ae1
--- /dev/null
+++ b/test/sql/namespace_rest_schema.test
@@ -0,0 +1,125 @@
+# name: test/sql/namespace_rest_schema.test
+# description: REST child namespaces are exposed as persistent DuckDB schemas
+# group: [sql]
+
+require-env LANCE_TEST_NAMESPACE 1
+
+test-env LANCE_NAMESPACE_ENDPOINT http://127.0.0.1:2333
+
+test-env LANCE_NAMESPACE_ID default
+
+test-env LANCE_TEST_SCHEMA rest_schema_crud
+
+require lance
+
+statement ok
+ATTACH '${LANCE_NAMESPACE_ID}' AS ns (TYPE LANCE, ENDPOINT '${LANCE_NAMESPACE_ENDPOINT}');
+
+statement ok
+BEGIN;
+
+statement error
+CREATE SCHEMA ns.rest_schema_in_transaction;
+----
+Not implemented Error: Lance schema DDL does not support explicit transactions
+
+statement ok
+ROLLBACK;
+
+statement error
+CREATE SCHEMA ns."x$y";
+----
+Invalid Input Error: Lance REST identifier segment 'x$y' contains the configured delimiter '$'
+
+statement ok
+CREATE SCHEMA ns."${LANCE_TEST_SCHEMA}";
+
+statement error
+CREATE TABLE ns."${LANCE_TEST_SCHEMA}"."x$y" (x INTEGER);
+----
+Invalid Input Error: Lance REST identifier segment 'x$y' contains the configured delimiter '$'
+
+statement error
+CREATE SCHEMA ns."${LANCE_TEST_SCHEMA}";
+----
+Catalog Error: Schema with name "${LANCE_TEST_SCHEMA}" already exists!
+
+statement error
+CREATE OR REPLACE SCHEMA ns."${LANCE_TEST_SCHEMA}";
+----
+Not implemented Error: CREATE OR REPLACE SCHEMA is not supported for Lance namespaces
+
+statement ok
+BEGIN;
+
+statement error
+CREATE TABLE ns."${LANCE_TEST_SCHEMA}".regular_in_transaction (x INTEGER);
+----
+Not implemented Error: Lance table DDL does not support explicit transactions
+
+statement ok
+ROLLBACK;
+
+statement ok
+BEGIN;
+
+statement error
+CREATE TABLE ns."${LANCE_TEST_SCHEMA}".ctas_in_transaction AS SELECT 1 AS x;
+----
+Not implemented Error: Lance table DDL does not support explicit transactions
+
+statement ok
+ROLLBACK;
+
+query T
+SELECT schema_name
+FROM duckdb_schemas()
+WHERE database_name = 'ns' AND schema_name = '${LANCE_TEST_SCHEMA}'
+----
+${LANCE_TEST_SCHEMA}
+
+statement ok
+DETACH ns;
+
+statement ok
+ATTACH '${LANCE_NAMESPACE_ID}' AS ns (TYPE LANCE, ENDPOINT '${LANCE_NAMESPACE_ENDPOINT}');
+
+query T
+SELECT schema_name
+FROM duckdb_schemas()
+WHERE database_name = 'ns' AND schema_name = '${LANCE_TEST_SCHEMA}'
+----
+${LANCE_TEST_SCHEMA}
+
+statement ok
+CREATE VIEW ns."${LANCE_TEST_SCHEMA}".dependency_view AS SELECT 42 AS x;
+
+statement error
+DROP SCHEMA ns."${LANCE_TEST_SCHEMA}";
+----
+Dependency Error: Cannot drop entry "${LANCE_TEST_SCHEMA}" because there are entries that depend on it.
+
+query I
+SELECT x FROM ns."${LANCE_TEST_SCHEMA}".dependency_view
+----
+42
+
+statement ok
+BEGIN;
+
+statement error
+DROP SCHEMA ns."${LANCE_TEST_SCHEMA}";
+----
+Not implemented Error: Lance schema DDL does not support explicit transactions
+
+statement ok
+ROLLBACK;
+
+statement ok
+DROP VIEW ns."${LANCE_TEST_SCHEMA}".dependency_view;
+
+statement ok
+DROP SCHEMA ns."${LANCE_TEST_SCHEMA}";
+
+statement ok
+DETACH ns;