diff --git a/Cargo.lock b/Cargo.lock index 0f264cc1f5..05c1b8c9e3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5151,6 +5151,7 @@ dependencies = [ "opendal-layer-logging", "opendal-layer-retry", "opendal-layer-timeout", + "opendal-service-azblob", "opendal-service-azdls", "opendal-service-fs", "opendal-service-gcs", @@ -5242,6 +5243,27 @@ dependencies = [ "tokio", ] +[[package]] +name = "opendal-service-azblob" +version = "0.58.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b3310fbbb48f111c6f590473c2cd15e1b7f8e384444b0d4e328f0464c864d767" +dependencies = [ + "base64 0.23.0", + "bytes", + "http 1.4.2", + "log", + "opendal-core", + "opendal-service-azure-common", + "quick-xml 0.41.0", + "reqsign-azure-storage", + "reqsign-core", + "reqsign-file-read-tokio", + "serde", + "sha2 0.11.0", + "uuid", +] + [[package]] name = "opendal-service-azdls" version = "0.58.1" diff --git a/bindings/python/DEPENDENCIES.rust.tsv b/bindings/python/DEPENDENCIES.rust.tsv index d38f10dd3f..2ae6d987c6 100644 --- a/bindings/python/DEPENDENCIES.rust.tsv +++ b/bindings/python/DEPENDENCIES.rust.tsv @@ -328,6 +328,7 @@ opendal-layer-concurrent-limit@0.57.0 X opendal-layer-logging@0.57.0 X opendal-layer-retry@0.57.0 X opendal-layer-timeout@0.57.0 X +opendal-service-azblob@0.57.0 X opendal-service-azdls@0.57.0 X opendal-service-azure-common@0.57.0 X opendal-service-fs@0.57.0 X diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 530aa4cbf3..88b6965326 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -689,6 +689,30 @@ pub fn iceberg::inspect::SnapshotsTable<'a>::new(table: &'a iceberg::table::Tabl pub async fn iceberg::inspect::SnapshotsTable<'a>::scan(&self) -> iceberg::Result pub fn iceberg::inspect::SnapshotsTable<'a>::schema(&self) -> iceberg::spec::Schema pub mod iceberg::io +pub struct iceberg::io::AzblobConfig +pub iceberg::io::AzblobConfig::account_key: core::option::Option +pub iceberg::io::AzblobConfig::account_name: core::option::Option +pub iceberg::io::AzblobConfig::endpoint: core::option::Option +pub iceberg::io::AzblobConfig::sas_token: core::option::Option +impl core::clone::Clone for iceberg::io::AzblobConfig +pub fn iceberg::io::AzblobConfig::clone(&self) -> iceberg::io::AzblobConfig +impl core::cmp::Eq for iceberg::io::AzblobConfig +impl core::cmp::PartialEq for iceberg::io::AzblobConfig +pub fn iceberg::io::AzblobConfig::eq(&self, other: &iceberg::io::AzblobConfig) -> bool +impl core::convert::TryFrom<&iceberg::io::StorageConfig> for iceberg::io::AzblobConfig +pub type iceberg::io::AzblobConfig::Error = iceberg::Error +pub fn iceberg::io::AzblobConfig::try_from(config: &iceberg::io::StorageConfig) -> iceberg::Result +impl core::default::Default for iceberg::io::AzblobConfig +pub fn iceberg::io::AzblobConfig::default() -> iceberg::io::AzblobConfig +impl core::fmt::Debug for iceberg::io::AzblobConfig +pub fn iceberg::io::AzblobConfig::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result +impl core::marker::StructuralPartialEq for iceberg::io::AzblobConfig +impl iceberg::io::AzblobConfig +pub fn iceberg::io::AzblobConfig::builder() -> AzblobConfigBuilder<((), (), (), ())> +impl serde_core::ser::Serialize for iceberg::io::AzblobConfig +pub fn iceberg::io::AzblobConfig::serialize<__S>(&self, __serializer: __S) -> core::result::Result<<__S as serde_core::ser::Serializer>::Ok, <__S as serde_core::ser::Serializer>::Error> where __S: serde_core::ser::Serializer +impl<'de> serde_core::de::Deserialize<'de> for iceberg::io::AzblobConfig +pub fn iceberg::io::AzblobConfig::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub struct iceberg::io::AzdlsConfig pub iceberg::io::AzdlsConfig::account_key: core::option::Option pub iceberg::io::AzdlsConfig::account_name: core::option::Option @@ -968,6 +992,9 @@ pub fn iceberg::io::StorageConfig::clone(&self) -> iceberg::io::StorageConfig impl core::cmp::Eq for iceberg::io::StorageConfig impl core::cmp::PartialEq for iceberg::io::StorageConfig pub fn iceberg::io::StorageConfig::eq(&self, other: &iceberg::io::StorageConfig) -> bool +impl core::convert::TryFrom<&iceberg::io::StorageConfig> for iceberg::io::AzblobConfig +pub type iceberg::io::AzblobConfig::Error = iceberg::Error +pub fn iceberg::io::AzblobConfig::try_from(config: &iceberg::io::StorageConfig) -> iceberg::Result impl core::convert::TryFrom<&iceberg::io::StorageConfig> for iceberg::io::AzdlsConfig pub type iceberg::io::AzdlsConfig::Error = iceberg::Error pub fn iceberg::io::AzdlsConfig::try_from(config: &iceberg::io::StorageConfig) -> iceberg::Result @@ -1000,6 +1027,10 @@ pub const iceberg::io::ADLS_CLIENT_SECRET: &str pub const iceberg::io::ADLS_CONNECTION_STRING: &str pub const iceberg::io::ADLS_SAS_TOKEN: &str pub const iceberg::io::ADLS_TENANT_ID: &str +pub const iceberg::io::AZBLOB_ACCOUNT_KEY: &str +pub const iceberg::io::AZBLOB_ACCOUNT_NAME: &str +pub const iceberg::io::AZBLOB_ENDPOINT: &str +pub const iceberg::io::AZBLOB_SAS_TOKEN: &str pub const iceberg::io::CLIENT_REGION: &str pub const iceberg::io::GCS_ALLOW_ANONYMOUS: &str pub const iceberg::io::GCS_CREDENTIALS_JSON: &str diff --git a/crates/iceberg/src/io/storage/config/azblob.rs b/crates/iceberg/src/io/storage/config/azblob.rs new file mode 100644 index 0000000000..a39659edc8 --- /dev/null +++ b/crates/iceberg/src/io/storage/config/azblob.rs @@ -0,0 +1,120 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Azure Blob Storage configuration. +//! +//! This module provides configuration constants and types for Azure Blob Storage. + +use serde::{Deserialize, Serialize}; +use typed_builder::TypedBuilder; + +use super::StorageConfig; +use crate::Result; + +/// Azure Blob Storage endpoint URL. +pub const AZBLOB_ENDPOINT: &str = "azblob.endpoint"; +/// Azure Blob Storage account name. +pub const AZBLOB_ACCOUNT_NAME: &str = "azblob.account-name"; +/// Azure Blob Storage account key. +pub const AZBLOB_ACCOUNT_KEY: &str = "azblob.account-key"; +/// Azure Blob Storage shared access signature. +pub const AZBLOB_SAS_TOKEN: &str = "azblob.sas-token"; + +/// Azure Blob Storage configuration. +/// +/// This struct contains all the configuration options for connecting to Azure Blob Storage. +/// Use the builder pattern via `AzblobConfig::builder()` to construct instances. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize, TypedBuilder)] +pub struct AzblobConfig { + /// Endpoint URL. + #[builder(default, setter(strip_option, into))] + pub endpoint: Option, + /// Account name. + #[builder(default, setter(strip_option, into))] + pub account_name: Option, + /// Account key. + #[builder(default, setter(strip_option, into))] + pub account_key: Option, + /// SAS token. + #[builder(default, setter(strip_option, into))] + pub sas_token: Option, +} + +impl TryFrom<&StorageConfig> for AzblobConfig { + type Error = crate::Error; + + fn try_from(config: &StorageConfig) -> Result { + let props = config.props(); + + let mut cfg = AzblobConfig::default(); + + if let Some(endpoint) = props.get(AZBLOB_ENDPOINT) { + cfg.endpoint = Some(endpoint.clone()); + } + if let Some(account_name) = props.get(AZBLOB_ACCOUNT_NAME) { + cfg.account_name = Some(account_name.clone()); + } + if let Some(account_key) = props.get(AZBLOB_ACCOUNT_KEY) { + cfg.account_key = Some(account_key.clone()); + } + if let Some(sas_token) = props.get(AZBLOB_SAS_TOKEN) { + cfg.sas_token = Some(sas_token.clone()); + } + + Ok(cfg) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_azblob_config_builder() { + let config = AzblobConfig::builder() + .endpoint("https://account.blob.core.windows.net") + .account_name("myaccount") + .account_key("my-account-key") + .build(); + + assert_eq!( + config.endpoint.as_deref(), + Some("https://account.blob.core.windows.net") + ); + assert_eq!(config.account_name.as_deref(), Some("myaccount")); + assert_eq!(config.account_key.as_deref(), Some("my-account-key")); + } + + #[test] + fn test_azblob_config_from_storage_config() { + let storage_config = StorageConfig::new() + .with_prop(AZBLOB_ENDPOINT, "https://account.blob.core.windows.net") + .with_prop(AZBLOB_ACCOUNT_NAME, "myaccount") + .with_prop(AZBLOB_ACCOUNT_KEY, "my-account-key") + .with_prop(AZBLOB_SAS_TOKEN, "my-sas-token"); + + let azblob_config = AzblobConfig::try_from(&storage_config).unwrap(); + + assert_eq!( + azblob_config.endpoint.as_deref(), + Some("https://account.blob.core.windows.net") + ); + assert_eq!(azblob_config.account_name.as_deref(), Some("myaccount")); + assert_eq!(azblob_config.account_key.as_deref(), Some("my-account-key")); + assert_eq!(azblob_config.sas_token.as_deref(), Some("my-sas-token")); + } +} diff --git a/crates/iceberg/src/io/storage/config/mod.rs b/crates/iceberg/src/io/storage/config/mod.rs index 2350aab6dd..d000ea6732 100644 --- a/crates/iceberg/src/io/storage/config/mod.rs +++ b/crates/iceberg/src/io/storage/config/mod.rs @@ -30,6 +30,7 @@ //! - [`OssConfig`]: Alibaba Cloud OSS specific configuration //! - [`AzdlsConfig`]: Azure Data Lake Storage specific configuration +mod azblob; mod azdls; mod gcs; mod hf; @@ -38,6 +39,7 @@ mod s3; use std::collections::HashMap; +pub use azblob::*; pub use azdls::*; pub use gcs::*; pub use hf::*; diff --git a/crates/storage/opendal/Cargo.toml b/crates/storage/opendal/Cargo.toml index e43e7845b3..e9ac9bb0d5 100644 --- a/crates/storage/opendal/Cargo.toml +++ b/crates/storage/opendal/Cargo.toml @@ -35,10 +35,12 @@ opendal-all = [ "opendal-s3", "opendal-gcs", "opendal-oss", + "opendal-azblob", "opendal-azdls", "opendal-hf", ] +opendal-azblob = ["opendal/services-azblob"] opendal-azdls = ["opendal/services-azdls"] opendal-fs = ["opendal/services-fs"] opendal-gcs = ["opendal/services-gcs"] diff --git a/crates/storage/opendal/DEPENDENCIES.rust.tsv b/crates/storage/opendal/DEPENDENCIES.rust.tsv index 3a960f193a..a06cc258fb 100644 --- a/crates/storage/opendal/DEPENDENCIES.rust.tsv +++ b/crates/storage/opendal/DEPENDENCIES.rust.tsv @@ -257,6 +257,7 @@ opendal-layer-concurrent-limit@0.57.0 X opendal-layer-logging@0.57.0 X opendal-layer-retry@0.57.0 X opendal-layer-timeout@0.57.0 X +opendal-service-azblob@0.57.0 X opendal-service-azdls@0.57.0 X opendal-service-azure-common@0.57.0 X opendal-service-fs@0.57.0 X diff --git a/crates/storage/opendal/README.md b/crates/storage/opendal/README.md index 5db4293e67..99b59d6896 100644 --- a/crates/storage/opendal/README.md +++ b/crates/storage/opendal/README.md @@ -31,6 +31,7 @@ OpenDAL-based storage backend implementations for [Apache Iceberg Rust](https:// | Google Cloud Storage | `opendal-gcs` | ✅ Stable | Google Cloud Storage | | Hugging Face | `opendal-hf` | ✅ Stable | Hugging Face buckets and repositories | | Alibaba Cloud OSS | `opendal-oss` | 🧪 Experimental | Alibaba Cloud Object Storage Service | +| Azure Blob Storage | `opendal-azblob` | 🧪 Experimental | Azure Blob Storage without requiring HNS | | Azure Datalake | `opendal-azdls` | 🧪 Experimental | Azure Datalake Storage v2 | You can enable all stable storage backends at once using the `opendal-all` feature flag. diff --git a/crates/storage/opendal/public-api.txt b/crates/storage/opendal/public-api.txt index d8c4ecdb38..559fdd3724 100644 --- a/crates/storage/opendal/public-api.txt +++ b/crates/storage/opendal/public-api.txt @@ -2,6 +2,8 @@ pub mod iceberg_storage_opendal pub use iceberg_storage_opendal::AwsCredential pub use iceberg_storage_opendal::ProvideCredential pub enum iceberg_storage_opendal::OpenDalStorage +pub iceberg_storage_opendal::OpenDalStorage::Azblob +pub iceberg_storage_opendal::OpenDalStorage::Azblob::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::Azdls pub iceberg_storage_opendal::OpenDalStorage::Azdls::config: alloc::sync::Arc pub iceberg_storage_opendal::OpenDalStorage::Gcs @@ -36,6 +38,7 @@ pub fn iceberg_storage_opendal::OpenDalStorage::serialize<__S>(&self, __serializ impl<'de> serde_core::de::Deserialize<'de> for iceberg_storage_opendal::OpenDalStorage pub fn iceberg_storage_opendal::OpenDalStorage::deserialize<__D>(__deserializer: __D) -> core::result::Result::Error> where __D: serde_core::de::Deserializer<'de> pub enum iceberg_storage_opendal::OpenDalStorageFactory +pub iceberg_storage_opendal::OpenDalStorageFactory::Azblob pub iceberg_storage_opendal::OpenDalStorageFactory::Azdls pub iceberg_storage_opendal::OpenDalStorageFactory::Fs pub iceberg_storage_opendal::OpenDalStorageFactory::Gcs diff --git a/crates/storage/opendal/src/azblob.rs b/crates/storage/opendal/src/azblob.rs new file mode 100644 index 0000000000..c8960fb41a --- /dev/null +++ b/crates/storage/opendal/src/azblob.rs @@ -0,0 +1,138 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Azure Blob Storage properties. + +use std::collections::HashMap; + +use iceberg::io::{AZBLOB_ACCOUNT_KEY, AZBLOB_ACCOUNT_NAME, AZBLOB_ENDPOINT, AZBLOB_SAS_TOKEN}; +use iceberg::{Error, ErrorKind, Result}; +use opendal::Operator; +use opendal::services::AzblobConfig; +use url::Url; + +use crate::utils::from_opendal_error; + +/// Parse azblob.* prefixed configuration properties. +pub(crate) fn azblob_config_parse(mut properties: HashMap) -> Result { + let mut config = AzblobConfig::default(); + + if let Some(endpoint) = properties.remove(AZBLOB_ENDPOINT) { + config.endpoint = Some(endpoint); + } + if let Some(account_name) = properties.remove(AZBLOB_ACCOUNT_NAME) { + config.account_name = Some(account_name); + } + if let Some(account_key) = properties.remove(AZBLOB_ACCOUNT_KEY) { + config.account_key = Some(account_key); + } + if let Some(sas_token) = properties.remove(AZBLOB_SAS_TOKEN) { + config.sas_token = Some(sas_token); + } + + Ok(config) +} + +/// Build an OpenDAL operator for an `azblob:///` URL. +pub(crate) fn azblob_create_operator<'a>( + path: &'a str, + config: &AzblobConfig, +) -> Result<(Operator, &'a str)> { + let (container, relative_path) = parse_azblob_path(path)?; + let mut config = config.clone(); + config.container = container; + + let operator = Operator::from_config(config).map_err(from_opendal_error)?; + Ok((operator, relative_path)) +} + +/// Extract the path relative to the Azure Blob container. +pub(crate) fn azblob_relative_path(path: &str) -> Result<&str> { + Ok(parse_azblob_path(path)?.1) +} + +fn parse_azblob_path(path: &str) -> Result<(String, &str)> { + let url = Url::parse(path)?; + let container = url.host_str().ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid azblob url: {path}, container is required"), + ) + })?; + let prefix = format!("azblob://{container}/"); + let relative_path = path.strip_prefix(&prefix).ok_or_else(|| { + Error::new( + ErrorKind::DataInvalid, + format!("Invalid azblob url: {path}, should start with {prefix}"), + ) + })?; + + Ok((container.to_string(), relative_path)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_azblob_config_parse() { + let config = azblob_config_parse(HashMap::from([ + ( + AZBLOB_ENDPOINT.to_string(), + "https://account.blob.core.windows.net".to_string(), + ), + (AZBLOB_ACCOUNT_NAME.to_string(), "account".to_string()), + (AZBLOB_ACCOUNT_KEY.to_string(), "key".to_string()), + (AZBLOB_SAS_TOKEN.to_string(), "token".to_string()), + ])) + .unwrap(); + + assert_eq!( + config.endpoint.as_deref(), + Some("https://account.blob.core.windows.net") + ); + assert_eq!(config.account_name.as_deref(), Some("account")); + assert_eq!(config.account_key.as_deref(), Some("key")); + assert_eq!(config.sas_token.as_deref(), Some("token")); + } + + #[test] + fn test_azblob_create_operator() { + let config = AzblobConfig { + endpoint: Some("https://account.blob.core.windows.net".to_string()), + account_name: Some("account".to_string()), + ..Default::default() + }; + + let (operator, relative_path) = + azblob_create_operator("azblob://container/path/to/file.parquet", &config).unwrap(); + + assert_eq!(operator.info().name(), "container"); + assert_eq!(relative_path, "path/to/file.parquet"); + } + + #[test] + fn test_azblob_relative_path_root() { + assert_eq!(azblob_relative_path("azblob://container/").unwrap(), ""); + } + + #[test] + fn test_azblob_relative_path_rejects_invalid_url() { + assert!(azblob_relative_path("s3://container/path").is_err()); + assert!(azblob_relative_path("azblob:///path").is_err()); + } +} diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 768d3ff77f..6cca1e4e88 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -43,6 +43,14 @@ use opendal::layers::{RetryLayer, TimeoutLayer}; use serde::{Deserialize, Serialize}; use utils::from_opendal_error; +cfg_if! { + if #[cfg(feature = "opendal-azblob")] { + mod azblob; + use azblob::*; + use opendal::services::AzblobConfig; + } +} + cfg_if! { if #[cfg(feature = "opendal-azdls")] { mod azdls; @@ -125,6 +133,9 @@ pub enum OpenDalStorageFactory { /// OSS storage factory. #[cfg(feature = "opendal-oss")] Oss, + /// Azure Blob Storage factory. + #[cfg(feature = "opendal-azblob")] + Azblob, /// Azure Data Lake Storage factory. #[cfg(feature = "opendal-azdls")] Azdls, @@ -159,6 +170,10 @@ impl StorageFactory for OpenDalStorageFactory { OpenDalStorageFactory::Oss => Ok(Arc::new(OpenDalStorage::Oss { config: oss_config_parse(config.props().clone())?.into(), })), + #[cfg(feature = "opendal-azblob")] + OpenDalStorageFactory::Azblob => Ok(Arc::new(OpenDalStorage::Azblob { + config: azblob_config_parse(config.props().clone())?.into(), + })), #[cfg(feature = "opendal-azdls")] OpenDalStorageFactory::Azdls => Ok(Arc::new(OpenDalStorage::Azdls { config: azdls_config_parse(config.props().clone())?.into(), @@ -173,6 +188,7 @@ impl StorageFactory for OpenDalStorageFactory { not(feature = "opendal-s3"), not(feature = "opendal-gcs"), not(feature = "opendal-oss"), + not(feature = "opendal-azblob"), not(feature = "opendal-azdls"), not(feature = "opendal-hf"), ))] @@ -223,6 +239,14 @@ pub enum OpenDalStorage { /// OSS configuration. config: Arc, }, + /// Azure Blob Storage variant. + /// + /// Accepts paths of the form `azblob:///`. + #[cfg(feature = "opendal-azblob")] + Azblob { + /// Azure Blob Storage configuration. + config: Arc, + }, /// Azure Data Lake Storage variant. /// /// Accepts paths of the form @@ -335,6 +359,8 @@ impl OpenDalStorage { )); } } + #[cfg(feature = "opendal-azblob")] + OpenDalStorage::Azblob { config } => azblob_create_operator(path, config)?, #[cfg(feature = "opendal-azdls")] OpenDalStorage::Azdls { config } => azdls_create_operator(path, config)?, #[cfg(feature = "opendal-hf")] @@ -344,6 +370,7 @@ impl OpenDalStorage { not(feature = "opendal-fs"), not(feature = "opendal-gcs"), not(feature = "opendal-oss"), + not(feature = "opendal-azblob"), not(feature = "opendal-azdls"), not(feature = "opendal-hf"), ))] @@ -453,6 +480,8 @@ impl OpenDalStorage { )) } } + #[cfg(feature = "opendal-azblob")] + OpenDalStorage::Azblob { .. } => azblob_relative_path(path), #[cfg(feature = "opendal-azdls")] OpenDalStorage::Azdls { config } => { let azure_path = path.parse::()?; @@ -472,6 +501,7 @@ impl OpenDalStorage { not(feature = "opendal-fs"), not(feature = "opendal-gcs"), not(feature = "opendal-oss"), + not(feature = "opendal-azblob"), not(feature = "opendal-azdls"), not(feature = "opendal-hf"), ))] @@ -750,6 +780,21 @@ mod tests { ); } + #[cfg(feature = "opendal-azblob")] + #[test] + fn test_relativize_path_azblob() { + let storage = OpenDalStorage::Azblob { + config: Arc::new(AzblobConfig::default()), + }; + + assert_eq!( + storage + .relativize_path("azblob://container/path/to/file.parquet") + .unwrap(), + "path/to/file.parquet" + ); + } + #[cfg(feature = "opendal-azdls")] #[test] fn test_relativize_path_azdls() { diff --git a/crates/storage/opendal/src/resolving.rs b/crates/storage/opendal/src/resolving.rs index 86993220a8..ca0a32d3f9 100644 --- a/crates/storage/opendal/src/resolving.rs +++ b/crates/storage/opendal/src/resolving.rs @@ -46,6 +46,7 @@ pub const SCHEME_S3N: &str = "s3n"; pub const SCHEME_GS: &str = "gs"; pub const SCHEME_GCS: &str = "gcs"; pub const SCHEME_OSS: &str = "oss"; +pub const SCHEME_AZBLOB: &str = "azblob"; pub const SCHEME_ABFSS: &str = "abfss"; pub const SCHEME_ABFS: &str = "abfs"; pub const SCHEME_WASBS: &str = "wasbs"; @@ -60,6 +61,7 @@ fn parse_scheme(scheme: &str) -> Result<&'static str> { SCHEME_S3 | SCHEME_S3A | SCHEME_S3N => Ok("s3"), SCHEME_GS | SCHEME_GCS => Ok("gcs"), SCHEME_OSS => Ok("oss"), + SCHEME_AZBLOB => Ok("azblob"), SCHEME_ABFSS | SCHEME_ABFS | SCHEME_WASBS | SCHEME_WASB => Ok("azdls"), SCHEME_HF => Ok("hf"), s => Err(Error::new( @@ -109,6 +111,13 @@ fn build_storage_for_scheme( config: Arc::new(config), }) } + #[cfg(feature = "opendal-azblob")] + "azblob" => { + let config = crate::azblob::azblob_config_parse(props.clone())?; + Ok(OpenDalStorage::Azblob { + config: Arc::new(config), + }) + } #[cfg(feature = "opendal-azdls")] "azdls" => { let config = crate::azdls::azdls_config_parse(props.clone())?; @@ -371,4 +380,13 @@ mod tests { "abfss and abfs should share one instance" ); } + + #[cfg(feature = "opendal-azblob")] + #[test] + fn test_resolve_azblob() { + let storage = empty_resolving_storage(); + + let azblob = storage.resolve("azblob://container/path").unwrap(); + assert!(matches!(azblob.as_ref(), OpenDalStorage::Azblob { .. })); + } } diff --git a/crates/storage/opendal/tests/file_io_azblob_test.rs b/crates/storage/opendal/tests/file_io_azblob_test.rs new file mode 100644 index 0000000000..7875293cdf --- /dev/null +++ b/crates/storage/opendal/tests/file_io_azblob_test.rs @@ -0,0 +1,120 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Integration tests for FileIO Azure Blob Storage. +//! +//! These tests require a real Azure Blob Storage account and are skipped when +//! the `TEST_ENV_*` variables used by storage integration tests are not set. + +#[cfg(feature = "opendal-azblob")] +mod tests { + use std::sync::Arc; + use std::time::{SystemTime, UNIX_EPOCH}; + + use bytes::Bytes; + use iceberg::io::{AZBLOB_ACCOUNT_KEY, AZBLOB_ACCOUNT_NAME, AZBLOB_ENDPOINT, FileIOBuilder}; + use iceberg_storage_opendal::OpenDalStorageFactory; + + macro_rules! require_env { + ($var:expr) => { + match std::env::var($var) { + Ok(value) if !value.is_empty() => value, + _ => { + eprintln!("Skipping Azure Blob test: {} not set", $var); + return; + } + } + }; + } + + #[tokio::test] + async fn test_file_io_azblob_non_hns_account() { + let provider = require_env!("TEST_ENV_CLOUD_PROVIDER"); + if !provider.eq_ignore_ascii_case("azure") { + eprintln!("Skipping Azure Blob test: TEST_ENV_CLOUD_PROVIDER is not azure"); + return; + } + + let account_name = require_env!("TEST_ENV_ACCESS_KEY"); + let account_key = require_env!("TEST_ENV_SECRET_KEY"); + let endpoint_suffix = require_env!("TEST_ENV_ADDRESS"); + let container = require_env!("TEST_ENV_BUCKET_NAME"); + let use_ssl = require_env!("TEST_ENV_USE_SSL"); + let http_scheme = if use_ssl.eq_ignore_ascii_case("true") { + "https" + } else { + "http" + }; + let endpoint = format!( + "{http_scheme}://{account_name}.blob.{}", + endpoint_suffix.trim_start_matches('.') + ); + + let file_io = FileIOBuilder::new(Arc::new(OpenDalStorageFactory::Azblob)) + .with_props(vec![ + (AZBLOB_ENDPOINT, endpoint), + (AZBLOB_ACCOUNT_NAME, account_name), + (AZBLOB_ACCOUNT_KEY, account_key), + ]) + .build(); + + let nonce = SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos(); + let prefix = format!( + "azblob://{container}/iceberg-rust-tests/non-hns-{}-{nonce}", + std::process::id() + ); + let file = format!("{prefix}/file.bin"); + let nested_file = format!("{prefix}/nested/file.bin"); + let content = Bytes::from_static(b"0123456789"); + + let _ = file_io.delete_prefix(&prefix).await; + assert!(!file_io.exists(&file).await.unwrap()); + + file_io + .new_output(&file) + .unwrap() + .write(content.clone()) + .await + .unwrap(); + assert!(file_io.exists(&file).await.unwrap()); + + let input = file_io.new_input(&file).unwrap(); + assert_eq!(input.metadata().await.unwrap().size, content.len() as u64); + assert_eq!(input.read().await.unwrap(), content); + assert_eq!( + input.reader().await.unwrap().read(2..6).await.unwrap(), + Bytes::from_static(b"2345") + ); + + file_io.delete(&file).await.unwrap(); + assert!(!file_io.exists(&file).await.unwrap()); + + file_io + .new_output(&nested_file) + .unwrap() + .write(Bytes::from_static(b"delete-prefix")) + .await + .unwrap(); + assert!(file_io.exists(&nested_file).await.unwrap()); + + file_io.delete_prefix(&prefix).await.unwrap(); + assert!(!file_io.exists(&nested_file).await.unwrap()); + } +}