fix: decide libhdfs routing from the scheme as written

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 (`<endpoint>/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;
 

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<String, String>,
-) -> Result<Url, ExecutionError> {
+) -> Result<NormalizedObjectStoreUrl, ExecutionError> {
     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, String>) -> 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<String, String> {
+        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

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<String, String>,
 ) -> 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<String, String> = 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}"
+        );
+    }
 }