fix: Allow HTTP object store for json/bson etc.

Resolves part of #2756

Signed-off-by: Vaibhav <vrongmeal@gmail.com>

crates/datasources/src/lake/mod.rs

@@ -16,6 +16,7 @@ use protogen::metastore::types::options::StorageOptions;
 use crate::common::url::{DatasourceUrl, DatasourceUrlType};
 use crate::object_store::azure::AzureStoreAccess;
 use crate::object_store::gcs::GcsStoreAccess;
+use crate::object_store::http::HttpStoreAccess;
 use crate::object_store::local::LocalStoreAccess;
 use crate::object_store::s3::S3StoreAccess;
 use crate::object_store::ObjStoreAccess;
@@ -101,9 +102,7 @@ pub fn storage_options_into_store_access(
             }))
         }
         DatasourceUrlType::File => Ok(Arc::new(LocalStoreAccess)),
-        DatasourceUrlType::Http => {
-            Err(LakeStorageOptionsError::UnsupportedObjectStore(url.clone()))
-        }
+        DatasourceUrlType::Http => Ok(Arc::new(HttpStoreAccess { url: url.as_url()? })),
     }
 }
 

crates/datasources/src/object_store/http.rs

@@ -9,7 +9,7 @@ use datafusion::execution::context::SessionState;
 use datafusion::execution::object_store::ObjectStoreUrl;
 use object_store::http::HttpBuilder;
 use object_store::path::Path as ObjectStorePath;
-use object_store::{ObjectMeta, ObjectStore};
+use object_store::{ClientConfigKey, ObjectMeta, ObjectStore};
 use url::Url;
 
 use super::glob_util::ResolvedPattern;
@@ -43,13 +43,17 @@ impl ObjStoreAccess for HttpStoreAccess {
             // To make path part of URL we make it a '/'.
             .replace('/', "__slash__")
             // TODO: Add more characters which might be invalid for domain.
-            .replace('%', "__percent__");
+            .replace('%', "__percent__")
+            .replace('?', "__question__")
+            .replace('=', "__equal__");
 
         Ok(ObjectStoreUrl::parse(u)?)
     }
 
     fn create_store(&self) -> Result<Arc<dyn ObjectStore>> {
-        let builder = HttpBuilder::new().with_url(self.url.to_string());
+        let builder = HttpBuilder::new()
+            .with_url(self.url.to_string())
+            .with_config(ClientConfigKey::AllowHttp, "true");
         let build = builder.build()?;
         Ok(Arc::new(build))
     }
@@ -94,17 +98,17 @@ impl ObjStoreAccess for HttpStoreAccess {
         }
         // reqwest doesn't check the content length header, instead looks at the contents
         // See: https://github.com/seanmonstar/reqwest/issues/843
-        let len: u64 = res
+        let len = res
             .headers()
             .get("Content-Length")
             .and_then(|v| v.to_str().ok())
-            .and_then(|v| v.parse().ok())
-            .unwrap_or_else(|| res.content_length().unwrap_or(0));
-        if len == 0 {
-            return Err(ObjectStoreSourceError::Static(
+            .and_then(|v| v.parse::<u64>().ok())
+            .or_else(|| res.content_length())
+            // TODO: Download the contents of the file and handle using local
+            // store (maybe outside of HTTP store).
+            .ok_or(ObjectStoreSourceError::Static(
                 "Missing content-length header",
-            ));
-        }
+            ))?;
 
         Ok(ObjectMeta {
             location: location.clone(),

crates/sqlbuiltins/src/functions/table/mod.rs

@@ -184,6 +184,7 @@ pub fn table_location_and_opts(
     let mut storage_options = StorageOptions::default();
     match (source_url.datasource_url_type(), maybe_cred_opts) {
         (DatasourceUrlType::File, None) => {} // no options fine in this case
+        (DatasourceUrlType::Http, None) => {} // no options fine in this case
         (DatasourceUrlType::File, _) => {
             return Err(ExtensionError::String(
                 "Credentials incorrectly provided when accessing local delta table".to_string(),
@@ -224,11 +225,6 @@ pub fn table_location_and_opts(
                 creds.access_key,
             );
         }
-        (DatasourceUrlType::Http, _) => {
-            return Err(ExtensionError::String(
-                "Accessing delta tables over http not supported".to_string(),
-            ))
-        }
         (datasource, creds) => {
             return Err(ExtensionError::String(format!(
                 "Invalid credentials for {datasource}, got {} creds",