From bcefd8a2ce5067f2028ff1730e79272a9fad6221 Mon Sep 17 00:00:00 2001 From: arrowbowang Date: Thu, 13 Aug 2026 19:30:50 +0800 Subject: [PATCH 01/11] fix(catalog): reject schemas in legacy directories Legacy directory attachments disable Lance namespace manifests and cannot persist child namespaces. Reject user-created schemas instead of falling back to transient DuckDB schema entries. Preserve the existing behavior that maps main directly to the catalog root. --- src/lance_storage.cpp | 11 ++++++ ...amespace_directory_schema_unsupported.test | 39 +++++++++++++++++++ 2 files changed, 50 insertions(+) create mode 100644 test/sql/namespace_directory_schema_unsupported.test diff --git a/src/lance_storage.cpp b/src/lance_storage.cpp index 4c52802e..578489e9 100644 --- a/src/lance_storage.cpp +++ b/src/lance_storage.cpp @@ -1115,6 +1115,17 @@ class LanceDuckCatalog final : public DuckCatalog { using DuckCatalog::PlanUpdate; + optional_ptr 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"); + } + return DuckCatalog::CreateSchema(transaction, info); + } + ErrorData SupportsCreateTable(BoundCreateTableInfo &info) override { auto &base = info.Base().Cast(); if (!base.partition_keys.empty()) { diff --git a/test/sql/namespace_directory_schema_unsupported.test b/test/sql/namespace_directory_schema_unsupported.test new file mode 100644 index 00000000..2aaf7b71 --- /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; From fe6c7891f88d23a4d3b4b768d8eecaf7602ccf59 Mon Sep 17 00:00:00 2001 From: arrowbowang Date: Thu, 13 Aug 2026 19:56:01 +0800 Subject: [PATCH 02/11] feat(catalog): expose REST child namespaces Reuse Lance namespace list, create, and drop operations to map REST child namespaces to DuckDB schemas. Rebuild schema entries and install their existing REST table generators on ATTACH so schemas persist across connections. --- rust/error.rs | 3 + rust/ffi/namespace.rs | 184 +++++++++++++++++++++++++++- src/include/lance_common.hpp | 13 ++ src/include/lance_ffi.hpp | 14 +++ src/lance_common.cpp | 67 ++++++++++ src/lance_storage.cpp | 151 +++++++++++++++++++++++ test/sql/namespace_rest_schema.test | 45 +++++++ 7 files changed, 476 insertions(+), 1 deletion(-) create mode 100644 test/sql/namespace_rest_schema.test diff --git a/rust/error.rs b/rust/error.rs index 70e41de3..5eb4c21a 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/namespace.rs b/rust/ffi/namespace.rs index cfba44e3..36e35b94 100644 --- a/rust/ffi/namespace.rs +++ b/rust/ffi/namespace.rs @@ -7,7 +7,8 @@ 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; @@ -86,6 +87,187 @@ fn storage_options_to_tsv(storage_options: std::collections::HashMap Vec { + if id.is_empty() { + Vec::new() + } else { + id.split(delimiter).map(ToString::to_string).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)> { + 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)) +} + +#[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, +) -> *const c_char { + 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 = namespace.list_namespaces(request).await.map_err(|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}")))? + })(); + match result { + Ok(namespaces) => { + clear_last_error(); + to_c_string(namespaces.join("\n")).into_raw() as *const c_char + } + Err(err) => { + set_last_error(err.code, err.message); + ptr::null() + } + } +} + +#[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, diff --git a/src/include/lance_common.hpp b/src/include/lance_common.hpp index 1d19d9f0..efb41f70 100644 --- a/src/include/lance_common.hpp +++ b/src/include/lance_common.hpp @@ -52,6 +52,19 @@ bool TryLanceNamespaceListTables(ClientContext &context, const string &endpoint, const string &api_key, const string &delimiter, const string &headers_tsv, vector &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 &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 &out_tables, diff --git a/src/include/lance_ffi.hpp b/src/include/lance_ffi.hpp index e6c0c1b4..324e43c9 100644 --- a/src/include/lance_ffi.hpp +++ b/src/include/lance_ffi.hpp @@ -52,6 +52,20 @@ 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); +const char * +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); +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( diff --git a/src/lance_common.cpp b/src/lance_common.cpp index a3d01745..dd1fa5d7 100644 --- a/src/lance_common.cpp +++ b/src/lance_common.cpp @@ -328,6 +328,73 @@ bool TryLanceNamespaceListTables( return true; } +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 &out_namespaces, + string &out_error) { + (void)context; + out_namespaces.clear(); + out_error.clear(); + auto *ptr = 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()); + if (!ptr) { + out_error = LanceConsumeLastError(); + return false; + } + string joined = ptr; + lance_free_string(ptr); + for (auto &name : StringUtil::Split(joined, '\n')) { + if (!name.empty()) { + out_namespaces.push_back(std::move(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 &out_keys, vector &out_values) { out_keys.clear(); diff --git a/src/lance_storage.cpp b/src/lance_storage.cpp index 578489e9..a1d64203 100644 --- a/src/lance_storage.cpp +++ b/src/lance_storage.cpp @@ -659,6 +659,10 @@ class LanceSchemaEntry final : public DuckSchemaEntry { : DuckSchemaEntry(catalog, info), directory_ns(std::move(directory_ns)), rest_ns(std::move(rest_ns)) {} + const shared_ptr &GetRestNamespace() const { + return rest_ns; + } + void SetTableDefaultGenerator(DefaultGenerator *generator) { table_default_generator = generator; } @@ -1123,9 +1127,84 @@ class LanceDuckCatalog final : public DuckCatalog { "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(); + string bearer_token; + string api_key; + ResolveRestAuth(context, bearer_token, api_key); + auto child_ns = MakeRestChildNamespace(info.schema); + 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 CreateRestSchemaEntry(transaction, info, std::move(child_ns), + bearer_token, api_key); + } return DuckCatalog::CreateSchema(transaction, info); } + void LoadRestSchemas(ClientContext &context, + CatalogTransaction transaction, + const vector &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); + 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; + } + 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); + } + if (!GetSchemaCatalogSet().DropEntry(transaction, existing->name, + info.cascade)) { + throw InternalException("Failed to drop Lance schema entry: " + + existing->name); + } + } + ErrorData SupportsCreateTable(BoundCreateTableInfo &info) override { auto &base = info.Base().Cast(); if (!base.partition_keys.empty()) { @@ -1712,6 +1791,65 @@ 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 + MakeRestChildNamespace(const string &schema_name) const { + auto child = make_shared_ptr(*rest_ns); + auto delimiter = rest_ns->delimiter.empty() ? "$" : rest_ns->delimiter; + child->namespace_id = rest_ns->namespace_id + delimiter + schema_name; + return child; + } + + void ResolveRestAuth(ClientContext &context, string &bearer_token, + string &api_key) const { + unordered_map 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 CreateRestSchemaEntry( + CatalogTransaction transaction, CreateSchemaInfo &info, + shared_ptr schema_ns, + const string &bearer_token, const string &api_key) { + auto &schemas = GetSchemaCatalogSet(); + LogicalDependencyList dependencies; + auto entry = + make_uniq(*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( + *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 directory_ns; shared_ptr rest_ns; }; @@ -1751,6 +1889,7 @@ LanceStorageAttach(optional_ptr, ClientContext &context, auto is_rest_namespace = !endpoint.empty(); string namespace_id; + vector discovered_namespaces; string bearer_token; string api_key; string bearer_token_override; @@ -1804,6 +1943,13 @@ LanceStorageAttach(optional_ptr, 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 @@ -1835,6 +1981,11 @@ LanceStorageAttach(optional_ptr, 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_rest_schema.test b/test/sql/namespace_rest_schema.test new file mode 100644 index 00000000..647eab58 --- /dev/null +++ b/test/sql/namespace_rest_schema.test @@ -0,0 +1,45 @@ +# 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 +CREATE SCHEMA ns."${LANCE_TEST_SCHEMA}"; + +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 +DROP SCHEMA ns."${LANCE_TEST_SCHEMA}"; + +statement ok +DETACH ns; From 2f4d84846a8fc36108c1d1b796a0dfed1a3737b4 Mon Sep 17 00:00:00 2001 From: arrowbowang Date: Thu, 13 Aug 2026 20:13:43 +0800 Subject: [PATCH 03/11] fix(catalog): route REST CTAS by bound schema Derive REST table operations from the bound Lance schema entry. Child-schema CTAS now targets its child namespace, while main retains the original namespace. Keep fully qualified table identifiers for namespace mutations and verify routing after reattachment. --- src/lance_storage.cpp | 50 +++++++++++++++--------- test/sql/namespace_rest_ctas_schema.test | 49 +++++++++++++++++++++++ 2 files changed, 81 insertions(+), 18 deletions(-) create mode 100644 test/sql/namespace_rest_ctas_schema.test diff --git a/src/lance_storage.cpp b/src/lance_storage.cpp index a1d64203..4b1780f2 100644 --- a/src/lance_storage.cpp +++ b/src/lance_storage.cpp @@ -795,7 +795,7 @@ class LanceSchemaEntry final : public DuckSchemaEntry { break; } if (StringUtil::CIEquals(t, leaf_id)) { - table_id_for_ops = leaf_id; + table_id_for_ops = prefixed_id.empty() ? leaf_id : prefixed_id; break; } } @@ -939,6 +939,9 @@ class LanceSchemaEntry final : public DuckSchemaEntry { } auto table_id_for_ops = exists ? existing_id : (prefixed_id.empty() ? leaf_id : prefixed_id); + if (!prefixed_id.empty()) { + table_id_for_ops = prefixed_id; + } if (create_info.on_conflict == OnCreateConflict::IGNORE_ON_CONFLICT && exists) { InvalidateTableDefaults(); @@ -1276,6 +1279,13 @@ class LanceDuckCatalog final : public DuckCatalog { "Lance ATTACH TYPE LANCE does not support TEMPORARY tables"); } if (rest_ns) { + auto *lance_schema = dynamic_cast(&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( @@ -1411,7 +1421,8 @@ class LanceDuckCatalog final : public DuckCatalog { break; } if (StringUtil::CIEquals(t, leaf_id)) { - state->table_id = leaf_id; + state->table_id = + prefixed_id.empty() ? leaf_id : prefixed_id; break; } } @@ -1593,31 +1604,33 @@ class LanceDuckCatalog final : public DuckCatalog { // Use LIST TABLES to implement conflict behavior in a side-effect-free // way. unordered_map 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 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() + auto delim = + schema_rest_ns->delimiter.empty() ? "$" : schema_rest_ns->delimiter; + auto prefix = schema_rest_ns->namespace_id.empty() ? string() - : (rest_ns->namespace_id + delim); + : (schema_rest_ns->namespace_id + delim); auto leaf_id = create_info.table; string prefixed_id; if (!prefix.empty() && !StringUtil::StartsWith(leaf_id, prefix)) { @@ -1653,11 +1666,12 @@ class LanceDuckCatalog final : public DuckCatalog { auto types = create_info.columns.GetColumnTypes(); string mode = CreateTableModeFromConflict(create_info.on_conflict); auto &create_as = planner.Make( - 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.estimated_cardinality); + 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, + 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; } diff --git a/test/sql/namespace_rest_ctas_schema.test b/test/sql/namespace_rest_ctas_schema.test new file mode 100644 index 00000000..6e6521b3 --- /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; From 5d99f7aa011f68cdd5fbbb4e2677178ac517ba5e Mon Sep 17 00:00:00 2001 From: arrowbowang Date: Thu, 13 Aug 2026 20:23:43 +0800 Subject: [PATCH 04/11] fix(catalog): refresh schemas after CTAS commit Invalidate only the target schema's existing default generator after a directory copy or REST writer finalization succeeds. This makes committed CTAS tables visible across connections without touching the dataset cache or invalidating during planning. --- src/lance_storage.cpp | 136 +++++++++++++----- .../namespace_ctas_catalog_visibility.test | 33 +++++ ...amespace_rest_ctas_catalog_visibility.test | 35 +++++ 3 files changed, 170 insertions(+), 34 deletions(-) create mode 100644 test/sql/namespace_ctas_catalog_visibility.test create mode 100644 test/sql/namespace_rest_ctas_catalog_visibility.test diff --git a/src/lance_storage.cpp b/src/lance_storage.cpp index 4b1780f2..8668cfa2 100644 --- a/src/lance_storage.cpp +++ b/src/lance_storage.cpp @@ -667,6 +667,12 @@ class LanceSchemaEntry final : public DuckSchemaEntry { 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); @@ -1097,18 +1103,76 @@ class LanceSchemaEntry final : public DuckSchemaEntry { } private: - void InvalidateTableDefaults() { - if (!table_default_generator) { - return; - } - table_default_generator->created_all_entries = false; - } - shared_ptr directory_ns; shared_ptr 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); + if (schema) { + auto *lance_schema = dynamic_cast(schema.get()); + if (lance_schema) { + lance_schema->InvalidateTableDefaults(); + } + } +} + +class PhysicalLanceCopyToFile final : public PhysicalCopyToFile { +public: + PhysicalLanceCopyToFile(PhysicalPlan &physical_plan, + vector types, CopyFunction function, + unique_ptr 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 types, CopyFunction function, + unique_ptr 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; @@ -1142,8 +1206,8 @@ class LanceDuckCatalog final : public DuckCatalog { 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); + throw IOException("Failed to create Lance schema '%s': %s", info.schema, + error); } return CreateRestSchemaEntry(transaction, info, std::move(child_ns), bearer_token, api_key); @@ -1151,8 +1215,7 @@ class LanceDuckCatalog final : public DuckCatalog { return DuckCatalog::CreateSchema(transaction, info); } - void LoadRestSchemas(ClientContext &context, - CatalogTransaction transaction, + void LoadRestSchemas(ClientContext &context, CatalogTransaction transaction, const vector &schema_names) { string bearer_token; string api_key; @@ -1292,9 +1355,10 @@ class LanceDuckCatalog final : public DuckCatalog { PhysicalPlan &physical_plan, vector 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 column_names_p, - vector 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 column_names_p, vector column_types_p, + idx_t estimated_cardinality) : PhysicalOperator(physical_plan, PhysicalOperatorType::EXTENSION, std::move(types_p), estimated_cardinality), endpoint(std::move(endpoint)), @@ -1303,6 +1367,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)), @@ -1421,8 +1487,7 @@ class LanceDuckCatalog final : public DuckCatalog { break; } if (StringUtil::CIEquals(t, leaf_id)) { - state->table_id = - prefixed_id.empty() ? leaf_id : prefixed_id; + state->table_id = prefixed_id.empty() ? leaf_id : prefixed_id; break; } } @@ -1542,7 +1607,6 @@ class LanceDuckCatalog final : public DuckCatalog { SinkFinalizeType Finalize(Pipeline &, Event &, ClientContext &context, OperatorSinkFinalizeInput &input) const override { - (void)context; auto &gstate = input.global_state.Cast(); { @@ -1556,6 +1620,8 @@ class LanceDuckCatalog final : public DuckCatalog { } } + InvalidateLanceSchema(context, catalog_name, schema_name); + return SinkFinalizeType::READY; } @@ -1594,6 +1660,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; @@ -1667,11 +1735,11 @@ class LanceDuckCatalog final : public DuckCatalog { string mode = CreateTableModeFromConflict(create_info.on_conflict); auto &create_as = planner.Make( op.types, schema_rest_ns->endpoint, schema_rest_ns->namespace_id, - schema_rest_ns->delimiter, - schema_rest_ns->bearer_token_override, + schema_rest_ns->delimiter, schema_rest_ns->bearer_token_override, schema_rest_ns->api_key_override, schema_rest_ns->headers_tsv, - create_info.table, mode, data_storage_version, std::move(names), - std::move(types), op.estimated_cardinality); + 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; } @@ -1739,10 +1807,11 @@ class LanceDuckCatalog final : public DuckCatalog { } if (execution_mode == CopyFunctionExecutionMode::BATCH_COPY_TO_FILE) { - auto © = planner.Make( + auto © = planner.Make( op.types, copy_function, std::move(bind_data), - op.estimated_cardinality); - auto &cast_copy = copy.Cast(); + op.estimated_cardinality, op.schema.catalog.GetName(), + op.schema.name); + auto &cast_copy = copy.Cast(); cast_copy.file_path = dataset_path; cast_copy.use_tmp_file = false; cast_copy.return_type = CopyFunctionReturnType::CHANGED_ROWS; @@ -1751,10 +1820,10 @@ class LanceDuckCatalog final : public DuckCatalog { return copy; } - auto © = planner.Make(op.types, copy_function, - std::move(bind_data), - op.estimated_cardinality); - auto &cast_copy = copy.Cast(); + auto © = planner.Make( + op.types, copy_function, std::move(bind_data), op.estimated_cardinality, + op.schema.catalog.GetName(), op.schema.name); + auto &cast_copy = copy.Cast(); cast_copy.file_path = dataset_path; cast_copy.use_tmp_file = false; cast_copy.filename_pattern = FilenamePattern(); @@ -1839,14 +1908,13 @@ class LanceDuckCatalog final : public DuckCatalog { bearer_token, api_key); } - optional_ptr CreateRestSchemaEntry( - CatalogTransaction transaction, CreateSchemaInfo &info, - shared_ptr schema_ns, - const string &bearer_token, const string &api_key) { + optional_ptr + CreateRestSchemaEntry(CatalogTransaction transaction, CreateSchemaInfo &info, + shared_ptr schema_ns, + const string &bearer_token, const string &api_key) { auto &schemas = GetSchemaCatalogSet(); LogicalDependencyList dependencies; - auto entry = - make_uniq(*this, info, nullptr, schema_ns); + auto entry = make_uniq(*this, info, nullptr, schema_ns); auto result = entry.get(); if (!schemas.CreateEntry(transaction, info.schema, std::move(entry), dependencies)) { diff --git a/test/sql/namespace_ctas_catalog_visibility.test b/test/sql/namespace_ctas_catalog_visibility.test new file mode 100644 index 00000000..7bca665d --- /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_rest_ctas_catalog_visibility.test b/test/sql/namespace_rest_ctas_catalog_visibility.test new file mode 100644 index 00000000..89c28196 --- /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; From 323925cd82f87a361fed800965f471f96f497523 Mon Sep 17 00:00:00 2001 From: arrowbowang Date: Thu, 13 Aug 2026 22:04:43 +0800 Subject: [PATCH 05/11] fix(catalog): preserve namespace identity after CTAS Refresh only the replaced table entry after a successful CTAS commit so existing connections rematerialize its schema without flushing dataset caches. Carry namespace identifiers across the FFI as length-delimited segments and remove leaf-name retries that could redirect qualified operations to another namespace. --- rust/ffi/namespace.rs | 153 +++++++---- rust/ffi/query_table.rs | 19 +- src/include/lance_common.hpp | 5 + src/lance_common.cpp | 77 +++++- src/lance_storage.cpp | 255 +++++++++--------- .../namespace_ctas_catalog_visibility.test | 27 ++ 6 files changed, 335 insertions(+), 201 deletions(-) diff --git a/rust/ffi/namespace.rs b/rust/ffi/namespace.rs index 36e35b94..b70d7928 100644 --- a/rust/ffi/namespace.rs +++ b/rust/ffi/namespace.rs @@ -87,12 +87,69 @@ fn storage_options_to_tsv(storage_options: std::collections::HashMap Vec { - if id.is_empty() { - Vec::new() - } else { - id.split(delimiter).map(ToString::to_string).collect() +const STRING_LIST_PREFIX: &str = "LID1;"; + +fn encode_string_list(values: &[String]) -> String { + let mut encoded = format!("{STRING_LIST_PREFIX}{};", values.len()); + for value in values { + encoded.push_str(&format!("{}:", value.len())); + encoded.push_str(value); } + encoded +} + +pub(crate) fn decode_id(id: &str, delimiter: &str) -> FfiResult> { + let Some(encoded) = id.strip_prefix(STRING_LIST_PREFIX) else { + return Ok(if id.is_empty() { + Vec::new() + } else { + id.split(delimiter).map(ToString::to_string).collect() + }); + }; + + let (count_text, mut encoded) = encoded.split_once(';').ok_or_else(|| { + FfiError::new( + ErrorCode::InvalidArgument, + "invalid Lance identifier list encoding", + ) + })?; + let count = count_text.parse::().map_err(|_| { + FfiError::new( + ErrorCode::InvalidArgument, + "invalid Lance identifier list count", + ) + })?; + let mut values = Vec::with_capacity(count); + for _ in 0..count { + let colon = encoded.find(':').ok_or_else(|| { + FfiError::new( + ErrorCode::InvalidArgument, + "invalid Lance identifier list encoding", + ) + })?; + let length = encoded[..colon].parse::().map_err(|_| { + FfiError::new( + ErrorCode::InvalidArgument, + "invalid Lance identifier length", + ) + })?; + encoded = &encoded[colon + 1..]; + if length > encoded.len() || !encoded.is_char_boundary(length) { + return Err(FfiError::new( + ErrorCode::InvalidArgument, + "invalid Lance identifier length", + )); + } + values.push(encoded[..length].to_string()); + encoded = &encoded[length..]; + } + if !encoded.is_empty() { + return Err(FfiError::new( + ErrorCode::InvalidArgument, + "trailing data in Lance identifier list", + )); + } + Ok(values) } fn namespace_operation_config( @@ -110,7 +167,7 @@ fn namespace_operation_config( 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 id = decode_id(namespace_id, &delimiter)?; let namespace = build_config( endpoint, bearer_token.as_deref(), @@ -167,7 +224,7 @@ pub unsafe extern "C" fn lance_namespace_list_namespaces( match result { Ok(namespaces) => { clear_last_error(); - to_c_string(namespaces.join("\n")).into_raw() as *const c_char + to_c_string(encode_string_list(&namespaces)).into_raw() as *const c_char } Err(err) => { set_last_error(err.code, err.message); @@ -284,6 +341,7 @@ 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_id = decode_id(namespace_id, &delimiter)?; let namespace = build_config( endpoint, bearer_token.as_deref(), @@ -298,11 +356,7 @@ fn list_tables_inner( let mut page_token: Option = 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_id.clone()); req.page_token = page_token.clone(); req.limit = Some(1000); let resp = namespace.list_tables(req).await.map_err(|err| { @@ -343,8 +397,7 @@ pub unsafe extern "C" fn lance_namespace_list_tables( ) { Ok(tables) => { clear_last_error(); - let joined = tables.join("\n"); - to_c_string(joined).into_raw() as *const c_char + to_c_string(encode_string_list(&tables)).into_raw() as *const c_char } Err(err) => { set_last_error(err.code, err.message); @@ -369,27 +422,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 = decode_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( @@ -484,19 +530,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 = decode_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 = 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); @@ -593,19 +636,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 = decode_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 = 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); @@ -667,24 +707,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 = decode_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| { @@ -772,21 +807,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 = decode_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 = table_id - .split(delimiter.as_str()) - .map(|s| s.to_string()) - .collect(); let (dataset, table_uri) = runtime::block_on(async move { record_namespace_describe(); @@ -940,3 +970,28 @@ pub unsafe extern "C" fn lance_json_arrow_schema_to_c( } } } + +#[cfg(test)] +mod tests { + use super::{decode_id, encode_string_list}; + + #[test] + fn identifier_list_round_trips_delimiters_and_newlines() { + let values = vec![ + "default".to_string(), + "a$b".to_string(), + "a\nb".to_string(), + "销售".to_string(), + ]; + let encoded = encode_string_list(&values); + assert_eq!(decode_id(&encoded, "$").unwrap(), values); + } + + #[test] + fn legacy_identifier_still_uses_configured_delimiter() { + assert_eq!( + decode_id("default$schema$table", "$").unwrap(), + vec!["default", "schema", "table"] + ); + } +} diff --git a/rust/ffi/query_table.rs b/rust/ffi/query_table.rs index 1cd55cc9..71ace1c6 100644 --- a/rust/ffi/query_table.rs +++ b/rust/ffi/query_table.rs @@ -18,6 +18,7 @@ use lance_namespace_impls::{DirectoryNamespaceBuilder, RestNamespaceBuilder}; use crate::error::{clear_last_error, set_last_error, ErrorCode}; use crate::runtime; +use super::namespace::decode_id; use super::types::StreamHandle; use super::util::{cstr_to_str, parse_optional_filter_ir, slice_from_ptr, FfiError, FfiResult}; @@ -230,15 +231,14 @@ unsafe fn parse_config( }) } -fn apply_base_request(config: &ParsedNamespaceQueryConfig, request: &mut QueryTableRequest) { +fn apply_base_request( + config: &ParsedNamespaceQueryConfig, + request: &mut QueryTableRequest, +) -> FfiResult<()> { request.id = Some(match &config.backend { NamespaceBackend::Rest { delimiter, .. } => { let delimiter = delimiter.as_deref().unwrap_or("$"); - config - .table_id - .split(delimiter) - .map(str::to_string) - .collect() + decode_id(&config.table_id, delimiter)? } NamespaceBackend::Directory { .. } => vec![config.table_id.clone()], }); @@ -251,6 +251,7 @@ fn apply_base_request(config: &ParsedNamespaceQueryConfig, request: &mut QueryTa if let Some(filter) = &config.filter { request.filter = Some(filter.clone()); } + Ok(()) } async fn execute_query_table( @@ -414,7 +415,7 @@ fn build_namespace_scan_request( })?; let mut request = QueryTableRequest::new(k, QueryTableRequestVector::new()); - apply_base_request(config, &mut request); + apply_base_request(config, &mut request)?; request.prefilter = None; if offset != 0 { request.offset = Some(offset); @@ -551,7 +552,7 @@ unsafe fn create_namespace_vector_search_stream_inner( let mut vector = QueryTableRequestVector::new(); vector.single_vector = Some(query_values.to_vec()); let mut request = QueryTableRequest::new(config.k, vector); - apply_base_request(&config, &mut request); + apply_base_request(&config, &mut request)?; request.vector_column = Some(vector_column.to_string()); if options.nprobes != 0 { request.nprobes = @@ -608,7 +609,7 @@ unsafe fn create_namespace_fts_search_stream_inner( let query = unsafe { cstr_to_str(options.query, "query")? }; let mut request = QueryTableRequest::new(config.k, QueryTableRequestVector::new()); - apply_base_request(&config, &mut request); + apply_base_request(&config, &mut request)?; let mut string_query = StringFtsQuery::new(query.to_string()); string_query.columns = Some(vec![text_column.to_string()]); diff --git a/src/include/lance_common.hpp b/src/include/lance_common.hpp index efb41f70..998dd372 100644 --- a/src/include/lance_common.hpp +++ b/src/include/lance_common.hpp @@ -46,6 +46,11 @@ void ResolveLanceNamespaceAuthOverrides( const unordered_map &options, string &out_bearer_token, string &out_api_key); +// Encode string lists crossing the C++/Rust FFI without relying on sentinel +// characters that may legally occur in namespace identifiers. +string LanceEncodeStringList(const vector &values); +vector LanceDecodeStringList(const string &encoded); + bool TryLanceNamespaceListTables(ClientContext &context, const string &endpoint, const string &namespace_id, const string &bearer_token, diff --git a/src/lance_common.cpp b/src/lance_common.cpp index dd1fa5d7..07760c59 100644 --- a/src/lance_common.cpp +++ b/src/lance_common.cpp @@ -293,6 +293,69 @@ void BuildStorageOptionPointerArrays(const vector &option_keys, } } +string LanceEncodeStringList(const vector &values) { + string result = "LID1;" + to_string(values.size()) + ";"; + for (const auto &value : values) { + result += to_string(value.size()) + ":" + value; + } + return result; +} + +vector LanceDecodeStringList(const string &encoded) { + constexpr const char *prefix = "LID1;"; + if (encoded.compare(0, strlen(prefix), prefix) != 0) { + vector values; + for (auto &value : StringUtil::Split(encoded, '\n')) { + if (!value.empty()) { + values.push_back(std::move(value)); + } + } + return values; + } + + idx_t offset = strlen(prefix); + auto read_size = [&](char terminator) { + if (offset >= encoded.size()) { + throw IOException("Invalid Lance identifier list encoding"); + } + idx_t value = 0; + bool has_digit = false; + while (offset < encoded.size() && encoded[offset] != terminator) { + auto ch = encoded[offset++]; + if (ch < '0' || ch > '9') { + throw IOException("Invalid Lance identifier list encoding"); + } + has_digit = true; + auto digit = NumericCast(ch - '0'); + if (value > (NumericLimits::Maximum() - digit) / 10) { + throw IOException("Lance identifier list length is too large"); + } + value = value * 10 + digit; + } + if (!has_digit || offset >= encoded.size()) { + throw IOException("Invalid Lance identifier list encoding"); + } + offset++; + return value; + }; + + auto count = read_size(';'); + vector values; + values.reserve(count); + for (idx_t value_idx = 0; value_idx < count; value_idx++) { + auto length = read_size(':'); + if (length > encoded.size() - offset) { + throw IOException("Invalid Lance identifier list encoding"); + } + values.push_back(encoded.substr(offset, length)); + offset += length; + } + if (offset != encoded.size()) { + throw IOException("Invalid Lance identifier list encoding"); + } + return values; +} + bool TryLanceNamespaceListTables( ClientContext &context, const string &endpoint, const string &namespace_id, const string &bearer_token, const string &api_key, const string &delimiter, @@ -318,13 +381,7 @@ bool TryLanceNamespaceListTables( } string joined = ptr; lance_free_string(ptr); - - vector parts = StringUtil::Split(joined, '\n'); - for (auto &p : parts) { - if (!p.empty()) { - out_tables.push_back(std::move(p)); - } - } + out_tables = LanceDecodeStringList(joined); return true; } @@ -348,11 +405,7 @@ bool TryLanceNamespaceListNamespaces( } string joined = ptr; lance_free_string(ptr); - for (auto &name : StringUtil::Split(joined, '\n')) { - if (!name.empty()) { - out_namespaces.push_back(std::move(name)); - } - } + out_namespaces = LanceDecodeStringList(joined); return true; } diff --git a/src/lance_storage.cpp b/src/lance_storage.cpp index 8668cfa2..2358146e 100644 --- a/src/lance_storage.cpp +++ b/src/lance_storage.cpp @@ -68,6 +68,39 @@ struct LanceRestNamespaceConfig { string headers_tsv; // Tab-separated key\tvalue pairs for custom headers }; +static string EffectiveNamespaceDelimiter(const string &delimiter) { + return delimiter.empty() ? "$" : delimiter; +} + +static vector DecodeRestIdentifier(const string &identifier, + const string &delimiter) { + if (identifier.compare(0, 5, "LID1;") == 0) { + return LanceDecodeStringList(identifier); + } + if (identifier.empty()) { + return {}; + } + return StringUtil::Split(identifier, EffectiveNamespaceDelimiter(delimiter)); +} + +static string EncodeRestIdentifier(const vector &segments) { + return LanceEncodeStringList(segments); +} + +static string AppendRestIdentifier(const string &identifier, + const string &delimiter, + const string &segment) { + auto segments = DecodeRestIdentifier(identifier, delimiter); + segments.push_back(segment); + return EncodeRestIdentifier(segments); +} + +static string DisplayRestIdentifier(const string &identifier, + const string &delimiter) { + return StringUtil::Join(DecodeRestIdentifier(identifier, delimiter), + EffectiveNamespaceDelimiter(delimiter)); +} + static string GetLanceNamespaceEndpoint(const AttachInfo &info) { for (auto &kv : info.options) { if (!StringUtil::CIEquals(kv.first, "endpoint") || kv.second.IsNull()) { @@ -249,13 +282,7 @@ ListRestNamespaceTables(const string &endpoint, const string &namespace_id, string joined = ptr; lance_free_string(ptr); - vector out; - for (auto &p : StringUtil::Split(joined, '\n')) { - if (!p.empty()) { - out.push_back(std::move(p)); - } - } - return out; + return LanceDecodeStringList(joined); } static bool @@ -449,15 +476,12 @@ class LanceRestNamespaceDefaultGenerator : public DefaultGenerator { resolved_api_key = api_key; } - // Build candidate table IDs (bare name + optional namespace-prefixed). - vector 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); - } - } + // Preserve identifier segment boundaries across the FFI. A lookup must + // never retry in a different namespace after a qualified request fails. + vector candidates = { + namespace_id.empty() + ? EncodeRestIdentifier({entry_name}) + : AppendRestIdentifier(namespace_id, delimiter, entry_name)}; // Fast path: describe_table with schema from REST API (skips S3 open). for (auto &table_id : candidates) { @@ -528,8 +552,8 @@ class LanceRestNamespaceDefaultGenerator : public DefaultGenerator { if (namespace_id.empty()) { return tables; } - auto delim = delimiter.empty() ? "$" : delimiter; - auto prefix = namespace_id + delim; + auto prefix = DisplayRestIdentifier(namespace_id, delimiter) + + EffectiveNamespaceDelimiter(delimiter); for (auto &t : tables) { if (StringUtil::StartsWith(t, prefix)) { t = t.substr(prefix.size()); @@ -776,14 +800,11 @@ 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 = + DisplayRestIdentifier(rest_ns->namespace_id, rest_ns->delimiter) + + EffectiveNamespaceDelimiter(rest_ns->delimiter) + leaf_id; vector discovered; string list_error; @@ -794,14 +815,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 = prefixed_id.empty() ? leaf_id : prefixed_id; break; } } @@ -911,14 +930,11 @@ 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 = + DisplayRestIdentifier(rest_ns->namespace_id, rest_ns->delimiter) + + EffectiveNamespaceDelimiter(rest_ns->delimiter) + leaf_id; vector discovered; string list_error; @@ -932,9 +948,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)) { @@ -943,11 +959,7 @@ class LanceSchemaEntry final : public DuckSchemaEntry { break; } } - auto table_id_for_ops = - exists ? existing_id : (prefixed_id.empty() ? leaf_id : prefixed_id); - if (!prefixed_id.empty()) { - table_id_for_ops = prefixed_id; - } + auto table_id_for_ops = qualified_id; if (create_info.on_conflict == OnCreateConflict::IGNORE_ON_CONFLICT && exists) { InvalidateTableDefaults(); @@ -975,27 +987,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( @@ -1108,17 +1102,37 @@ class LanceSchemaEntry final : public DuckSchemaEntry { DefaultGenerator *table_default_generator = nullptr; }; -static void InvalidateLanceSchema(ClientContext &context, - const string &catalog_name, - const string &schema_name) { +static void RefreshLanceSchemaTable(ClientContext &context, + const string &catalog_name, + const string &schema_name, + const string &table_name) { auto schema = Catalog::GetSchema(context, catalog_name, schema_name, OnEntryNotFound::RETURN_NULL); - if (schema) { - auto *lance_schema = dynamic_cast(schema.get()); - if (lance_schema) { - lance_schema->InvalidateTableDefaults(); - } + auto *lance_schema = + schema ? dynamic_cast(schema.get()) : nullptr; + if (!lance_schema) { + return; + } + lance_schema->InvalidateTableDefaults(); + + auto &set = lance_schema->GetCatalogSet(CatalogType::TABLE_ENTRY); + auto transaction = + CatalogTransaction::GetSystemTransaction(schema->catalog.GetDatabase()); + auto existing_entry = set.GetEntry(transaction, table_name); + if (!existing_entry) { + return; + } + if (existing_entry->type != CatalogType::TABLE_ENTRY && + existing_entry->type != CatalogType::VIEW_ENTRY) { + throw InternalException( + "Unexpected catalog entry type for Lance table '%s': %s", table_name, + CatalogTypeToString(existing_entry->type)); } + if (!set.DropEntry(transaction, existing_entry->name, false, true)) { + throw InternalException( + "Could not refresh catalog entry for Lance table '%s'", table_name); + } + set.CleanupEntry(*existing_entry); } class PhysicalLanceCopyToFile final : public PhysicalCopyToFile { @@ -1127,23 +1141,25 @@ class PhysicalLanceCopyToFile final : public PhysicalCopyToFile { vector types, CopyFunction function, unique_ptr bind_data, idx_t estimated_cardinality, string catalog_name, - string schema_name) + string schema_name, string table_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)) {} + schema_name(std::move(schema_name)), table_name(std::move(table_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); + RefreshLanceSchemaTable(context, catalog_name, schema_name, table_name); return result; } private: string catalog_name; string schema_name; + string table_name; }; class PhysicalLanceBatchCopyToFile final : public PhysicalBatchCopyToFile { @@ -1152,25 +1168,27 @@ class PhysicalLanceBatchCopyToFile final : public PhysicalBatchCopyToFile { vector types, CopyFunction function, unique_ptr bind_data, idx_t estimated_cardinality, string catalog_name, - string schema_name) + string schema_name, string table_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)) {} + schema_name(std::move(schema_name)), table_name(std::move(table_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); + RefreshLanceSchemaTable(context, catalog_name, schema_name, table_name); return result; } private: string catalog_name; string schema_name; + string table_name; }; class LanceDuckCatalog final : public DuckCatalog { @@ -1459,15 +1477,12 @@ 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 = + DisplayRestIdentifier(state->namespace_id, state->delimiter) + + EffectiveNamespaceDelimiter(state->delimiter) + leaf_id; vector discovered; string list_error; @@ -1480,14 +1495,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 = prefixed_id.empty() ? leaf_id : prefixed_id; break; } } @@ -1511,26 +1524,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( @@ -1620,7 +1616,8 @@ class LanceDuckCatalog final : public DuckCatalog { } } - InvalidateLanceSchema(context, catalog_name, schema_name); + RefreshLanceSchemaTable(context, catalog_name, schema_name, + table_name); return SinkFinalizeType::READY; } @@ -1694,23 +1691,18 @@ class LanceDuckCatalog final : public DuckCatalog { (list_error.empty() ? "unknown error" : list_error)); } - auto delim = - schema_rest_ns->delimiter.empty() ? "$" : schema_rest_ns->delimiter; - auto prefix = schema_rest_ns->namespace_id.empty() - ? string() - : (schema_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 = + DisplayRestIdentifier(schema_rest_ns->namespace_id, + schema_rest_ns->delimiter) + + EffectiveNamespaceDelimiter(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)) { @@ -1809,8 +1801,8 @@ class LanceDuckCatalog final : public DuckCatalog { if (execution_mode == CopyFunctionExecutionMode::BATCH_COPY_TO_FILE) { auto © = planner.Make( op.types, copy_function, std::move(bind_data), - op.estimated_cardinality, op.schema.catalog.GetName(), - op.schema.name); + op.estimated_cardinality, op.schema.catalog.GetName(), op.schema.name, + create_info.table); auto &cast_copy = copy.Cast(); cast_copy.file_path = dataset_path; cast_copy.use_tmp_file = false; @@ -1822,7 +1814,7 @@ class LanceDuckCatalog final : public DuckCatalog { auto © = planner.Make( op.types, copy_function, std::move(bind_data), op.estimated_cardinality, - op.schema.catalog.GetName(), op.schema.name); + op.schema.catalog.GetName(), op.schema.name, create_info.table); auto &cast_copy = copy.Cast(); cast_copy.file_path = dataset_path; cast_copy.use_tmp_file = false; @@ -1890,8 +1882,8 @@ class LanceDuckCatalog final : public DuckCatalog { shared_ptr MakeRestChildNamespace(const string &schema_name) const { auto child = make_shared_ptr(*rest_ns); - auto delimiter = rest_ns->delimiter.empty() ? "$" : rest_ns->delimiter; - child->namespace_id = rest_ns->namespace_id + delimiter + schema_name; + child->namespace_id = AppendRestIdentifier(rest_ns->namespace_id, + rest_ns->delimiter, schema_name); return child; } @@ -1999,11 +1991,12 @@ LanceStorageAttach(optional_ptr, 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 = EncodeRestIdentifier( + StringUtil::Split(attach_path, EffectiveNamespaceDelimiter(delimiter))); ResolveLanceNamespaceAuth(context, endpoint, info.options, bearer_token, api_key); ResolveLanceNamespaceAuthOverrides(info.options, bearer_token_override, diff --git a/test/sql/namespace_ctas_catalog_visibility.test b/test/sql/namespace_ctas_catalog_visibility.test index 7bca665d..dd522381 100644 --- a/test/sql/namespace_ctas_catalog_visibility.test +++ b/test/sql/namespace_ctas_catalog_visibility.test @@ -29,5 +29,32 @@ SELECT sum(id) FROM ns.main.ctas_new ---- 42 +statement ok con1 +CREATE TABLE ns.main.replace_schema AS SELECT 1::INTEGER AS a; + +query TT con1 +SELECT column_name, data_type FROM duckdb_columns() +WHERE database_name = 'ns' AND schema_name = 'main' + AND table_name = 'replace_schema' +ORDER BY column_index +---- +a INTEGER + +statement ok con2 +CREATE OR REPLACE TABLE ns.main.replace_schema AS SELECT 'x'::VARCHAR AS b; + +query TT con1 +SELECT column_name, data_type FROM duckdb_columns() +WHERE database_name = 'ns' AND schema_name = 'main' + AND table_name = 'replace_schema' +ORDER BY column_index +---- +b VARCHAR + +query T con1 +SELECT b FROM ns.main.replace_schema +---- +x + statement ok con1 DETACH ns; From dfa069ead9b9d884c8b4a18ce190ccc81bf52af2 Mon Sep 17 00:00:00 2001 From: arrowbowang Date: Thu, 13 Aug 2026 22:52:19 +0800 Subject: [PATCH 06/11] fix(catalog): coordinate namespace state across connections Use a database-shared generation map to invalidate only replaced dataset keys while preserving unrelated cross-query cache entries. Keep optional child-namespace discovery from blocking main-only backends, and reject remote schema DDL forms whose transaction or replacement semantics cannot be preserved. --- rust/ffi/namespace.rs | 23 +++-- src/lance_dataset_cache.cpp | 92 ++++++++++++++----- src/lance_storage.cpp | 34 +++++-- .../namespace_ctas_catalog_visibility.test | 5 + test/sql/namespace_rest_schema.test | 43 +++++++++ 5 files changed, 163 insertions(+), 34 deletions(-) diff --git a/rust/ffi/namespace.rs b/rust/ffi/namespace.rs index b70d7928..32d66309 100644 --- a/rust/ffi/namespace.rs +++ b/rust/ffi/namespace.rs @@ -179,6 +179,13 @@ fn namespace_operation_config( Ok((namespace, id)) } +fn is_unsupported_namespace_operation(error: &LanceError) -> bool { + matches!(error, LanceError::NotSupported { .. }) + // 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, @@ -205,12 +212,16 @@ pub unsafe extern "C" fn lance_namespace_list_namespaces( request.id = Some(id.clone()); request.page_token = page_token.clone(); request.limit = Some(1000); - let response = namespace.list_namespaces(request).await.map_err(|err| { - FfiError::new( - ErrorCode::NamespaceListNamespaces, - format!("namespace list_namespaces: {err}"), - ) - })?; + 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), diff --git a/src/lance_dataset_cache.cpp b/src/lance_dataset_cache.cpp index 316acf41..69875b11 100644 --- a/src/lance_dataset_cache.cpp +++ b/src/lance_dataset_cache.cpp @@ -7,6 +7,7 @@ #include "duckdb/common/types/hash.hpp" #include "duckdb/main/client_context_state.hpp" +#include "duckdb/storage/object_cache.hpp" #include @@ -14,29 +15,63 @@ namespace duckdb { static constexpr const char *LANCE_DATASET_CACHE_STATE_KEY = "lance_dataset_cache_state"; +static constexpr const char *LANCE_DATASET_CACHE_GENERATIONS_KEY = + "lance.dataset_cache_generations.v1"; + +class LanceDatasetCacheGenerations final : public ObjectCacheEntry { +public: + static string ObjectType() { return "lance_dataset_cache_generations"; } + string GetObjectType() override { return ObjectType(); } + optional_idx GetEstimatedCacheMemory() const override { + return optional_idx(); + } + + idx_t Get(const string &key) { + lock_guard guard(lock); + return generations[key]; + } + + void Bump(const string &key) { + lock_guard guard(lock); + generations[key]++; + } + +private: + mutex lock; + unordered_map generations; +}; + +struct LanceCachedDataset { + idx_t generation; + shared_ptr entry; +}; class LanceDatasetCacheState final : public ClientContextState { public: - shared_ptr Get(const string &key) { + shared_ptr Get(const string &key, idx_t generation) { lock_guard guard(lock); auto entry = entries.find(key); - if (entry == entries.end()) { + if (entry == entries.end() || entry->second.generation != generation) { + if (entry != entries.end()) { + entries.erase(entry); + } query_misses++; return nullptr; } query_hits++; - return entry->second; + return entry->second.entry; } shared_ptr - PutOrGetExisting(const string &key, + PutOrGetExisting(const string &key, idx_t generation, shared_ptr entry) { lock_guard guard(lock); auto existing = entries.find(key); - if (existing != entries.end()) { - return existing->second; + if (existing != entries.end() && + existing->second.generation == generation) { + return existing->second.entry; } - entries[key] = entry; + entries[key] = {generation, entry}; return entry; } @@ -59,7 +94,7 @@ class LanceDatasetCacheState final : public ClientContextState { private: mutex lock; - unordered_map> entries; + unordered_map entries; idx_t query_hits = 0; idx_t query_misses = 0; }; @@ -81,6 +116,13 @@ GetOrCreateLanceDatasetCacheState(ClientContext &context) { LANCE_DATASET_CACHE_STATE_KEY); } +static shared_ptr +GetOrCreateLanceDatasetCacheGenerations(ClientContext &context) { + return ObjectCache::GetObjectCache(context) + .GetOrCreate( + LANCE_DATASET_CACHE_GENERATIONS_KEY); +} + static void AppendCacheKeyPart(string &key, const string &value) { key += to_string(value.size()); key += ':'; @@ -232,22 +274,29 @@ static shared_ptr GetOrOpenDatasetCacheEntry( const std::function()> &open_dataset, bool *out_cache_hit) { auto state = GetOrCreateLanceDatasetCacheState(context); - auto entry = state->Get(cache_key); - if (entry) { - if (out_cache_hit) { - *out_cache_hit = true; + auto generations = GetOrCreateLanceDatasetCacheGenerations(context); + while (true) { + auto generation = generations->Get(cache_key); + auto entry = state->Get(cache_key, generation); + if (entry) { + if (out_cache_hit) { + *out_cache_hit = true; + } + return entry; } - return entry; - } - auto opened = open_dataset(); - if (!opened) { - return nullptr; - } - if (out_cache_hit) { - *out_cache_hit = false; + auto opened = open_dataset(); + if (!opened) { + return nullptr; + } + if (generations->Get(cache_key) != generation) { + continue; + } + if (out_cache_hit) { + *out_cache_hit = false; + } + return state->PutOrGetExisting(cache_key, generation, opened); } - return state->PutOrGetExisting(cache_key, opened); } shared_ptr @@ -383,6 +432,7 @@ string LanceBuildDatasetCacheKeyForTable(ClientContext &context, void LanceInvalidateDatasetCache(ClientContext &context, const string &cache_key) { + GetOrCreateLanceDatasetCacheGenerations(context)->Bump(cache_key); auto state = context.registered_state->Get( LANCE_DATASET_CACHE_STATE_KEY); if (state) { diff --git a/src/lance_storage.cpp b/src/lance_storage.cpp index 2358146e..a1497207 100644 --- a/src/lance_storage.cpp +++ b/src/lance_storage.cpp @@ -1128,6 +1128,10 @@ static void RefreshLanceSchemaTable(ClientContext &context, "Unexpected catalog entry type for Lance table '%s': %s", table_name, CatalogTypeToString(existing_entry->type)); } + auto *lance_table = dynamic_cast(existing_entry.get()); + if (lance_table) { + LanceInvalidateDatasetCacheForTable(context, *lance_table); + } if (!set.DropEntry(transaction, existing_entry->name, false, true)) { throw InternalException( "Could not refresh catalog entry for Lance table '%s'", table_name); @@ -1215,10 +1219,23 @@ class LanceDuckCatalog final : public DuckCatalog { 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) { + return nullptr; + } string error; if (!TryLanceNamespaceCreateNamespace( context, rest_ns->endpoint, child_ns->namespace_id, bearer_token, @@ -1227,8 +1244,7 @@ class LanceDuckCatalog final : public DuckCatalog { throw IOException("Failed to create Lance schema '%s': %s", info.schema, error); } - return CreateRestSchemaEntry(transaction, info, std::move(child_ns), - bearer_token, api_key); + return result; } return DuckCatalog::CreateSchema(transaction, info); } @@ -1262,6 +1278,10 @@ class LanceDuckCatalog final : public DuckCatalog { 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) { @@ -1270,6 +1290,11 @@ class LanceDuckCatalog final : public DuckCatalog { } 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); @@ -1282,11 +1307,6 @@ class LanceDuckCatalog final : public DuckCatalog { throw IOException("Failed to drop Lance schema '%s': %s", info.name, error); } - if (!GetSchemaCatalogSet().DropEntry(transaction, existing->name, - info.cascade)) { - throw InternalException("Failed to drop Lance schema entry: " + - existing->name); - } } ErrorData SupportsCreateTable(BoundCreateTableInfo &info) override { diff --git a/test/sql/namespace_ctas_catalog_visibility.test b/test/sql/namespace_ctas_catalog_visibility.test index dd522381..e377d764 100644 --- a/test/sql/namespace_ctas_catalog_visibility.test +++ b/test/sql/namespace_ctas_catalog_visibility.test @@ -40,6 +40,11 @@ ORDER BY column_index ---- a INTEGER +query I con1 +SELECT a FROM ns.main.replace_schema +---- +1 + statement ok con2 CREATE OR REPLACE TABLE ns.main.replace_schema AS SELECT 'x'::VARCHAR AS b; diff --git a/test/sql/namespace_rest_schema.test b/test/sql/namespace_rest_schema.test index 647eab58..5fd05527 100644 --- a/test/sql/namespace_rest_schema.test +++ b/test/sql/namespace_rest_schema.test @@ -15,9 +15,25 @@ 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 ok CREATE SCHEMA ns."${LANCE_TEST_SCHEMA}"; +statement error +CREATE OR REPLACE SCHEMA ns."${LANCE_TEST_SCHEMA}"; +---- +Not implemented Error: CREATE OR REPLACE SCHEMA is not supported for Lance namespaces + query T SELECT schema_name FROM duckdb_schemas() @@ -38,6 +54,33 @@ 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}"; From 911de5c6752dac37883bae031a1910031bc8f209 Mon Sep 17 00:00:00 2001 From: arrowbowang Date: Thu, 13 Aug 2026 23:32:44 +0800 Subject: [PATCH 07/11] fix(catalog): preserve transactional namespace DDL Recognize structured unsupported namespace responses, preserve DuckDB duplicate-schema conflict semantics, and reject REST table DDL before remote side effects inside explicit transactions. --- rust/ffi/namespace.rs | 8 +++++++- src/lance_storage.cpp | 16 ++++++++++++++++ test/sql/namespace_rest_schema.test | 27 +++++++++++++++++++++++++++ 3 files changed, 50 insertions(+), 1 deletion(-) diff --git a/rust/ffi/namespace.rs b/rust/ffi/namespace.rs index 32d66309..8b40a9ab 100644 --- a/rust/ffi/namespace.rs +++ b/rust/ffi/namespace.rs @@ -11,7 +11,7 @@ use lance_namespace::models::{ 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}; @@ -181,6 +181,12 @@ fn namespace_operation_config( fn is_unsupported_namespace_operation(error: &LanceError) -> bool { matches!(error, LanceError::NotSupported { .. }) + || match error { + LanceError::Namespace { source, .. } => source + .downcast_ref::() + .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") diff --git a/src/lance_storage.cpp b/src/lance_storage.cpp index a1497207..5ac52858 100644 --- a/src/lance_storage.cpp +++ b/src/lance_storage.cpp @@ -786,6 +786,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 overrides; if (!rest_ns->bearer_token_override.empty()) { overrides["bearer_token"] = Value(rest_ns->bearer_token_override); @@ -916,6 +920,10 @@ class LanceSchemaEntry final : public DuckSchemaEntry { vector option_values; if (rest_ns) { + if (!context.transaction.IsAutoCommit()) { + throw NotImplementedException( + "Lance table DDL does not support explicit transactions"); + } unordered_map overrides; if (!rest_ns->bearer_token_override.empty()) { overrides["bearer_token"] = Value(rest_ns->bearer_token_override); @@ -1234,6 +1242,10 @@ class LanceDuckCatalog final : public DuckCatalog { 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; @@ -1380,6 +1392,10 @@ 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(&op.schema); if (!lance_schema || !lance_schema->GetRestNamespace()) { throw InternalException( diff --git a/test/sql/namespace_rest_schema.test b/test/sql/namespace_rest_schema.test index 5fd05527..f3403ed3 100644 --- a/test/sql/namespace_rest_schema.test +++ b/test/sql/namespace_rest_schema.test @@ -29,11 +29,38 @@ ROLLBACK; statement ok CREATE SCHEMA ns."${LANCE_TEST_SCHEMA}"; +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() From 0ac6d0df967942d2424f9524ffd6aa9834036102 Mon Sep 17 00:00:00 2001 From: arrowbowang Date: Fri, 14 Aug 2026 10:39:08 +0800 Subject: [PATCH 08/11] refactor(cache): defer cross-connection freshness Keep PR 243 focused on namespace routing and catalog discovery. Restore connection-local dataset caching and move the database-wide generation boundary plus replacement freshness coverage to the dedicated cache-coherency work. --- src/lance_dataset_cache.cpp | 92 +++++-------------- src/lance_storage.cpp | 4 - .../namespace_ctas_catalog_visibility.test | 32 ------- 3 files changed, 21 insertions(+), 107 deletions(-) diff --git a/src/lance_dataset_cache.cpp b/src/lance_dataset_cache.cpp index 69875b11..316acf41 100644 --- a/src/lance_dataset_cache.cpp +++ b/src/lance_dataset_cache.cpp @@ -7,7 +7,6 @@ #include "duckdb/common/types/hash.hpp" #include "duckdb/main/client_context_state.hpp" -#include "duckdb/storage/object_cache.hpp" #include @@ -15,63 +14,29 @@ namespace duckdb { static constexpr const char *LANCE_DATASET_CACHE_STATE_KEY = "lance_dataset_cache_state"; -static constexpr const char *LANCE_DATASET_CACHE_GENERATIONS_KEY = - "lance.dataset_cache_generations.v1"; - -class LanceDatasetCacheGenerations final : public ObjectCacheEntry { -public: - static string ObjectType() { return "lance_dataset_cache_generations"; } - string GetObjectType() override { return ObjectType(); } - optional_idx GetEstimatedCacheMemory() const override { - return optional_idx(); - } - - idx_t Get(const string &key) { - lock_guard guard(lock); - return generations[key]; - } - - void Bump(const string &key) { - lock_guard guard(lock); - generations[key]++; - } - -private: - mutex lock; - unordered_map generations; -}; - -struct LanceCachedDataset { - idx_t generation; - shared_ptr entry; -}; class LanceDatasetCacheState final : public ClientContextState { public: - shared_ptr Get(const string &key, idx_t generation) { + shared_ptr Get(const string &key) { lock_guard guard(lock); auto entry = entries.find(key); - if (entry == entries.end() || entry->second.generation != generation) { - if (entry != entries.end()) { - entries.erase(entry); - } + if (entry == entries.end()) { query_misses++; return nullptr; } query_hits++; - return entry->second.entry; + return entry->second; } shared_ptr - PutOrGetExisting(const string &key, idx_t generation, + PutOrGetExisting(const string &key, shared_ptr entry) { lock_guard guard(lock); auto existing = entries.find(key); - if (existing != entries.end() && - existing->second.generation == generation) { - return existing->second.entry; + if (existing != entries.end()) { + return existing->second; } - entries[key] = {generation, entry}; + entries[key] = entry; return entry; } @@ -94,7 +59,7 @@ class LanceDatasetCacheState final : public ClientContextState { private: mutex lock; - unordered_map entries; + unordered_map> entries; idx_t query_hits = 0; idx_t query_misses = 0; }; @@ -116,13 +81,6 @@ GetOrCreateLanceDatasetCacheState(ClientContext &context) { LANCE_DATASET_CACHE_STATE_KEY); } -static shared_ptr -GetOrCreateLanceDatasetCacheGenerations(ClientContext &context) { - return ObjectCache::GetObjectCache(context) - .GetOrCreate( - LANCE_DATASET_CACHE_GENERATIONS_KEY); -} - static void AppendCacheKeyPart(string &key, const string &value) { key += to_string(value.size()); key += ':'; @@ -274,29 +232,22 @@ static shared_ptr GetOrOpenDatasetCacheEntry( const std::function()> &open_dataset, bool *out_cache_hit) { auto state = GetOrCreateLanceDatasetCacheState(context); - auto generations = GetOrCreateLanceDatasetCacheGenerations(context); - while (true) { - auto generation = generations->Get(cache_key); - auto entry = state->Get(cache_key, generation); - if (entry) { - if (out_cache_hit) { - *out_cache_hit = true; - } - return entry; - } - - auto opened = open_dataset(); - if (!opened) { - return nullptr; - } - if (generations->Get(cache_key) != generation) { - continue; - } + auto entry = state->Get(cache_key); + if (entry) { if (out_cache_hit) { - *out_cache_hit = false; + *out_cache_hit = true; } - return state->PutOrGetExisting(cache_key, generation, opened); + return entry; + } + + auto opened = open_dataset(); + if (!opened) { + return nullptr; + } + if (out_cache_hit) { + *out_cache_hit = false; } + return state->PutOrGetExisting(cache_key, opened); } shared_ptr @@ -432,7 +383,6 @@ string LanceBuildDatasetCacheKeyForTable(ClientContext &context, void LanceInvalidateDatasetCache(ClientContext &context, const string &cache_key) { - GetOrCreateLanceDatasetCacheGenerations(context)->Bump(cache_key); auto state = context.registered_state->Get( LANCE_DATASET_CACHE_STATE_KEY); if (state) { diff --git a/src/lance_storage.cpp b/src/lance_storage.cpp index 5ac52858..826b87f4 100644 --- a/src/lance_storage.cpp +++ b/src/lance_storage.cpp @@ -1136,10 +1136,6 @@ static void RefreshLanceSchemaTable(ClientContext &context, "Unexpected catalog entry type for Lance table '%s': %s", table_name, CatalogTypeToString(existing_entry->type)); } - auto *lance_table = dynamic_cast(existing_entry.get()); - if (lance_table) { - LanceInvalidateDatasetCacheForTable(context, *lance_table); - } if (!set.DropEntry(transaction, existing_entry->name, false, true)) { throw InternalException( "Could not refresh catalog entry for Lance table '%s'", table_name); diff --git a/test/sql/namespace_ctas_catalog_visibility.test b/test/sql/namespace_ctas_catalog_visibility.test index e377d764..7bca665d 100644 --- a/test/sql/namespace_ctas_catalog_visibility.test +++ b/test/sql/namespace_ctas_catalog_visibility.test @@ -29,37 +29,5 @@ SELECT sum(id) FROM ns.main.ctas_new ---- 42 -statement ok con1 -CREATE TABLE ns.main.replace_schema AS SELECT 1::INTEGER AS a; - -query TT con1 -SELECT column_name, data_type FROM duckdb_columns() -WHERE database_name = 'ns' AND schema_name = 'main' - AND table_name = 'replace_schema' -ORDER BY column_index ----- -a INTEGER - -query I con1 -SELECT a FROM ns.main.replace_schema ----- -1 - -statement ok con2 -CREATE OR REPLACE TABLE ns.main.replace_schema AS SELECT 'x'::VARCHAR AS b; - -query TT con1 -SELECT column_name, data_type FROM duckdb_columns() -WHERE database_name = 'ns' AND schema_name = 'main' - AND table_name = 'replace_schema' -ORDER BY column_index ----- -b VARCHAR - -query T con1 -SELECT b FROM ns.main.replace_schema ----- -x - statement ok con1 DETACH ns; From 2af75bb584a5a828e6cf953592f26898095fcc3a Mon Sep 17 00:00:00 2001 From: arrowbowang Date: Fri, 14 Aug 2026 11:09:34 +0800 Subject: [PATCH 09/11] refactor(namespace): reuse REST delimiter encoding Remove the private LID1 FFI format because Lance REST still serializes identifiers with its configured delimiter. Preserve qualified child-namespace routing and fail-closed lookups while retaining the pre-existing REST identifier contract. --- rust/ffi/namespace.rs | 110 +++++------------------------------ rust/ffi/query_table.rs | 19 +++--- src/include/lance_common.hpp | 5 -- src/lance_common.cpp | 75 ++++-------------------- src/lance_storage.cpp | 68 +++++++--------------- 5 files changed, 54 insertions(+), 223 deletions(-) diff --git a/rust/ffi/namespace.rs b/rust/ffi/namespace.rs index 8b40a9ab..de80fc18 100644 --- a/rust/ffi/namespace.rs +++ b/rust/ffi/namespace.rs @@ -87,69 +87,12 @@ fn storage_options_to_tsv(storage_options: std::collections::HashMap String { - let mut encoded = format!("{STRING_LIST_PREFIX}{};", values.len()); - for value in values { - encoded.push_str(&format!("{}:", value.len())); - encoded.push_str(value); - } - encoded -} - -pub(crate) fn decode_id(id: &str, delimiter: &str) -> FfiResult> { - let Some(encoded) = id.strip_prefix(STRING_LIST_PREFIX) else { - return Ok(if id.is_empty() { - Vec::new() - } else { - id.split(delimiter).map(ToString::to_string).collect() - }); - }; - - let (count_text, mut encoded) = encoded.split_once(';').ok_or_else(|| { - FfiError::new( - ErrorCode::InvalidArgument, - "invalid Lance identifier list encoding", - ) - })?; - let count = count_text.parse::().map_err(|_| { - FfiError::new( - ErrorCode::InvalidArgument, - "invalid Lance identifier list count", - ) - })?; - let mut values = Vec::with_capacity(count); - for _ in 0..count { - let colon = encoded.find(':').ok_or_else(|| { - FfiError::new( - ErrorCode::InvalidArgument, - "invalid Lance identifier list encoding", - ) - })?; - let length = encoded[..colon].parse::().map_err(|_| { - FfiError::new( - ErrorCode::InvalidArgument, - "invalid Lance identifier length", - ) - })?; - encoded = &encoded[colon + 1..]; - if length > encoded.len() || !encoded.is_char_boundary(length) { - return Err(FfiError::new( - ErrorCode::InvalidArgument, - "invalid Lance identifier length", - )); - } - values.push(encoded[..length].to_string()); - encoded = &encoded[length..]; +fn split_id(id: &str, delimiter: &str) -> Vec { + if id.is_empty() { + Vec::new() + } else { + id.split(delimiter).map(ToString::to_string).collect() } - if !encoded.is_empty() { - return Err(FfiError::new( - ErrorCode::InvalidArgument, - "trailing data in Lance identifier list", - )); - } - Ok(values) } fn namespace_operation_config( @@ -167,7 +110,7 @@ fn namespace_operation_config( 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 = decode_id(namespace_id, &delimiter)?; + let id = split_id(namespace_id, &delimiter); let namespace = build_config( endpoint, bearer_token.as_deref(), @@ -241,7 +184,7 @@ pub unsafe extern "C" fn lance_namespace_list_namespaces( match result { Ok(namespaces) => { clear_last_error(); - to_c_string(encode_string_list(&namespaces)).into_raw() as *const c_char + to_c_string(namespaces.join("\n")).into_raw() as *const c_char } Err(err) => { set_last_error(err.code, err.message); @@ -358,7 +301,7 @@ 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_id = decode_id(namespace_id, &delimiter)?; + let namespace_id = split_id(namespace_id, &delimiter); let namespace = build_config( endpoint, bearer_token.as_deref(), @@ -414,7 +357,7 @@ pub unsafe extern "C" fn lance_namespace_list_tables( ) { Ok(tables) => { clear_last_error(); - to_c_string(encode_string_list(&tables)).into_raw() as *const c_char + to_c_string(tables.join("\n")).into_raw() as *const c_char } Err(err) => { set_last_error(err.code, err.message); @@ -439,7 +382,7 @@ 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 = decode_id(table_id, &delimiter)?; + let table_id_segments = split_id(table_id, &delimiter); let namespace = build_config( endpoint, bearer_token.as_deref(), @@ -547,7 +490,7 @@ 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 = decode_id(table_id, &delimiter)?; + let table_id_segments = split_id(table_id, &delimiter); let namespace = build_config( endpoint, bearer_token.as_deref(), @@ -653,7 +596,7 @@ 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 = decode_id(table_id, &delimiter)?; + let table_id_segments = split_id(table_id, &delimiter); let namespace = build_config( endpoint, bearer_token.as_deref(), @@ -724,7 +667,7 @@ 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 = decode_id(table_id, &delimiter)?; + let table_id_segments = split_id(table_id, &delimiter); let namespace = build_config( endpoint, bearer_token.as_deref(), @@ -824,7 +767,7 @@ 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 = decode_id(table_id, &delimiter)?; + let table_id_segments = split_id(table_id, &delimiter); let namespace = build_config( endpoint, bearer_token.as_deref(), @@ -987,28 +930,3 @@ pub unsafe extern "C" fn lance_json_arrow_schema_to_c( } } } - -#[cfg(test)] -mod tests { - use super::{decode_id, encode_string_list}; - - #[test] - fn identifier_list_round_trips_delimiters_and_newlines() { - let values = vec![ - "default".to_string(), - "a$b".to_string(), - "a\nb".to_string(), - "销售".to_string(), - ]; - let encoded = encode_string_list(&values); - assert_eq!(decode_id(&encoded, "$").unwrap(), values); - } - - #[test] - fn legacy_identifier_still_uses_configured_delimiter() { - assert_eq!( - decode_id("default$schema$table", "$").unwrap(), - vec!["default", "schema", "table"] - ); - } -} diff --git a/rust/ffi/query_table.rs b/rust/ffi/query_table.rs index 71ace1c6..1cd55cc9 100644 --- a/rust/ffi/query_table.rs +++ b/rust/ffi/query_table.rs @@ -18,7 +18,6 @@ use lance_namespace_impls::{DirectoryNamespaceBuilder, RestNamespaceBuilder}; use crate::error::{clear_last_error, set_last_error, ErrorCode}; use crate::runtime; -use super::namespace::decode_id; use super::types::StreamHandle; use super::util::{cstr_to_str, parse_optional_filter_ir, slice_from_ptr, FfiError, FfiResult}; @@ -231,14 +230,15 @@ unsafe fn parse_config( }) } -fn apply_base_request( - config: &ParsedNamespaceQueryConfig, - request: &mut QueryTableRequest, -) -> FfiResult<()> { +fn apply_base_request(config: &ParsedNamespaceQueryConfig, request: &mut QueryTableRequest) { request.id = Some(match &config.backend { NamespaceBackend::Rest { delimiter, .. } => { let delimiter = delimiter.as_deref().unwrap_or("$"); - decode_id(&config.table_id, delimiter)? + config + .table_id + .split(delimiter) + .map(str::to_string) + .collect() } NamespaceBackend::Directory { .. } => vec![config.table_id.clone()], }); @@ -251,7 +251,6 @@ fn apply_base_request( if let Some(filter) = &config.filter { request.filter = Some(filter.clone()); } - Ok(()) } async fn execute_query_table( @@ -415,7 +414,7 @@ fn build_namespace_scan_request( })?; let mut request = QueryTableRequest::new(k, QueryTableRequestVector::new()); - apply_base_request(config, &mut request)?; + apply_base_request(config, &mut request); request.prefilter = None; if offset != 0 { request.offset = Some(offset); @@ -552,7 +551,7 @@ unsafe fn create_namespace_vector_search_stream_inner( let mut vector = QueryTableRequestVector::new(); vector.single_vector = Some(query_values.to_vec()); let mut request = QueryTableRequest::new(config.k, vector); - apply_base_request(&config, &mut request)?; + apply_base_request(&config, &mut request); request.vector_column = Some(vector_column.to_string()); if options.nprobes != 0 { request.nprobes = @@ -609,7 +608,7 @@ unsafe fn create_namespace_fts_search_stream_inner( let query = unsafe { cstr_to_str(options.query, "query")? }; let mut request = QueryTableRequest::new(config.k, QueryTableRequestVector::new()); - apply_base_request(&config, &mut request)?; + apply_base_request(&config, &mut request); let mut string_query = StringFtsQuery::new(query.to_string()); string_query.columns = Some(vec![text_column.to_string()]); diff --git a/src/include/lance_common.hpp b/src/include/lance_common.hpp index 998dd372..efb41f70 100644 --- a/src/include/lance_common.hpp +++ b/src/include/lance_common.hpp @@ -46,11 +46,6 @@ void ResolveLanceNamespaceAuthOverrides( const unordered_map &options, string &out_bearer_token, string &out_api_key); -// Encode string lists crossing the C++/Rust FFI without relying on sentinel -// characters that may legally occur in namespace identifiers. -string LanceEncodeStringList(const vector &values); -vector LanceDecodeStringList(const string &encoded); - bool TryLanceNamespaceListTables(ClientContext &context, const string &endpoint, const string &namespace_id, const string &bearer_token, diff --git a/src/lance_common.cpp b/src/lance_common.cpp index 07760c59..3dca89e9 100644 --- a/src/lance_common.cpp +++ b/src/lance_common.cpp @@ -293,69 +293,6 @@ void BuildStorageOptionPointerArrays(const vector &option_keys, } } -string LanceEncodeStringList(const vector &values) { - string result = "LID1;" + to_string(values.size()) + ";"; - for (const auto &value : values) { - result += to_string(value.size()) + ":" + value; - } - return result; -} - -vector LanceDecodeStringList(const string &encoded) { - constexpr const char *prefix = "LID1;"; - if (encoded.compare(0, strlen(prefix), prefix) != 0) { - vector values; - for (auto &value : StringUtil::Split(encoded, '\n')) { - if (!value.empty()) { - values.push_back(std::move(value)); - } - } - return values; - } - - idx_t offset = strlen(prefix); - auto read_size = [&](char terminator) { - if (offset >= encoded.size()) { - throw IOException("Invalid Lance identifier list encoding"); - } - idx_t value = 0; - bool has_digit = false; - while (offset < encoded.size() && encoded[offset] != terminator) { - auto ch = encoded[offset++]; - if (ch < '0' || ch > '9') { - throw IOException("Invalid Lance identifier list encoding"); - } - has_digit = true; - auto digit = NumericCast(ch - '0'); - if (value > (NumericLimits::Maximum() - digit) / 10) { - throw IOException("Lance identifier list length is too large"); - } - value = value * 10 + digit; - } - if (!has_digit || offset >= encoded.size()) { - throw IOException("Invalid Lance identifier list encoding"); - } - offset++; - return value; - }; - - auto count = read_size(';'); - vector values; - values.reserve(count); - for (idx_t value_idx = 0; value_idx < count; value_idx++) { - auto length = read_size(':'); - if (length > encoded.size() - offset) { - throw IOException("Invalid Lance identifier list encoding"); - } - values.push_back(encoded.substr(offset, length)); - offset += length; - } - if (offset != encoded.size()) { - throw IOException("Invalid Lance identifier list encoding"); - } - return values; -} - bool TryLanceNamespaceListTables( ClientContext &context, const string &endpoint, const string &namespace_id, const string &bearer_token, const string &api_key, const string &delimiter, @@ -381,7 +318,11 @@ bool TryLanceNamespaceListTables( } string joined = ptr; lance_free_string(ptr); - out_tables = LanceDecodeStringList(joined); + for (auto &table : StringUtil::Split(joined, '\n')) { + if (!table.empty()) { + out_tables.push_back(std::move(table)); + } + } return true; } @@ -405,7 +346,11 @@ bool TryLanceNamespaceListNamespaces( } string joined = ptr; lance_free_string(ptr); - out_namespaces = LanceDecodeStringList(joined); + for (auto &namespace_name : StringUtil::Split(joined, '\n')) { + if (!namespace_name.empty()) { + out_namespaces.push_back(std::move(namespace_name)); + } + } return true; } diff --git a/src/lance_storage.cpp b/src/lance_storage.cpp index 826b87f4..8ed691ca 100644 --- a/src/lance_storage.cpp +++ b/src/lance_storage.cpp @@ -72,33 +72,13 @@ static string EffectiveNamespaceDelimiter(const string &delimiter) { return delimiter.empty() ? "$" : delimiter; } -static vector DecodeRestIdentifier(const string &identifier, - const string &delimiter) { - if (identifier.compare(0, 5, "LID1;") == 0) { - return LanceDecodeStringList(identifier); - } - if (identifier.empty()) { - return {}; - } - return StringUtil::Split(identifier, EffectiveNamespaceDelimiter(delimiter)); -} - -static string EncodeRestIdentifier(const vector &segments) { - return LanceEncodeStringList(segments); -} - static string AppendRestIdentifier(const string &identifier, const string &delimiter, const string &segment) { - auto segments = DecodeRestIdentifier(identifier, delimiter); - segments.push_back(segment); - return EncodeRestIdentifier(segments); -} - -static string DisplayRestIdentifier(const string &identifier, - const string &delimiter) { - return StringUtil::Join(DecodeRestIdentifier(identifier, delimiter), - EffectiveNamespaceDelimiter(delimiter)); + if (identifier.empty()) { + return segment; + } + return identifier + EffectiveNamespaceDelimiter(delimiter) + segment; } static string GetLanceNamespaceEndpoint(const AttachInfo &info) { @@ -282,7 +262,13 @@ ListRestNamespaceTables(const string &endpoint, const string &namespace_id, string joined = ptr; lance_free_string(ptr); - return LanceDecodeStringList(joined); + vector tables; + for (auto &table : StringUtil::Split(joined, '\n')) { + if (!table.empty()) { + tables.push_back(std::move(table)); + } + } + return tables; } static bool @@ -476,12 +462,10 @@ class LanceRestNamespaceDefaultGenerator : public DefaultGenerator { resolved_api_key = api_key; } - // Preserve identifier segment boundaries across the FFI. A lookup must - // never retry in a different namespace after a qualified request fails. + // A qualified lookup must never retry in a different namespace after it + // fails. vector candidates = { - namespace_id.empty() - ? EncodeRestIdentifier({entry_name}) - : AppendRestIdentifier(namespace_id, delimiter, entry_name)}; + AppendRestIdentifier(namespace_id, delimiter, entry_name)}; // Fast path: describe_table with schema from REST API (skips S3 open). for (auto &table_id : candidates) { @@ -552,8 +536,7 @@ class LanceRestNamespaceDefaultGenerator : public DefaultGenerator { if (namespace_id.empty()) { return tables; } - auto prefix = DisplayRestIdentifier(namespace_id, delimiter) + - EffectiveNamespaceDelimiter(delimiter); + auto prefix = namespace_id + EffectiveNamespaceDelimiter(delimiter); for (auto &t : tables) { if (StringUtil::StartsWith(t, prefix)) { t = t.substr(prefix.size()); @@ -806,9 +789,7 @@ class LanceSchemaEntry final : public DuckSchemaEntry { auto leaf_id = info.name; auto qualified_id = AppendRestIdentifier(rest_ns->namespace_id, rest_ns->delimiter, leaf_id); - auto qualified_display = - DisplayRestIdentifier(rest_ns->namespace_id, rest_ns->delimiter) + - EffectiveNamespaceDelimiter(rest_ns->delimiter) + leaf_id; + auto qualified_display = qualified_id; vector discovered; string list_error; @@ -940,9 +921,7 @@ class LanceSchemaEntry final : public DuckSchemaEntry { auto leaf_id = create_info.table; auto qualified_id = AppendRestIdentifier(rest_ns->namespace_id, rest_ns->delimiter, leaf_id); - auto qualified_display = - DisplayRestIdentifier(rest_ns->namespace_id, rest_ns->delimiter) + - EffectiveNamespaceDelimiter(rest_ns->delimiter) + leaf_id; + auto qualified_display = qualified_id; vector discovered; string list_error; @@ -1512,9 +1491,7 @@ class LanceDuckCatalog final : public DuckCatalog { auto leaf_id = state->table_name; auto qualified_id = AppendRestIdentifier(state->namespace_id, state->delimiter, leaf_id); - auto qualified_display = - DisplayRestIdentifier(state->namespace_id, state->delimiter) + - EffectiveNamespaceDelimiter(state->delimiter) + leaf_id; + auto qualified_display = qualified_id; vector discovered; string list_error; @@ -1724,10 +1701,8 @@ class LanceDuckCatalog final : public DuckCatalog { } auto leaf_id = create_info.table; - auto qualified_display = - DisplayRestIdentifier(schema_rest_ns->namespace_id, - schema_rest_ns->delimiter) + - EffectiveNamespaceDelimiter(schema_rest_ns->delimiter) + leaf_id; + auto qualified_display = AppendRestIdentifier( + schema_rest_ns->namespace_id, schema_rest_ns->delimiter, leaf_id); bool exists = false; string existing_id; @@ -2027,8 +2002,7 @@ LanceStorageAttach(optional_ptr, ClientContext &context, throw InvalidInputException( "ATTACH TYPE LANCE with ENDPOINT requires a non-empty namespace id"); } - namespace_id = EncodeRestIdentifier( - StringUtil::Split(attach_path, EffectiveNamespaceDelimiter(delimiter))); + namespace_id = attach_path; ResolveLanceNamespaceAuth(context, endpoint, info.options, bearer_token, api_key); ResolveLanceNamespaceAuthOverrides(info.options, bearer_token_override, From c3ce19e2e1ae33ec5d7f174f63390a56c04b9750 Mon Sep 17 00:00:00 2001 From: arrowbowang Date: Fri, 14 Aug 2026 11:28:24 +0800 Subject: [PATCH 10/11] refactor(catalog): limit CTAS refresh to discovery Invalidate only the target schema's lazy table generator after a successful CTAS commit. Leave existing table-entry replacement and dataset freshness semantics to the dedicated replacement and cache-coherency work. --- src/lance_storage.cpp | 56 ++++++++++++------------------------------- 1 file changed, 15 insertions(+), 41 deletions(-) diff --git a/src/lance_storage.cpp b/src/lance_storage.cpp index 8ed691ca..42aa076d 100644 --- a/src/lance_storage.cpp +++ b/src/lance_storage.cpp @@ -1089,37 +1089,16 @@ class LanceSchemaEntry final : public DuckSchemaEntry { DefaultGenerator *table_default_generator = nullptr; }; -static void RefreshLanceSchemaTable(ClientContext &context, - const string &catalog_name, - const string &schema_name, - const string &table_name) { +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(schema.get()) : nullptr; - if (!lance_schema) { - return; + if (lance_schema) { + lance_schema->InvalidateTableDefaults(); } - lance_schema->InvalidateTableDefaults(); - - auto &set = lance_schema->GetCatalogSet(CatalogType::TABLE_ENTRY); - auto transaction = - CatalogTransaction::GetSystemTransaction(schema->catalog.GetDatabase()); - auto existing_entry = set.GetEntry(transaction, table_name); - if (!existing_entry) { - return; - } - if (existing_entry->type != CatalogType::TABLE_ENTRY && - existing_entry->type != CatalogType::VIEW_ENTRY) { - throw InternalException( - "Unexpected catalog entry type for Lance table '%s': %s", table_name, - CatalogTypeToString(existing_entry->type)); - } - if (!set.DropEntry(transaction, existing_entry->name, false, true)) { - throw InternalException( - "Could not refresh catalog entry for Lance table '%s'", table_name); - } - set.CleanupEntry(*existing_entry); } class PhysicalLanceCopyToFile final : public PhysicalCopyToFile { @@ -1128,25 +1107,23 @@ class PhysicalLanceCopyToFile final : public PhysicalCopyToFile { vector types, CopyFunction function, unique_ptr bind_data, idx_t estimated_cardinality, string catalog_name, - string schema_name, string table_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)), table_name(std::move(table_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); - RefreshLanceSchemaTable(context, catalog_name, schema_name, table_name); + InvalidateLanceSchema(context, catalog_name, schema_name); return result; } private: string catalog_name; string schema_name; - string table_name; }; class PhysicalLanceBatchCopyToFile final : public PhysicalBatchCopyToFile { @@ -1155,27 +1132,25 @@ class PhysicalLanceBatchCopyToFile final : public PhysicalBatchCopyToFile { vector types, CopyFunction function, unique_ptr bind_data, idx_t estimated_cardinality, string catalog_name, - string schema_name, string table_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)), table_name(std::move(table_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); - RefreshLanceSchemaTable(context, catalog_name, schema_name, table_name); + InvalidateLanceSchema(context, catalog_name, schema_name); return result; } private: string catalog_name; string schema_name; - string table_name; }; class LanceDuckCatalog final : public DuckCatalog { @@ -1625,8 +1600,7 @@ class LanceDuckCatalog final : public DuckCatalog { } } - RefreshLanceSchemaTable(context, catalog_name, schema_name, - table_name); + InvalidateLanceSchema(context, catalog_name, schema_name); return SinkFinalizeType::READY; } @@ -1808,8 +1782,8 @@ class LanceDuckCatalog final : public DuckCatalog { if (execution_mode == CopyFunctionExecutionMode::BATCH_COPY_TO_FILE) { auto © = planner.Make( op.types, copy_function, std::move(bind_data), - op.estimated_cardinality, op.schema.catalog.GetName(), op.schema.name, - create_info.table); + op.estimated_cardinality, op.schema.catalog.GetName(), + op.schema.name); auto &cast_copy = copy.Cast(); cast_copy.file_path = dataset_path; cast_copy.use_tmp_file = false; @@ -1821,7 +1795,7 @@ class LanceDuckCatalog final : public DuckCatalog { auto © = planner.Make( op.types, copy_function, std::move(bind_data), op.estimated_cardinality, - op.schema.catalog.GetName(), op.schema.name, create_info.table); + op.schema.catalog.GetName(), op.schema.name); auto &cast_copy = copy.Cast(); cast_copy.file_path = dataset_path; cast_copy.use_tmp_file = false; From b4a7fbfc64384f89c3c950fd89c41a587e9cc477 Mon Sep 17 00:00:00 2001 From: arrowbowang Date: Fri, 14 Aug 2026 15:52:05 +0800 Subject: [PATCH 11/11] fix(namespace): preserve REST identifier boundaries Return namespace and table listings through a shared typed string-list FFI instead of lossy newline framing. Reuse the ownership API for directory listings and scalar-index discovery. Normalize full table identifiers relative to the requested namespace, while accepting leaf-only compatibility responses and excluding deeper descendants. Reject appended REST segments containing the active delimiter before remote operations so quoted names cannot silently change hierarchy. --- rust/ffi/dir_namespace.rs | 19 ++--- rust/ffi/index.rs | 50 +++---------- rust/ffi/namespace.rs | 105 ++++++++++++++++++++-------- rust/ffi/util.rs | 103 ++++++++++++++++++++++++++- src/include/lance_common.hpp | 3 + src/include/lance_ffi.hpp | 36 ++++++---- src/lance_common.cpp | 64 ++++++++++------- src/lance_scan.cpp | 22 +++--- src/lance_storage.cpp | 62 ++++++---------- test/sql/namespace_rest_schema.test | 10 +++ 10 files changed, 301 insertions(+), 173 deletions(-) diff --git a/rust/ffi/dir_namespace.rs b/rust/ffi/dir_namespace.rs index 0b1cb414..ef9cd6cd 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 3533f695..499c680a 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 *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> { diff --git a/rust/ffi/namespace.rs b/rust/ffi/namespace.rs index de80fc18..2f90b90e 100644 --- a/rust/ffi/namespace.rs +++ b/rust/ffi/namespace.rs @@ -20,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( @@ -95,6 +95,33 @@ fn split_id(id: &str, delimiter: &str) -> Vec { } } +fn normalize_listed_tables( + tables: Vec, + namespace_id: &str, + delimiter: &str, +) -> Vec { + 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, @@ -143,7 +170,8 @@ pub unsafe extern "C" fn lance_namespace_list_namespaces( api_key: *const c_char, delimiter: *const c_char, headers_tsv: *const c_char, -) -> *const c_char { + out: *mut LanceStringList, +) -> i32 { let result = (|| { let (namespace, id) = namespace_operation_config( endpoint, @@ -181,16 +209,7 @@ pub unsafe extern "C" fn lance_namespace_list_namespaces( }) .map_err(|err| FfiError::new(ErrorCode::Runtime, format!("runtime: {err}")))? })(); - match result { - Ok(namespaces) => { - clear_last_error(); - to_c_string(namespaces.join("\n")).into_raw() as *const c_char - } - Err(err) => { - set_last_error(err.code, err.message); - ptr::null() - } - } + unsafe { export_string_list(result, out) } } #[no_mangle] @@ -301,14 +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_id = split_id(namespace_id, &delimiter); + 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 { @@ -316,7 +335,7 @@ fn list_tables_inner( let mut page_token: Option = None; loop { let mut req = ListTablesRequest::new(); - req.id = Some(namespace_id.clone()); + 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| { @@ -335,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] @@ -346,24 +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(); - to_c_string(tables.join("\n")).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( @@ -930,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 96fa4f57..01d5c0fd 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 = Result; +#[repr(C)] +pub struct LanceStringList { + pub items: *mut *mut c_char, + pub count: usize, +} + +pub(crate) unsafe fn write_string_list( + values: Vec, + 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::>(); + 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>, + 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) -> 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 efb41f70..14b0f43e 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,6 +47,8 @@ void ResolveLanceNamespaceAuthOverrides( const unordered_map &options, string &out_bearer_token, string &out_api_key); +vector LanceConsumeStringList(LanceStringList &list); + bool TryLanceNamespaceListTables(ClientContext &context, const string &endpoint, const string &namespace_id, const string &bearer_token, diff --git a/src/include/lance_ffi.hpp b/src/include/lance_ffi.hpp index 324e43c9..95002ed8 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,14 +55,17 @@ 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); -const char * +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); + 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, @@ -111,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, @@ -352,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 3dca89e9..6b53ab7b 100644 --- a/src/lance_common.cpp +++ b/src/lance_common.cpp @@ -293,6 +293,24 @@ void BuildStorageOptionPointerArrays(const vector &option_keys, } } +vector LanceConsumeStringList(LanceStringList &list) { + vector 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,23 +324,18 @@ 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); - for (auto &table : StringUtil::Split(joined, '\n')) { - if (!table.empty()) { - out_tables.push_back(std::move(table)); - } - } + out_tables = LanceConsumeStringList(list); return true; } @@ -334,20 +347,21 @@ bool TryLanceNamespaceListNamespaces( (void)context; out_namespaces.clear(); out_error.clear(); - auto *ptr = lance_namespace_list_namespaces( + 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()); - if (!ptr) { + headers_tsv.empty() ? nullptr : headers_tsv.c_str(), &list); + if (rc != 0) { out_error = LanceConsumeLastError(); return false; } - string joined = ptr; - lance_free_string(ptr); - for (auto &namespace_name : StringUtil::Split(joined, '\n')) { - if (!namespace_name.empty()) { + 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)); } } @@ -536,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"; @@ -547,11 +563,7 @@ bool TryLanceDirNamespaceListTables(ClientContext &context, const string &root, return false; } - string joined = ptr; - lance_free_string(ptr); - - vector 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 8058a919..46d3c830 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 42aa076d..6724a0cb 100644 --- a/src/lance_storage.cpp +++ b/src/lance_storage.cpp @@ -75,10 +75,17 @@ static string EffectiveNamespaceDelimiter(const string &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 + EffectiveNamespaceDelimiter(delimiter) + segment; + return identifier + effective_delimiter + segment; } static string GetLanceNamespaceEndpoint(const AttachInfo &info) { @@ -223,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 out; - for (auto &p : StringUtil::Split(joined, '\n')) { - if (!p.empty()) { - out.push_back(std::move(p)); - } - } - return out; + return LanceConsumeStringList(list); } static vector @@ -252,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 tables; - for (auto &table : StringUtil::Split(joined, '\n')) { - if (!table.empty()) { - tables.push_back(std::move(table)); - } - } - return tables; + return LanceConsumeStringList(list); } static bool @@ -531,18 +523,8 @@ class LanceRestNamespaceDefaultGenerator : public DefaultGenerator { } vector GetDefaultEntries() override { - auto tables = ListRestNamespaceTables(endpoint, namespace_id, bearer_token, - api_key, delimiter, headers_tsv); - if (namespace_id.empty()) { - return tables; - } - auto prefix = namespace_id + EffectiveNamespaceDelimiter(delimiter); - 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: diff --git a/test/sql/namespace_rest_schema.test b/test/sql/namespace_rest_schema.test index f3403ed3..b1f6ae11 100644 --- a/test/sql/namespace_rest_schema.test +++ b/test/sql/namespace_rest_schema.test @@ -26,9 +26,19 @@ 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}"; ----