Repository: roapi/roapi (language: python)

Issue:
Support for azure blob storage
To my understanding, this is not supported yet.

Would it be something along the lines of adding an adapter for azure the same way as https://github.com/roapi/roapi/blob/main/columnq/src/io/s3.rs works with s3?

Maybe also related to #74 



Discussion / hints:

Fix patch:
diff --git a/.github/workflows/build.yml b/.github/workflows/build.yml
index 6d5cac5fd..b21932115 100644
--- a/.github/workflows/build.yml
+++ b/.github/workflows/build.yml
@@ -162,13 +162,26 @@ jobs:
       run: |
         curl -XPOST --data-binary '{"name":"test-data"}' -H "Content-Type: application/json" "http://localhost:4443/storage/v1/b"
         curl -XPOST --data-binary '@test_data/blogs.parquet' "http://localhost:4443/storage/v1/b/test-data/o?uploadType=media&name=blogs.parquet"
+    - name: Install azure storage server (Azure)
+      run: |
+        docker run -d -p 10000:10000 -p 10001:10001 -p 10002:10002 mcr.microsoft.com/azure-storage/azurite
+    - name: Create bucket and copy test data (Azure)
+      run: |
+        az storage container create -n test-data --connection-string 'DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://localhost:10000/devstoreaccount1;QueueEndpoint=http://localhost:10001/devstoreaccount1;'
+        az storage blob upload -f test_data/blogs.parquet -c test-data -n blogs.parquet --connection-string 'DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://localhost:10000/devstoreaccount1;QueueEndpoint=http://localhost:10001/devstoreaccount1;'
     - name: Start roapi and Query
       env:
         AWS_DEFAULT_REGION: "us-east-1"
         AWS_ACCESS_KEY_ID: minioadmin
         AWS_SECRET_ACCESS_KEY: minioadmin
-        AWS_ENDPOINT_URL: http://127.0.0.1:9000
+        AWS_ENDPOINT_URL: "http://127.0.0.1:9000"
         GOOGLE_SERVICE_ACCOUNT: "/tmp/gcs.json"
+        AZURITE_BLOB_STORAGE_URL: "http://127.0.0.1:10000"
+        AZURE_STORAGE_USE_EMULATOR: "true"
+        AZURE_ALLOW_HTTP: "true"
+        # https://docs.microsoft.com/en-us/azure/storage/common/storage-use-azurite?tabs=visual-studio#http-connection-strings
+        AZURE_STORAGE_ACCOUNT_NAME: "devstoreaccount1"
+        AZURE_STORAGE_ACCOUNT_KEY: "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="
       uses: BerniWittmann/background-server-action@v1.0.4
       with:
         command: /home/runner/work/roapi/roapi/test_end_to_end/query_blogs.sh
@@ -225,6 +238,13 @@ jobs:
       run: |
         curl -XPOST --data-binary '{"name":"test-data"}' -H "Content-Type: application/json" "http://localhost:4443/storage/v1/b"
         curl -XPOST --data-binary '@test_data/blogs.parquet' "http://localhost:4443/storage/v1/b/test-data/o?uploadType=media&name=blogs.parquet"
+    - name: Install azure storage server (Azure)
+      run: |
+        docker run -d -p 10000:10000 -p 10001:10001 -p 10002:10002 mcr.microsoft.com/azure-storage/azurite
+    - name: Create bucket and copy test data (Azure)
+      run: |
+        az storage container create -n test-data --connection-string 'DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://localhost:10000/devstoreaccount1;QueueEndpoint=http://localhost:10001/devstoreaccount1;'
+        az storage blob upload -f test_data/blogs.parquet -c test-data -n blogs.parquet --connection-string 'DefaultEndpointsProtocol=http;AccountName=devstoreaccount1;AccountKey=Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==;BlobEndpoint=http://localhost:10000/devstoreaccount1;QueueEndpoint=http://localhost:10001/devstoreaccount1;'
     - name: Start roapi and Query
       env:
         AWS_DEFAULT_REGION: "us-east-1"
@@ -232,6 +252,12 @@ jobs:
         AWS_SECRET_ACCESS_KEY: minioadmin
         AWS_ENDPOINT_URL: http://127.0.0.1:9000
         GOOGLE_SERVICE_ACCOUNT: "/tmp/gcs.json"
+        AZURITE_BLOB_STORAGE_URL: "http://127.0.0.1:10000"
+        AZURE_STORAGE_USE_EMULATOR: "true"
+        AZURE_ALLOW_HTTP: "true"
+        # https://docs.microsoft.com/en-us/azure/storage/common/storage-use-azurite?tabs=visual-studio#http-connection-strings
+        AZURE_STORAGE_ACCOUNT_NAME: "devstoreaccount1"
+        AZURE_STORAGE_ACCOUNT_KEY: "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="
       uses: BerniWittmann/background-server-action@v1.0.4
       with:
         command: /home/runner/work/roapi/roapi/test_end_to_end/query_blogs.sh
diff --git a/README.md b/README.md
index 4b40ee71d..b3fc8bb67 100644
--- a/README.md
+++ b/README.md
@@ -306,6 +306,7 @@ Data layer:
 - [x] HTTP/HTTPS
 - [x] S3
 - [x] GCS
+- [x] Azure Storage
 - [x] Google spreadsheet
 - [x] MySQL
 - [x] SQLite
diff --git a/columnq/Cargo.toml b/columnq/Cargo.toml
index 0be671285..04135bb87 100644
--- a/columnq/Cargo.toml
+++ b/columnq/Cargo.toml
@@ -16,7 +16,7 @@ path = "src/lib.rs"
 arrow-schema = { version ="26", features = ["serde"] }
 
 datafusion = "14"
-object_store = { version = "0.5.3", features = ["aws_profile", "gcp"] }
+object_store = { version = "0.5.3", features = ["aws_profile", "gcp", "azure"] }
 url = "2.2"
 
 log = "0"
diff --git a/columnq/src/columnq.rs b/columnq/src/columnq.rs
index 7d91e5fc6..934cce821 100644
--- a/columnq/src/columnq.rs
+++ b/columnq/src/columnq.rs
@@ -1,3 +1,5 @@
+use std::convert::TryFrom;
+use crate::io::BlobStoreType;
 use std::collections::hash_map::Entry;
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -14,6 +16,7 @@ use datafusion::physical_plan::collect;
 
 use object_store::aws::AmazonS3Builder;
 use object_store::gcp::GoogleCloudStorageBuilder;
+use object_store::azure::MicrosoftAzureBuilder;
 use crate::error::{ColumnQError, QueryError};
 use crate::query;
 use crate::table::{self, KeyValueSource, TableSource};
@@ -22,32 +25,46 @@ use url::Url;
 pub struct ColumnQObjectStoreProvider {}
 impl ObjectStoreProvider for ColumnQObjectStoreProvider {
     fn get_by_url(&self, url: &Url) -> DatafusionResult<Arc<dyn object_store::ObjectStore>> {
-        let url_schema = url.scheme();
-        match url_schema {
-            "s3" => {
-                let host = url.host_str().unwrap();
-                let mut s3_builder = AmazonS3Builder::from_env().with_bucket_name(host);
-                // for minio in CI
-                s3_builder = s3_builder.with_allow_http(true);
-
-                match s3_builder.build() {
-                    Ok(s3) => Ok(Arc::new(s3)),
-                    Err(err) => Err(DataFusionError::External(Box::new(err))),
-                }
-            },
-            "gs" => {
-                let host = url.host_str().unwrap();
-                let gcs_builder = GoogleCloudStorageBuilder::from_env().with_bucket_name(host);
-                match gcs_builder.build() {
-                    Ok(gcs) => Ok(Arc::new(gcs)),
+        match url.host_str() {
+            None => Err(DataFusionError::Execution(format!("Missing bucket name: {}", url.as_str()))),
+            Some(host) => {
+                let url_schema = url.scheme();
+                match BlobStoreType::try_from(url_schema) {
                     Err(err) => Err(DataFusionError::External(Box::new(err))),
+                    Ok(blob_type) => match blob_type {
+                        BlobStoreType::S3 => {
+                            let mut s3_builder = AmazonS3Builder::from_env().with_bucket_name(host);
+                            // for minio in CI
+                            s3_builder = s3_builder.with_allow_http(true);
+
+                            match s3_builder.build() {
+                                Ok(s3) => Ok(Arc::new(s3)),
+                                Err(err) => Err(DataFusionError::External(Box::new(err))),
+                            }
+                        },
+                        BlobStoreType::GCS => {
+                            let gcs_builder = GoogleCloudStorageBuilder::from_env().with_bucket_name(host);
+                            match gcs_builder.build() {
+                                Ok(gcs) => Ok(Arc::new(gcs)),
+                                Err(err) => Err(DataFusionError::External(Box::new(err))),
+                            }
+
+                        },
+                        BlobStoreType::Azure => {
+                            let azure_builder = MicrosoftAzureBuilder::from_env().with_container_name(host);
+                            match azure_builder.build() {
+                                Ok(azure) => Ok(Arc::new(azure)),
+                                Err(err) => Err(DataFusionError::External(Box::new(err))),
+                            }
+                        },
+                        _ => Err(DataFusionError::Execution(format!(
+                            "Unsupported scheme: {}",
+                            url_schema
+                        ))),
+                    }
                 }
 
-            },
-            _ => Err(DataFusionError::Execution(format!(
-                "Unsupported object store scheme {}",
-                url_schema
-            ))),
+            }
         }
     }
 }
@@ -219,6 +236,17 @@ mod tests {
         env::remove_var("AWS_REGION");
     }
 
+    #[test]
+    fn s3_object_store_type_no_bucket() {
+        let host_url = "s3://";
+        let provider = ColumnQObjectStoreProvider {};
+
+        let err = provider
+            .get_by_url(&Url::from_str(host_url).unwrap())
+            .unwrap_err();
+        assert!(err.to_string().contains("Missing bucket name: s3://"));
+    }
+
     #[tokio::test]
     async fn gcs_object_store_type() -> anyhow::Result<()> {
         let host_url = "gs://bucket_name/path";
@@ -248,13 +276,33 @@ mod tests {
 
     #[test]
     fn azure_object_store_type() {
-        let unknown = "az://bucket_name/path";
+        let host_url = "az://bucket_name/path";
+        let provider = ColumnQObjectStoreProvider {};
+        // https://docs.microsoft.com/en-us/azure/storage/common/storage-use-azurite?tabs=visual-studio#http-connection-strings
+        env::set_var("AZURE_STORAGE_ACCOUNT_NAME", "devstoreaccount1");
+        env::set_var("AZURE_STORAGE_ACCOUNT_KEY", "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==");
+
+        let res = provider
+            .get_by_url(&Url::from_str(host_url).unwrap());
+        let msg = match res {
+            Err(e) => format!("{}", e),
+            Ok(_) => "".to_string(),
+        };
+        assert_eq!("".to_string(), msg);
+
+        env::remove_var("AZURE_STORAGE_ACCOUNT_NAME");
+        env::remove_var("AZURE_STORAGE_ACCOUNT_KEY");
+    }
+
+    #[test]
+    fn unknown_object_store_type() {
+        let unknown = "unknown://bucket_name/path";
         let provider = ColumnQObjectStoreProvider {};
         let err = provider
             .get_by_url(&Url::from_str(unknown).unwrap())
             .unwrap_err();
         assert!(err
             .to_string()
-            .contains("Unsupported object store scheme az"))
+            .contains("External error: Invalid table URI: Unsupported scheme: \"unknown\""))
     }
 }
diff --git a/columnq/src/io/mod.rs b/columnq/src/io/mod.rs
index ec75134a3..d2bfdc930 100644
--- a/columnq/src/io/mod.rs
+++ b/columnq/src/io/mod.rs
@@ -12,6 +12,7 @@ pub enum BlobStoreType {
     Http,
     S3,
     GCS,
+    Azure,
     FileSystem,
     Memory,
 }
@@ -26,16 +27,24 @@ impl TryFrom<Option<&uriparse::Scheme<'_>>> for BlobStoreType {
                 Ok(BlobStoreType::FileSystem)
             }
             Some(uriparse::Scheme::HTTP) | Some(uriparse::Scheme::HTTPS) => Ok(BlobStoreType::Http),
-            Some(uriparse::Scheme::Unregistered(s)) => match s.as_str() {
-                "s3" => Ok(BlobStoreType::S3),
-                "gs" => Ok(BlobStoreType::GCS),
-                "memory" => Ok(BlobStoreType::Memory),
-                _ => Err(ColumnQError::InvalidUri(format!(
-                    "Unsupported scheme: {:?}",
-                    scheme
-                ))),
-            },
+            Some(uriparse::Scheme::Unregistered(s)) => BlobStoreType::try_from(s.as_str()),
+            _ => Err(ColumnQError::InvalidUri(format!(
+                "Unsupported scheme: {:?}",
+                scheme
+            ))),
+        }
+    }
+}
 
+impl TryFrom<&str> for BlobStoreType {
+    type Error = ColumnQError;
+
+    fn try_from(scheme: &str) -> Result<Self, Self::Error> {
+        match scheme {
+            "s3" => Ok(BlobStoreType::S3),
+            "gs" => Ok(BlobStoreType::GCS),
+            "az" | "adl" | "adfs" | "adfss" | "azure" => Ok(BlobStoreType::Azure),
+            "memory" => Ok(BlobStoreType::Memory),
             _ => Err(ColumnQError::InvalidUri(format!(
                 "Unsupported scheme: {:?}",
                 scheme
diff --git a/columnq/src/lib.rs b/columnq/src/lib.rs
index 847da1773..d8faed236 100644
--- a/columnq/src/lib.rs
+++ b/columnq/src/lib.rs
@@ -18,7 +18,7 @@ macro_rules! partitions_from_table_source {
             io::BlobStoreType::Http => {
                 io::http::partitions_from_uri(&$table_source, uri, $call_with_r).await
             }
-            io::BlobStoreType::S3 | io::BlobStoreType::GCS => {
+            io::BlobStoreType::S3 | io::BlobStoreType::GCS | io::BlobStoreType::Azure => {
                 io::object_store::partitions_from_uri(&$table_source, uri, $call_with_r).await
             }
             io::BlobStoreType::Memory => {
diff --git a/columnq/src/table/delta.rs b/columnq/src/table/delta.rs
index 45a1f4707..94350bca2 100644
--- a/columnq/src/table/delta.rs
+++ b/columnq/src/table/delta.rs
@@ -92,7 +92,7 @@ pub async fn to_mem_table(
                 read_partition::<std::fs::File>(r, batch_size)
             },
         )?,
-        io::BlobStoreType::S3 | io::BlobStoreType::GCS => {
+        io::BlobStoreType::S3 | io::BlobStoreType::GCS | io::BlobStoreType::Azure => {
             io::object_store::partitions_from_path_iterator(
                 path_iter,
                 |r| -> Result<Vec<RecordBatch>, ColumnQError> {


Test patch:
diff --git a/test_end_to_end/query_blogs.sh b/test_end_to_end/query_blogs.sh
index 8e5c328b5..6f60ac9fb 100755
--- a/test_end_to_end/query_blogs.sh
+++ b/test_end_to_end/query_blogs.sh
@@ -32,4 +32,15 @@ else
     echo "success"
 fi
 
+http_status=$(curl -o /dev/null -s -w "%{http_code}" -X POST -d "SELECT count(1) from azure_blogs" 127.0.0.1:8000/api/sql)
+
+echo $http_status
+if [[ $http_status != "200" ]]
+then
+    echo "error"
+    exit 1
+else
+    echo "success"
+fi
+
 exit 0
diff --git a/test_end_to_end/test_object_store_direct.yml b/test_end_to_end/test_object_store_direct.yml
index b6c3aae48..8beac8cb2 100644
--- a/test_end_to_end/test_object_store_direct.yml
+++ b/test_end_to_end/test_object_store_direct.yml
@@ -21,3 +21,8 @@ tables:
     option:
       format: "parquet"
       use_memory_table: false
+  - name: "azure_blogs"
+    uri: "az://test-data/blogs.parquet"
+    option:
+      format: "parquet"
+      use_memory_table: false
diff --git a/test_end_to_end/test_object_store_memory.yml b/test_end_to_end/test_object_store_memory.yml
index ae1d1f1b8..35561ce49 100644
--- a/test_end_to_end/test_object_store_memory.yml
+++ b/test_end_to_end/test_object_store_memory.yml
@@ -19,3 +19,8 @@ tables:
     option:
       format: "parquet"
       use_memory_table: true
+  - name: "azure_blogs"
+    uri: "az://test-data/blogs.parquet"
+    option:
+      format: "parquet"
+      use_memory_table: true