From 6d3f02c06cc4cabf602e441df97a003e24da462c Mon Sep 17 00:00:00 2001 From: comphead Date: Sat, 8 Aug 2026 11:56:05 -0700 Subject: [PATCH] feat: support S3 compliant filesystems --- .../src/execution/operators/iceberg_scan.rs | 18 ++- native/core/src/execution/planner.rs | 32 +++-- native/core/src/lib.rs | 5 +- .../src/parquet/objectstore/blob_alias.rs | 130 ++++++++++++++++++ native/core/src/parquet/objectstore/mod.rs | 1 + native/core/src/parquet/objectstore/s3.rs | 8 +- native/core/src/parquet/parquet_support.rs | 81 +++++++++-- .../comet/parquet/CometFileKeyUnwrapper.java | 14 +- .../comet/objectstore/NativeConfig.scala | 78 ++++++++++- .../apache/comet/rules/CometScanRule.scala | 37 +++-- .../parquet/TestCometFileKeyUnwrapper.java | 59 ++++++++ .../comet/objectstore/NativeConfigSuite.scala | 96 ++++++++++++- .../rules/CometScanSchemeFallbackSuite.scala | 36 +++++ 13 files changed, 557 insertions(+), 38 deletions(-) create mode 100644 native/core/src/parquet/objectstore/blob_alias.rs create mode 100644 spark/src/test/java/org/apache/comet/parquet/TestCometFileKeyUnwrapper.java diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs index e727294fd9e..f56fbd20537 100644 --- a/native/core/src/execution/operators/iceberg_scan.rs +++ b/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()) } } diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index c179c3b57c5..eea9c24b45a 100644 --- a/native/core/src/execution/planner.rs +++ b/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, ExecutionError>; @@ -388,6 +390,10 @@ impl PhysicalPlanner { partition: &SparkFilePartition, ) -> Result, 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 = 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 + // (`/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, diff --git a/native/core/src/lib.rs b/native/core/src/lib.rs index 6cfe33223f1..2178e4be44f 100644 --- a/native/core/src/lib.rs +++ b/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) }) diff --git a/native/core/src/parquet/objectstore/blob_alias.rs b/native/core/src/parquet/objectstore/blob_alias.rs new file mode 100644 index 00000000000..f183f401f19 --- /dev/null +++ b/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, +) -> Result { + 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 { + 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"); + } +} diff --git a/native/core/src/parquet/objectstore/mod.rs b/native/core/src/parquet/objectstore/mod.rs index bae5ac6f51f..605eb275cb1 100644 --- a/native/core/src/parquet/objectstore/mod.rs +++ b/native/core/src/parquet/objectstore/mod.rs @@ -16,4 +16,5 @@ // under the License. pub mod azure; +pub mod blob_alias; pub mod s3; diff --git a/native/core/src/parquet/objectstore/s3.rs b/native/core/src/parquet/objectstore/s3.rs index f9c5f7a8855..c10fc60816e 100644 --- a/native/core/src/parquet/objectstore/s3.rs +++ b/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()); diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index 2ee1230ed87..a35b6ea6bb8 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/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, ) -> 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 = 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 = 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")); + } } diff --git a/spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java b/spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java index 1e71c25a07a..a07dd9f1318 100644 --- a/spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java +++ b/spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java @@ -103,21 +103,27 @@ public class CometFileKeyUnwrapper { /** * Normalizes S3 URI schemes to a canonical form. S3 can be accessed via multiple schemes (s3://, - * s3a://, s3n://) that refer to the same logical filesystem. This method ensures consistent cache - * lookups regardless of which scheme is used. + * s3a://, s3n://, blob://) that refer to the same logical filesystem. This method ensures + * consistent cache lookups regardless of which scheme is used. The put and get sides must agree, + * because the JVM store side is called with the user-facing scheme (e.g. blob://) while the + * native side JNIs back with the scheme after `prepare_object_store_with_configs` has already + * rewritten aliases to s3://. * * @param filePath The file path that may contain an S3 URI * @return The file path with normalized S3 scheme (s3a://) */ private String normalizeS3Scheme(final String filePath) { - // Normalize s3:// and s3n:// to s3a:// for consistent cache lookups - // This handles the case where ObjectStoreUrl uses s3:// but Spark uses s3a:// + // Normalize s3://, s3n://, and blob:// to s3a:// for consistent cache lookups + // This handles the case where ObjectStoreUrl uses s3:// but Spark uses s3a:// or blob:// String s3Prefix = "s3://"; String s3nPrefix = "s3n://"; + String blobPrefix = "blob://"; if (filePath.startsWith(s3Prefix)) { return "s3a://" + filePath.substring(s3Prefix.length()); } else if (filePath.startsWith(s3nPrefix)) { return "s3a://" + filePath.substring(s3nPrefix.length()); + } else if (filePath.startsWith(blobPrefix)) { + return "s3a://" + filePath.substring(blobPrefix.length()); } return filePath; } diff --git a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala index 6c2436ac39a..81988dbbb47 100644 --- a/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala +++ b/spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala @@ -30,9 +30,11 @@ import org.apache.comet.CometConf.COMET_LIBHDFS_SCHEMES_KEY object NativeConfig { private val objectStoreConfigPrefixes = Map( - // Amazon S3 configurations + // Amazon S3 configurations. `blob` is a Comet-recognized synonym for `s3` and shares the + // same Hadoop `fs.s3a.*` credential surface (see `prepare_object_store_with_configs`). "s3" -> Seq("fs.s3a."), "s3a" -> Seq("fs.s3a."), + "blob" -> Seq("fs.s3a."), // Google Cloud Storage configurations "gs" -> Seq("fs.gs."), // Azure Blob Storage configurations (can use both prefixes) @@ -42,6 +44,55 @@ object NativeConfig { "abfs" -> Seq("fs.azure.", "fs.abfs."), "abfss" -> Seq("fs.azure.", "fs.abfss.", "fs.abfs.")) + private val blobKeyPattern = "^fs\\.blob\\.([^.]+)\\.(.+)$".r + + // Some blob:// filesystem implementations fall back to the literal string "default" as the + // authority when the URI has none. For `blob:///bucket/key`, the filesystem therefore looks + // up `fs.blob.default.*`, while the actual S3 bucket comes from the URL path. Translate + // `fs.blob.default.*` to the GLOBAL `fs.s3a.*` key (not the per-bucket + // `fs.s3a.bucket.default.*`) so the credentials/endpoint apply to whichever bucket the URL + // path resolves to, matching those implementations' semantics. + private val blobDefaultAuthority = "default" + + /** + * Translates vendor-style `fs.blob..` keys into the `fs.s3a.*` shape that + * object_store's AmazonS3Builder reads. Some blob:// connectors use per-authority keys and + * never set a region -- an endpoint alone is enough for the AWS SDK v1 client they build -- and + * their endpoints are typically path-style against non-AWS services, so an `endpoint` key also + * enables `path.style.access` on the same scope. + * + * `fs.blob..*` is the authoritative source for `blob://` URLs, so callers should + * apply these translations AFTER a plain `fs.s3a.*` pass so blob-supplied values override any + * unrelated `fs.s3a.*` the user set for a different workload (see 403 misdirect in the class + * docstring). + */ + private def translateBlobKeys(hadoopConf: Configuration): Map[String, String] = { + import scala.jdk.CollectionConverters._ + val out = scala.collection.mutable.Map[String, String]() + hadoopConf.iterator().asScala.foreach { entry => + entry.getKey match { + case blobKeyPattern(authority, property) => + val s3aSuffix = property match { + case "endpoint" => "endpoint" + case "awsAccessKeyId" => "access.key" + case "awsSecretAccessKey" => "secret.key" + case _ => null + } + if (s3aSuffix != null) { + val scope = + if (authority == blobDefaultAuthority) "fs.s3a" + else s"fs.s3a.bucket.$authority" + out(s"$scope.$s3aSuffix") = entry.getValue + if (s3aSuffix == "endpoint") { + out.getOrElseUpdate(s"$scope.path.style.access", "true") + } + } + case _ => + } + } + out.toMap + } + /** * Extract object store configurations from Hadoop configuration for native DataFusion usage. * This includes S3, GCS, Azure and other cloud storage configurations. @@ -51,6 +102,21 @@ object NativeConfig { * fs.s3a.bucket.{bucket-name}.access.key). The native code will prioritize per-bucket * configurations over global ones when both are present. * + * For `blob://` URIs it also translates vendor-style `fs.blob..endpoint`, + * `fs.blob..awsAccessKeyId`, and `fs.blob..awsSecretAccessKey` into the + * equivalent `fs.s3a.bucket..endpoint`, `access.key`, and `secret.key`. Because + * those blob:// backends usually talk path-style to non-AWS endpoints and never set a region, + * the translation also enables `fs.s3a.bucket..path.style.access=true` when a blob + * endpoint is present. + * + * `fs.blob..*` is the authoritative source for `blob://` URLs: the user may have + * `fs.s3a.*` keys targeting a completely different s3a-scheme workload in the same Spark + * session, and leaking those into blob-scheme connections silently redirects credentials to the + * wrong service (produces a 403 "The access key Id you provided does not exist in our + * records"). For `blob://` URIs, blob translations therefore OVERRIDE any conflicting + * `fs.s3a.*` values the user also set. If you actually want to override a blob endpoint, change + * the `fs.blob..endpoint` value itself. + * * The configurations are passed to the native code which uses object_store's parse_url_opts for * consistent and standardized cloud storage support across all providers. */ @@ -75,13 +141,17 @@ object NativeConfig { // Extract all configurations that match the object store prefixes hadoopConf.iterator().asScala.foreach { entry => val key = entry.getKey - val value = entry.getValue - // Check if key starts with any of the prefixes for this scheme if (prefixes.get.exists(prefix => key.startsWith(prefix))) { - options(key) = value + options(key) = entry.getValue } } + // For blob:// URIs, apply vendor-key translation AFTER the fs.s3a.* pass so blob values + // override any unrelated fs.s3a.* the user set for a different workload. + if (scheme == "blob") { + translateBlobKeys(hadoopConf).foreach { case (k, v) => options(k) = v } + } + options.toMap } } diff --git a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala index 2c15bb5e4e7..ac8d83ceae7 100644 --- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala +++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala @@ -591,9 +591,9 @@ case class CometScanRule(session: SparkSession) val allSupportedFilesystems = if (taskValidation.unsupportedSchemes.isEmpty) { true } else { - fallbackReasons += "Iceberg scan contains files with unsupported filesystem " + - s"schemes: ${taskValidation.unsupportedSchemes.mkString(", ")}. " + - "Comet only supports: file, s3, s3a, gs, gcs, oss, abfss, abfs, wasbs, wasb" + fallbackReasons += "Iceberg scan contains files with filesystem schemes not " + + "recognized by Comet's native object_store: " + + s"${taskValidation.unsupportedSchemes.toSeq.sorted.mkString(", ")}" false } @@ -975,6 +975,26 @@ object CometScanRule extends Logging { val SKIP_COMET_SCAN_TAG: org.apache.spark.sql.catalyst.trees.TreeNodeTag[Unit] = org.apache.spark.sql.catalyst.trees.TreeNodeTag[Unit]("comet.skipCometScan") + /** + * Schemes readable by iceberg-rust's OpenDAL storage factory that `ObjectStoreScheme::parse` + * does NOT recognize, so `isNativelyReadableScheme` alone under-admits for iceberg scans. + * Mirror of the extra `match` arms in + * `native/core/src/execution/operators/iceberg_scan.rs::storage_factory_for` -- add here what + * you add there. Currently Aliyun OSS via `OpenDalStorageFactory::Oss`. + */ + private val icebergExtraSchemes: Set[String] = Set("oss") + + /** + * Scheme gate for the Iceberg scan path. Accepts anything the Parquet scan gate accepts, plus + * schemes iceberg-rust reads via OpenDAL that object_store's parser doesn't recognize. + */ + private[rules] def isIcebergReadableScheme(uri: URI): Boolean = { + if (isNativelyReadableScheme(uri)) return true + Option(uri.getScheme) + .map(_.toLowerCase(Locale.ROOT)) + .exists(icebergExtraSchemes.contains) + } + /** * Single-pass validation of Iceberg FileScanTasks. * @@ -999,9 +1019,6 @@ object CometScanRule extends Logging { val deletesMethod = IcebergReflection.getMethod(fileScanTaskClass, "deletes") val termMethod = IcebergReflection.getMethod(unboundPredicateClass, "term") - val supportedSchemes = - Set("file", "s3", "s3a", "gs", "gcs", "oss", "abfss", "abfs", "wasbs", "wasb") - var allParquet = true val unsupportedSchemes = mutable.Set[String]() var nonIdentityTransform: Option[String] = None @@ -1016,12 +1033,14 @@ object CometScanRule extends Logging { allParquet = false } - // Filesystem scheme check for data file + // Filesystem scheme check for data file. Delegated to the native gate + // (`isNativelyReadableScheme` -> `NativeBase.isObjectStoreSchemeSupported`) so the iceberg + // and Parquet-scan paths share a single source of truth. try { val filePath = pathMethod.invoke(dataFile).toString val uri = new URI(filePath) val scheme = uri.getScheme - if (scheme != null && !supportedSchemes.contains(scheme)) { + if (scheme != null && !isIcebergReadableScheme(uri)) { unsupportedSchemes += scheme } } catch { @@ -1059,7 +1078,7 @@ object CometScanRule extends Logging { try { val deleteUri = new URI(deletePath) val deleteScheme = deleteUri.getScheme - if (deleteScheme != null && !supportedSchemes.contains(deleteScheme)) { + if (deleteScheme != null && !isIcebergReadableScheme(deleteUri)) { unsupportedSchemes += deleteScheme } } catch { diff --git a/spark/src/test/java/org/apache/comet/parquet/TestCometFileKeyUnwrapper.java b/spark/src/test/java/org/apache/comet/parquet/TestCometFileKeyUnwrapper.java new file mode 100644 index 00000000000..9cbdbe33f84 --- /dev/null +++ b/spark/src/test/java/org/apache/comet/parquet/TestCometFileKeyUnwrapper.java @@ -0,0 +1,59 @@ +/* + * 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. + */ + +package org.apache.comet.parquet; + +import java.lang.reflect.Method; + +import org.junit.Test; + +import static org.junit.Assert.assertEquals; + +/** + * Regression guard for {@link CometFileKeyUnwrapper#normalizeS3Scheme}. The put side is called with + * the user-facing URI (e.g. {@code blob://...}), while the native side JNIs back with the URI after + * Comet has rewritten it to {@code s3://} in {@code prepare_object_store_with_configs}. Both sides + * must normalize to the SAME canonical form ({@code s3a://}) or encrypted Parquet reads over {@code + * blob://} tables fail with {@code Failed to find DecryptionKeyRetriever}. + */ +public class TestCometFileKeyUnwrapper { + + private static String normalize(String filePath) throws Exception { + Method m = CometFileKeyUnwrapper.class.getDeclaredMethod("normalizeS3Scheme", String.class); + m.setAccessible(true); + return (String) m.invoke(new CometFileKeyUnwrapper(), filePath); + } + + @Test + public void allS3AliasesNormalizeToS3a() throws Exception { + String suffix = "bucket/foo/part-0.parquet"; + String canonical = "s3a://" + suffix; + assertEquals(canonical, normalize("s3://" + suffix)); + assertEquals(canonical, normalize("s3a://" + suffix)); + assertEquals(canonical, normalize("s3n://" + suffix)); + assertEquals(canonical, normalize("blob://" + suffix)); + } + + @Test + public void nonS3SchemesPassThrough() throws Exception { + assertEquals( + "hdfs://nn/warehouse/part-0.parquet", normalize("hdfs://nn/warehouse/part-0.parquet")); + assertEquals("file:///tmp/part-0.parquet", normalize("file:///tmp/part-0.parquet")); + } +} diff --git a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala index b49b958f588..4147b656be8 100644 --- a/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala +++ b/spark/src/test/scala/org/apache/comet/objectstore/NativeConfigSuite.scala @@ -44,7 +44,10 @@ class NativeConfigSuite extends AnyFunSuite with Matchers { hadoopConf.set("fs.azure.account.key.testaccount.blob.core.windows.net", "azure-key") // Should extract s3 options - Seq("s3a://test-bucket/test-object", "s3://test-bucket/test-object").foreach { path => + Seq( + "s3a://test-bucket/test-object", + "s3://test-bucket/test-object", + "blob://test-bucket/test-object").foreach { path => val options = NativeConfig.extractObjectStoreOptions(hadoopConf, new URI(path)) assert(options("fs.s3a.access.key") == "s3-access-key") assert(options("fs.s3a.secret.key") == "s3-secret-key") @@ -108,4 +111,95 @@ class NativeConfigSuite extends AnyFunSuite with Matchers { s"oauth provider type should be forwarded for $path") } } + + test("extractObjectStoreOptions - blob:// forwards vendor fs.blob..* keys as s3a") { + // Some blob:// connectors use per-authority Hadoop keys and never set a region: the AWS + // SDK v1 client they build is happy with just an endpoint. Comet's native S3 path goes + // through object_store's AmazonS3Builder, which reads `fs.s3a.bucket..`. + // When a user routes a bucket via `blob://` with an existing S3-compliant storage config, + // Comet must translate those keys automatically or the read fails with `Failed to resolve + // region: Bucket not found` (the AWS auto-detect HEAD). + val hadoopConf = new Configuration() + hadoopConf.set("fs.blob.mybucket.endpoint", "https://s3-compat.example.internal") + hadoopConf.set("fs.blob.mybucket.awsAccessKeyId", "AKIA-blob") + hadoopConf.set("fs.blob.mybucket.awsSecretAccessKey", "secret-blob") + // A different authority to make sure translation is per-authority. + hadoopConf.set("fs.blob.other.endpoint", "https://other.example.internal") + + val opts = NativeConfig.extractObjectStoreOptions( + hadoopConf, + new URI("blob://mybucket/dataset/part-0.parquet")) + + assert(opts("fs.s3a.bucket.mybucket.endpoint") == "https://s3-compat.example.internal") + assert(opts("fs.s3a.bucket.mybucket.access.key") == "AKIA-blob") + assert(opts("fs.s3a.bucket.mybucket.secret.key") == "secret-blob") + // Path-style is a common requirement of these S3-compatible services and must be + // propagated so signing targets the endpoint's path form rather than + // .. + assert(opts("fs.s3a.bucket.mybucket.path.style.access") == "true") + // Per-authority translation must also apply to the second bucket seen in the same config. + assert(opts("fs.s3a.bucket.other.endpoint") == "https://other.example.internal") + // No region is set by the vendor connector; Comet must not synthesize one -- object_store's + // builder will default it, and non-AWS services typically accept any region in the SigV4 + // credential. + assert(!opts.contains("fs.s3a.bucket.mybucket.endpoint.region")) + } + + test("extractObjectStoreOptions - blob:// endpoint wins over conflicting fs.s3a.*") { + // For a blob:// URL, `fs.blob..*` is the authoritative namespace. A user's + // `fs.s3a.*` in the same Spark session usually targets an unrelated s3a-scheme workload; + // leaking that endpoint into the blob connection silently sends credentials to the wrong + // service and produces a misleading 403 "access key Id you provided does not exist". + val hadoopConf = new Configuration() + hadoopConf.set("fs.blob.default.endpoint", "s3-compat.example.com") + hadoopConf.set("fs.blob.default.awsAccessKeyId", "AKIA-blob") + hadoopConf.set("fs.blob.default.awsSecretAccessKey", "secret-blob") + // Unrelated s3a-scheme endpoint the user also configured -- must NOT influence blob://. + hadoopConf.set("fs.s3a.endpoint", "other-s3.example.com") + hadoopConf.set("fs.s3a.endpoint.region", "other-region") + + val opts = NativeConfig.extractObjectStoreOptions( + hadoopConf, + new URI("blob:///mybucket/dataset/part-0.parquet")) + assert(opts("fs.s3a.endpoint") == "s3-compat.example.com") + assert(opts("fs.s3a.access.key") == "AKIA-blob") + assert(opts("fs.s3a.secret.key") == "secret-blob") + assert(opts("fs.s3a.path.style.access") == "true") + } + + test("extractObjectStoreOptions - blob translation does not fire for s3:// / s3a://") { + // The fs.blob.* namespace is scoped to blob:// URIs; do not surface it on plain s3/s3a URIs. + val hadoopConf = new Configuration() + hadoopConf.set("fs.blob.mybucket.endpoint", "https://s3-compat.example.internal") + val opts = NativeConfig.extractObjectStoreOptions(hadoopConf, new URI("s3a://mybucket/x")) + assert(!opts.contains("fs.s3a.bucket.mybucket.endpoint")) + } + + test("extractObjectStoreOptions - blob:// maps fs.blob.default.* to global fs.s3a.*") { + // Some blob:// filesystem implementations fall back to authority=\"default\" whenever the + // URI has none. The actual S3 bucket in that case comes from the URL path, so per-bucket + // fs.s3a.bucket.default.* would never match at runtime. Translate to global fs.s3a.* so the + // credentials/endpoint apply to whichever bucket the URL path resolves to. + val hadoopConf = new Configuration() + hadoopConf.set("fs.blob.default.endpoint", "https://s3-compat.example.internal") + hadoopConf.set("fs.blob.default.awsAccessKeyId", "AKIA-default") + hadoopConf.set("fs.blob.default.awsSecretAccessKey", "secret-default") + + // `blob:///mybucket/...` -- URL authority is empty, bucket is in the path. + val opts = NativeConfig.extractObjectStoreOptions( + hadoopConf, + new URI("blob:///mybucket/dataset/part-0.parquet")) + + // GLOBAL keys, not per-bucket. + assert(opts("fs.s3a.endpoint") == "https://s3-compat.example.internal") + assert(opts("fs.s3a.access.key") == "AKIA-default") + assert(opts("fs.s3a.secret.key") == "secret-default") + // Vendor backends typically use path-style; propagate globally so the real-bucket request + // signs path-style even though the user did not set fs.s3a.path.style.access explicitly. + assert(opts("fs.s3a.path.style.access") == "true") + // Must NOT surface as fs.s3a.bucket.default.* -- that key would never match the real bucket + // and the auto-region HEAD to AWS would still fire. + assert(!opts.contains("fs.s3a.bucket.default.endpoint")) + assert(!opts.contains("fs.s3a.bucket.default.access.key")) + } } diff --git a/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala b/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala index fac2e77a106..41ed927d80b 100644 --- a/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala +++ b/spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala @@ -20,6 +20,7 @@ package org.apache.comet.rules import java.io.File +import java.net.URI import java.nio.file.Files import java.util.UUID @@ -100,6 +101,41 @@ class CometScanSchemeFallbackSuite extends CometTestBase { } } + test("blob:// is a Comet-recognized synonym for s3://") { + // `blob` is not in object_store::ObjectStoreScheme, so the native JNI must special-case it + // (native/core/src/lib.rs) for `isNativelyReadableScheme` to return true. Without that + // disjunct, `CometScanRule.transformV1Scan` emits `Unsupported filesystem schemes: blob` + // and falls back to Spark -- the exact regression this suite guards against. Both the + // Parquet-scan and Iceberg-scan gates delegate to `isNativelyReadableScheme`, so this one + // assertion covers both. + assert( + CometScanRule.isNativelyReadableScheme(new URI("blob://bucket/key.parquet")), + "blob:// must be recognized natively; native JNI or the s3-alias rewrite has regressed") + } + + test("iceberg gate: oss:// admitted (regression guard for #)") { + // Aliyun OSS is readable by iceberg-rust via `OpenDalStorageFactory::Oss` + // (native/core/src/execution/operators/iceberg_scan.rs::storage_factory_for), but + // `object_store::ObjectStoreScheme::parse` does NOT recognize it. If the iceberg gate + // delegates to only `isNativelyReadableScheme`, `oss` would fall back to Spark even though + // native can read it. `isIcebergReadableScheme` supplements the JNI check with schemes + // iceberg-rust supports on top of object_store; keep the two in lockstep with the arms in + // `storage_factory_for`. + assert( + CometScanRule.isIcebergReadableScheme(new URI("oss://bucket/key.parquet")), + "oss:// must remain iceberg-readable; icebergExtraSchemes has regressed") + } + + test("iceberg gate: wasbs:// rejected (correctness tightening)") { + // The pre-delegation hardcoded set falsely admitted wasbs/wasb/gcs, none of which + // `storage_factory_for` in native/core/src/execution/operators/iceberg_scan.rs actually + // handles at runtime. Delegating to the native gate closes that false positive so scans + // fall back to Spark up-front instead of failing later during native setup. + assert( + !CometScanRule.isIcebergReadableScheme(new URI("wasbs://container@acct/key.parquet")), + "wasbs:// must not be iceberg-readable; native storage_factory_for has no wasbs arm") + } + test("native scan claims hdfs:// when libhdfs.schemes is unset (native-default lockstep)") { // Native's `is_hdfs_scheme` treats `hdfs` as readable when `fs.comet.libhdfs.schemes` is unset, // and `create_hdfs_object_store` is in the default build. The JVM gate must agree: with the