Support http(s) stores

pg_parquet supports reading/writing from/to public http(s) parquet uris to/from Postgres tables.

Author: aykut-bozkurt

.devcontainer/.env

@@ -18,5 +18,9 @@ AZURE_TEST_CONTAINER_NAME=testcontainer
 AZURE_TEST_READ_ONLY_SAS="se=2100-05-05&sp=r&sv=2022-11-02&sr=c&sig=YMPFnAHKe9y0o3hFegncbwQTXtAyvsJEgPB2Ne1b9CQ%3D"
 AZURE_TEST_READ_WRITE_SAS="se=2100-05-05&sp=rcw&sv=2022-11-02&sr=c&sig=TPz2jEz0t9L651t6rTCQr%2BOjmJHkM76tnCGdcyttnlA%3D"
 
+# http(s) tests
+ALLOW_HTTP=true
+HTTP_ENDPOINT=http://localhost:8080
+
 # Others
 RUST_TEST_THREADS=1

.devcontainer/docker-compose.yml

@@ -20,6 +20,7 @@ services:
     depends_on:
       - minio
       - azurite
+      - webdav
 
   minio:
     image: minio/minio
@@ -47,3 +48,16 @@ services:
       interval: 6s
       timeout: 2s
       retries: 3
+
+  webdav:
+    image: rclone/rclone
+    command: ["serve", "webdav", "/data", "--addr", ":8080"]
+    env_file:
+      - .env
+    network_mode: host
+    restart: unless-stopped
+    healthcheck:
+      test: ["CMD", "curl", "http://localhost:8080"]
+      interval: 6s
+      timeout: 2s
+      retries: 3

.github/workflows/ci.yml

@@ -140,6 +140,18 @@ jobs:
           az storage container create -n $AZURE_TEST_CONTAINER_NAME --connection-string $AZURE_STORAGE_CONNECTION_STRING
           az storage container create -n ${AZURE_TEST_CONTAINER_NAME}2 --connection-string $AZURE_STORAGE_CONNECTION_STRING
 
+      - name: Start local web server for http(s) tests
+        run: |
+          docker run -d \
+            --env-file .devcontainer/.env \
+            -p 8080:80 \
+            rclone/rclone serve webdav /data --addr :80
+
+          while ! curl $HTTP_ENDPOINT; do
+            echo "Waiting for $HTTP_ENDPOINT..."
+            sleep 1
+          done
+
       - name: Run tests
         run: |
           # Run tests with coverage tool

Cargo.toml

@@ -28,7 +28,7 @@ aws-credential-types = {version = "1", default-features = false}
 azure_storage = {version = "0.21", default-features = false}
 futures = "0.3"
 home = "0.5"
-object_store = {version = "0.11", default-features = false, features = ["aws", "azure"]}
+object_store = {version = "0.11", default-features = false, features = ["aws", "azure", "http"]}
 once_cell = "1"
 parquet = {version = "54", default-features = false, features = [
     "arrow",

README.md

@@ -158,7 +158,7 @@ SELECT uri, encode(key, 'escape') as key, encode(value, 'escape') as value FROM
 ```
 
 ## Object Store Support
-`pg_parquet` supports reading and writing Parquet files from/to `S3` and `Azure Blob Storage` object stores.
+`pg_parquet` supports reading and writing Parquet files from/to `S3`, `Azure Blob Storage` and `http(s)` object stores.
 
 > [!NOTE]
 > To be able to write into a object store location, you need to grant `parquet_object_store_write` role to your current postgres user.
@@ -241,6 +241,10 @@ Supported authorization methods' priority order is shown below:
 2. Sas token,
 3. Storage key.
 
+#### Http(s) Storage
+
+`Https` uris are supported by default. You can set `ALLOW_HTTP` environment variable to allow `http` uris.
+
 ## Copy Options
 `pg_parquet` supports the following options in the `COPY TO` command:
 - `format parquet`: you need to specify this option to read or write Parquet files which does not end with `.parquet[.<compression>]` extension,

src/arrow_parquet/uri_utils.rs

@@ -20,7 +20,7 @@ use url::Url;
 use crate::{
     arrow_parquet::parquet_writer::DEFAULT_ROW_GROUP_SIZE,
     object_store::{
-        aws::parse_s3_bucket, azure::parse_azure_blob_container,
+        aws::parse_s3_bucket, azure::parse_azure_blob_container, http::parse_http_base_uri,
         object_store_cache::get_or_create_object_store,
     },
     PG_BACKEND_TOKIO_RUNTIME,
@@ -51,7 +51,7 @@ impl ParsedUriInfo {
     fn try_parse_scheme(uri: &Url) -> Result<(ObjectStoreScheme, Path), String> {
         ObjectStoreScheme::parse(uri).map_err(|_| {
             format!(
-                "unrecognized uri {}. pg_parquet supports local paths, s3:// or azure:// schemes.",
+                "unrecognized uri {}. pg_parquet supports local paths, https://, s3:// or az:// schemes.",
                 uri
             )
         })
@@ -65,8 +65,11 @@ impl ParsedUriInfo {
             ObjectStoreScheme::MicrosoftAzure => parse_azure_blob_container(uri)
                 .ok_or(format!("unsupported azure blob storage uri: {uri}"))
                 .map(Some),
+            ObjectStoreScheme::Http => parse_http_base_uri(uri).
+                ok_or(format!("unsupported http storage uri: {uri}"))
+                .map(Some),
             ObjectStoreScheme::Local => Ok(None),
-            _ => Err(format!("unsupported scheme {} in uri {}. pg_parquet supports local paths, s3:// or azure:// schemes.",
+            _ => Err(format!("unsupported scheme {} in uri {}. pg_parquet supports local paths, https://, s3:// or az:// schemes.",
                             uri.scheme(), uri))
         }
     }

src/object_store.rs

@@ -9,5 +9,6 @@ use crate::{
 
 pub(crate) mod aws;
 pub(crate) mod azure;
+pub(crate) mod http;
 pub(crate) mod local_file;
 pub(crate) mod object_store_cache;

src/object_store/http.rs

@@ -0,0 +1,42 @@
+use std::sync::Arc;
+
+use object_store::{http::HttpBuilder, ClientOptions};
+use url::Url;
+
+use super::object_store_cache::ObjectStoreWithExpiration;
+
+// create_http_object_store creates a http(s) object store with the given bucket name.
+pub(crate) fn create_http_object_store(uri: &Url) -> ObjectStoreWithExpiration {
+    let base_uri = parse_http_base_uri(uri).unwrap_or_else(|| {
+        panic!("unsupported http uri: {}", uri);
+    });
+
+    let allow_http = std::env::var("ALLOW_HTTP").is_ok();
+
+    let client_options = ClientOptions::new()
+        .with_allow_http2()
+        .with_allow_http(allow_http);
+
+    let http_builder = HttpBuilder::new()
+        .with_url(base_uri)
+        .with_client_options(client_options);
+
+    let object_store = http_builder.build().unwrap_or_else(|e| panic!("{}", e));
+
+    let expire_at = None;
+
+    ObjectStoreWithExpiration {
+        object_store: Arc::new(object_store),
+        expire_at,
+    }
+}
+
+pub(crate) fn parse_http_base_uri(uri: &Url) -> Option<String> {
+    let scheme = uri.scheme();
+
+    let host = uri.host_str().expect("http uri missing host");
+
+    let port = uri.port().map(|p| format!(":{}", p)).unwrap_or_default();
+
+    Some(format!("{}://{}{}", scheme, host, port))
+}

src/object_store/object_store_cache.rs

@@ -12,7 +12,10 @@ use url::Url;
 
 use crate::arrow_parquet::uri_utils::ParsedUriInfo;
 
-use super::{create_azure_object_store, create_local_file_object_store, create_s3_object_store};
+use super::{
+    create_azure_object_store, create_local_file_object_store, create_s3_object_store,
+    http::create_http_object_store,
+};
 
 // OBJECT_STORE_CACHE is a global cache for object stores per Postgres session.
 // It caches object stores based on the scheme and bucket.
@@ -86,9 +89,10 @@ impl ObjectStoreCache {
         match scheme {
             ObjectStoreScheme::AmazonS3 => create_s3_object_store(uri),
             ObjectStoreScheme::MicrosoftAzure => create_azure_object_store(uri),
+            ObjectStoreScheme::Http => create_http_object_store(uri),
             ObjectStoreScheme::Local => create_local_file_object_store(uri, copy_from),
             _ => panic!(
-                    "unsupported scheme {} in uri {}. pg_parquet supports local paths, s3:// or azure:// schemes.",
+                    "unsupported scheme {} in uri {}. pg_parquet supports local paths, https, s3:// or az:// schemes.",
                     uri.scheme(),
                     uri
                 ),

src/pgrx_tests/object_store.rs

@@ -630,6 +630,26 @@ mod tests {
         Spi::run("copy test_table from 'https://ACCOUNT.dfs.fabric.microsoft.com';").unwrap();
     }
 
+    #[pg_test]
+    fn test_http_uri() {
+        object_store_cache_clear();
+
+        let http_endpoint: String =
+            std::env::var("HTTP_ENDPOINT").expect("HTTP_ENDPOINT not found");
+
+        let http_uris = [
+            format!("{http_endpoint}/pg_parquet_test.parquet"),
+            format!("{http_endpoint}/dummy/pg_parquet_test"),
+        ];
+
+        for http_uri in http_uris {
+            let test_table = TestTable::<i32>::new("int4".into()).with_uri(http_uri);
+
+            test_table.insert("INSERT INTO test_expected (a) VALUES (1), (2), (null);");
+            test_table.assert_expected_and_result_rows();
+        }
+    }
+
     #[pg_test]
     #[should_panic(expected = "relative path not allowed")]
     fn test_copy_to_unsupported_scheme() {
@@ -759,5 +779,50 @@ mod tests {
                 ("MicrosoftAzure", "testcontainer2", None)
             ]
         );
+
+        // https scheme and base uri
+        let http_uri =
+            "https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2024-03.parquet";
+        Spi::run(format!("SELECT * FROM parquet.schema('{http_uri}')").as_str()).unwrap();
+
+        assert_eq!(
+            object_store_cache_items(),
+            vec![
+                ("AmazonS3", "testbucket", None),
+                ("AmazonS3", "testbucket2", None),
+                ("MicrosoftAzure", "testcontainer", None),
+                ("HttpStore", "https://d37ci6vzurychx.cloudfront.net", None)
+            ]
+        );
+
+        // https scheme and same base uri
+        let http_uri =
+            "https://d37ci6vzurychx.cloudfront.net/trip-data/yellow_tripdata_2024-04.parquet";
+        Spi::run(format!("SELECT * FROM parquet.schema('{http_uri}')").as_str()).unwrap();
+
+        assert_eq!(
+            object_store_cache_items(),
+            vec![
+                ("AmazonS3", "testbucket", None),
+                ("AmazonS3", "testbucket2", None),
+                ("MicrosoftAzure", "testcontainer", None),
+                ("HttpStore", "https://d37ci6vzurychx.cloudfront.net", None)
+            ]
+        );
+
+        // https scheme and different base uri
+        let http_uri = "https://www.filesampleshub.com/download/code/parquet/sample1.parquet";
+        Spi::run(format!("SELECT * FROM parquet.schema('{http_uri}')").as_str()).unwrap();
+
+        assert_eq!(
+            object_store_cache_items(),
+            vec![
+                ("AmazonS3", "testbucket", None),
+                ("AmazonS3", "testbucket2", None),
+                ("MicrosoftAzure", "testcontainer", None),
+                ("Http", "https://d37ci6vzurychx.cloudfront.net", None),
+                ("HttpStore", "https://www.filesampleshub.com", None)
+            ]
+        );
     }
 }