SS-452 Add sql option for iceberg storage provider to iceberg catalog connection

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"] }

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",

src/sql-lexer/src/keywords.txt

@@ -380,6 +380,7 @@ Progress
 Projection
 Protobuf
 Protocol
+Provider
 Public
 Publication
 Pushdown

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",

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,

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)
                                                ^
 

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,
                         })
                     }
                 };

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<WithOptionValue<Aug>> for IcebergStorageProvider {
+    fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
+        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<WithOptionValue<Aug>> {
+        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<Self, PlanError> {
+        sql_bail!("must provide an iceberg storage provider")
+    }
+}
+
 impl TryFromValue<WithOptionValue<Aug>> for IcebergAccessDelegation {
     fn try_from_value(v: WithOptionValue<Aug>) -> Result<Self, PlanError> {
         match String::try_from_value(v)? {

src/storage-types/src/connections.rs

@@ -671,6 +671,11 @@ pub struct RestIcebergCatalog<C: ConnectionAccess = InlinedConnection> {
     /// a catalog can reject our whole request,
     /// even if we have our own storage credentials to fall back on.
     pub access_delegation: Option<IcebergAccessDelegation>,
+    /// 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<R: ConnectionResolver> IntoInlineConnection<RestIcebergCatalog, R>
             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<C: ConnectionAccess = InlinedConnection> {
     Rest(RestIcebergCatalog<C>),
@@ -1109,14 +1145,25 @@ impl IcebergCatalogConnection<InlinedConnection> {
                 };
 
                 (
-                    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`).