azblob

Author: chenzl25

Cargo.lock

@@ -30,14 +30,15 @@ repository = { workspace = true }
 
 [features]
 default = ["storage-memory", "storage-fs", "storage-s3", "tokio"]
-storage-all = ["storage-memory", "storage-fs", "storage-s3", "storage-gcs"]
+storage-all = ["storage-memory", "storage-fs", "storage-s3", "storage-gcs", "storage-azblob"]
 
 storage-azdls = ["opendal/services-azdls"]
 storage-fs = ["opendal/services-fs"]
 storage-gcs = ["opendal/services-gcs"]
 storage-memory = ["opendal/services-memory"]
 storage-oss = ["opendal/services-oss"]
 storage-s3 = ["opendal/services-s3", "reqsign"]
+storage-azblob = ["opendal/services-azblob"]
 
 smol = ["dep:smol"]
 tokio = ["tokio/rt-multi-thread"]

crates/iceberg/Cargo.toml

@@ -44,6 +44,7 @@ use crate::{Error, ErrorKind, Result};
 /// | GCS                | `storage-gcs`     | `gs`, `gcs`                      | `gs://<bucket>/path/to/file`  |
 /// | OSS                | `storage-oss`     | `oss`                            | `oss://<bucket>/path/to/file` |
 /// | Azure Datalake     | `storage-azdls`   | `abfs`, `abfss`, `wasb`, `wasbs` | `abfs://<filesystem>@<account>.dfs.core.windows.net/path/to/file` or `wasb://<container>@<account>.blob.core.windows.net/path/to/file` |
+/// | AZBLOB             | `storage-azblob`  | `azblob`                         | `azblob://<container>/path/to/file` |
 #[derive(Clone, Debug)]
 pub struct FileIO {
     builder: FileIOBuilder,

crates/iceberg/src/io/file_io.rs

@@ -98,6 +98,11 @@ pub use storage_oss::*;
 #[cfg(feature = "storage-s3")]
 pub use storage_s3::*;
 
+#[cfg(feature = "storage-azblob")]
+mod storage_azblob;
+#[cfg(feature = "storage-azblob")]
+pub use storage_azblob::*;
+
 pub(crate) fn is_truthy(value: &str) -> bool {
     ["true", "t", "1", "on"].contains(&value.to_lowercase().as_str())
 }

crates/iceberg/src/io/mod.rs

@@ -18,6 +18,8 @@
 use std::sync::Arc;
 
 use opendal::layers::RetryLayer;
+#[cfg(feature = "storage-azblob")]
+use opendal::services::AzblobConfig;
 #[cfg(feature = "storage-azdls")]
 use opendal::services::AzdlsConfig;
 #[cfg(feature = "storage-gcs")]
@@ -53,6 +55,8 @@ pub(crate) enum Storage {
     },
     #[cfg(feature = "storage-gcs")]
     Gcs { config: Arc<GcsConfig> },
+    #[cfg(feature = "storage-azblob")]
+    Azblob { config: Arc<AzblobConfig> },
     #[cfg(feature = "storage-oss")]
     Oss { config: Arc<OssConfig> },
     /// Expects paths of the form
@@ -90,6 +94,10 @@ impl Storage {
             Scheme::Gcs => Ok(Self::Gcs {
                 config: super::gcs_config_parse(props)?.into(),
             }),
+            #[cfg(feature = "storage-azblob")]
+            Scheme::Azblob => Ok(Self::Azblob {
+                config: super::azblob_config_parse(props)?.into(),
+            }),
             #[cfg(feature = "storage-oss")]
             Scheme::Oss => Ok(Self::Oss {
                 config: super::oss_config_parse(props)?.into(),
@@ -179,6 +187,19 @@ impl Storage {
                     ))
                 }
             }
+            #[cfg(feature = "storage-azblob")]
+            Storage::Azblob { config } => {
+                let operator = super::azblob_config_build(config, path)?;
+                let prefix = format!("azblob://{}/", operator.info().name());
+                if path.starts_with(&prefix) {
+                    Ok((operator, &path[prefix.len()..]))
+                } else {
+                    Err(Error::new(
+                        ErrorKind::DataInvalid,
+                        format!("Invalid azblob url: {}, should start with {}", path, prefix),
+                    ))
+                }
+            }
             #[cfg(feature = "storage-oss")]
             Storage::Oss { config } => {
                 let op = super::oss_config_build(config, path)?;
@@ -203,6 +224,7 @@ impl Storage {
                 not(feature = "storage-s3"),
                 not(feature = "storage-fs"),
                 not(feature = "storage-gcs"),
+                not(feature = "storage-azblob"),
                 not(feature = "storage-oss"),
                 not(feature = "storage-azdls"),
             ))]

crates/iceberg/src/io/storage.rs

@@ -0,0 +1,63 @@
+// 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 opendal::Operator;
+use opendal::services::AzblobConfig;
+use url::Url;
+
+use crate::{Error, ErrorKind, Result};
+
+/// Azure blob account name.
+pub const AZBLOB_ACCOUNT_NAME: &str = "azblob.account-name";
+/// Azure blob account key.
+pub const AZBLOB_ACCOUNT_KEY: &str = "azblob.account-key";
+/// Azure blob account endpoint.
+pub const AZBLOB_ENDPOINT: &str = "azblob.endpoint";
+
+/// Parse iceberg properties to [`AzblobConfig`].
+pub(crate) fn azblob_config_parse(mut m: HashMap<String, String>) -> Result<AzblobConfig> {
+    let mut cfg = AzblobConfig::default();
+
+    if let Some(account_name) = m.remove(AZBLOB_ACCOUNT_NAME) {
+        cfg.account_name = Some(account_name);
+    };
+    if let Some(account_key) = m.remove(AZBLOB_ACCOUNT_KEY) {
+        cfg.account_key = Some(account_key);
+    };
+    if let Some(endpoint) = m.remove(AZBLOB_ENDPOINT) {
+        cfg.endpoint = Some(endpoint);
+    };
+
+    Ok(cfg)
+}
+
+/// Build a new OpenDAL [`Operator`] based on a provided [`AzblobConfig`].
+pub(crate) fn azblob_config_build(cfg: &AzblobConfig, path: &str) -> Result<Operator> {
+    let url = Url::parse(path)?;
+    let container = url.host_str().ok_or_else(|| {
+        Error::new(
+            ErrorKind::DataInvalid,
+            format!("Invalid azblob url: {}, container is required", path),
+        )
+    })?;
+
+    let mut cfg = cfg.clone();
+    cfg.container = container.to_string();
+    Ok(Operator::from_config(cfg)?.finish())
+}