feat: support S3 compliant filesystems

native/core/src/execution/operators/iceberg_scan.rs

@@ -255,7 +255,7 @@ impl IcebergScanExec {
         };
         match scheme {
             "file" => Ok(Arc::new(OpenDalStorageFactory::Fs)),
-            "s3" | "s3a" => {
+            "s3" | "s3a" | "blob" => {
                 let customized_credential_load =
                     build_s3_credential_loader(path, catalog_properties, catalog_name);
                 Ok(Arc::new(OpenDalStorageFactory::S3 {
@@ -368,6 +368,22 @@ impl IcebergScanExec {
             }
         }
 
+        // Object-store's AmazonS3Builder defaults the SigV4 region to `us-east-1` when unset;
+        // iceberg-storage-opendal's S3 factory instead errors with `region is missing. Please
+        // find it by S3::detect_region() or set them in env.` Non-AWS S3-compliant storage
+        // services accept any region in the credential, so default to `us-east-1` when the
+        // catalog didn't ship one. Both key spellings iceberg-rust reads (`client.region`
+        // wins over `s3.region`).
+        let region_forwarded = catalog_properties.contains_key("s3.region")
+            || catalog_properties.contains_key("client.region");
+        let is_s3_family = matches!(
+            metadata_location.split_once("://"),
+            Some(("s3" | "s3a" | "blob", _))
+        );
+        if !region_forwarded && is_s3_family {
+            file_io_builder = file_io_builder.with_prop("s3.region", "us-east-1");
+        }
+
         Ok(file_io_builder.build())
     }
 }

native/core/src/execution/planner.rs

@@ -85,6 +85,9 @@ use iceberg::expr::Bind;
 use crate::execution::operators::ExecutionError::GeneralError;
 use crate::execution::shuffle::{CometPartitioning, CompressionCodec};
 use crate::execution::spark_plan::SparkPlan;
+use crate::parquet::objectstore::blob_alias::{
+    normalize_object_store_url, normalize_object_store_url_string,
+};
 use crate::parquet::parquet_support::prepare_object_store_with_configs;
 use datafusion::common::scalar::ScalarStructBuilder;
 use datafusion::common::{
@@ -143,7 +146,6 @@ use num::{BigInt, ToPrimitive};
 use object_store::path::Path;
 use std::cmp::max;
 use std::{collections::HashMap, sync::Arc};
-use url::Url;
 
 // For clippy error on type_complexity.
 type PhyAggResult = Result<Vec<AggregateFunctionExpr>, ExecutionError>;
@@ -388,6 +390,10 @@ impl PhysicalPlanner {
         partition: &SparkFilePartition,
     ) -> Result<Vec<PartitionedFile>, ExecutionError> {
         let mut files = Vec::with_capacity(partition.partitioned_file.len());
+        // Empty object-store configs are fine here: the only thing `normalize_object_store_url`
+        // consults them for is `is_hdfs_scheme`, and blob/s3a rewriting is the only shape
+        // we care about for the object-store key. HDFS-routed schemes are handled elsewhere.
+        let empty_configs: HashMap<String, String> = HashMap::new();
         partition.partitioned_file.iter().try_for_each(|file| {
             assert!(file.start + file.length <= file.file_size);
 
@@ -398,10 +404,12 @@ impl PhysicalPlanner {
                 file.start + file.length,
             );
 
-            // Spark sends the path over as URL-encoded, parse that first.
-            let url =
-                Url::parse(file.file_path.as_ref()).map_err(|e| GeneralError(e.to_string()))?;
-            // Convert that to a Path object to use in the PartitionedFile.
+            // Apply the same blob/s3a scheme rewrite and three-slash normalization the object
+            // store construction uses, so the object-store key we hand DataFusion is stripped of
+            // the bucket prefix. Skipping this here means paths like `blob:///bucket/key` end up
+            // with `bucket/key` as the object key, and path-style S3 GETs double the bucket
+            // (`<endpoint>/bucket/bucket/key`).
+            let url = normalize_object_store_url(&file.file_path, &empty_configs)?;
             let path = Path::from_url_path(url.path()).map_err(|e| GeneralError(e.to_string()))?;
             partitioned_file.object_meta.location = path;
 
@@ -1752,7 +1760,13 @@ impl PhysicalPlanner {
                     .iter()
                     .map(|(k, v)| (k.clone(), v.clone()))
                     .collect();
-                let metadata_location = common.metadata_location.clone();
+                // Normalize blob/s3a aliases (including Java's `blob:/bucket/key` single-slash
+                // form that Iceberg manifests store) to canonical `s3://bucket/key`. Without
+                // this, storage_factory_for's `contains("://")` check falls back to LocalFs and
+                // iceberg-storage-opendal's fs branch strips one char with `&path[1..]`,
+                // producing `lob:/...` errors.
+                let metadata_location =
+                    normalize_object_store_url_string(&common.metadata_location)?;
                 let catalog_name = common.catalog_name.clone();
                 let tasks = parse_file_scan_tasks_from_common(common, &scan.file_scan_tasks)?;
                 let data_file_concurrency_limit = common.data_file_concurrency_limit as usize;
@@ -3919,7 +3933,8 @@ fn parse_file_scan_tasks_from_common(
             };
 
             Ok(iceberg::scan::FileScanTaskDeleteFile {
-                file_path: del.file_path.clone(),
+                // Normalize blob/s3a aliases so iceberg-rust routes through S3, not LocalFs.
+                file_path: normalize_object_store_url_string(&del.file_path)?,
                 file_type,
                 // Not serialized; filled in by IcebergScanExec::fill_delete_file_sizes.
                 file_size_in_bytes: 0,
@@ -4140,7 +4155,8 @@ fn parse_file_scan_tasks_from_common(
 
             Ok(iceberg::scan::FileScanTask {
                 file_size_in_bytes: proto_task.file_size_in_bytes,
-                data_file_path: proto_task.data_file_path.clone(),
+                // Normalize blob/s3a aliases so iceberg-rust routes through S3, not LocalFs.
+                data_file_path: normalize_object_store_url_string(&proto_task.data_file_path)?,
                 start: proto_task.start,
                 length: proto_task.length,
                 record_count: proto_task.record_count,

native/core/src/lib.rs

@@ -169,6 +169,9 @@ pub extern "system" fn Java_org_apache_comet_NativeBase_isFeatureEnabled(
 /// answering from `ObjectStoreScheme::parse` here, the planner can decline early without
 /// hardcoding -- and drifting from -- the object_store-supported scheme set. (hdfs / libhdfs
 /// schemes are handled separately on the JVM side via the user's libhdfs scheme config.)
+///
+/// `blob` is not recognized by `ObjectStoreScheme::parse`, but Comet treats it as a synonym
+/// for `s3` (rewritten in `prepare_object_store_with_configs`), so report it as supported.
 #[no_mangle]
 pub extern "system" fn Java_org_apache_comet_NativeBase_isObjectStoreSchemeSupported(
     env: EnvUnowned,
@@ -179,7 +182,7 @@ pub extern "system" fn Java_org_apache_comet_NativeBase_isObjectStoreSchemeSuppo
         let url_str: String = url.try_to_string(env)?;
         let supported = url::Url::parse(&url_str)
             .ok()
-            .map(|u| object_store::ObjectStoreScheme::parse(&u).is_ok())
+            .map(|u| u.scheme() == "blob" || object_store::ObjectStoreScheme::parse(&u).is_ok())
             .unwrap_or(false);
         Ok(supported)
     })

native/core/src/parquet/objectstore/blob_alias.rs

@@ -0,0 +1,130 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! S3-compatible filesystem alias handling.
+//!
+//! Comet treats `blob://` and `s3a://` as aliases for `s3://` -- they route through the same
+//! `object_store::AmazonS3` / `iceberg-storage-opendal::S3` code paths but the aliased forms are
+//! not directly recognized by those crates. This module rewrites the URL shape (scheme +
+//! `blob:/bucket/key` single-slash form that Java's `URI.toString()` produces from
+//! `blob:///bucket/key` and Iceberg manifests store) so downstream code can treat everything as
+//! canonical `s3://bucket/key`.
+//!
+//! The actual S3 store construction lives in [`super::s3`]; this module is only about URL shape
+//! so all the alias code has one home instead of drifting across `parquet_support.rs`,
+//! `execution/planner.rs`, and `execution/operators/iceberg_scan.rs`.
+
+use std::collections::HashMap;
+
+use url::Url;
+
+use crate::execution::operators::ExecutionError;
+use crate::parquet::parquet_support::is_hdfs_scheme;
+
+/// Parses `url_str` and rewrites `blob`/`s3a` schemes (and the awkward three-slash
+/// `blob:///bucket/key` form that `ObjectStoreScheme::parse` rejects because host=None) to the
+/// canonical `s3://bucket/key`. Non-alias schemes are returned unchanged.
+///
+/// `object_store_configs` is consulted only via `is_hdfs_scheme`: if the user routed `s3a`
+/// through libhdfs via `fs.comet.libhdfs.schemes`, we must NOT rewrite it to `s3` -- HDFS
+/// handling takes over.
+pub(crate) fn normalize_object_store_url(
+    url_str: &str,
+    object_store_configs: &HashMap<String, String>,
+) -> Result<Url, ExecutionError> {
+    let mut url = Url::parse(url_str)
+        .map_err(|e| ExecutionError::GeneralError(format!("Error parsing URL {url_str}: {e}")))?;
+    if is_hdfs_scheme(&url, object_store_configs) {
+        return Ok(url);
+    }
+    let scheme = url.scheme();
+    if scheme != "s3a" && scheme != "blob" {
+        return Ok(url);
+    }
+    let original = scheme.to_string();
+    let needs_host_promotion = url.host_str().is_none();
+    url.set_scheme("s3").map_err(|_| {
+        ExecutionError::GeneralError(format!("Could not convert scheme from {original} to s3"))
+    })?;
+    if needs_host_promotion {
+        // Some deployments emit `blob:///bucket/key` (three slashes, empty authority) or Java
+        // collapses that to `blob:/bucket/key` (opaque form) in Iceberg manifests. In both,
+        // `url::Url` reports host=None and path=`/bucket/key`, but `ObjectStoreScheme::parse`
+        // requires a non-empty host. Lift the first path segment into the host.
+        let trimmed = url.path().trim_start_matches('/').to_string();
+        let (bucket, key) = match trimmed.split_once('/') {
+            Some((b, k)) => (b.to_string(), k.to_string()),
+            None => (trimmed, String::new()),
+        };
+        if bucket.is_empty() {
+            return Err(ExecutionError::GeneralError(format!(
+                "{original}:// URL is missing bucket name: {url}"
+            )));
+        }
+        url = Url::parse(&format!("s3://{bucket}/{key}")).map_err(|e| {
+            ExecutionError::GeneralError(format!("Could not normalize {original}:// URL: {e}"))
+        })?;
+    }
+    Ok(url)
+}
+
+/// String-returning wrapper: iceberg-rust stores paths as owned strings on
+/// `FileScanTask`/`FileScanTaskDeleteFile`, and iceberg-storage-opendal's `storage_factory_for`
+/// picks the LocalFs backend for any path without `://` -- so a single-slash `blob:/...` path
+/// lands in the fs backend and gets its first character stripped (`&path[1..]`), producing
+/// errors like `path: lob:/...`. Rewrite before handing paths to iceberg-rust.
+pub(crate) fn normalize_object_store_url_string(path: &str) -> Result<String, ExecutionError> {
+    Ok(normalize_object_store_url(path, &HashMap::new())?.to_string())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_normalize_object_store_url_string_blob_single_slash_form() {
+        // Java's URI.toString() collapses `blob:///bucket/key` to `blob:/bucket/key` (opaque form
+        // with a leading slash on the path), and that's exactly what Iceberg manifests store.
+        // Iceberg-storage-opendal's `storage_factory_for` uses `path.contains("://")` to detect
+        // the scheme, so a `blob:/...` path routes to the LocalFs backend and its `&path[1..]`
+        // fallback strips the first char, producing `lob:/...` errors. Guard: this string helper
+        // must produce the canonical `s3://bucket/key` before the path reaches iceberg-rust.
+        let out = normalize_object_store_url_string(
+            "blob:/mybucket/tmp/warehouse/db/test_table/data/part-0.parquet",
+        )
+        .expect("single-slash blob URL should normalize");
+        assert_eq!(
+            out,
+            "s3://mybucket/tmp/warehouse/db/test_table/data/part-0.parquet"
+        );
+
+        // Two-slash form (authority present) also normalizes to s3://.
+        let out = normalize_object_store_url_string("blob://bucket/key.parquet").unwrap();
+        assert_eq!(out, "s3://bucket/key.parquet");
+
+        // s3a shares the alias path and gets rewritten too.
+        let out = normalize_object_store_url_string("s3a://bucket/key.parquet").unwrap();
+        assert_eq!(out, "s3://bucket/key.parquet");
+
+        // Non-alias schemes pass through unchanged so file:// / memory:/ Iceberg paths keep
+        // working with iceberg-storage-opendal's Fs / Memory backends.
+        let out = normalize_object_store_url_string("file:///tmp/warehouse/db/t").unwrap();
+        assert_eq!(out, "file:///tmp/warehouse/db/t");
+        let out = normalize_object_store_url_string("s3://bucket/key.parquet").unwrap();
+        assert_eq!(out, "s3://bucket/key.parquet");
+    }
+}

native/core/src/parquet/objectstore/mod.rs

@@ -16,4 +16,5 @@
 // under the License.
 
 pub mod azure;
+pub mod blob_alias;
 pub mod s3;

native/core/src/parquet/objectstore/s3.rs

@@ -120,7 +120,13 @@ pub fn create_store(
             .block_on(resolve_bucket_region(bucket))
             .map_err(|e| object_store::Error::Generic {
                 store: "S3",
-                source: format!("Failed to resolve region: {e}").into(),
+                source: format!(
+                    "Failed to resolve region: {e}. If '{bucket}' is on a non-AWS S3-compatible \
+                     service, set fs.s3a.endpoint (and optionally fs.s3a.endpoint.region, \
+                     fs.s3a.path.style.access) or the per-bucket variants \
+                     fs.s3a.bucket.{bucket}.endpoint[.region] so Comet skips the AWS HEAD probe."
+                )
+                .into(),
             })?;
         debug!("resolved region: {region:?}");
         builder = builder.with_config(AmazonS3ConfigKey::Region, region.to_string());

native/core/src/parquet/parquet_support.rs

@@ -553,16 +553,12 @@ pub(crate) fn prepare_object_store_with_configs(
     url: String,
     object_store_configs: &HashMap<String, String>,
 ) -> Result<(ObjectStoreUrl, Path), ExecutionError> {
-    let mut url = Url::parse(url.as_str())
-        .map_err(|e| ExecutionError::GeneralError(format!("Error parsing URL {url}: {e}")))?;
+    let url = super::objectstore::blob_alias::normalize_object_store_url(
+        url.as_str(),
+        object_store_configs,
+    )?;
     let is_hdfs_scheme = is_hdfs_scheme(&url, object_store_configs);
-    let mut scheme = url.scheme();
-    if !is_hdfs_scheme && scheme == "s3a" {
-        scheme = "s3";
-        url.set_scheme("s3").map_err(|_| {
-            ExecutionError::GeneralError("Could not convert scheme from s3a to s3".to_string())
-        })?;
-    }
+    let scheme = url.scheme();
     let url_key = format!(
         "{}://{}",
         scheme,
@@ -683,4 +679,71 @@ mod tests {
             }
         }
     }
+
+    #[cfg(not(feature = "hdfs-opendal"))]
+    #[test]
+    #[cfg_attr(miri, ignore)] // AWS credential providers and object_store call foreign functions
+    fn test_prepare_object_store_rewrites_blob_to_s3() {
+        // `blob` is a Comet-recognized synonym for `s3` -- the alias must be rewritten inside
+        // `prepare_object_store_with_configs`, otherwise ObjectStoreScheme::parse rejects the
+        // URL and the native scan fails at execution time (the JVM gate having already claimed
+        // the scan via `NativeBase.isObjectStoreSchemeSupported`). Regressing the rewrite would
+        // resurface `Unsupported filesystem schemes: blob` at execution rather than planning.
+        use crate::parquet::parquet_support::prepare_object_store_with_configs;
+        let mut configs: HashMap<String, String> = HashMap::new();
+        configs.insert(
+            "fs.s3a.aws.credentials.provider".to_string(),
+            "org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider".to_string(),
+        );
+        configs.insert(
+            "fs.s3a.endpoint.region".to_string(),
+            "us-east-1".to_string(),
+        );
+        let (object_store_url, path) = prepare_object_store_with_configs(
+            Arc::new(RuntimeEnv::default()),
+            "blob://test_bucket/comet/spark-warehouse/part-00000.snappy.parquet".to_string(),
+            &configs,
+        )
+        .expect("blob:// URL should be rewritten to s3:// and accepted");
+        assert_eq!(
+            object_store_url,
+            ObjectStoreUrl::parse("s3://test_bucket").unwrap()
+        );
+        assert_eq!(
+            path,
+            Path::from("/comet/spark-warehouse/part-00000.snappy.parquet")
+        );
+    }
+
+    #[cfg(not(feature = "hdfs-opendal"))]
+    #[test]
+    #[cfg_attr(miri, ignore)] // AWS credential providers and object_store call foreign functions
+    fn test_prepare_object_store_promotes_first_path_segment_when_blob_url_has_empty_authority() {
+        // Some deployments emit `blob:///bucket/key` (three slashes, empty authority) rather than
+        // the canonical `blob://bucket/key`. `ObjectStoreScheme::parse` in object_store 0.13
+        // requires a Some(host), so a naive rewrite to `s3:///bucket/key` fails at execution
+        // with `Generic URL error: Unable to recognise URL`. `prepare_object_store_with_configs`
+        // must lift the first path segment into the host to match S3's canonical shape.
+        use crate::parquet::parquet_support::prepare_object_store_with_configs;
+        let mut configs: HashMap<String, String> = HashMap::new();
+        configs.insert(
+            "fs.s3a.aws.credentials.provider".to_string(),
+            "org.apache.hadoop.fs.s3a.AnonymousAWSCredentialsProvider".to_string(),
+        );
+        configs.insert(
+            "fs.s3a.endpoint.region".to_string(),
+            "us-east-1".to_string(),
+        );
+        let (object_store_url, path) = prepare_object_store_with_configs(
+            Arc::new(RuntimeEnv::default()),
+            "blob:///mybucket/warehouse/data/part-0.snappy.parquet".to_string(),
+            &configs,
+        )
+        .expect("blob:///bucket/... should be normalized to s3://bucket/...");
+        assert_eq!(
+            object_store_url,
+            ObjectStoreUrl::parse("s3://mybucket").unwrap()
+        );
+        assert_eq!(path, Path::from("warehouse/data/part-0.snappy.parquet"));
+    }
 }