fix: read each native scan partition through the object store its files live in

.github/workflows/pr_build_linux.yml

@@ -551,6 +551,7 @@ jobs:
               org.apache.comet.rules.CometScanRuleSuite
               org.apache.comet.rules.CometScanContribSuite
               org.apache.comet.rules.CometScanSchemeFallbackSuite
+              org.apache.comet.rules.CometMultiStoreScanSuite
               org.apache.comet.rules.CometExecRuleSuite
               org.apache.spark.sql.comet.CometPlanEqualitySuite
               org.apache.comet.rules.CometEmptyRelationExecRuleSuite

.github/workflows/pr_build_macos.yml

@@ -253,6 +253,7 @@ jobs:
               org.apache.comet.rules.CometScanRuleSuite
               org.apache.comet.rules.CometScanContribSuite
               org.apache.comet.rules.CometScanSchemeFallbackSuite
+              org.apache.comet.rules.CometMultiStoreScanSuite
               org.apache.comet.rules.CometExecRuleSuite
               org.apache.spark.sql.comet.CometPlanEqualitySuite
               org.apache.comet.rules.CometEmptyRelationExecRuleSuite

dev/ci/check-suites.py

@@ -35,6 +35,7 @@ def file_to_class_name(path: Path) -> str | None:
     ignore_list = [
         "org.apache.comet.parquet.ParquetReadSuite", # abstract
         "org.apache.comet.parquet.ParquetReadFromS3Suite", # manual test suite
+        "org.apache.comet.csv.CsvReadFromS3Suite", # manual test suite
         "org.apache.comet.parquet.ParquetReadFromFakeHadoopFsSuite", # manual test suite (loads libhdfs, see #5023)
         "org.apache.comet.IcebergReadFromS3Suite", # manual test suite
         "org.apache.comet.cloud.s3.CometS3CredentialBridgeSuite", # manual test suite

docs/source/contributor-guide/s3-credential-provider-design.md

@@ -154,7 +154,7 @@ The cache holds the `FileIO` and its bridge. The bridge reuses its credential un
 
 ## Property-bag handling on the Parquet path
 
-The Parquet path forwards the full `fs.s3a.*` config subset as `catalog_properties`, so an SPI provider sees the same `fs.s3a.*` config Spark would. `forward_catalog_properties` in `native/core/src/parquet/objectstore/s3.rs` keeps every `fs.s3a.*` key, including the static credentials (`*.access.key`, `*.secret.key`, `*.session.token`). Forwarding them is deliberate: `AWSCredentialProviderList` skips a provider that throws `NoAwsCredentialsException` and moves to the next entry, so stripping the static keys would let a chain such as `SimpleAWSCredentialsProvider,customProvider` silently resolve through a different provider than Spark, reading data as a different principal. These keys already cross JNI for the non-adapter native path (`build_credential_provider` reads them), so this matches existing behavior rather than widening exposure. Because `NativeConfig.extractObjectStoreOptions` only collects `fs.s3a.*` (plus the `fs.comet.*` scheme keys), config a provider reads that is not under `fs.s3a.*` -- notably `hadoop.security.credential.provider.path` -- never crosses JNI. `AdapterSupport.toConfiguration` therefore seeds the Hadoop `Configuration` from the executor's own Spark-derived Hadoop conf (`spark.hadoop.*`) and overlays the forwarded `fs.s3a.*` on top, so those keys are present.
+The Parquet path forwards the full `fs.s3a.*` config subset as `catalog_properties`, so an SPI provider sees the same `fs.s3a.*` config Spark would. `forward_catalog_properties` in `native/core/src/parquet/objectstore/s3.rs` keeps every `fs.s3a.*` key, including the static credentials (`*.access.key`, `*.secret.key`, `*.session.token`). Forwarding them is deliberate: `AWSCredentialProviderList` skips a provider that throws `NoAwsCredentialsException` and moves to the next entry, so stripping the static keys would let a chain such as `SimpleAWSCredentialsProvider,customProvider` silently resolve through a different provider than Spark, reading data as a different principal. These keys already cross JNI for the non-adapter native path (`build_credential_provider` reads them), so this matches existing behavior rather than widening exposure. Because `NativeConfig.extractObjectStoreOptions` only collects the object store prefixes of the schemes the scan reads (plus the `fs.comet.*` scheme keys), config a provider reads that is not under `fs.s3a.*` -- notably `hadoop.security.credential.provider.path` -- never crosses JNI. `AdapterSupport.toConfiguration` therefore seeds the Hadoop `Configuration` from the executor's own Spark-derived Hadoop conf (`spark.hadoop.*`) and overlays the forwarded `fs.s3a.*` on top, so those keys are present.
 
 ## Built-in adapters
 

docs/source/user-guide/latest/datasources.md

@@ -104,6 +104,29 @@ See the dedicated [Comet and Iceberg Guide](iceberg.md).
 
 Comet supports most standard storage systems, such as local file system and object storage.
 
+### Scans over more than one object store
+
+A native Parquet or CSV scan reads each of its partitions through one object store, such as one S3
+bucket, one GCS bucket, or one HDFS name node. When a scan's files live in more than one store,
+Comet puts the files of each store into separate partitions, so such a scan can run with a different
+number of partitions than Spark would use. A native Parquet scan packs each store's files on their
+own, so `spark.sql.files.maxPartitionNum` applies to each store's files separately. A native CSV
+scan that would split a partition falls back to Spark when Spark planned the operators above it for
+the scan's own partitioning, as Spark 3.4 does for a scan of one partition.
+
+Such a scan forwards the object store settings of every scheme it reads, so its files can mix
+schemes such as `s3a://`, `gs://` and `hdfs://`. It falls back to Spark when it reads one bucket
+through an S3-compliant alias and through another scheme (for example `blob://bucket` and
+`s3a://bucket`), because the alias settings would then apply to both. It also falls back when an
+alias path names no bucket (`blob:///`) next to other S3 paths, because that path's alias settings
+would apply to every bucket. An alias scheme that is also listed in `fs.comet.libhdfs.schemes` is
+read through libhdfs, so Comet translates none of its settings and such a path does not cause
+either fallback. ABFS is the exception to mixing stores: Comet resolves the credentials of one
+container and account per scan, so a scan whose files span more than one ABFS container or account,
+or that mixes `abfs`/`abfss` paths with other schemes, falls back to Spark (see [Azure](#azure)). A
+native Parquet scan of a bucketed table also falls back when its files span more than one store,
+because each table bucket is read as one partition.
+
 ### HDFS
 
 The Apache DataFusion Comet Rust-based reader seamlessly scans files from remote HDFS for [supported formats](#supported-spark-data-sources)
@@ -333,11 +356,12 @@ above. The recognized vendor-style properties and their `fs.s3a.*` targets are:
 
 Unrecognized `fs.<s>.<authority>.*` properties are ignored.
 
-A URL with no authority, such as the triple-slash `blob:///bucket/key` form, reports its authority
-as the literal string `default`. For that case, list the keys under `fs.<s>.default.<property>`.
-Comet resolves `default` to the bucket taken from the URL path (`bucket` here) and applies the
-translated settings at that bucket's scope, exactly as if they had been written under
-`fs.<s>.bucket.<property>`.
+Keys under the literal authority `default`, `fs.<s>.default.<property>`, apply to every bucket of
+that scheme that has no `fs.<s>.<bucket>.<property>` key of its own, whether the URL names the
+bucket as its authority (`blob://bucket/key`) or in its path. A URL with no authority, such as the
+triple-slash `blob:///bucket/key` form, reports its authority as `default`, so these keys are the
+way to configure it. Comet applies them at the scope of the bucket the URL names (`bucket` here),
+exactly as if they had been written under `fs.<s>.bucket.<property>`.
 
 When `fs.<s>.<authority>.endpoint` is set, Comet defaults path-style access to enabled for that
 bucket, since most non-AWS S3-compatible services require it. This is only a default: set
@@ -350,9 +374,9 @@ credential providers and options documented above also apply to alias-scheme URL
 translation feeds the native Iceberg scan; see
 [Object store configuration (S3)](iceberg.md#object-store-configuration-s3) in the Iceberg guide.
 
-A native Parquet scan whose alias-scheme paths span more than one bucket falls back to Spark. Alias
-schemes apply to native scans only: a native Iceberg write to an alias-scheme location falls back to
-iceberg-java.
+A native Parquet or CSV scan whose alias-scheme paths span more than one bucket resolves the alias
+settings for each bucket. Alias schemes apply to native scans only: a native Iceberg write to an
+alias-scheme location falls back to iceberg-java.
 
 ### Examples
 

native/core/src/execution/planner.rs

@@ -102,7 +102,9 @@ use iceberg::expr::Bind;
 use crate::execution::operators::ExecutionError::GeneralError;
 use crate::execution::spark_plan::SparkPlan;
 use crate::parquet::objectstore::s3_blob_fs_support::normalize_object_store_url;
-use crate::parquet::parquet_support::{prepare_object_store_with_configs, ObjectStoreBackend};
+use crate::parquet::parquet_support::{
+    object_store_url_key, prepare_object_store_with_configs, ObjectStoreBackend,
+};
 use datafusion::common::scalar::ScalarStructBuilder;
 use datafusion::common::{
     tree_node::{Transformed, TransformedResult, TreeNode, TreeNodeRecursion, TreeNodeRewriter},
@@ -696,7 +698,8 @@ impl PhysicalPlanner {
             .map(Some)
     }
 
-    /// get DataFusion PartitionedFiles from a Spark FilePartition
+    /// get DataFusion PartitionedFiles from a Spark FilePartition. The scan reads a partition
+    /// through a single object store, so every file must resolve to the same one.
     fn get_partitioned_files(
         &self,
         partition: &SparkFilePartition,
@@ -705,6 +708,7 @@ impl PhysicalPlanner {
         let mut files = Vec::with_capacity(partition.partitioned_file.len());
         // Create an empty input schema for partition values because they are all literals.
         let empty_schema = Arc::new(Schema::empty());
+        let mut partition_store: Option<(String, bool)> = None;
         partition.partitioned_file.iter().try_for_each(|file| {
             assert!(file.start + file.length <= file.file_size);
 
@@ -719,7 +723,28 @@ 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)?.url;
+            let normalized = normalize_object_store_url(&file.file_path, object_store_options)?;
+            let store = object_store_url_key(&normalized);
+            match &partition_store {
+                None => partition_store = Some(store),
+                Some(first) if *first != store => {
+                    let describe = |(key, is_hdfs): &(String, bool)| {
+                        if *is_hdfs {
+                            format!("{key} (libhdfs)")
+                        } else {
+                            key.clone()
+                        }
+                    };
+                    return Err(GeneralError(format!(
+                        "A native scan partition holds files from two object stores ({} and {}); \
+                         each partition is read through one store",
+                        describe(first),
+                        describe(&store)
+                    )));
+                }
+                Some(_) => {}
+            }
+            let url = normalized.url;
             let path = Path::from_url_path(url.path()).map_err(|e| GeneralError(e.to_string()))?;
             partitioned_file.object_meta.location = path;
 
@@ -2139,21 +2164,30 @@ impl PhysicalPlanner {
                     .iter()
                     .map(|(k, v)| (k.clone(), v.clone()))
                     .collect();
-                let one_file = scan
-                    .file_partitions
-                    .first()
-                    .and_then(|f| f.partitioned_file.first())
-                    .map(|f| f.file_path.clone())
-                    .ok_or(GeneralError("Failed to locate file".to_string()))?;
-                let (object_store_url, _, _) = prepare_object_store_with_configs(
-                    self.session_ctx.runtime_env(),
-                    one_file,
-                    &object_store_options,
-                )?;
-                let files = self.get_partitioned_files(
-                    &scan.file_partitions[self.partition as usize],
-                    &object_store_options,
-                )?;
+                let partition = usize::try_from(self.partition)
+                    .ok()
+                    .and_then(|p| scan.file_partitions.get(p))
+                    .ok_or_else(|| {
+                        GeneralError(format!(
+                            "CsvScan has no file partition {} ({} partitions)",
+                            self.partition,
+                            scan.file_partitions.len()
+                        ))
+                    })?;
+                // Build the store from this partition's files; other partitions may read
+                // from a different one.
+                let object_store_url = match partition.partitioned_file.first() {
+                    Some(file) => {
+                        prepare_object_store_with_configs(
+                            self.session_ctx.runtime_env(),
+                            file.file_path.clone(),
+                            &object_store_options,
+                        )?
+                        .0
+                    }
+                    None => ObjectStoreUrl::local_filesystem(),
+                };
+                let files = self.get_partitioned_files(partition, &object_store_options)?;
                 let file_groups: Vec<Vec<PartitionedFile>> = vec![files];
                 let scan = init_csv_datasource_exec(
                     object_store_url,
@@ -5288,6 +5322,7 @@ fn needs_fields_coercion(sig: &TypeSignature) -> bool {
 #[cfg(test)]
 mod tests {
     mod empty_native_scan;
+    mod multi_store_scan;
 
     use futures::{poll, StreamExt};
     use std::{

native/core/src/execution/planner/delta_spark_scan.rs

@@ -108,14 +108,12 @@ fn plan_delta_spark_scan(
             .collect::<Result<Vec<_>, _>>()?,
     };
 
-    // Defense-in-depth against a stale or bypassed JVM gate: DeltaScanSupport.declineReason
-    // (multiStoreReason) already declines data files spanning multiple object-store authorities
-    // at planning time, but prepare_scan_store_and_files below resolves this whole partition's
-    // ObjectStoreUrl from the FIRST file only and then strips every other file down to its bare
-    // object-store path -- a file that actually lives under a different authority would
-    // silently read through the first file's store handle. Checked here rather than inside
-    // prepare_scan_store_and_files itself, which is shared with plain NativeScan and out of
-    // scope for this Delta-specific invariant.
+    // Defense in depth behind the JVM gate (DeltaScanSupport.declineReason, multiStoreReason),
+    // which declines data files spanning multiple object-store authorities. The shared
+    // get_partitioned_files also rejects a mixed-store partition, but it compares normalized store
+    // keys and runs after the first file's store is resolved. This check runs before any store is
+    // resolved and compares scheme, userinfo, host and port (see ObjectStoreAuthority), so a
+    // bypassed gate fails with the JVM decline's wording.
     check_same_object_store_authority(&spark_partition.partitioned_file)?;
 
     let (object_store_url, object_store_backend, mut files) =
@@ -232,18 +230,21 @@ fn plan_delta_spark_scan(
 /// from it may be case-sensitive and it is safer to draw more authority distinctions than fewer;
 /// port is compared as `Option<u16>` so an explicit port never collapses into an absent one.
 /// Mirrors `DeltaScanSupport.uriAuthority`'s normalization on the JVM side, which folds scheme,
-/// userinfo, host, and port into one lowercased `getAuthority`-derived key -- both sides must
-/// treat two URIs as the same authority in exactly the same cases so the JVM-side gate
-/// (`multiStoreReason`, which declines) always fires before this native check (which errors) ever
-/// would.
+/// userinfo, host, and port into one lowercased `getAuthority`-derived key. The JVM side also
+/// lowercases userinfo, but it declines every userinfo-bearing URI
+/// (`userInfoBearingAuthorityReason`) before `multiStoreReason` runs, so the two agree on every
+/// URI that reaches the JVM authority comparison and the JVM gate (which declines) fires before
+/// this native check (which errors) would.
 type ObjectStoreAuthority = (String, String, String, Option<u16>);
 
 /// Errors unless every file in `files` shares the first file's [`ObjectStoreAuthority`]. The
 /// `url` crate does NOT lowercase the host for opaque (non-"special") schemes like
 /// `s3a`/`abfss`/`hdfs`, so comparing `url[BeforeHost..AfterPort]` verbatim would treat two
 /// spellings of the same bucket (`s3a://Bucket-A/..` vs `s3a://bucket-a/..`) as different
-/// authorities and hard-error instead of gracefully declining. See the call site's comment for
-/// why this defensive check exists alongside the JVM-side gate.
+/// authorities and hard-error instead of gracefully declining. Two spellings that differ only in
+/// host case therefore pass here, and if they share a partition `get_partitioned_files` rejects
+/// it, since its store key keeps an opaque scheme's host as written. See the call site's comment
+/// for why this defensive check exists alongside the JVM-side gate.
 fn check_same_object_store_authority(files: &[SparkPartitionedFile]) -> Result<(), ExecutionError> {
     let mut first: Option<(ObjectStoreAuthority, &str)> = None;
     for file in files {