diff --git a/Cargo.lock b/Cargo.lock index 80bc64224e23a..3052342c2636f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9847,6 +9847,7 @@ dependencies = [ "opendal-core", "opendal-layer-retry", "opendal-layer-timeout", + "opendal-service-azdls", "opendal-service-gcs", "opendal-service-s3", ] @@ -9900,6 +9901,36 @@ dependencies = [ "tokio", ] +[[package]] +name = "opendal-service-azdls" +version = "0.57.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dea4908d490143a9b0b7f7a790e139ff829b06a023f670455ed3d44f664b361" +dependencies = [ + "base64 0.22.1", + "bytes", + "http 1.4.2", + "log", + "opendal-core", + "opendal-service-azure-common", + "quick-xml 0.39.4", + "reqsign-azure-storage", + "reqsign-core", + "reqsign-file-read-tokio", + "serde", + "serde_json", +] + +[[package]] +name = "opendal-service-azure-common" +version = "0.57.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9b489f13c42e69d69bdd72952b634356ec43a7881a20259b38b540fcecdf4051" +dependencies = [ + "http 1.4.2", + "opendal-core", +] + [[package]] name = "opendal-service-gcs" version = "0.57.0" @@ -11685,6 +11716,27 @@ dependencies = [ "serde", ] +[[package]] +name = "reqsign-azure-storage" +version = "3.1.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2824e7da3c2cc42ac3406c674eb57c89127fdcd97f3a73c608cfc680505ea134" +dependencies = [ + "anyhow", + "base64 0.23.1", + "bytes", + "form_urlencoded", + "http 1.4.2", + "log", + "pem 3.0.6", + "percent-encoding", + "reqsign-core", + "rsa", + "serde", + "serde_json", + "sha1 0.11.0", +] + [[package]] name = "reqsign-core" version = "3.3.0" diff --git a/Cargo.toml b/Cargo.toml index d6de0f6ce2336..4da5cc64393e3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -383,7 +383,7 @@ hyper-util = "0.1.20" tower-service = "0.3.3" iceberg = "0.10.1" iceberg-catalog-rest = "0.10.1" -iceberg-storage-opendal = { version = "0.10.1", default-features = false, features = ["opendal-s3", "opendal-gcs"] } +iceberg-storage-opendal = { version = "0.10.1", default-features = false, features = ["opendal-s3", "opendal-gcs", "opendal-azdls"] } imbl = { version = "7.0.0", features = ["serde"] } include_dir = "0.7.4" indexmap = { version = "2.10.0", default-features = false, features = ["std"] } diff --git a/deny.toml b/deny.toml index ec3704b00ef8a..0a72f4bde8c0a 100644 --- a/deny.toml +++ b/deny.toml @@ -235,6 +235,7 @@ wrappers = [ "native-tls", "opendal-core", "opendal-layer-retry", + "opendal-service-azdls", "opendal-service-gcs", "opendal-service-s3", "os_info", @@ -245,6 +246,7 @@ wrappers = [ "rdkafka", "reqsign-aws-core", "reqsign-aws-v4", + "reqsign-azure-storage", "reqsign-core", "reqsign-google", "reqwest", diff --git a/src/sql-lexer/src/keywords.txt b/src/sql-lexer/src/keywords.txt index a69287cc3ce23..18763dd47871e 100644 --- a/src/sql-lexer/src/keywords.txt +++ b/src/sql-lexer/src/keywords.txt @@ -380,6 +380,7 @@ Progress Projection Protobuf Protocol +Provider Public Publication Pushdown diff --git a/src/sql-parser/src/ast/defs/ddl.rs b/src/sql-parser/src/ast/defs/ddl.rs index 5a591458609b7..c3626f9631ffa 100644 --- a/src/sql-parser/src/ast/defs/ddl.rs +++ b/src/sql-parser/src/ast/defs/ddl.rs @@ -931,6 +931,7 @@ pub enum ConnectionOptionName { SslMode, SessionToken, CatalogType, + StorageProvider, Url, User, Warehouse, @@ -978,6 +979,7 @@ impl ConnectionOptionName { | ConnectionOptionName::SshTunnel | ConnectionOptionName::SslMode | ConnectionOptionName::CatalogType + | ConnectionOptionName::StorageProvider | ConnectionOptionName::Url | ConnectionOptionName::Warehouse => false, } @@ -1027,6 +1029,7 @@ impl AstDisplay for ConnectionOptionName { ConnectionOptionName::SslMode => "SSL MODE", ConnectionOptionName::SessionToken => "SESSION TOKEN", ConnectionOptionName::CatalogType => "CATALOG TYPE", + ConnectionOptionName::StorageProvider => "STORAGE PROVIDER", ConnectionOptionName::Url => "URL", ConnectionOptionName::User => "USER", ConnectionOptionName::Warehouse => "WAREHOUSE", diff --git a/src/sql-parser/src/parser.rs b/src/sql-parser/src/parser.rs index 96e4ba1efad03..76af586641bf9 100644 --- a/src/sql-parser/src/parser.rs +++ b/src/sql-parser/src/parser.rs @@ -2989,6 +2989,7 @@ impl<'a> Parser<'a> { SESSION, SSH, SSL, + STORAGE, URL, USER, USERNAME, @@ -3102,6 +3103,10 @@ impl<'a> Parser<'a> { MODE => ConnectionOptionName::SslMode, _ => unreachable!(), }, + STORAGE => { + self.expect_keyword(PROVIDER)?; + ConnectionOptionName::StorageProvider + } URL => ConnectionOptionName::Url, // TYPE => ConnectionOptionName::CatalogType, WAREHOUSE => ConnectionOptionName::Warehouse, diff --git a/src/sql-parser/tests/testdata/ddl b/src/sql-parser/tests/testdata/ddl index 37c4c04cb1711..cfeaf705a4337 100644 --- a/src/sql-parser/tests/testdata/ddl +++ b/src/sql-parser/tests/testdata/ddl @@ -627,6 +627,20 @@ CREATE CONNECTION icebergconn TO ICEBERG CATALOG (CATALOG TYPE = 'rest', URL = ' => CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("icebergconn")]), connection_type: IcebergCatalog, if_not_exists: false, values: [ConnectionOption { name: CatalogType, value: Some(Value(String("rest"))) }, ConnectionOption { name: Url, value: Some(Value(String("https://example/api/catalog"))) }, ConnectionOption { name: Credential, value: Some(Value(String("id:secret"))) }, ConnectionOption { name: Oauth2ServerUrl, value: Some(Value(String("https://example/oidc/v1/token"))) }, ConnectionOption { name: Scope, value: Some(Value(String("all-apis"))) }, ConnectionOption { name: Warehouse, value: Some(Value(String("wh"))) }], with_options: [] }) +parse-statement +CREATE CONNECTION icebergconn TO ICEBERG CATALOG (CATALOG TYPE 'rest', URL 'https://example/api/catalog', CREDENTIAL 'id:secret', STORAGE PROVIDER 'gcs', WAREHOUSE 'wh') +---- +CREATE CONNECTION icebergconn TO ICEBERG CATALOG (CATALOG TYPE = 'rest', URL = 'https://example/api/catalog', CREDENTIAL = 'id:secret', STORAGE PROVIDER = 'gcs', WAREHOUSE = 'wh') +=> +CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("icebergconn")]), connection_type: IcebergCatalog, if_not_exists: false, values: [ConnectionOption { name: CatalogType, value: Some(Value(String("rest"))) }, ConnectionOption { name: Url, value: Some(Value(String("https://example/api/catalog"))) }, ConnectionOption { name: Credential, value: Some(Value(String("id:secret"))) }, ConnectionOption { name: StorageProvider, value: Some(Value(String("gcs"))) }, ConnectionOption { name: Warehouse, value: Some(Value(String("wh"))) }], with_options: [] }) + +parse-statement +CREATE CONNECTION icebergconn TO ICEBERG CATALOG (CATALOG TYPE 'rest', URL 'https://example/api/catalog', CREDENTIAL 'id:secret', STORAGE 'gcs') +---- +error: Expected PROVIDER, found string literal "gcs" +CREATE CONNECTION icebergconn TO ICEBERG CATALOG (CATALOG TYPE 'rest', URL 'https://example/api/catalog', CREDENTIAL 'id:secret', STORAGE 'gcs') + ^ + parse-statement CREATE CONNECTION privatelinkconn TO AWS PRIVATELINK (SERVICE NAME 'com.amazonaws.vpce.us-east-1.vpce-svc-0e123abc123198abc', AVAILABILITY ZONES ('use1-az1', 'use1-az4')) ---- @@ -3082,7 +3096,7 @@ CreateConnection(CreateConnectionStatement { name: UnresolvedItemName([Ident("my parse-statement CREATE CONNECTION my_ssh_tunnel FOR SSH TUNNEL (PUBLIC KEY 3 = nope) ---- -error: Expected one of ACCESS or ASSUME or AVAILABILITY or AWS or BROKER or BROKERS or CATALOG or CREDENTIAL or DATABASE or ENDPOINT or GCP or HOST or OAUTH2 or PASSWORD or PORT or PUBLIC or PROGRESS or REGION or REGISTRY or SASL or SCOPE or SECRET or SECURITY or SERVICE or SESSION or SSH or SSL or URL or USER or USERNAME or WAREHOUSE, found left parenthesis +error: Expected one of ACCESS or ASSUME or AVAILABILITY or AWS or BROKER or BROKERS or CATALOG or CREDENTIAL or DATABASE or ENDPOINT or GCP or HOST or OAUTH2 or PASSWORD or PORT or PUBLIC or PROGRESS or REGION or REGISTRY or SASL or SCOPE or SECRET or SECURITY or SERVICE or SESSION or SSH or SSL or STORAGE or URL or USER or USERNAME or WAREHOUSE, found left parenthesis CREATE CONNECTION my_ssh_tunnel FOR SSH TUNNEL (PUBLIC KEY 3 = nope) ^ diff --git a/src/sql/src/plan/statement/ddl/connection.rs b/src/sql/src/plan/statement/ddl/connection.rs index cf333d0b1ab93..eccbfed25b732 100644 --- a/src/sql/src/plan/statement/ddl/connection.rs +++ b/src/sql/src/plan/statement/ddl/connection.rs @@ -36,9 +36,10 @@ use mz_storage_types::connections::{ AwsPrivatelink, AwsPrivatelinkConnection, AwsPrivatelinkRule, CsrConnection, CsrConnectionHttpAuth, GlueSchemaRegistryConnection, IcebergAccessDelegation, IcebergCatalogAuth, IcebergCatalogConnection, IcebergCatalogImpl, IcebergCatalogType, - KafkaConnection, KafkaSaslConfig, KafkaTlsConfig, KafkaTopicOptions, MySqlConnection, - MySqlSslMode, PostgresConnection, RestIcebergCatalog, S3TablesRestIcebergCatalog, - SqlServerConnectionDetails, SshConnection, SshTunnel, TlsIdentity, Tunnel, + IcebergStorageProvider, KafkaConnection, KafkaSaslConfig, KafkaTlsConfig, KafkaTopicOptions, + MySqlConnection, MySqlSslMode, PostgresConnection, RestIcebergCatalog, + S3TablesRestIcebergCatalog, SqlServerConnectionDetails, SshConnection, SshTunnel, TlsIdentity, + Tunnel, }; use crate::names::Aug; @@ -50,6 +51,7 @@ use crate::session::vars; generate_extracted_config!( ConnectionOption, (AccessDelegation, IcebergAccessDelegation), + (StorageProvider, IcebergStorageProvider), (AccessKeyId, StringOrSecret), (AssumeRoleArn, String), (AssumeRoleSessionName, String), @@ -199,6 +201,7 @@ pub(super) fn validate_options_per_connection_type( GcpConnection, Oauth2ServerUrl, Scope, + StorageProvider, Url, Warehouse, ], @@ -711,6 +714,11 @@ impl ConnectionOptionExtracted { "invalid CONNECTION: ICEBERG s3tablesrest connections do not support ACCESS DELEGATION" ); } + if self.storage_provider.is_some() { + sql_bail!( + "invalid CONNECTION: ICEBERG s3tablesrest connections do not support STORAGE PROVIDER" + ); + } let Some(warehouse) = warehouse else { sql_bail!( "invalid CONNECTION: ICEBERG s3tablesrest connections must specify WAREHOUSE" @@ -753,6 +761,18 @@ impl ConnectionOptionExtracted { "invalid CONNECTION: ICEBERG GCP CONNECTION does not support ACCESS DELEGATION" ); } + // A GCP connection authenticates to GCS, so the + // store is already determined. Reject a + // contradicting value rather than silently + // overriding it. + if let Some(provider) = self.storage_provider + && provider != IcebergStorageProvider::Gcs + { + sql_bail!( + "invalid CONNECTION: ICEBERG GCP CONNECTION implies STORAGE PROVIDER 'gcs', not '{}'", + provider.as_str() + ); + } /// All BigLake Iceberg REST Catalogs use the same catalog URI. const BIGLAKE_CATALOG_URI: &str = "https://biglake.googleapis.com/iceberg/v1/restcatalog"; @@ -769,10 +789,19 @@ impl ConnectionOptionExtracted { ), }; + // A GCP connection only authenticates to GCS, so it + // fixes the store regardless of what was written. + let storage_provider = if matches!(auth, IcebergCatalogAuth::Gcp(_)) { + IcebergStorageProvider::Gcs + } else { + self.storage_provider.unwrap_or_default() + }; + IcebergCatalogImpl::Rest(RestIcebergCatalog { auth, warehouse, access_delegation: self.access_delegation, + storage_provider, }) } }; diff --git a/src/sql/src/plan/with_options.rs b/src/sql/src/plan/with_options.rs index 2c7342b0ae7b6..0bf582c02fb05 100644 --- a/src/sql/src/plan/with_options.rs +++ b/src/sql/src/plan/with_options.rs @@ -21,7 +21,9 @@ use mz_sql_parser::ast::{ NetworkPolicyRuleDefinition, RefreshOptionValue, ReplicaDefinition, }; use mz_storage_types::connections::string_or_secret::StringOrSecret; -use mz_storage_types::connections::{IcebergAccessDelegation, IcebergCatalogType}; +use mz_storage_types::connections::{ + IcebergAccessDelegation, IcebergCatalogType, IcebergStorageProvider, +}; use serde::{Deserialize, Serialize}; use crate::ast::{AstInfo, UnresolvedItemName, Value, WithOptionValue}; @@ -68,6 +70,33 @@ impl ImpliedValue for IcebergCatalogType { } } +impl TryFromValue> for IcebergStorageProvider { + fn try_from_value(v: WithOptionValue) -> Result { + match String::try_from_value(v)? { + s if s.eq_ignore_ascii_case("s3") => Ok(IcebergStorageProvider::S3), + s if s.eq_ignore_ascii_case("gcs") => Ok(IcebergStorageProvider::Gcs), + s if s.eq_ignore_ascii_case("adls") => Ok(IcebergStorageProvider::Adls), + _ => sql_bail!("invalid iceberg storage provider, expected 's3', 'gcs', or 'adls'"), + } + } + + fn try_into_value(self, _catalog: &dyn SessionCatalog) -> Option> { + Some(WithOptionValue::Value(Value::String( + self.as_str().to_string(), + ))) + } + + fn name() -> String { + "iceberg storage provider".to_string() + } +} + +impl ImpliedValue for IcebergStorageProvider { + fn implied_value() -> Result { + sql_bail!("must provide an iceberg storage provider") + } +} + impl TryFromValue> for IcebergAccessDelegation { fn try_from_value(v: WithOptionValue) -> Result { match String::try_from_value(v)? { diff --git a/src/storage-types/src/connections.rs b/src/storage-types/src/connections.rs index aa48f6a599c23..628af8478aba5 100644 --- a/src/storage-types/src/connections.rs +++ b/src/storage-types/src/connections.rs @@ -671,6 +671,11 @@ pub struct RestIcebergCatalog { /// a catalog can reject our whole request, /// even if we have our own storage credentials to fall back on. pub access_delegation: Option, + /// Which object store the catalog's tables live in. + /// + /// Defaults to S3. A REST catalog does not tell us this, so a table backed + /// by GCS or ADLS is unreadable until the connection says so. + pub storage_provider: IcebergStorageProvider, } /// The value Materialize sends in the Iceberg REST `X-Iceberg-Access-Delegation` @@ -725,6 +730,7 @@ impl IntoInlineConnection auth: self.auth.into_inline_connection(&r), warehouse: self.warehouse, access_delegation: self.access_delegation, + storage_provider: self.storage_provider, } } } @@ -746,6 +752,36 @@ pub enum IcebergCatalogType { S3TablesRest, } +/// Which object store holds the data files of a REST catalog's tables. +/// +/// The catalog protocol says nothing about this: a REST catalog hands back +/// storage locations and credentials, and the client has to already know how to +/// talk to that store. So it is configured per connection rather than +/// discovered. +#[derive(Clone, Copy, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)] +pub enum IcebergStorageProvider { + S3, + Gcs, + Adls, +} + +impl IcebergStorageProvider { + /// The name as spelled in SQL. + pub fn as_str(&self) -> &'static str { + match self { + IcebergStorageProvider::S3 => "s3", + IcebergStorageProvider::Gcs => "gcs", + IcebergStorageProvider::Adls => "adls", + } + } +} + +impl Default for IcebergStorageProvider { + fn default() -> Self { + IcebergStorageProvider::S3 + } +} + #[derive(Clone, Debug, Eq, PartialEq, Hash, Serialize, Deserialize)] pub enum IcebergCatalogImpl { Rest(RestIcebergCatalog), @@ -1109,14 +1145,25 @@ impl IcebergCatalogConnection { }; ( - OpenDalStorageFactory::S3 { + // The catalog tells us where the data lives but not what + // kind of store it is, so the connection has to say. + match rest.storage_provider { // When used with MinIO, Polaris returns a config with: // s3.access-key-id, s3.secret-access-key, s3.endpoint, ... // `iceberg-rust` forwards these props to `opendal`. When the catalog // vends instead, it returns per-table `storage-credentials` that // `iceberg-rust` wires into the same FileIO. // N.B. This is not confirmed to work with other catalog & storage implementations. - customized_credential_load, + IcebergStorageProvider::S3 => OpenDalStorageFactory::S3 { + customized_credential_load, + }, + // Both take their credentials from the catalog's + // config, which `iceberg-rust` forwards to `opendal` + // the same way. Neither has an equivalent of the S3 + // credential loader, so vended credentials for these + // stores only work through those props. + IcebergStorageProvider::Gcs => OpenDalStorageFactory::Gcs, + IcebergStorageProvider::Adls => OpenDalStorageFactory::Azdls, }, // NOTE: We construct our own OAuth authenticator for the Catalog client instead of using the one built in. // This means we ignore auth overrides from `/v1/config` (e.g. `oauth2-server-uri`). diff --git a/test/iceberg/mzcompose.py b/test/iceberg/mzcompose.py index d0da220de564e..f681670208f37 100644 --- a/test/iceberg/mzcompose.py +++ b/test/iceberg/mzcompose.py @@ -130,6 +130,19 @@ def workflow_oauth2_server_url(c: Composition) -> None: ) +def workflow_storage_provider(c: Composition) -> None: + """STORAGE PROVIDER names the object store behind a REST catalog's tables. + The catalog protocol does not report it, so a table on GCS or ADLS is + unreadable until the connection says so. This exercises connection planning + only and needs no Iceberg backend.""" + c.down(destroy_volumes=True) + c.up("materialized") + + c.run_testdrive_files( + "storage-provider.td", + ) + + def workflow_mode_append(c: Composition) -> None: key = _setup(c) diff --git a/test/iceberg/storage-provider.td b/test/iceberg/storage-provider.td new file mode 100644 index 0000000000000..b09a99efdb5e6 --- /dev/null +++ b/test/iceberg/storage-provider.td @@ -0,0 +1,100 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +# STORAGE PROVIDER names the object store holding a REST catalog's data files. +# The catalog protocol does not report this: it hands back locations and +# credentials and expects the client to already know how to reach that store. +# +# Everything asserted here is decided at plan time, so skip the +# connect-on-create validation and point at hosts that do not exist. +$ postgres-execute connection=postgres://mz_system:materialize@${testdrive.materialize-internal-sql-addr} +ALTER SYSTEM SET enable_default_connection_validation = false + +> CREATE SECRET sp_cred AS 'client-id:client-secret' + +# Omitting the option leaves the default of S3 in place. It is not echoed back, +# so an unset provider stays unset in the catalog rather than being normalized +# to an explicit 's3'. +> CREATE CONNECTION sp_default TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'https://example.invalid/api/catalog', + CREDENTIAL = SECRET sp_cred, + WAREHOUSE = 'wh' + ); + +> SHOW CREATE CONNECTION sp_default +name create_sql +--------------------------------- +materialize.public.sp_default "CREATE CONNECTION materialize.public.sp_default TO ICEBERG CATALOG (CATALOG TYPE = 'rest', CREDENTIAL = SECRET materialize.public.sp_cred, URL = 'https://example.invalid/api/catalog', WAREHOUSE = 'wh');" + +# Each accepted value round-trips through the catalog. +> CREATE CONNECTION sp_s3 TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'https://example.invalid/api/catalog', + CREDENTIAL = SECRET sp_cred, + STORAGE PROVIDER = 's3', + WAREHOUSE = 'wh' + ); + +> CREATE CONNECTION sp_gcs TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'https://example.invalid/api/catalog', + CREDENTIAL = SECRET sp_cred, + STORAGE PROVIDER = 'gcs', + WAREHOUSE = 'wh' + ); + +> CREATE CONNECTION sp_adls TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'https://example.invalid/api/catalog', + CREDENTIAL = SECRET sp_cred, + STORAGE PROVIDER = 'adls', + WAREHOUSE = 'wh' + ); + +> SELECT name FROM mz_connections WHERE name LIKE 'sp\_%' ORDER BY name +sp_adls +sp_default +sp_gcs +sp_s3 + +# The value is case-insensitive, like CATALOG TYPE. +> CREATE CONNECTION sp_upper TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'https://example.invalid/api/catalog', + CREDENTIAL = SECRET sp_cred, + STORAGE PROVIDER = 'GCS', + WAREHOUSE = 'wh' + ); + +! CREATE CONNECTION sp_bad TO ICEBERG CATALOG ( + CATALOG TYPE = 'REST', + URL = 'https://example.invalid/api/catalog', + CREDENTIAL = SECRET sp_cred, + STORAGE PROVIDER = 'azure', + WAREHOUSE = 'wh' + ); +contains:invalid iceberg storage provider, expected 's3', 'gcs', or 'adls' + +# S3 Tables is always backed by S3, so the option is meaningless there rather +# than merely redundant. +! CREATE CONNECTION sp_on_s3tables TO ICEBERG CATALOG ( + CATALOG TYPE = 's3tablesrest', + URL = 'https://example.invalid/api/catalog', + STORAGE PROVIDER = 's3', + WAREHOUSE = 'wh' + ); +contains:s3tablesrest connections do not support STORAGE PROVIDER + +> DROP CONNECTION sp_upper +> DROP CONNECTION sp_adls +> DROP CONNECTION sp_gcs +> DROP CONNECTION sp_s3 +> DROP CONNECTION sp_default +> DROP SECRET sp_cred