diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs index fe01bdcfeec..3b2080db57d 100644 --- a/native/core/src/execution/planner.rs +++ b/native/core/src/execution/planner.rs @@ -468,7 +468,7 @@ impl PhysicalPlanner { // object-store key we hand DataFusion is stripped of the bucket prefix. Skipping this // would leave `bucket/key` as the object key, and path-style S3 GETs would double the // bucket (`/bucket/bucket/key`). - let url = normalize_object_store_url(&file.file_path, object_store_options)?; + let url = normalize_object_store_url(&file.file_path, object_store_options)?.url; let path = Path::from_url_path(url.path()).map_err(|e| GeneralError(e.to_string()))?; partitioned_file.object_meta.location = path; diff --git a/native/core/src/parquet/objectstore/s3_blob_fs_support.rs b/native/core/src/parquet/objectstore/s3_blob_fs_support.rs index 137ffa626aa..0ec4c572519 100644 --- a/native/core/src/parquet/objectstore/s3_blob_fs_support.rs +++ b/native/core/src/parquet/objectstore/s3_blob_fs_support.rs @@ -56,23 +56,43 @@ use url::Url; use crate::execution::operators::ExecutionError; use crate::parquet::parquet_support::{is_hdfs_scheme, scheme_in_list}; +/// A URL after alias normalization, paired with the libhdfs routing decision that produced it. +/// +/// [`Self::is_hdfs`] is answered on the URL AS WRITTEN, before any rewrite, and callers must read +/// it from here rather than re-running [`is_hdfs_scheme`] on [`Self::url`]. Normalization can land +/// an alias on a scheme the user listed in `fs.comet.libhdfs.schemes`: `s3a://bucket/k` becomes +/// `s3://bucket/k`, so a `fs.comet.libhdfs.schemes=s3` that never mentioned `s3a` would capture it +/// on the second look and push an S3 read through the libhdfs bridge. That would also break +/// lockstep with the JVM gate, which matches `libhdfsSchemes` against the scheme the user wrote +/// (`CometScanRule.classifyRootPaths`) and so admits the scan as object_store-native. +pub(crate) struct NormalizedObjectStoreUrl { + pub(crate) url: Url, + pub(crate) is_hdfs: bool, +} + /// Rewrites `s3a` and the configured s3-compliant aliases to `s3://bucket/key`, promoting a missing /// authority into the host. Non-alias schemes are returned unchanged. `s3a` routed through libhdfs /// (`fs.comet.libhdfs.schemes`) is left alone. pub(crate) fn normalize_object_store_url( url_str: &str, object_store_configs: &HashMap, -) -> Result { +) -> Result { let 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); + return Ok(NormalizedObjectStoreUrl { url, is_hdfs: true }); } let scheme = url.scheme(); if scheme != "s3a" && !is_s3_compliant_alias_scheme(scheme, object_store_configs) { - return Ok(url); + return Ok(NormalizedObjectStoreUrl { + url, + is_hdfs: false, + }); } - rewrite_alias_to_s3(url) + Ok(NormalizedObjectStoreUrl { + url: rewrite_alias_to_s3(url)?, + is_hdfs: false, + }) } /// True if `scheme` is a configured s3-compliant alias (`fs.comet.s3Compliant.schemes`; empty or @@ -261,6 +281,7 @@ mod tests { fn normalized(url: &str, configs: &HashMap) -> String { normalize_object_store_url(url, configs) .unwrap() + .url .as_str() .to_string() } @@ -320,6 +341,84 @@ mod tests { } } + /// Config routing `schemes` through libhdfs, optionally opting `aliases` in as S3 aliases. + fn libhdfs_configs(schemes: &str, aliases: Option<&str>) -> HashMap { + let mut configs = HashMap::new(); + configs.insert("fs.comet.libhdfs.schemes".to_string(), schemes.to_string()); + if let Some(aliases) = aliases { + configs.insert( + "fs.comet.s3Compliant.schemes".to_string(), + aliases.to_string(), + ); + } + configs + } + + #[test] + fn test_libhdfs_routing_uses_the_scheme_as_written() { + // The libhdfs decision must be read off the URL the user wrote, never off the normalized + // one. `fs.comet.libhdfs.schemes=s3` asks for `s3://` to go through libhdfs and says + // nothing about `s3a` or aliases, but normalization rewrites both onto `s3://`, so a + // second `is_hdfs_scheme` call on the result would capture them. `CometScanRule` matches + // the scheme as written too, and would already have admitted such a scan as + // object_store-native, so the recompute also desyncs the planner from the executor. + for (input, configs, expect_hdfs, expect_url) in [ + // `s3a` and an opted-in alias both normalize onto `s3`, which IS in the libhdfs list. + // They must still route to the S3 store. + ( + "s3a://bucket/f.parquet", + libhdfs_configs("s3", None), + false, + "s3://bucket/f.parquet", + ), + ( + "blob://bucket/f.parquet", + libhdfs_configs("s3", Some("blob")), + false, + "s3://bucket/f.parquet", + ), + // Listing `s3a` itself is the supported way to route it through libhdfs: honored, and + // the URL is left alone so the name node keeps the scheme the user configured. + ( + "s3a://bucket/f.parquet", + libhdfs_configs("s3a", None), + true, + "s3a://bucket/f.parquet", + ), + // With the config unset only `hdfs` routes to libhdfs, so the default is unaffected. + ( + "s3a://bucket/f.parquet", + HashMap::new(), + false, + "s3://bucket/f.parquet", + ), + ( + "hdfs://nn:8020/f.parquet", + HashMap::new(), + true, + "hdfs://nn:8020/f.parquet", + ), + ] { + let normalized = normalize_object_store_url(input, &configs).unwrap(); + assert_eq!(normalized.is_hdfs, expect_hdfs, "is_hdfs for {input}"); + assert_eq!(normalized.url.as_str(), expect_url, "url for {input}"); + // The flag always agrees with the scheme as written. That is the whole contract. + let as_written = Url::parse(input).unwrap(); + assert_eq!( + is_hdfs_scheme(&as_written, &configs), + expect_hdfs, + "as-written scheme for {input}" + ); + } + + // Pin the trap itself: re-deriving the flag from the normalized URL DOES flip, which is + // why `NormalizedObjectStoreUrl` carries it rather than leaving callers to recompute. + let configs = libhdfs_configs("s3", None); + let normalized = normalize_object_store_url("s3a://bucket/f.parquet", &configs).unwrap(); + assert!(!normalized.is_hdfs); + assert!(is_hdfs_scheme(&normalized.url, &configs)); + } + #[test] fn test_is_s3_compliant_alias_scheme() { // Opt-in and case-insensitive; `s3a` is intentionally NOT reported (callers special-case diff --git a/native/core/src/parquet/parquet_support.rs b/native/core/src/parquet/parquet_support.rs index a461e8a16c0..dd7eb402668 100644 --- a/native/core/src/parquet/parquet_support.rs +++ b/native/core/src/parquet/parquet_support.rs @@ -49,7 +49,9 @@ use std::{fmt::Debug, hash::Hash, sync::Arc}; use url::Url; use super::objectstore; -use super::objectstore::s3_blob_fs_support::normalize_object_store_url; +use super::objectstore::s3_blob_fs_support::{ + normalize_object_store_url, NormalizedObjectStoreUrl, +}; // This file originates from cast.rs. While developing native scan support and implementing // SparkSchemaAdapter we observed that Spark's type conversion logic on Parquet reads does not @@ -665,8 +667,13 @@ pub(crate) fn prepare_object_store_with_configs( url: String, object_store_configs: &HashMap, ) -> Result<(ObjectStoreUrl, Path), ExecutionError> { - let url = normalize_object_store_url(url.as_str(), object_store_configs)?; - let is_hdfs_scheme = is_hdfs_scheme(&url, object_store_configs); + // `is_hdfs` comes back from normalization because it must be decided on the URL as written. + // Re-deriving it from the normalized URL would let an `s3a`/alias rewrite land on an `s3` + // entry in `fs.comet.libhdfs.schemes` and route an S3 read through libhdfs. + let NormalizedObjectStoreUrl { + url, + is_hdfs: is_hdfs_scheme, + } = normalize_object_store_url(url.as_str(), object_store_configs)?; let scheme = url.scheme(); let url_key = format!( "{}://{}", @@ -1144,4 +1151,61 @@ mod tests { assert_eq!(path, Path::from(expected_path)); } } + + #[cfg(not(feature = "hdfs-opendal"))] + #[test] + #[cfg_attr(miri, ignore)] // AWS credential providers and object_store call foreign functions + fn test_prepare_object_store_keeps_s3a_off_libhdfs_when_only_s3_is_listed() { + // `fs.comet.libhdfs.schemes=s3` routes `s3://` through libhdfs and says nothing about + // `s3a` or the opted-in aliases. Both normalize onto `s3://`, so deciding libhdfs from the + // normalized URL would hand these scans to `create_hdfs_object_store` -- which in this + // build is the "not enabled" stub, and in a default build would point libhdfs at a name + // node of `s3://bucket`. The JVM gate classifies them as object_store-native and admits + // them, so this dispatch is what keeps native in lockstep with the planner. + use crate::parquet::parquet_support::prepare_object_store_with_configs; + let mut configs: HashMap = HashMap::new(); + configs.insert("fs.comet.libhdfs.schemes".to_string(), "s3".to_string()); + configs.insert( + "fs.comet.s3Compliant.schemes".to_string(), + "blob".to_string(), + ); + 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(), + ); + + for input in [ + "s3a://test_bucket/comet/part-00000.snappy.parquet", + "blob://test_bucket/comet/part-00000.snappy.parquet", + ] { + let (object_store_url, path) = prepare_object_store_with_configs( + Arc::new(RuntimeEnv::default()), + input.to_string(), + &configs, + ) + .unwrap_or_else(|e| panic!("{input} must build an S3 store, not libhdfs: {e}")); + assert_eq!( + object_store_url, + ObjectStoreUrl::parse("s3://test_bucket").unwrap() + ); + assert_eq!(path, Path::from("/comet/part-00000.snappy.parquet")); + } + + // Listing `s3a` is the supported way to route it through libhdfs, and still does. + configs.insert("fs.comet.libhdfs.schemes".to_string(), "s3a".to_string()); + let err = prepare_object_store_with_configs( + Arc::new(RuntimeEnv::default()), + "s3a://test_bucket/comet/part-00000.snappy.parquet".to_string(), + &configs, + ) + .expect_err("an explicitly listed s3a must reach the libhdfs backend"); + assert!( + err.to_string().contains("Hdfs support is not enabled"), + "unexpected error: {err}" + ); + } }