From 5620efb09740ff1b106a7c978b6d08bb6e9c91af Mon Sep 17 00:00:00 2001 From: jiaqizho Date: Mon, 27 Jul 2026 16:08:33 +0800 Subject: [PATCH 1/2] feat(storage-azblob): Add Azure Blob Storage support Add a dedicated OpenDAL Azblob backend for Azure Blob Storage accounts without hierarchical namespaces. Support azblob:/// locations with endpoint, account name, account key, and SAS token configuration. Enable selection through both OpenDalStorageFactory and OpenDalResolvingStorage while keeping existing ABFS and WASB locations routed to Azdls. Add unit coverage and an environment-gated integration test for file operations against a non-HNS Azure account. Signed-off-by: jiaqizho --- Cargo.lock | 22 +++ bindings/python/DEPENDENCIES.rust.tsv | 1 + crates/iceberg/public-api.txt | 4 + .../iceberg/src/io/storage/config/azblob.rs | 27 ++++ crates/iceberg/src/io/storage/config/mod.rs | 2 + crates/storage/opendal/Cargo.toml | 2 + crates/storage/opendal/DEPENDENCIES.rust.tsv | 1 + crates/storage/opendal/README.md | 1 + crates/storage/opendal/public-api.txt | 3 + crates/storage/opendal/src/azblob.rs | 140 ++++++++++++++++++ crates/storage/opendal/src/lib.rs | 45 ++++++ crates/storage/opendal/src/resolving.rs | 18 +++ .../opendal/tests/file_io_azblob_test.rs | 120 +++++++++++++++ 13 files changed, 386 insertions(+) create mode 100644 crates/iceberg/src/io/storage/config/azblob.rs create mode 100644 crates/storage/opendal/src/azblob.rs create mode 100644 crates/storage/opendal/tests/file_io_azblob_test.rs diff --git a/Cargo.lock b/Cargo.lock index 0f264cc1f5..5d1717122d 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.57.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0030644366ef5d8cbe3a4a5822bf99a4aafddc1666e9d24b44d158d9062fc76a" +dependencies = [ + "base64", + "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", + "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..1dbd9d9e8b 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -1000,6 +1000,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..18198da29d --- /dev/null +++ b/crates/iceberg/src/io/storage/config/azblob.rs @@ -0,0 +1,27 @@ +// 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. + +/// 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"; 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..9b7246a412 --- /dev/null +++ b/crates/storage/opendal/src/azblob.rs @@ -0,0 +1,140 @@ +// 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)? + .finish(); + 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()); + } +} From eb1431d282db9a19cdfcd4ca8ae3c8f852d6c34d Mon Sep 17 00:00:00 2001 From: jiaqizho Date: Thu, 6 Aug 2026 17:25:41 +0800 Subject: [PATCH 2/2] address comments and rebase Signed-off-by: jiaqizho --- Cargo.lock | 8 +- crates/iceberg/public-api.txt | 27 ++++++ .../iceberg/src/io/storage/config/azblob.rs | 93 +++++++++++++++++++ crates/storage/opendal/src/azblob.rs | 4 +- 4 files changed, 125 insertions(+), 7 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 5d1717122d..05c1b8c9e3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5245,17 +5245,17 @@ dependencies = [ [[package]] name = "opendal-service-azblob" -version = "0.57.0" +version = "0.58.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0030644366ef5d8cbe3a4a5822bf99a4aafddc1666e9d24b44d158d9062fc76a" +checksum = "b3310fbbb48f111c6f590473c2cd15e1b7f8e384444b0d4e328f0464c864d767" dependencies = [ - "base64", + "base64 0.23.0", "bytes", "http 1.4.2", "log", "opendal-core", "opendal-service-azure-common", - "quick-xml 0.39.4", + "quick-xml 0.41.0", "reqsign-azure-storage", "reqsign-core", "reqsign-file-read-tokio", diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index 1dbd9d9e8b..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 diff --git a/crates/iceberg/src/io/storage/config/azblob.rs b/crates/iceberg/src/io/storage/config/azblob.rs index 18198da29d..a39659edc8 100644 --- a/crates/iceberg/src/io/storage/config/azblob.rs +++ b/crates/iceberg/src/io/storage/config/azblob.rs @@ -16,6 +16,14 @@ // 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"; @@ -25,3 +33,88 @@ pub const AZBLOB_ACCOUNT_NAME: &str = "azblob.account-name"; 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/storage/opendal/src/azblob.rs b/crates/storage/opendal/src/azblob.rs index 9b7246a412..c8960fb41a 100644 --- a/crates/storage/opendal/src/azblob.rs +++ b/crates/storage/opendal/src/azblob.rs @@ -56,9 +56,7 @@ pub(crate) fn azblob_create_operator<'a>( let mut config = config.clone(); config.container = container; - let operator = Operator::from_config(config) - .map_err(from_opendal_error)? - .finish(); + let operator = Operator::from_config(config).map_err(from_opendal_error)?; Ok((operator, relative_path)) }