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

## Which issue does this PR close?

- Closes #1360.

## What changes are included in this PR?

This PR adds an integration for the Azure Datalake storage service. At
it's core, it adds parsing logic for configuration properties. The
finished config struct is simply passed down to OpenDAL. In addition it
adds logic to parse fully qualified file URIs, and matches it against
expected (previously configured) values.

It also creates a new `Storage::Azdls` enum variant based on OpenDAL's
existing `Scheme::Azdls` enum variant. It then fits the parsing logic
into the existing framework to build the storage integration from an
`io::FileIOBuilder`.

### Note on WASB support
Other Iceberg ADLS integrations ([pyiceberg +
Java](https://github.com/apache/iceberg-go/pull/313/files#r2021460617))
also support the `wasb://` and `wasbs://` schemes.
WASB refers to a client-side implementation of hierarchical namespaces
on top of Blob Storage. ADLS(v2) on the other hand is a service offered
by Azure, also built on top of Blob Storage.
IIUC we can accept both schemes because objects written to Blob Storage
via `wasb://` will also be accessible via `adfs://` (which operates on
the same Blob Storage).
Even though the URIs slightly differ in format when they refer to the
same object, we can largely reuse existing logic.
```diff
-wasb[s]://<containername>@<accountname>.blob.core.windows.net/<path>
+adfs[s]://<filesystemname>@<accountname>.dfs.core.windows.net/<path>
```

## Are these changes tested?

### Unit
I added minor unit tests to validate the configuration property parsing
logic.

### Integration
I decided **not** to add integration tests because
1. ADLS is not S3-compatible which means that we can't reuse our Minio
setup
2. the Azure-specific alternative to local testing - Azurite - doesn't
support ADLS

### End-to-end
I have yet to test it in a functioning environment.

---------

Signed-off-by: Jannik Steinmann <jannik.steinmann@datadoghq.com>

Author: DerGut

Cargo.lock

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

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/Cargo.toml

@@ -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/README.md

@@ -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/file_io.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/mod.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>()?),
         }
     }