Add Support for Microsoft Fabric / OneLake

object_store/src/azure/mod.rs

@@ -726,7 +726,9 @@ impl MicrosoftAzureBuilder {
                 // or the convention for the hadoop driver abfs[s]://<file_system>@<account_name>.dfs.core.windows.net/<path>
                 if parsed.username().is_empty() {
                     self.container_name = Some(validate(host)?);
-                } else if let Some(a) = host.strip_suffix(".dfs.core.windows.net") {
+                } else if let Some(a) = host.strip_suffix(".dfs.core.windows.net") 
+                                .or_else(|| host.strip_suffix(".dfs.fabric.microsoft.com"))
+                {
                     self.container_name = Some(validate(parsed.username())?);
                     self.account_name = Some(validate(a)?);
                 } else {
@@ -735,7 +737,12 @@ impl MicrosoftAzureBuilder {
             }
             "https" => match host.split_once('.') {
                 Some((a, "dfs.core.windows.net"))
-                | Some((a, "blob.core.windows.net")) => {
+                | Some((a, "blob.core.windows.net"))              
+                => {
+                    self.account_name = Some(validate(a)?);
+                }
+                Some((a, "dfs.fabric.microsoft.com"))
+                | Some((a, "blob.fabric.microsoft.com")) =>{
                     self.account_name = Some(validate(a)?);
                 }
                 _ => return Err(UrlNotRecognisedSnafu { url }.build().into()),
@@ -889,7 +896,17 @@ impl MicrosoftAzureBuilder {
             self.parse_url(&url)?;
         }
 
-        let container = self.container_name.ok_or(Error::MissingContainerName {})?;
+        let use_ok_or = match &self.account_name {
+            Some(account_name) => !account_name.contains("onelake"),
+            None => true,
+        };
+
+        let container = if use_ok_or {
+            self.container_name.ok_or(Error::MissingContainerName {})?
+        } else {
+            self.container_name.unwrap_or_default()
+        };
+
         let static_creds = |credential: AzureCredential| -> AzureCredentialProvider {
             Arc::new(StaticCredentialProvider::new(credential))
         };
@@ -911,7 +928,13 @@ impl MicrosoftAzureBuilder {
             (true, url, credential, account_name)
         } else {
             let account_name = self.account_name.ok_or(Error::MissingAccount {})?;
-            let account_url = format!("https://{}.blob.core.windows.net", &account_name);
+            let account_url = if account_name.contains("onelake") {
+                format!("https://{}.blob.fabric.microsoft.com", &account_name)
+            } else {
+                format!("https://{}.blob.core.windows.net", &account_name)
+            };
+
+
             let url = Url::parse(&account_url)
                 .context(UnableToParseUrlSnafu { url: account_url })?;
 
@@ -1144,9 +1167,16 @@ mod tests {
         assert_eq!(builder.account_name, Some("account".to_string()));
         assert_eq!(builder.container_name, Some("file_system".to_string()));
 
+        let mut builder = MicrosoftAzureBuilder::new();
+        builder
+            .parse_url("abfss://file_system@account.dfs.fabric.microsoft.com/")
+            .unwrap();
+        assert_eq!(builder.account_name, Some("account".to_string()));
+        assert_eq!(builder.container_name, Some("file_system".to_string()));
+
         let mut builder = MicrosoftAzureBuilder::new();
         builder.parse_url("abfs://container/path").unwrap();
-        assert_eq!(builder.container_name, Some("container".to_string()));
+        assert_eq!(builder.container_name, Some("container".to_string()));   
 
         let mut builder = MicrosoftAzureBuilder::new();
         builder.parse_url("az://container").unwrap();
@@ -1168,6 +1198,18 @@ mod tests {
             .unwrap();
         assert_eq!(builder.account_name, Some("account".to_string()));
 
+        let mut builder = MicrosoftAzureBuilder::new();
+        builder
+            .parse_url("https://account.dfs.fabric.microsoft.com/")
+            .unwrap();
+        assert_eq!(builder.account_name, Some("account".to_string()));
+
+        let mut builder = MicrosoftAzureBuilder::new();
+        builder
+            .parse_url("https://account.blob.fabric.microsoft.com/")
+            .unwrap();
+        assert_eq!(builder.account_name, Some("account".to_string()));
+
         let err_cases = [
             "mailto://account.blob.core.windows.net/",
             "az://blob.mydomain/",

object_store/tests/azure_object_store.rs

@@ -0,0 +1,114 @@
+use async_trait::async_trait;
+use bytes::Bytes;
+use futures::stream::BoxStream;
+use object_store::azure::{MicrosoftAzureBuilder, MicrosoftAzure};
+use object_store::path::Path;
+use object_store::{
+    GetOptions, GetResult, ListResult, MultipartId, ObjectMeta, ObjectStore,
+};
+use std::fmt::Formatter;
+use tokio::io::AsyncWrite;
+
+#[derive(Debug)]
+struct AzureStore(MicrosoftAzure);
+
+impl std::fmt::Display for AzureStore {
+    fn fmt(&self, _: &mut Formatter<'_>) -> std::fmt::Result {
+        todo!()
+    }
+}
+
+#[async_trait]
+impl ObjectStore for AzureStore {
+    async fn put(&self, path: &Path, data: Bytes) -> object_store::Result<()> {
+        self.0.put(path, data).await
+    }
+
+    async fn put_multipart(
+        &self,
+        _: &Path,
+    ) -> object_store::Result<(MultipartId, Box<dyn AsyncWrite + Unpin + Send>)> {
+        todo!()
+    }
+
+    async fn abort_multipart(
+        &self,
+        _: &Path,
+        _: &MultipartId,
+    ) -> object_store::Result<()> {
+        todo!()
+    }
+
+    async fn get_opts(
+        &self,
+        location: &Path,
+        options: GetOptions,
+    ) -> object_store::Result<GetResult> {
+        self.0.get_opts(location, options).await
+    }
+
+    async fn head(&self, _: &Path) -> object_store::Result<ObjectMeta> {
+        todo!()
+    }
+
+    async fn delete(&self, _: &Path) -> object_store::Result<()> {
+        todo!()
+    }
+
+    async fn list(
+        &self,
+        _: Option<&Path>,
+    ) -> object_store::Result<BoxStream<'_, object_store::Result<ObjectMeta>>> {
+        todo!()
+    }
+
+    async fn list_with_delimiter(
+        &self,
+        _: Option<&Path>,
+    ) -> object_store::Result<ListResult> {
+        todo!()
+    }
+
+    async fn copy(&self, _: &Path, _: &Path) -> object_store::Result<()> {
+        todo!()
+    }
+
+    async fn copy_if_not_exists(&self, _: &Path, _: &Path) -> object_store::Result<()> {
+        todo!()
+    }
+}
+
+
+#[tokio::test]
+async fn test_fabric() {        
+    //Format:https://onelake.dfs.fabric.microsoft.com/<workspaceGUID>/<itemGUID>/Files/test.csv
+    //Example:https://onelake.dfs.fabric.microsoft.com/86bc63cf-5086-42e0-b16d-6bc580d1dc87/17d3977c-d46e-4bae-8fed-ff467e674aed/Files/SampleCustomerList.csv
+    //Account Name : onelake
+    //Container Name : workspaceGUID
+
+    let daily_store = AzureStore(
+        MicrosoftAzureBuilder::new()
+        .with_container_name("86bc63cf-5086-42e0-b16d-6bc580d1dc87")
+        .with_account("onelake")
+        .with_bearer_token_authorization("jwt-token")
+        .build()
+        .unwrap());
+
+    let path = Path::from("17d3977c-d46e-4bae-8fed-ff467e674aed/Files/SampleCustomerList.csv");
+
+    read_write_test(&daily_store, &path).await;
+}
+
+
+async fn read_write_test(store: &AzureStore, path: &Path) {
+    let expected = Bytes::from_static(b"hello world");
+    store.put(path, expected.clone()).await.unwrap();
+    let fetched = store.get(path).await.unwrap().bytes().await.unwrap();
+    assert_eq!(expected, fetched);
+
+    for range in [0..10, 3..5, 0..expected.len()] {
+        let data = store.get_range(path, range.clone()).await.unwrap();
+        assert_eq!(&data[..], &expected[range])
+    }
+}
+