From edcaab0c90d8552c177ec5362bb2c163e8fb1992 Mon Sep 17 00:00:00 2001 From: Tishj Date: Thu, 16 Jul 2026 19:35:40 +0200 Subject: [PATCH] separate attach logic from the rest of the catalog --- CMakeLists.txt | 1 + src/catalog/rest/iceberg_catalog.cpp | 263 +---------------- src/iceberg_attach.cpp | 268 ++++++++++++++++++ src/iceberg_extension.cpp | 3 +- src/include/catalog/rest/iceberg_catalog.hpp | 10 +- .../rest/storage/iceberg_authorization.hpp | 31 +- src/include/iceberg_attach.hpp | 48 ++++ 7 files changed, 324 insertions(+), 300 deletions(-) create mode 100644 src/iceberg_attach.cpp create mode 100644 src/include/iceberg_attach.hpp diff --git a/CMakeLists.txt b/CMakeLists.txt index 3fd0f75de..ce43213e4 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -102,6 +102,7 @@ set(EXTENSION_SOURCES src/iceberg_extension.cpp src/iceberg_logging.cpp src/iceberg_options.cpp + src/iceberg_attach.cpp src/maintenance/maintenance_table_loader.cpp src/maintenance/rewrite_data_files_executor.cpp src/maintenance/rewrite_data_files_operator.cpp diff --git a/src/catalog/rest/iceberg_catalog.cpp b/src/catalog/rest/iceberg_catalog.cpp index 8eecdec7c..c9d635146 100644 --- a/src/catalog/rest/iceberg_catalog.cpp +++ b/src/catalog/rest/iceberg_catalog.cpp @@ -8,7 +8,6 @@ #include "duckdb/main/attached_database.hpp" #include "duckdb/planner/operator/logical_create_table.hpp" #include "duckdb/common/exception/conversion_exception.hpp" -#include "regex" #include "catalog/rest/catalog_entry/schema/iceberg_schema_entry.hpp" #include "catalog/rest/catalog_entry/table/iceberg_table_entry.hpp" @@ -18,10 +17,6 @@ #include "common/iceberg_utils.hpp" #include "iceberg_logging.hpp" #include "catalog/rest/api/api_utils.hpp" -#include "catalog/rest/storage/iceberg_authorization.hpp" -#include "catalog/rest/storage/authorization/oauth2.hpp" -#include "catalog/rest/storage/authorization/sigv4.hpp" -#include "catalog/rest/storage/authorization/none.hpp" #include "rest_catalog/objects/catalog_config.hpp" using namespace duckdb_yyjson; @@ -416,264 +411,10 @@ void IcebergCatalog::GetConfig(ClientContext &context, IcebergEndpointType &endp // Attach //===--------------------------------------------------------------------===// -// namespace -namespace { - -static IcebergEndpointType EndpointTypeFromString(const string &input) { - D_ASSERT(StringUtil::Lower(input) == input); - - static const case_insensitive_map_t mapping {{"glue", IcebergEndpointType::AWS_GLUE}, - {"s3_tables", IcebergEndpointType::AWS_S3TABLES}}; - - for (auto &entry : mapping) { - if (entry.first == input) { - return entry.second; - } - } - set options; - for (auto &entry : mapping) { - options.insert(entry.first); - } - throw InvalidConfigurationException("Unrecognized 'endpoint_type' (%s), accepted options are: %s", input, - StringUtil::Join(options, ", ")); -} - -} // namespace - //! Streamlined initialization for recognized catalog types -static void S3OrGlueAttachInternal(IcebergAttachOptions &input, const string &service, const string ®ion) { - if (input.authorization_type != IcebergAuthorizationType::INVALID) { - throw InvalidConfigurationException("'endpoint_type' can not be combined with 'authorization_type'"); - } - - input.authorization_type = IcebergAuthorizationType::SIGV4; - input.endpoint = StringUtil::Format("%s.%s.amazonaws.com/iceberg", service, region); -} - -static void S3TablesAttach(IcebergAttachOptions &input) { - // extract region from the amazon ARN - auto substrings = StringUtil::Split(input.warehouse, ":"); - if (substrings.size() != 6) { - throw InvalidInputException("Could not parse S3 Tables ARN warehouse value"); - } - auto region = substrings[3]; - // Populate sigv4_region so it can be used as a fallback region when creating storage secrets - input.options.emplace("sigv4_region", Value(region)); - S3OrGlueAttachInternal(input, "s3tables", region); -} - -static bool SanityCheckGlueWarehouse(const string &warehouse) { - // See: https://docs.aws.amazon.com/glue/latest/dg/connect-glu-iceberg-rest.html#prefix-catalog-path-parameters - - const std::regex patterns[] = { - std::regex("^:$"), // Default catalog ":" in current account - std::regex("^\\d{12}$"), // Default catalog in a specific account - std::regex("^\\d{12}:[^:/]+$"), // Specific catalog in a specific account - std::regex("^[^:]+/[^:]+$"), // Nested catalog in the current account - std::regex("^\\d{12}:[^/]+/[^:]+$") // Nested catalog in a specific account - }; - - for (const auto &pattern : patterns) { - if (std::regex_match(warehouse, pattern)) { - return true; - } - } - - throw InvalidConfigurationException( - "Invalid Glue Catalog Format: '%s'. Expected format: ':', '12-digit account ID', " - "'catalog1/catalog2', or '12-digit accountId:catalog1/catalog2'.", - warehouse); -} - -static void GlueAttach(ClientContext &context, IcebergAttachOptions &input) { - SanityCheckGlueWarehouse(input.warehouse); - - string secret; - auto secret_it = input.options.find("secret"); - if (secret_it != input.options.end()) { - secret = secret_it->second.ToString(); - } - - // look up any s3 secret - - // if there is no secret, an error will be thrown - auto secret_entry = IcebergCatalog::GetStorageSecret(context, secret); - auto kv_secret = dynamic_cast(*secret_entry->secret); - auto region = kv_secret.TryGetValue("region"); - - if (region.IsNull()) { - throw InvalidConfigurationException("Assumed catalog secret '%s' for catalog '%s' does not have a region", - secret_entry->secret->GetName(), input.name); - } - S3OrGlueAttachInternal(input, "glue", region.ToString()); -} - -void IcebergCatalog::SetAWSCatalogOptions(IcebergAttachOptions &attach_options, - case_insensitive_set_t &set_by_attach_options) { - if (set_by_attach_options.find("remove_files_on_delete") == set_by_attach_options.end()) { - attach_options.remove_files_on_delete = false; - } - if (set_by_attach_options.find("stage_create_tables") == set_by_attach_options.end()) { - attach_options.stage_create_tables = false; - } - if (set_by_attach_options.find("purge_requested") == set_by_attach_options.end()) { - attach_options.purge_requested = true; - } -} - -unique_ptr IcebergCatalog::Attach(optional_ptr storage_info, ClientContext &context, - AttachedDatabase &db, const string &name, AttachInfo &info, - AttachOptions &options) { - IcebergAttachOptions attach_options; - attach_options.warehouse = info.path; - attach_options.name = name; - - // check if we have a secret provided - string default_schema; - string endpoint_type_string; - string authorization_type_string; - string access_mode_string; - case_insensitive_set_t set_by_attach_options; - //! First handle generic attach options - for (auto &entry : info.options) { - auto lower_name = StringUtil::Lower(entry.first); - if (lower_name == "type" || lower_name == "read_only") { - continue; - } - - if (lower_name == "endpoint_type") { - endpoint_type_string = StringUtil::Lower(entry.second.ToString()); - } else if (lower_name == "authorization_type") { - authorization_type_string = StringUtil::Lower(entry.second.ToString()); - } else if (lower_name == "access_delegation_mode") { - access_mode_string = StringUtil::Lower(entry.second.ToString()); - } else if (lower_name == "endpoint") { - attach_options.endpoint = entry.second.ToString(); - StringUtil::RTrim(attach_options.endpoint, "/"); - } else if (lower_name == "stage_create_tables") { - auto result = entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue(); - attach_options.stage_create_tables = result; - set_by_attach_options.insert("stage_create_tables"); - } else if (lower_name == "disable_multi_table_commit") { - attach_options.disable_multi_table_commit = - entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue(); - } else if (lower_name == "skip_create_table_metadata_updates") { - attach_options.skip_create_table_metadata_updates = - entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue(); - } else if (lower_name == "remove_files_on_delete") { - attach_options.remove_files_on_delete = entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue(); - set_by_attach_options.insert("remove_files_on_delete"); - } else if (lower_name == "support_nested_namespaces") { - attach_options.support_nested_namespaces = - entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue(); - set_by_attach_options.insert("support_nested_namespaces"); - } else if (lower_name == "purge_requested") { - attach_options.purge_requested = entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue(); - set_by_attach_options.insert("purge_requested"); - } else if (lower_name == "default_schema") { - default_schema = entry.second.ToString(); - } else if (lower_name == "encode_entire_prefix") { - attach_options.encode_entire_prefix = true; - } else if (lower_name == "max_table_staleness") { - auto interval_option = entry.second.DefaultCastAs(LogicalType::INTERVAL); - auto interval_value = interval_option.GetValue(); - int64_t interval_in_micros = 0; - if (!Interval::TryGetMicro(interval_value, interval_in_micros)) { - throw ConversionException("Could not get interval information from %s", interval_option.ToString()); - } - attach_options.max_table_staleness_micros = interval_in_micros; - } else { - attach_options.options.emplace(std::move(entry)); - } - } - IcebergEndpointType endpoint_type = IcebergEndpointType::INVALID; - //! Then check any if the 'endpoint_type' is set, for any well known catalogs - if (!endpoint_type_string.empty()) { - endpoint_type = EndpointTypeFromString(endpoint_type_string); - switch (endpoint_type) { - case IcebergEndpointType::AWS_GLUE: { - GlueAttach(context, attach_options); - endpoint_type = IcebergEndpointType::AWS_GLUE; - SetAWSCatalogOptions(attach_options, set_by_attach_options); - break; - } - case IcebergEndpointType::AWS_S3TABLES: { - S3TablesAttach(attach_options); - endpoint_type = IcebergEndpointType::AWS_S3TABLES; - SetAWSCatalogOptions(attach_options, set_by_attach_options); - break; - } - default: - throw InternalException("Endpoint type (%s) not implemented", endpoint_type_string); - } - } - - //! Then check the authorization type - if (!authorization_type_string.empty()) { - if (attach_options.authorization_type != IcebergAuthorizationType::INVALID) { - throw InvalidConfigurationException("'authorization_type' can not be combined with 'endpoint_type'"); - } - attach_options.authorization_type = IcebergAuthorization::TypeFromString(authorization_type_string); - } - if (!access_mode_string.empty()) { - if (access_mode_string == "vended_credentials") { - attach_options.access_mode = IRCAccessDelegationMode::VENDED_CREDENTIALS; - } else if (access_mode_string == "none") { - attach_options.access_mode = IRCAccessDelegationMode::NONE; - } else { - throw InvalidInputException( - "Unrecognized access mode '%s'. Supported options are 'vended_credentials' and 'none'", - access_mode_string); - } - } - if (attach_options.authorization_type == IcebergAuthorizationType::INVALID) { - attach_options.authorization_type = IcebergAuthorizationType::OAUTH2; - } - - //! Finally, create the auth_handler class from the authorization_type and the remaining options - unique_ptr auth_handler; - switch (attach_options.authorization_type) { - case IcebergAuthorizationType::OAUTH2: { - auth_handler = OAuth2Authorization::FromAttachOptions(db, context, attach_options); - break; - } - case IcebergAuthorizationType::SIGV4: { - auth_handler = SIGV4Authorization::FromAttachOptions(db, attach_options); - break; - } - case IcebergAuthorizationType::NONE: { - auth_handler = NoneAuthorization::FromAttachOptions(db, attach_options); - break; - } - default: - throw InternalException("Authorization Type (%s) not implemented", authorization_type_string); - } - - //! We throw if there are any additional options not handled by previous steps - if (!attach_options.options.empty()) { - set unrecognized_options; - for (auto &entry : attach_options.options) { - unrecognized_options.insert(entry.first); - } - throw InvalidConfigurationException("Unhandled options found: %s", - StringUtil::Join(unrecognized_options, ", ")); - } - - if (attach_options.endpoint.empty()) { - throw InvalidConfigurationException("Missing 'endpoint' option for Iceberg attach"); - } - - D_ASSERT(auth_handler); - auto catalog = - make_uniq(db, options.access_mode, std::move(auth_handler), attach_options, default_schema); - //! Remember the raw attach options so that a later ATTACH OR REPLACE can detect when they change. - catalog->raw_attach_options.insert(options.options.begin(), options.options.end()); - catalog->GetConfig(context, endpoint_type); - if (!default_schema.empty() && !IRCAPI::VerifySchemaExistence(context, *catalog, default_schema)) { - throw InvalidConfigurationException("default_schema '%s' does not exist", default_schema); - } - return std::move(catalog); +void IcebergCatalog::SetAttachOptions(const unordered_map &options) { + raw_attach_options.insert(options.begin(), options.end()); } bool IcebergCatalog::HasConflictingAttachOptions(const string &path, const AttachOptions &options) { diff --git a/src/iceberg_attach.cpp b/src/iceberg_attach.cpp new file mode 100644 index 000000000..1997ebefb --- /dev/null +++ b/src/iceberg_attach.cpp @@ -0,0 +1,268 @@ +#include "iceberg_attach.hpp" +#include "catalog/rest/iceberg_catalog.hpp" + +#include "catalog/rest/storage/iceberg_authorization.hpp" +#include "catalog/rest/storage/authorization/oauth2.hpp" +#include "catalog/rest/storage/authorization/sigv4.hpp" +#include "catalog/rest/storage/authorization/none.hpp" +#include "regex" + +namespace duckdb { + +namespace { + +static IcebergEndpointType EndpointTypeFromString(const string &input) { + D_ASSERT(StringUtil::Lower(input) == input); + + static const case_insensitive_map_t mapping {{"glue", IcebergEndpointType::AWS_GLUE}, + {"s3_tables", IcebergEndpointType::AWS_S3TABLES}}; + + for (auto &entry : mapping) { + if (entry.first == input) { + return entry.second; + } + } + set options; + for (auto &entry : mapping) { + options.insert(entry.first); + } + throw InvalidConfigurationException("Unrecognized 'endpoint_type' (%s), accepted options are: %s", input, + StringUtil::Join(options, ", ")); +} + +static void S3OrGlueAttachInternal(IcebergAttachOptions &input, const string &service, const string ®ion) { + if (input.authorization_type != IcebergAuthorizationType::INVALID) { + throw InvalidConfigurationException("'endpoint_type' can not be combined with 'authorization_type'"); + } + + input.authorization_type = IcebergAuthorizationType::SIGV4; + input.endpoint = StringUtil::Format("%s.%s.amazonaws.com/iceberg", service, region); +} + +static void S3TablesAttach(IcebergAttachOptions &input) { + // extract region from the amazon ARN + auto substrings = StringUtil::Split(input.warehouse, ":"); + if (substrings.size() != 6) { + throw InvalidInputException("Could not parse S3 Tables ARN warehouse value"); + } + auto region = substrings[3]; + // Populate sigv4_region so it can be used as a fallback region when creating storage secrets + input.options.emplace("sigv4_region", Value(region)); + S3OrGlueAttachInternal(input, "s3tables", region); +} + +static bool SanityCheckGlueWarehouse(const string &warehouse) { + // See: https://docs.aws.amazon.com/glue/latest/dg/connect-glu-iceberg-rest.html#prefix-catalog-path-parameters + + const std::regex patterns[] = { + std::regex("^:$"), // Default catalog ":" in current account + std::regex("^\\d{12}$"), // Default catalog in a specific account + std::regex("^\\d{12}:[^:/]+$"), // Specific catalog in a specific account + std::regex("^[^:]+/[^:]+$"), // Nested catalog in the current account + std::regex("^\\d{12}:[^/]+/[^:]+$") // Nested catalog in a specific account + }; + + for (const auto &pattern : patterns) { + if (std::regex_match(warehouse, pattern)) { + return true; + } + } + + throw InvalidConfigurationException( + "Invalid Glue Catalog Format: '%s'. Expected format: ':', '12-digit account ID', " + "'catalog1/catalog2', or '12-digit accountId:catalog1/catalog2'.", + warehouse); +} + +static void GlueAttach(ClientContext &context, IcebergAttachOptions &input) { + SanityCheckGlueWarehouse(input.warehouse); + + string secret; + auto secret_it = input.options.find("secret"); + if (secret_it != input.options.end()) { + secret = secret_it->second.ToString(); + } + + // look up any s3 secret + + // if there is no secret, an error will be thrown + auto secret_entry = IcebergCatalog::GetStorageSecret(context, secret); + auto kv_secret = dynamic_cast(*secret_entry->secret); + auto region = kv_secret.TryGetValue("region"); + + if (region.IsNull()) { + throw InvalidConfigurationException("Assumed catalog secret '%s' for catalog '%s' does not have a region", + secret_entry->secret->GetName(), input.name); + } + S3OrGlueAttachInternal(input, "glue", region.ToString()); +} + +static void SetAWSCatalogOptions(IcebergAttachOptions &attach_options, case_insensitive_set_t &set_by_attach_options) { + if (set_by_attach_options.find("remove_files_on_delete") == set_by_attach_options.end()) { + attach_options.remove_files_on_delete = false; + } + if (set_by_attach_options.find("stage_create_tables") == set_by_attach_options.end()) { + attach_options.stage_create_tables = false; + } + if (set_by_attach_options.find("purge_requested") == set_by_attach_options.end()) { + attach_options.purge_requested = true; + } +} + +} // namespace + +unique_ptr IcebergAttach::Attach(optional_ptr storage_info, ClientContext &context, + AttachedDatabase &db, const string &name, AttachInfo &info, + AttachOptions &options) { + IcebergAttachOptions attach_options; + attach_options.warehouse = info.path; + attach_options.name = name; + + // check if we have a secret provided + string default_schema; + string endpoint_type_string; + string authorization_type_string; + string access_mode_string; + case_insensitive_set_t set_by_attach_options; + //! First handle generic attach options + for (auto &entry : info.options) { + auto lower_name = StringUtil::Lower(entry.first); + if (lower_name == "type" || lower_name == "read_only") { + continue; + } + + if (lower_name == "endpoint_type") { + endpoint_type_string = StringUtil::Lower(entry.second.ToString()); + } else if (lower_name == "authorization_type") { + authorization_type_string = StringUtil::Lower(entry.second.ToString()); + } else if (lower_name == "access_delegation_mode") { + access_mode_string = StringUtil::Lower(entry.second.ToString()); + } else if (lower_name == "endpoint") { + attach_options.endpoint = entry.second.ToString(); + StringUtil::RTrim(attach_options.endpoint, "/"); + } else if (lower_name == "stage_create_tables") { + auto result = entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue(); + attach_options.stage_create_tables = result; + set_by_attach_options.insert("stage_create_tables"); + } else if (lower_name == "disable_multi_table_commit") { + attach_options.disable_multi_table_commit = + entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue(); + } else if (lower_name == "skip_create_table_metadata_updates") { + attach_options.skip_create_table_metadata_updates = + entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue(); + } else if (lower_name == "remove_files_on_delete") { + attach_options.remove_files_on_delete = entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue(); + set_by_attach_options.insert("remove_files_on_delete"); + } else if (lower_name == "support_nested_namespaces") { + attach_options.support_nested_namespaces = + entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue(); + set_by_attach_options.insert("support_nested_namespaces"); + } else if (lower_name == "purge_requested") { + attach_options.purge_requested = entry.second.DefaultCastAs(LogicalType::BOOLEAN).GetValue(); + set_by_attach_options.insert("purge_requested"); + } else if (lower_name == "default_schema") { + default_schema = entry.second.ToString(); + } else if (lower_name == "encode_entire_prefix") { + attach_options.encode_entire_prefix = true; + } else if (lower_name == "max_table_staleness") { + auto interval_option = entry.second.DefaultCastAs(LogicalType::INTERVAL); + auto interval_value = interval_option.GetValue(); + int64_t interval_in_micros = 0; + if (!Interval::TryGetMicro(interval_value, interval_in_micros)) { + throw ConversionException("Could not get interval information from %s", interval_option.ToString()); + } + attach_options.max_table_staleness_micros = interval_in_micros; + } else { + attach_options.options.emplace(std::move(entry)); + } + } + IcebergEndpointType endpoint_type = IcebergEndpointType::INVALID; + //! Then check any if the 'endpoint_type' is set, for any well known catalogs + if (!endpoint_type_string.empty()) { + endpoint_type = EndpointTypeFromString(endpoint_type_string); + switch (endpoint_type) { + case IcebergEndpointType::AWS_GLUE: { + GlueAttach(context, attach_options); + endpoint_type = IcebergEndpointType::AWS_GLUE; + SetAWSCatalogOptions(attach_options, set_by_attach_options); + break; + } + case IcebergEndpointType::AWS_S3TABLES: { + S3TablesAttach(attach_options); + endpoint_type = IcebergEndpointType::AWS_S3TABLES; + SetAWSCatalogOptions(attach_options, set_by_attach_options); + break; + } + default: + throw InternalException("Endpoint type (%s) not implemented", endpoint_type_string); + } + } + + //! Then check the authorization type + if (!authorization_type_string.empty()) { + if (attach_options.authorization_type != IcebergAuthorizationType::INVALID) { + throw InvalidConfigurationException("'authorization_type' can not be combined with 'endpoint_type'"); + } + attach_options.authorization_type = IcebergAuthorization::TypeFromString(authorization_type_string); + } + if (!access_mode_string.empty()) { + if (access_mode_string == "vended_credentials") { + attach_options.access_mode = IRCAccessDelegationMode::VENDED_CREDENTIALS; + } else if (access_mode_string == "none") { + attach_options.access_mode = IRCAccessDelegationMode::NONE; + } else { + throw InvalidInputException( + "Unrecognized access mode '%s'. Supported options are 'vended_credentials' and 'none'", + access_mode_string); + } + } + if (attach_options.authorization_type == IcebergAuthorizationType::INVALID) { + attach_options.authorization_type = IcebergAuthorizationType::OAUTH2; + } + + //! Finally, create the auth_handler class from the authorization_type and the remaining options + unique_ptr auth_handler; + switch (attach_options.authorization_type) { + case IcebergAuthorizationType::OAUTH2: { + auth_handler = OAuth2Authorization::FromAttachOptions(db, context, attach_options); + break; + } + case IcebergAuthorizationType::SIGV4: { + auth_handler = SIGV4Authorization::FromAttachOptions(db, attach_options); + break; + } + case IcebergAuthorizationType::NONE: { + auth_handler = NoneAuthorization::FromAttachOptions(db, attach_options); + break; + } + default: + throw InternalException("Authorization Type (%s) not implemented", authorization_type_string); + } + + //! We throw if there are any additional options not handled by previous steps + if (!attach_options.options.empty()) { + set unrecognized_options; + for (auto &entry : attach_options.options) { + unrecognized_options.insert(entry.first); + } + throw InvalidConfigurationException("Unhandled options found: %s", + StringUtil::Join(unrecognized_options, ", ")); + } + + if (attach_options.endpoint.empty()) { + throw InvalidConfigurationException("Missing 'endpoint' option for Iceberg attach"); + } + + D_ASSERT(auth_handler); + auto catalog = + make_uniq(db, options.access_mode, std::move(auth_handler), attach_options, default_schema); + //! Remember the raw attach options so that a later ATTACH OR REPLACE can detect when they change. + catalog->SetAttachOptions(options.options); + catalog->GetConfig(context, endpoint_type); + if (!default_schema.empty() && !IRCAPI::VerifySchemaExistence(context, *catalog, default_schema)) { + throw InvalidConfigurationException("default_schema '%s' does not exist", default_schema); + } + return std::move(catalog); +} + +} // namespace duckdb diff --git a/src/iceberg_extension.cpp b/src/iceberg_extension.cpp index a4d10ba79..b859ca5e5 100644 --- a/src/iceberg_extension.cpp +++ b/src/iceberg_extension.cpp @@ -22,6 +22,7 @@ #include "catalog/rest/storage/authorization/sigv4.hpp" #include "common/iceberg_utils.hpp" #include "iceberg_logging.hpp" +#include "iceberg_attach.hpp" #include "iceberg_options.hpp" #include "function/copy/iceberg_copy_function.hpp" #include "duckdb/optimizer/optimizer_extension.hpp" @@ -38,7 +39,7 @@ static unique_ptr CreateTransactionManager(optional_ptr Attach(optional_ptr storage_info, ClientContext &context, - AttachedDatabase &db, const string &name, AttachInfo &info, - AttachOptions &options); - public: void Initialize(bool load_builtin) override; string GetCatalogType() override { @@ -192,6 +185,7 @@ class IcebergCatalog : public Catalog { string GetDBPath() override; //! Allow ATTACH OR REPLACE to actually re-attach when iceberg-specific options change bool HasConflictingAttachOptions(const string &path, const AttachOptions &options) override; + void SetAttachOptions(const unordered_map &options); static string GetOnlyMergeOnReadSupportedErrorMessage(const string &table_name, const string &property, const string &property_value); @@ -216,7 +210,7 @@ class IcebergCatalog : public Catalog { case_insensitive_map_t defaults; case_insensitive_map_t overrides; //! raw attach options (after core stripping) used to detect a conflicting ATTACH OR REPLACE - case_insensitive_map_t raw_attach_options; + unordered_map raw_attach_options; public: unordered_set supported_urls; diff --git a/src/include/catalog/rest/storage/iceberg_authorization.hpp b/src/include/catalog/rest/storage/iceberg_authorization.hpp index 9e45a4998..fe406b696 100644 --- a/src/include/catalog/rest/storage/iceberg_authorization.hpp +++ b/src/include/catalog/rest/storage/iceberg_authorization.hpp @@ -4,41 +4,12 @@ #include "duckdb/common/http_util.hpp" #include "duckdb/main/client_context_state.hpp" +#include "iceberg_attach.hpp" #include "catalog/rest/api/catalog_utils.hpp" #include "catalog/rest/api/url_utils.hpp" namespace duckdb { -enum class IcebergEndpointType : uint8_t { AWS_S3TABLES, AWS_GLUE, INVALID }; - -enum class IcebergAuthorizationType : uint8_t { OAUTH2, SIGV4, NONE, INVALID }; - -enum class IRCAccessDelegationMode : uint8_t { NONE, VENDED_CREDENTIALS }; - -struct IcebergAttachOptions { - string endpoint; - string warehouse; - string secret; - string name; - // some catalogs do not yet support stage create - bool stage_create_tables = true; - // some catalogs reject the multi-table transactions/commit endpoint; opt out of it here - bool disable_multi_table_commit = false; - // some catalogs fully initialize metadata during non-staged CREATE TABLE and reject follow-up metadata updates - bool skip_create_table_metadata_updates = false; - // if the catalog allows manual cleaning up of storage files. - bool remove_files_on_delete = true; - bool support_nested_namespaces = false; - bool encode_entire_prefix = false; - // in rest api spec, purge requested defaults to false. - bool purge_requested = false; - IRCAccessDelegationMode access_mode = IRCAccessDelegationMode::VENDED_CREDENTIALS; - IcebergAuthorizationType authorization_type = IcebergAuthorizationType::INVALID; - unordered_map options; - // max staleness for cached table metadata in minutes (optional - if not set, always request fresh metadata) - optional_idx max_table_staleness_micros; -}; - //! Hold the pre-initialized HTTPClient for a given connection struct IcebergAuthorizationContextState : public ClientContextState { public: diff --git a/src/include/iceberg_attach.hpp b/src/include/iceberg_attach.hpp new file mode 100644 index 000000000..02073e9e7 --- /dev/null +++ b/src/include/iceberg_attach.hpp @@ -0,0 +1,48 @@ +#pragma once + +#include "duckdb/common/string.hpp" +#include "duckdb/common/optional.hpp" +#include "duckdb/common/named_parameter_map.hpp" +#include "duckdb/storage/storage_extension.hpp" +#include "duckdb/main/attached_database.hpp" +#include "duckdb/main/client_context.hpp" + +namespace duckdb { + +enum class IcebergEndpointType : uint8_t { AWS_S3TABLES, AWS_GLUE, INVALID }; + +enum class IcebergAuthorizationType : uint8_t { OAUTH2, SIGV4, NONE, INVALID }; + +enum class IRCAccessDelegationMode : uint8_t { NONE, VENDED_CREDENTIALS }; + +struct IcebergAttachOptions { + string endpoint; + string warehouse; + string secret; + string name; + // some catalogs do not yet support stage create + bool stage_create_tables = true; + // some catalogs reject the multi-table transactions/commit endpoint; opt out of it here + bool disable_multi_table_commit = false; + // some catalogs fully initialize metadata during non-staged CREATE TABLE and reject follow-up metadata updates + bool skip_create_table_metadata_updates = false; + // if the catalog allows manual cleaning up of storage files. + bool remove_files_on_delete = true; + bool support_nested_namespaces = false; + bool encode_entire_prefix = false; + // in rest api spec, purge requested defaults to false. + bool purge_requested = false; + IRCAccessDelegationMode access_mode = IRCAccessDelegationMode::VENDED_CREDENTIALS; + IcebergAuthorizationType authorization_type = IcebergAuthorizationType::INVALID; + unordered_map options; + // max staleness for cached table metadata in minutes (optional - if not set, always request fresh metadata) + optional_idx max_table_staleness_micros; +}; + +struct IcebergAttach { + static unique_ptr Attach(optional_ptr storage_info, ClientContext &context, + AttachedDatabase &db, const string &name, AttachInfo &info, + AttachOptions &options); +}; + +} // namespace duckdb