feat: [branch-1.1] per-location credentials for native Iceberg reads and writes (#6478)

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

@@ -95,13 +95,27 @@ The locations come from asking for the bucket's whole list rather than which pre
 
 The store itself caches no credentials. Each location's bridge reuses its own credential until shortly before its expiry, as [Credential reuse, bounded by the reported expiry](#credential-reuse-bounded-by-the-reported-expiry) describes. What it keeps is the provider's location list and a store for each location that has been read, so it grows with the vendor's policy list, not with the paths read. The list is only fetched again after a 403 or a credential failure, so a change that causes neither is not seen until the executor builds a new store.
 
-The dispatcher returns `null` for a provider that does not implement the interface without calling it, and Comet builds the same plain store as before. The Iceberg path does not use locations. Operations other than reads route by path without the retry, since Comet only reads through these stores.
+The dispatcher returns `null` for a provider that does not implement the interface without calling it, and Comet builds the same plain store as before. The Iceberg path routes by the same locations; see the next section. Operations other than reads route by path without the retry, since Comet only reads through these stores.
+
+## Location-scoped credentials on the Iceberg path
+
+iceberg-rust's OpenDAL S3 storage attaches one credential loader to the operator it builds for every file, and Comet builds that loader from one reference path: the table's metadata location for a scan, its data location for a write. A base provider therefore gets one credential per table, requested with that path. That is wrong for a table whose files span locations with different policies: data outside the table location (`write.data.path`, the object-storage layout, files added by `add_files` or `migrate`), a narrower policy nested under the table location, or files in another bucket.
+
+For a location-scoped provider, `storage_factory_for` returns a `LocationScopedS3StorageFactory` (`execution/operators/iceberg_location_scoped.rs`) instead of `OpenDalStorageFactory::S3`, following the pattern of `BlobHostPromotingS3StorageFactory`. Its storage implements iceberg-rust's `Storage`, whose every method receives the path it operates on:
+
+- Each call is routed by its path to the longest location covering it in that path's bucket, using the same `LocationIndex` as the Parquet store (`cloud/s3/policy_locations.rs`), and delegated to an OpenDAL S3 storage whose loader is a bridge bound to that location. Bucket snapshots and location storages are built on first use and belong to a `SharedLocations` that every `FileIO` of the provider registration shares; see [Shared locations on the Iceberg path](#shared-locations-on-the-iceberg-path).
+- The key a call is routed by is the key the OpenDAL storage asks S3 for: the rest of its path after `{scheme}://{bucket}/`, normalized as opendal normalizes every path, which trims whitespace and drops leading and empty segments. The Parquet store percent-decodes a request's URI to get its key, but an Iceberg path is not a URI: a partition value's escape, such as the `%3A` in `ts=2024-01-01T00%3A00`, is part of the key. Routing stops at the first segment no location can have, such as `..`, since no location covers the key beyond it.
+- The reference bucket's locations are fetched when the registration's first `FileIO` is built, on the thread that loads it, as `create_store` does on the Parquet path. A later `FileIO` whose reference bucket is new to the registration fetches that bucket's on its own loading thread. Another bucket's are fetched when a call first names it, inside an async storage call, so the JVM call runs in `tokio::task::block_in_place`. Bridges for other locations and buckets are derived from the reference bridge and reuse its dispatcher handle: on this path the dispatch key is the catalog name, so one registration serves every bucket a table touches.
+- `new_input` and `new_output` return files bound to this storage rather than to a location's, so every later read of a file is routed and can retry. `reader` returns a `FileRead` that routes each range read at the current snapshot, as a fresh request is routed, and reopens the file when that snapshot routes it elsewhere. So a refresh by another request moves an open reader too, and a failed range read shares only a refresh attempted after it was routed, which keeps a burst of failures on one refresh.
+- The refresh semantics are the Parquet store's, and `PolicyLocations` shares the per-generation bookkeeping between the two: a request that gets a 403, or that fails because the provider could not produce its location's credential, fetches the bucket's locations again, once per snapshot generation, and retries once if its path now routes elsewhere. The second trigger looks different here. reqsign's credential chain logs a provider exception and reports no credential (see [the error message fidelity caveat](#iceberg-path-error-message-fidelity-caveat)), so the signer fails with `CredentialInvalid` and the request is never sent. The storage looks for opendal's `PermissionDenied` or reqsign's `CredentialInvalid` in the source chain of the iceberg error. Unlike the Parquet store, which Comet only reads through, this storage also serves native Iceberg writes, so writes and deletes refresh and retry the same way. A streaming `writer` is the exception: bytes it already handed to a location's writer cannot be sent again without buffering the whole file, so a failed writer fetches the locations but returns the error, and Spark's retry of the task writes the file through the new route. `delete_stream` deletes a failed batch again, through the new routes, if any of its paths now routes elsewhere.
+- For an s3-compliant alias scheme, `BlobHostPromotingS3Storage` wraps this storage, so routing sees the promoted host as the bucket. A hostless reference path, such as `blob:///bucket/...`, is not covered: `build_s3_access` finds no bucket in it and returns `S3Access::Loader(None)` before it looks up the provider, so the table uses opendal's default chain, as it does with a base provider.
+- A provider whose `getPolicyLocations` throws when the factory is built fails the read or write, as on the Parquet path, rather than falling back to one credential for every file.
 
 ## The IRSA web-identity provider keeps its own cache
 
 The bridge's reuse above is bounded by the expiry a vendor reports. The EKS/IRSA web-identity provider in `native/core/src/cloud/s3/web_identity.rs` is Comet's own and keeps a fuller cache. It is not a vendor path -- there is no JVM SPI involved -- so the reasoning above does not apply.
 
-It exists because the implicit IRSA credential path on the **Iceberg scan** mishandles STS throttling. `AssumeRoleWithWebIdentity` is resolved per reader thread; a concurrent startup burst throttles STS; opendal's default reqsign chain does not retry the throttle and falls through to the EKS node instance role, which lacks bucket access, turning a transient throttle into a hard `403` (this is the reported failure). The raw-Parquet path is intentionally left alone: its AWS SDK default chain already retries and stops on a provider error rather than downgrading, and — because `NativeConfig.extractObjectStoreOptions` forwards Hadoop's `core-default.xml` default for `fs.s3a.aws.credentials.provider` — a Parquet take-over gated on "no provider configured" could never engage anyway. So the take-over is Iceberg-only, and covers both reads and writes (both build their `FileIO` through `load_file_io`): when both `AWS_WEB_IDENTITY_TOKEN_FILE` and `AWS_ROLE_ARN` are set, a region is set, and the catalog configures no explicit credentials, `iceberg_common.rs::build_s3_credential_loader` installs `WebIdentityCredentialProvider` (via a `CustomAwsCredentialLoader`) instead of returning `Ok(None)`. It:
+It exists because the implicit IRSA credential path on the **Iceberg scan** mishandles STS throttling. `AssumeRoleWithWebIdentity` is resolved per reader thread; a concurrent startup burst throttles STS; opendal's default reqsign chain does not retry the throttle and falls through to the EKS node instance role, which lacks bucket access, turning a transient throttle into a hard `403` (this is the reported failure). The raw-Parquet path is intentionally left alone: its AWS SDK default chain already retries and stops on a provider error rather than downgrading, and — because `NativeConfig.extractObjectStoreOptions` forwards Hadoop's `core-default.xml` default for `fs.s3a.aws.credentials.provider` — a Parquet take-over gated on "no provider configured" could never engage anyway. So the take-over is Iceberg-only, and covers both reads and writes (both build their `FileIO` through `load_file_io`): when both `AWS_WEB_IDENTITY_TOKEN_FILE` and `AWS_ROLE_ARN` are set, a region is set, and the catalog configures no explicit credentials, `iceberg_common.rs::build_s3_access` installs `WebIdentityCredentialProvider` (via a `CustomAwsCredentialLoader`) instead of returning `S3Access::Loader(None)`. It:
 
 - builds its STS client from the AWS SDK's fully-resolved `SdkConfig` (`aws_config::defaults(...).load()`) with a raised `RetryConfig`, and calls `AssumeRoleWithWebIdentity` on it; because the client comes from the resolved config rather than a hand-assembled one, it honors region, `use_fips`, `use_dual_stack` and any profile/custom STS endpoint with no per-setting copying to keep in sync,
 - reports the credential's real STS expiry (not an early deadline). reqsign's signer refuses to sign within ~10s of the reported expiry and reloads within 120s of it, so an early deadline would open a signing dead zone before every refresh; instead our own cache margin (`min_ttl`, floored to 120s) refreshes at or before the signer's reload point, and a still-valid credential keeps being served if a refresh is throttled,
@@ -132,6 +146,10 @@ The full unfiltered FileIO property bag crosses JNI as `catalog_properties`. The
 
 `IcebergScanExec` derives a redacting `Debug` so plan dumps and tracing do not leak the property bag.
 
+## Shared locations on the Iceberg path
+
+`load_file_io` in `iceberg_common.rs` builds a `FileIO` for each scan and write task. A location-scoped provider's bucket snapshots and location storages are not kept per `FileIO`. `build_s3_access` keeps them per provider registration and access mode, keyed by access mode, dispatch key and the whole catalog property bag, so the tasks of a table, and of the tables whose catalog properties match, share one set. `CometScanRule` adds each table's FileIO properties to the bag, so a REST catalog that vends per-table properties, such as per-table credentials, gets one set per table. Sharing across different bags would cross provider instances, since `ensureInitialized` registers one per bag. Builds for one registration run one at a time, so concurrent tasks on a new snapshot wait for the first rather than each asking the provider. The registry holds weak references, so the state lives as long as some task's `FileIO` uses it: tasks that run at the same time share one set, and a task that starts after the registration's last `FileIO` was dropped asks the provider again.
+
 ## 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.
@@ -155,4 +173,4 @@ The SPI returns a `CometS3Credentials` or throws. There is no sentinel "I do not
 
 ## Iceberg path: error message fidelity caveat
 
-When the bridge is wired into `iceberg-rust`, the outer `reqsign-core::ProvideCredentialChain` currently swallows thrown exceptions into "no credential" before the request reaches `opendal`. The credential is still not issued and the request still fails, but the message is degraded to an opaque anonymous-request failure. No Comet change fixes this; it is resolved upstream when `iceberg-rust` stops wrapping custom loaders in its outer chain or moves its S3 backend to `object_store`.
+When the bridge is wired into `iceberg-rust`, the outer `reqsign-core::ProvideCredentialChain` currently swallows thrown exceptions into "no credential" before the request reaches `opendal`. The credential is still not issued and the request still fails, but the message is degraded to reqsign's generic `failed to load signing credential`, and the request is never sent. No Comet change fixes this; it is resolved upstream when `iceberg-rust` stops wrapping custom loaders in its outer chain or moves its S3 backend to `object_store`.