feat(storage-azdls): Add Azure Datalake Storage support

Cargo.toml

@@ -87,7 +87,7 @@ motore-macros = "0.4.3"
 murmur3 = "0.5.2"
 num-bigint = "0.4.6"
 once_cell = "1.20"
-opendal = "0.53.0"
+opendal = "0.53.3"
 ordered-float = "4"
 parquet = "55"
 pilota = "0.11.2"

crates/iceberg/Cargo.toml

@@ -32,6 +32,7 @@ repository = { workspace = true }
 default = ["storage-memory", "storage-fs", "storage-s3", "tokio"]
 storage-all = ["storage-memory", "storage-fs", "storage-s3", "storage-gcs"]
 
+storage-azdls = ["opendal/services-azdls"]
 storage-fs = ["opendal/services-fs"]
 storage-gcs = ["opendal/services-gcs"]
 storage-memory = ["opendal/services-memory"]

crates/iceberg/README.md

@@ -62,17 +62,18 @@ async fn main() -> Result<()> {
 
 Iceberg Rust provides various storage backends through feature flags. Here are the currently supported storage backends:
 
-| Storage Backend | Feature Flag | Status | Description |
-|----------------|--------------|--------|-------------|
-| Memory | `storage-memory` | ✅ Stable | In-memory storage for testing and development |
-| Local Filesystem | `storage-fs` | ✅ Stable | Local filesystem storage |
-| Amazon S3 | `storage-s3` | ✅ Stable | Amazon S3 storage |
-| Google Cloud Storage | `storage-gcs` | ✅ Stable | Google Cloud Storage |
-| Alibaba Cloud OSS | `storage-oss` | 🧪 Experimental | Alibaba Cloud Object Storage Service |
+| Storage Backend      | Feature Flag     | Status         | Description                                   |
+| -------------------- | ---------------- | -------------- | --------------------------------------------- |
+| Memory               | `storage-memory` | ✅ Stable       | In-memory storage for testing and development |
+| Local Filesystem     | `storage-fs`     | ✅ Stable       | Local filesystem storage                      |
+| Amazon S3            | `storage-s3`     | ✅ Stable       | Amazon S3 storage                             |
+| Google Cloud Storage | `storage-gcs`    | ✅ Stable       | Google Cloud Storage                          |
+| Alibaba Cloud OSS    | `storage-oss`    | 🧪 Experimental | Alibaba Cloud Object Storage Service          |
+| Azure Datalake       | `storage-azdls`  | 🧪 Experimental | Azure Datalake Storage v2                     |
 
 You can enable all stable storage backends at once using the `storage-all` feature flag. 
 
-> Note that `storage-oss` is currently experimental and not included in `storage-all`.
+> Note that `storage-oss` and `storage-azdls` are currently experimental and not included in `storage-all`.
 
 Example usage in `Cargo.toml`:
 

crates/iceberg/src/io/file_io.rs

@@ -35,12 +35,14 @@ use crate::{Error, ErrorKind, Result};
 ///
 /// Supported storages:
 ///
-/// | Storage            | Feature Flag     | Schemes    |
-/// |--------------------|-------------------|------------|
-/// | Local file system  | `storage-fs`      | `file`     |
-/// | Memory             | `storage-memory`  | `memory`   |
-/// | S3                 | `storage-s3`      | `s3`, `s3a`|
-/// | GCS                | `storage-gcs`     | `gs`, `gcs`|
+/// | Storage            | Feature Flag      | Expected Path Format             | Schemes                       |
+/// |--------------------|-------------------|----------------------------------| ------------------------------|
+/// | Local file system  | `storage-fs`      | `file`                           | `file://path/to/file`         |
+/// | Memory             | `storage-memory`  | `memory`                         | `memory://path/to/file`       |
+/// | S3                 | `storage-s3`      | `s3`, `s3a`                      | `s3://<bucket>/path/to/file`  |
+/// | 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` |
 #[derive(Clone, Debug)]
 pub struct FileIO {
     builder: FileIOBuilder,

crates/iceberg/src/io/mod.rs

@@ -67,32 +67,36 @@
 //! - `new_output`: Create output file for writing.
 
 mod file_io;
+mod storage;
+
 pub use file_io::*;
+pub(crate) mod object_cache;
 
-mod storage;
+#[cfg(feature = "storage-azdls")]
+mod storage_azdls;
+#[cfg(feature = "storage-fs")]
+mod storage_fs;
+#[cfg(feature = "storage-gcs")]
+mod storage_gcs;
 #[cfg(feature = "storage-memory")]
 mod storage_memory;
-#[cfg(feature = "storage-memory")]
-use storage_memory::*;
+#[cfg(feature = "storage-oss")]
+mod storage_oss;
 #[cfg(feature = "storage-s3")]
 mod storage_s3;
-#[cfg(feature = "storage-s3")]
-pub use storage_s3::*;
-pub(crate) mod object_cache;
-#[cfg(feature = "storage-fs")]
-mod storage_fs;
 
+#[cfg(feature = "storage-azdls")]
+pub use storage_azdls::*;
 #[cfg(feature = "storage-fs")]
 use storage_fs::*;
 #[cfg(feature = "storage-gcs")]
-mod storage_gcs;
-#[cfg(feature = "storage-gcs")]
 pub use storage_gcs::*;
-
-#[cfg(feature = "storage-oss")]
-mod storage_oss;
+#[cfg(feature = "storage-memory")]
+use storage_memory::*;
 #[cfg(feature = "storage-oss")]
 pub use storage_oss::*;
+#[cfg(feature = "storage-s3")]
+pub use storage_s3::*;
 
 pub(crate) fn is_truthy(value: &str) -> bool {
     ["true", "t", "1", "on"].contains(&value.to_lowercase().as_str())

crates/iceberg/src/io/storage.rs

@@ -18,6 +18,8 @@
 use std::sync::Arc;
 
 use opendal::layers::RetryLayer;
+#[cfg(feature = "storage-azdls")]
+use opendal::services::AzdlsConfig;
 #[cfg(feature = "storage-gcs")]
 use opendal::services::GcsConfig;
 #[cfg(feature = "storage-oss")]
@@ -26,6 +28,8 @@ use opendal::services::OssConfig;
 use opendal::services::S3Config;
 use opendal::{Operator, Scheme};
 
+#[cfg(feature = "storage-azdls")]
+use super::AzureStorageScheme;
 use super::FileIOBuilder;
 use crate::{Error, ErrorKind};
 
@@ -36,17 +40,28 @@ pub(crate) enum Storage {
     Memory(Operator),
     #[cfg(feature = "storage-fs")]
     LocalFs,
+    /// Expects paths of the form `s3[a]://<bucket>/<path>`.
     #[cfg(feature = "storage-s3")]
     S3 {
         /// s3 storage could have `s3://` and `s3a://`.
         /// Storing the scheme string here to return the correct path.
-        scheme_str: String,
+        configured_scheme: String,
         config: Arc<S3Config>,
     },
-    #[cfg(feature = "storage-oss")]
-    Oss { config: Arc<OssConfig> },
     #[cfg(feature = "storage-gcs")]
     Gcs { config: Arc<GcsConfig> },
+    #[cfg(feature = "storage-oss")]
+    Oss { config: Arc<OssConfig> },
+    /// Expects paths of the form
+    /// `abfs[s]://<filesystem>@<account>.dfs.<endpoint-suffix>/<path>` or
+    /// `wasb[s]://<container>@<account>.blob.<endpoint-suffix>/<path>`.
+    #[cfg(feature = "storage-azdls")]
+    Azdls {
+        /// Because Azdls accepts multiple possible schemes, we store the full
+        /// passed scheme here to later validate schemes passed via paths.
+        configured_scheme: AzureStorageScheme,
+        config: Arc<AzdlsConfig>,
+    },
 }
 
 impl Storage {
@@ -62,7 +77,7 @@ impl Storage {
             Scheme::Fs => Ok(Self::LocalFs),
             #[cfg(feature = "storage-s3")]
             Scheme::S3 => Ok(Self::S3 {
-                scheme_str,
+                configured_scheme: scheme_str,
                 config: super::s3_config_parse(props)?.into(),
             }),
             #[cfg(feature = "storage-gcs")]
@@ -73,6 +88,14 @@ impl Storage {
             Scheme::Oss => Ok(Self::Oss {
                 config: super::oss_config_parse(props)?.into(),
             }),
+            #[cfg(feature = "storage-azdls")]
+            Scheme::Azdls => {
+                let scheme = scheme_str.parse::<AzureStorageScheme>()?;
+                Ok(Self::Azdls {
+                    config: super::azdls_config_parse(props)?.into(),
+                    configured_scheme: scheme,
+                })
+            }
             // Update doc on [`FileIO`] when adding new schemes.
             _ => Err(Error::new(
                 ErrorKind::FeatureUnsupported,
@@ -118,12 +141,15 @@ impl Storage {
                 }
             }
             #[cfg(feature = "storage-s3")]
-            Storage::S3 { scheme_str, config } => {
+            Storage::S3 {
+                configured_scheme,
+                config,
+            } => {
                 let op = super::s3_config_build(config, path)?;
                 let op_info = op.info();
 
                 // Check prefix of s3 path.
-                let prefix = format!("{}://{}/", scheme_str, op_info.name());
+                let prefix = format!("{}://{}/", configured_scheme, op_info.name());
                 if path.starts_with(&prefix) {
                     Ok((op, &path[prefix.len()..]))
                 } else {
@@ -133,7 +159,6 @@ impl Storage {
                     ))
                 }
             }
-
             #[cfg(feature = "storage-gcs")]
             Storage::Gcs { config } => {
                 let operator = super::gcs_config_build(config, path)?;
@@ -162,11 +187,17 @@ impl Storage {
                     ))
                 }
             }
+            #[cfg(feature = "storage-azdls")]
+            Storage::Azdls {
+                configured_scheme,
+                config,
+            } => super::azdls_create_operator(path, config, configured_scheme),
             #[cfg(all(
                 not(feature = "storage-s3"),
                 not(feature = "storage-fs"),
                 not(feature = "storage-gcs"),
-                not(feature = "storage-oss")
+                not(feature = "storage-oss"),
+                not(feature = "storage-azdls"),
             ))]
             _ => Err(Error::new(
                 ErrorKind::FeatureUnsupported,
@@ -189,6 +220,7 @@ impl Storage {
             "s3" | "s3a" => Ok(Scheme::S3),
             "gs" | "gcs" => Ok(Scheme::Gcs),
             "oss" => Ok(Scheme::Oss),
+            "abfss" | "abfs" | "wasbs" | "wasb" => Ok(Scheme::Azdls),
             s => Ok(s.parse::<Scheme>()?),
         }
     }