You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by mb...@apache.org on 2026/05/28 11:37:22 UTC
(datafusion-comet) branch main updated: feat: vendor-pluggable S3 credentials for native scans (#4309)
This is an automated email from the ASF dual-hosted git repository.
mbutrovich pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
The following commit(s) were added to refs/heads/main by this push:
new a08cb4eca7 feat: vendor-pluggable S3 credentials for native scans (#4309)
a08cb4eca7 is described below
commit a08cb4eca7dc4fbf107cf10bb6dc7f32e56f2aa1
Author: Matt Butrovich <mb...@users.noreply.github.com>
AuthorDate: Thu May 28 07:37:18 2026 -0400
feat: vendor-pluggable S3 credentials for native scans (#4309)
---
dev/ci/check-suites.py | 1 +
docs/source/contributor-guide/index.md | 1 +
.../s3-credential-provider-design.md | 106 ++++++
docs/source/user-guide/latest/index.rst | 1 +
.../user-guide/latest/s3-credential-providers.md | 292 +++++++++++++++++
native/Cargo.lock | 308 +++++++++++++-----
native/Cargo.toml | 5 +-
native/core/Cargo.toml | 1 +
native/core/src/cloud/mod.rs | 23 ++
native/core/src/cloud/s3/credential_bridge.rs | 361 +++++++++++++++++++++
native/core/src/cloud/s3/mod.rs | 18 +
.../core/src/execution/operators/iceberg_scan.rs | 125 ++++++-
native/core/src/execution/planner.rs | 2 +
native/core/src/lib.rs | 1 +
native/core/src/parquet/mod.rs | 2 +-
native/core/src/parquet/objectstore/s3.rs | 45 ++-
.../src/comet_s3_credential_dispatcher.rs | 90 +++++
native/jni-bridge/src/lib.rs | 5 +
native/proto/src/proto/operator.proto | 6 +
pom.xml | 19 +-
.../apache/comet/cloud/s3/CometS3AccessMode.java | 28 ++
.../comet/cloud/s3/CometS3CredentialContext.java | 61 ++++
.../cloud/s3/CometS3CredentialDispatcher.java | 205 ++++++++++++
.../comet/cloud/s3/CometS3CredentialProvider.java | 54 +++
.../apache/comet/cloud/s3/CometS3Credentials.java | 59 ++++
.../java/org/apache/comet/util/ClassLoaders.java | 43 +++
.../apache/comet/iceberg/IcebergReflection.scala | 78 ++++-
.../org/apache/comet/rules/CometScanRule.scala | 9 +-
.../serde/operator/CometIcebergNativeScan.scala | 16 +-
.../cloud/s3/CometS3CredentialDispatcherTest.java | 270 +++++++++++++++
.../cloud/s3/MinioCometS3CredentialProvider.java | 130 ++++++++
.../apache/comet/cloud/s3/NoNoArgCtorProvider.java | 32 ++
.../comet/cloud/s3/NotACredentialProvider.java | 25 ++
.../cloud/s3/TestCometS3CredentialProvider.java | 88 +++++
.../scala/org/apache/comet/CometS3TestBase.scala | 20 ++
.../org/apache/comet/IcebergReadFromS3Suite.scala | 12 +-
.../cloud/s3/CometS3CredentialBridgeSuite.scala | 237 ++++++++++++++
.../cloud/s3/IcebergRESTVendedS3Provider.java | 72 ++++
.../cloud/s3/IcebergRESTVendedS3ProviderTest.java | 97 ++++++
39 files changed, 2808 insertions(+), 140 deletions(-)
diff --git a/dev/ci/check-suites.py b/dev/ci/check-suites.py
index 279c6a89c9..b7369d1707 100644
--- a/dev/ci/check-suites.py
+++ b/dev/ci/check-suites.py
@@ -35,6 +35,7 @@ if __name__ == "__main__":
"org.apache.comet.parquet.ParquetReadSuite", # abstract
"org.apache.comet.parquet.ParquetReadFromS3Suite", # manual test suite
"org.apache.comet.IcebergReadFromS3Suite", # manual test suite
+ "org.apache.comet.cloud.s3.CometS3CredentialBridgeSuite", # manual test suite
"org.apache.spark.sql.comet.CometPlanStabilitySuite", # abstract
"org.apache.spark.sql.comet.ParquetDatetimeRebaseSuite", # abstract
"org.apache.comet.exec.CometColumnarShuffleSuite" # abstract
diff --git a/docs/source/contributor-guide/index.md b/docs/source/contributor-guide/index.md
index c601dab87d..55b6552371 100644
--- a/docs/source/contributor-guide/index.md
+++ b/docs/source/contributor-guide/index.md
@@ -49,6 +49,7 @@ Arrow FFI <ffi>
JVM Shuffle <jvm_shuffle>
Native Shuffle <native_shuffle>
ANSI Error Propagation <sql_error_propagation>
+S3 Credential Provider Design <s3-credential-provider-design>
```
```{toctree}
diff --git a/docs/source/contributor-guide/s3-credential-provider-design.md b/docs/source/contributor-guide/s3-credential-provider-design.md
new file mode 100644
index 0000000000..00504c3ea3
--- /dev/null
+++ b/docs/source/contributor-guide/s3-credential-provider-design.md
@@ -0,0 +1,106 @@
+<!--
+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 Credential Provider SPI: Design Notes
+
+This page captures why the `org.apache.comet.cloud.s3.CometS3CredentialProvider` SPI is shaped the way it is. The user-facing contract and operator setup live in the user guide page on S3 credential providers; this page is for maintainers and reviewers who want the design rationale.
+
+## The gap the SPI fills
+
+Comet's native scan paths (`object_store` for raw Parquet, `opendal` via `iceberg-rust` for Iceberg) bypass Spark's Hadoop S3A code path. That means credentials cannot flow through any of the contracts that vendors typically wire into for S3A:
+
+- `org.apache.spark.deploy.security.cloud.CloudCredentialsProvider` yields a single JWT per service name. No path argument, no AWS credential.
+- Hadoop S3A custom signers hide path-aware logic inside `Signer.sign(request, credentials)`. The credential never leaves the signing pipeline, and the underlying secret is an HMAC key that is not present in the signed output, so running the signer against a synthesized request cannot recover it.
+- `AWSCredentialsProvider.getCredentials()` (AWS SDK v1) and `AwsCredentialsProvider.resolveCredentials()` (v2) are parameterless. They cannot vend per-path credentials.
+- Reflecting into vendor singletons would encode per-vendor class names and lifecycles in Comet and would silently break on vendor upgrades.
+
+A Comet-specific SPI is the narrowest fit: a single Java method that takes a `CometS3CredentialContext` (today wrapping `bucket`, `path`, and access `mode`; new fields can be added without breaking vendors compiled against earlier versions) and returns `CometS3Credentials`.
+
+## Why config-driven activation, not `META-INF/services`
+
+An earlier iteration used `ServiceLoader` discovery. That was rejected because:
+
+- Peer SPIs in the same space (Hadoop `AWSCredentialsProvider`, AWS SDK v2 `AwsCredentialsProvider`, Iceberg `AwsClientFactory`, S3A custom signers) are all class-name-in-config. Vendors are already familiar with that model.
+- ServiceLoader makes activation implicit on classpath presence. A vendor JAR drifting onto a cluster could silently change S3 auth behavior. The config key makes activation explicit.
+- The activation key (`fs.s3a.comet.credential.provider.class`, with per-bucket override) follows the same shape as `fs.s3a.bucket.<name>.aws.credentials.provider`, so operators do not learn a new pattern.
+
+Activation is modeled on `parquet.crypto.factory.class` (Parquet Modular Encryption KMS, see Comet #2447): the user names a single vendor class and the vendor dispatches across multiple credential backends inside that class if they need to. This mirrors how Iceberg's `DecryptionPropertiesFactory` already behaves for Parquet keys.
+
+## Why `(FQCN, dispatchKey, catalogProperties)` keying
+
+Comet caches one provider instance per `(FQCN, dispatchKey, catalogProperties)` triple. The dispatch key is the Spark V2 catalog name on the Iceberg path and the bucket on the Parquet path.
+
+- Two catalogs that share one provider class (typical in multi-tenant deployments) need isolated `initialize` maps because their `catalogProperties` differ. Without `dispatchKey`, the second `initialize` would either overwrite the first or be silently skipped.
+- The bucket as `dispatchKey` for Parquet gives vendors per-bucket isolation when the same provider is named under several `fs.s3a.bucket.<name>.comet.credential.provider.class` keys.
+- `catalogProperties` enters the key to handle multi-tenant JVMs (Spark Connect, Thrift Server, `SparkSession.newSession()`) where two sessions can configure the same provider class against the same `dispatchKey` but with different REST endpoints, OAuth tokens, or vendor keys. Without it the second session would silently use the first session's credentials.
+- Keying solely by FQCN would force vendors to encode multi-tenant routing in static state. The triple-key keeps each call site independent.
+
+`ensureInitialized` returns a `long` handle that the native bridge stashes and replays on every per-request call. Routing per-request lookups by handle avoids re-sending the property bag across JNI on the hot path and unambiguously selects the right provider when the same `(FQCN, dispatchKey)` pair maps to multiple instances.
+
+## Why fresh construction in `initialize`, not probing a JVM-wide static
+
+A provider implementation might be tempted to probe an existing static populated elsewhere (e.g. by a Hadoop S3A signer's `registerStore` callback) and reuse the credential cache that the Hadoop path uses. That fails on Comet-only executors:
+
+- The driver JVM hits `S3AFileSystem.initialize` during analysis (raw `s3a://` paths) or during Hadoop catalog manifest reads (Iceberg with Hadoop catalog), so the static is populated there.
+- The driver may not hit `S3AFileSystem` at all under Iceberg with REST catalog plus `S3FileIO`, because `S3FileIO` calls AWS SDK directly without going through the Hadoop layer. The static stays null.
+- Executors with Comet-only reads never instantiate `S3AFileSystem`. The data path is `object_store` (raw Parquet) or `opendal` via `iceberg-rust` (Iceberg native scan). Neither touches Hadoop S3A. The static stays null on every executor.
+
+Constructing a fresh provider from `catalogProperties` plus `SparkEnv` is the only strategy that works across all four cases. The trade-off is that on the driver (and any JVM where Hadoop S3A is also active), two credential caches now exist for the same identity: one inside the Hadoop signer's provider, one inside the SPI implementation's. The vendor pays for this with a small number of extra AS round-trips on cold starts and TTL boundaries. A future optional optimization could probe the [...]
+
+## Why no Comet-side cache
+
+Comet's bridge does not maintain a TTL cache, schedule refresh, or broadcast catalog state. All of that is the vendor's responsibility:
+
+- Iceberg vendors get `software.amazon.awssdk.utils.cache.CachedSupplier` for free inside `org.apache.iceberg.aws.s3.VendedCredentialsProvider`.
+- Custom-STS vendors write whatever cache fits their refresh model.
+- Driver-only state is distributed via `initialize`'s `catalogProperties` (Iceberg path) or read from Hadoop conf via `SparkEnv` (Parquet path). Both are plan-time snapshots: Comet does not re-execute the catalog or push fresh values to running scans. Vendors that need a refreshing bearer compose with Spark's `HadoopDelegationTokenProvider`, which mints and renews on the driver and propagates to executors via `UserGroupInformation`. The two SPIs are orthogonal: Spark covers bearer lifecy [...]
+
+A Comet-side cache would have to either expose a tuning knob (TTL, max size, eviction policy) and grow over time, or be hardcoded and surprise vendors whose policies disagree. The bridge intentionally has neither and forwards every call.
+
+## Path-specific behavior
+
+`object_store::CredentialProvider` and `reqsign_core::ProvideCredential` differ in what they consume:
+
+| Concern | Parquet (`object_store`) | Iceberg (`opendal` via `reqsign-core`) |
+| ----------------------- | -------------------------------------------------- | ------------------------------------------------------------ |
+| Trait method | `get_credential() -> AwsCredential` | `provide_credential(...) -> Option<IcebergAwsCredential>` |
+| Returns expiry? | No (only key/secret/token) | Yes (`expires_in: Option<Timestamp>`) |
+| Comet-side TTL wrapper? | None. Bridge passed straight to `with_credentials` | None. `opendal` schedules the next refresh from `expires_in` |
+| When SPI is called | Every `get_credential()` call | When `expires_in` is exceeded |
+| Vendor returns 0 expiry | Field has no use | Bridge substitutes 5 minutes to bound staleness |
+
+The 5-minute fallback is a safety net so a vendor that omits expiry cannot leave `opendal` caching a stale token indefinitely. It is intentionally not a configuration knob.
+
+## Property-bag handling on the Iceberg path
+
+The full unfiltered FileIO property bag crosses JNI as `catalog_properties`. The storage-prefix filter (`s3.`/`gcs.`/`adls.`/`client.`) is applied native-side in `iceberg_scan.rs::load_file_io` immediately before `FileIOBuilder.with_prop`. This means the bridge sees `credentials.uri`, OAuth tokens, and any vendor-custom keys with no parallel field on the operator and no driver-side broadcast. Vendors set their own keys on the catalog config and read them back inside `initialize(Map)`.
+
+`IcebergScanExec` derives a redacting `Debug` so plan dumps and tracing do not leak the property bag.
+
+## Returns or throws, not a fall-through value
+
+The SPI returns a `CometS3Credentials` or throws. There is no sentinel "I do not know" return. Vendors that are only authoritative for some paths resolve the default AWS chain themselves for the rest and return the result. This matches the contract on every other AWS credential SPI in the JVM ecosystem (AWS SDK v1/v2, Hadoop S3A, Iceberg `VendedCredentialsProvider`).
+
+## Lifecycle: `AutoCloseable` plus a JVM shutdown hook
+
+`CometS3CredentialProvider` extends `AutoCloseable` with a default no-op `close()`. The dispatcher installs a JVM shutdown hook that iterates every cached instance and calls `close()`, swallowing per-provider exceptions so a slow or buggy vendor cannot block other providers from cleaning up. Stateless providers ignore this entirely; vendors that hold long-lived resources (HTTP clients, scheduled-refresh executors, STS connection pools) override `close()` to release them. Shutdown hooks a [...]
+
+## 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`.
diff --git a/docs/source/user-guide/latest/index.rst b/docs/source/user-guide/latest/index.rst
index 236f284ce0..00f770c27e 100644
--- a/docs/source/user-guide/latest/index.rst
+++ b/docs/source/user-guide/latest/index.rst
@@ -79,6 +79,7 @@ to read more.
:hidden:
Iceberg Guide <iceberg>
+ S3 Credential Providers <s3-credential-providers>
Kubernetes Guide <kubernetes>
.. toctree::
diff --git a/docs/source/user-guide/latest/s3-credential-providers.md b/docs/source/user-guide/latest/s3-credential-providers.md
new file mode 100644
index 0000000000..33777c497a
--- /dev/null
+++ b/docs/source/user-guide/latest/s3-credential-providers.md
@@ -0,0 +1,292 @@
+<!--
+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 Credential Providers
+
+Comet's native S3 readers normally fetch credentials from the standard AWS credential chain (static keys, instance profiles, environment variables, etc.). Some clusters use a vendor-managed mechanism instead, where credentials are issued per request based on a JWT or per S3 path. For those clusters, Comet supports loading a vendor-supplied bridge class that routes every native credential request through the vendor's Java code.
+
+## Do I need this?
+
+You don't, if any of the following describe your cluster:
+
+- You use static AWS credentials (`fs.s3a.access.key` / `fs.s3a.secret.key`).
+- You use EC2 instance profiles, EKS pod identities, ECS task roles, or environment variables.
+- Your S3 access works in Spark today via the default AWS credential chain.
+
+You probably do, if any of these are true:
+
+- You have a Hadoop S3A custom signer configured (`fs.s3a.custom.signers=...`).
+- You have a Spark `CloudCredentialsProvider` that issues a JWT for a vendor STS service.
+- You have a custom Iceberg `client.factory` that injects a configured S3 client.
+- Spark queries against your S3 paths work, but the same queries with Comet enabled fail with 403.
+
+## Enabling a bridge
+
+A bridge is activated by naming the vendor's class in a Spark config. Putting a JAR on the classpath alone has no effect; the config key must be set.
+
+For raw Parquet (the `object_store` path), set the Hadoop S3A config:
+
+```
+spark.hadoop.fs.s3a.comet.credential.provider.class=com.vendor.MyCometCredentialProvider
+```
+
+A per-bucket override is supported and follows the same shape as `fs.s3a.bucket.<name>.aws.credentials.provider`:
+
+```
+spark.hadoop.fs.s3a.bucket.<bucket>.comet.credential.provider.class=com.vendor.MyCometCredentialProvider
+```
+
+For Iceberg (the `opendal` path), set the per-catalog property under the `s3.` namespace Iceberg already uses for its S3 settings:
+
+```
+spark.sql.catalog.<catalog>.s3.comet.credential.provider.class=com.vendor.MyCometCredentialProvider
+```
+
+Add the vendor JAR to your Spark executor classpath:
+
+```sh
+spark-submit --jars vendor-comet-bridge.jar ...
+```
+
+OSS Comet ships no vendor-specific bridges. Get one from the same vendor that supplies your Hadoop S3A signer or Iceberg client factory. If they do not yet provide one, send them to the "Writing a bridge" section below.
+
+## Wire encryption
+
+Catalog configuration you set under `spark.sql.catalog.<catalog>.*` (which may include endpoint URLs, OAuth tokens, and other vendor keys your provider reads) is sent from the driver to executors over Spark's Netty RPC channel so the provider can run there. Live AWS credentials fetched by the provider stay on the executor that fetched them.
+
+Spark's RPC channel is plaintext by default and offers two opt-in encryption mechanisms, both off by default:
+
+- [`spark.network.crypto.enabled`](https://spark.apache.org/docs/latest/security.html#authentication-and-encryption) for AES-based RPC encryption keyed off the auth shared secret.
+- [`spark.ssl.rpc.enabled`](https://spark.apache.org/docs/latest/security.html#ssl-configuration) for TLS on the same channel (Spark 3.4+).
+
+The same defaults apply to Hadoop delegation tokens, `CloudCredentialsProvider` JWTs, and any other secrets Spark already ships driver-to-executor, so this is a deployment-wide call rather than something specific to this SPI. See Spark's [security guide](https://spark.apache.org/docs/latest/security.html) for the full set of knobs.
+
+## Verification
+
+With the config set and the JAR on the classpath, executor logs show on first S3 access:
+
+- Info level: `Instantiated CometS3CredentialProvider <fully.qualified.VendorClassName>`
+- Debug level: `Fetching credentials via <class> (dispatchKey=<key>) for bucket=... path=... mode=...`
+
+Without the config set, no credential-related log lines appear at startup; native readers use the default AWS credential chain.
+
+## Troubleshooting
+
+**`CometS3CredentialProvider class not found: <name>`**. The class named in the config is not on the executor classpath. Re-check `--jars` / `spark.jars`. On YARN or Kubernetes, confirm the JAR actually reached the executor and not only the driver.
+
+**`<class> does not implement org.apache.comet.cloud.s3.CometS3CredentialProvider`**. The configured class exists but does not implement the SPI. Double-check the FQCN against the vendor's documentation.
+
+**`<class> must declare a public no-arg constructor`**. Vendor classes are instantiated reflectively with `Class.forName(name).getDeclaredConstructor().newInstance()`. A non-default constructor is not supported; ask the vendor to expose a no-arg form that reads any state it needs from `initialize(Map)` or environment.
+
+**`CometS3CredentialProvider <class> (dispatchKey=...) was not initialized`**. `initialize(Map)` was not called before a credential request. Comet should always invoke `ensureInitialized` synchronously when it builds the bridge at plan time, so this indicates the bridge skipped the init call (a Comet bug) or the vendor's `initialize` threw and the bridge fell through to the default chain.
+
+**`403 AccessDenied` with the bridge configured.** The provider returned credentials but they were rejected by S3. Most often a region mismatch (see Iceberg section below) or expired session token; enable debug logging on the vendor's class to confirm what it returned.
+
+**Credentials silently going stale during long-running jobs.** When a vendor returns `expirationEpochMillis=0`, the bridge substitutes a 5-minute expiry before handing the credential to `opendal`, so `opendal`'s cache cannot hold a stale credential indefinitely. Returning a real expiry is preferred; the 5-minute fallback is a safety net, not a knob.
+
+## Iceberg: explicit S3 region required
+
+With the bridge configured, Comet wires a custom credential loader into `iceberg-storage-opendal`. `opendal`'s built-in S3 region auto-detection only runs when no custom loader is configured, so on the bridge path the region (and endpoint for non-AWS) must be set explicitly on the Spark catalog:
+
+```
+spark.sql.catalog.<catalog>.s3.region = us-east-1
+spark.sql.catalog.<catalog>.s3.endpoint = https://... (non-AWS only)
+spark.sql.catalog.<catalog>.s3.path-style-access = true (path-style endpoints only)
+```
+
+If you hit `region is missing. Please find it by S3::detect_region() or set them in env`, this is the missing config.
+
+## Writing a bridge
+
+Comet's native scan paths (`object_store` for raw Parquet, `opendal` via `iceberg-rust` for Iceberg) bypass Hadoop S3A entirely. The standard `AWSCredentialsProvider.getCredentials()` has no path argument, so vendors that issue per-path STS credentials cannot expose them through it. The `CometS3CredentialProvider` SPI fills that gap.
+
+Implement `org.apache.comet.cloud.s3.CometS3CredentialProvider`:
+
+```java
+package org.apache.comet.cloud.s3;
+
+public interface CometS3CredentialProvider extends AutoCloseable {
+ /** Called once per (FQCN, dispatchKey, catalogProperties) before any per-request call. Optional. */
+ default void initialize(java.util.Map<String, String> catalogProperties) {}
+
+ CometS3Credentials getCredentialsForPath(CometS3CredentialContext context) throws Exception;
+
+ /** Invoked from the dispatcher's JVM shutdown hook. Default is a no-op. */
+ @Override default void close() throws Exception {}
+}
+```
+
+The class must have a public no-arg constructor. `getCredentialsForPath` may be invoked concurrently from many native tokio worker threads; the implementation must be thread-safe. The supplied `CometS3CredentialContext` exposes `getBucket()`, `getPath()`, and `getMode()`; future Comet releases may add accessors here without changing the method signature, so vendors compiled against today's API stay binary-compatible.
+
+### Lifecycle
+
+Comet keys provider instances by `(FQCN, dispatchKey, catalogProperties)`. The dispatch key is the Spark V2 catalog name on the Iceberg path and the S3 bucket name on the Parquet path. The first time a given key is seen on an executor, Comet reflects the class, calls `initialize(Map)` exactly once, and caches the instance for the JVM lifetime. Two catalogs sharing one provider FQCN therefore get isolated instances with their own `initialize` maps. Including `catalogProperties` in the key [...]
+
+`initialize` should be cheap and non-blocking. Defer real credential fetches (REST round-trips, STS calls) to the first `getCredentialsForPath` invocation. On the Iceberg path the supplied `catalogProperties` carries the unfiltered FileIO bag, including REST-vended fields like `credentials.uri`, OAuth tokens, and any vendor-custom keys you set on the catalog config. The map may contain secrets, so do not log it.
+
+`close()` is invoked from a JVM shutdown hook installed by the dispatcher. The default no-op is fine for stateless providers. Override it to release HTTP clients, scheduled-refresh executors, or STS connection pools. Shutdown hooks are best-effort: a `SIGKILL` or abrupt JVM termination skips them, so do not depend on `close()` for correctness.
+
+### Caching, refresh, and distribution
+
+Comet does not maintain a TTL cache, broadcast catalog state, or schedule refresh. Vendors decide:
+
+- Whether to cache credentials and for how long. Iceberg vendors get `software.amazon.awssdk.utils.cache.CachedSupplier` for free inside `VendedCredentialsProvider`; vendors with custom STS write whatever cache fits.
+- When to refresh: proactive timer, on-demand at expiry, on `403` retry, etc.
+- How to distribute and refresh state. `catalogProperties` is a plan-time snapshot. Comet does not re-execute the catalog or push fresh values to running scans, so any state that must refresh during a scan is the vendor's responsibility. Options: call back to a vendor service from the executor on each `getCredentialsForPath`, run your own Spark broadcast inside the class, or compose with Spark's [`HadoopDelegationTokenProvider`](https://spark.apache.org/docs/latest/security.html#kerberos [...]
+
+When using `HadoopDelegationTokenProvider`, Spark mints and renews the token on the driver and propagates it to executors via `UserGroupInformation`. The provider reads the current value inside `getCredentialsForPath` and exchanges it for AWS credentials:
+
+```java
+@Override
+public CometS3Credentials getCredentialsForPath(CometS3CredentialContext ctx) throws Exception {
+ Token<?> token = UserGroupInformation.getCurrentUser()
+ .getCredentials()
+ .getToken(new Text("vendor-service"));
+ return vendorSts.exchangeForS3Credentials(token, ctx.getBucket(), ctx.getPath(), ctx.getMode());
+}
+```
+
+Spark delegation token propagation is supported on YARN and Kubernetes only. Standalone deployments need a different refresh path, typically a vendor-side service callback authenticated by long-lived state in `catalogProperties` or Hadoop conf.
+
+`expirationEpochMillis` only matters on the Iceberg/`opendal` path. There the bridge implements `reqsign_core::ProvideCredential`, which carries an `expires_in` field that `opendal` uses to schedule the next refresh. Publish a real expiry when you have one. `0` means "unknown"; the bridge then substitutes a 5-minute expiry to bound staleness.
+
+The Parquet/`object_store` path has no expiry concept: `object_store::CredentialProvider` returns just `AwsCredential` (key/secret/token). The bridge is passed to `with_credentials` without a TTL wrapper, so `object_store` calls into the SPI on every request and relies on the vendor's own cache for hit rates. Expiry handling is fully the vendor's responsibility: the vendor decides when its internal cache refreshes. If `object_store` receives a 403 from an expired session token, its retry [...]
+
+### Returned fields
+
+| Field | Notes |
+| ----------------------- | ------------------------------------------------------------------------------------------------------------------------- |
+| `accessKeyId` | Required. |
+| `secretAccessKey` | Required. |
+| `sessionToken` | `null` for non-STS credentials. |
+| `expirationEpochMillis` | Iceberg path only. `0` means "unknown"; the bridge substitutes a 5-minute expiry. The Parquet path has no expiry concept. |
+
+Provide a real `expirationEpochMillis` whenever you have one on the Iceberg path. The Parquet path's `object_store::CredentialProvider` does not consume an expiry, and the bridge invokes the SPI on every `get_credential()` call.
+
+### Returns or throws
+
+The SPI follows the same shape as the other JVM AWS-credential SPIs (AWS SDK v1/v2, Hadoop S3A, Iceberg `VendedCredentialsProvider`): return credentials or throw. There is no "fall-through" return value.
+
+If your provider is authoritative only for some paths, resolve the default AWS chain yourself for the rest:
+
+```java
+private final DefaultCredentialsProvider defaultChain = DefaultCredentialsProvider.create();
+
+@Override
+public CometS3Credentials getCredentialsForPath(CometS3CredentialContext ctx) throws Exception {
+ if (handlesPath(ctx.getBucket(), ctx.getPath())) {
+ return mintFromMyVendorService(ctx.getBucket(), ctx.getPath(), ctx.getMode());
+ }
+ AwsCredentials c = defaultChain.resolveCredentials();
+ String token = (c instanceof AwsSessionCredentials)
+ ? ((AwsSessionCredentials) c).sessionToken()
+ : null;
+ return new CometS3Credentials(c.accessKeyId(), c.secretAccessKey(), token, 0L);
+}
+```
+
+### Composing multiple credential backends
+
+A single configured provider class is the dispatcher. If a vendor needs to route across several credential backends (per bucket, per path prefix, per tenant), the dispatch lives inside the vendor's class:
+
+```java
+public final class MyCometCredentialProvider implements CometS3CredentialProvider {
+ private final ProdVendor prod = ...;
+ private final StsVendor sts = ...;
+ private final DefaultVendor fallback = ...;
+
+ @Override
+ public CometS3Credentials getCredentialsForPath(CometS3CredentialContext ctx) throws Exception {
+ if (ctx.getBucket().startsWith("data-prod-")) return prod.fetch(ctx);
+ if (ctx.getBucket().equals("partner-shared")) return sts.assumeRole(ctx);
+ return fallback.fetch(ctx.getBucket(), ctx.getPath());
+ }
+}
+```
+
+Per-bucket Hadoop overrides (`fs.s3a.bucket.<name>.comet.credential.provider.class`) are also available if you prefer to ship multiple vendor classes and pick by bucket in config rather than in code.
+
+For Iceberg deployments where two catalogs share one provider class but need isolated state, configure the same FQCN on both catalogs and read your discriminator from `initialize`'s `catalogProperties`. Each catalog gets its own provider instance because Comet keys by `(FQCN, catalogName, catalogProperties)`:
+
+```java
+public final class MyMultiTenantProvider implements CometS3CredentialProvider {
+ private volatile String tenantId;
+
+ @Override
+ public void initialize(Map<String, String> catalogProperties) {
+ this.tenantId = catalogProperties.get("vendor.tenant-id");
+ }
+
+ @Override
+ public CometS3Credentials getCredentialsForPath(CometS3CredentialContext ctx) {
+ return mintForTenant(tenantId, ctx.getBucket(), ctx.getPath(), ctx.getMode());
+ }
+}
+```
+
+### Reference implementation: Iceberg REST vended credentials
+
+For Iceberg REST catalogs that vend AWS credentials (`LoadTableResponse.credentials`), the canonical implementation wraps Iceberg's existing `VendedCredentialsProvider`:
+
+```java
+public final class IcebergRESTVendedS3Provider implements CometS3CredentialProvider {
+ private volatile VendedCredentialsProvider provider;
+
+ @Override
+ public void initialize(Map<String, String> catalogProperties) {
+ this.provider = VendedCredentialsProvider.create(catalogProperties);
+ }
+
+ @Override
+ public CometS3Credentials getCredentialsForPath(CometS3CredentialContext ctx) {
+ AwsCredentials c = provider.resolveCredentials();
+ String token = (c instanceof AwsSessionCredentials)
+ ? ((AwsSessionCredentials) c).sessionToken() : null;
+ return new CometS3Credentials(c.accessKeyId(), c.secretAccessKey(), token, 0L);
+ }
+}
+```
+
+`VendedCredentialsProvider` reads `credentials.uri`, the catalog endpoint, and OAuth tokens from the supplied map (Comet forwards the unfiltered FileIO bag to `initialize`), and refreshes through its own `CachedSupplier`. Caching, refresh-near-expiry, and the REST round-trip all live in Iceberg, not in Comet. Comet ships a copy of this class under `spark/src/test` as a reference; copy it into your runtime jar alongside `iceberg-aws` and AWS SDK v2.
+
+### Access mode
+
+| Value | Used for |
+| ------- | -------------------------------------------------------------------------- |
+| `READ` | All native scan paths (raw Parquet, Iceberg). Comet today only sends READ. |
+| `WRITE` | Reserved for future native write paths. |
+
+A `WRITE` credential is not implicitly read-capable. Vendors that need read-during-write workflows include the required read permissions in the IAM policy attached to their `WRITE` credentials.
+
+### Build setup
+
+Vendor implementations need the Comet SPI classes at compile time only. Use `provided`-scope:
+
+```xml
+<dependency>
+ <groupId>org.apache.datafusion</groupId>
+ <artifactId>comet-spark-spark${spark.version.short}_${scala.binary.version}</artifactId>
+ <version>${comet.version}</version>
+ <scope>provided</scope>
+</dependency>
+```
+
+### Iceberg path: error message fidelity
+
+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, only the message is degraded to an opaque anonymous-request failure. No Comet change fixes this; it is resolved upstream if `iceberg-rust` stops wrapping custom loaders in its outer chain or moves its S3 backend to `object_store`.
diff --git a/native/Cargo.lock b/native/Cargo.lock
index 11e9b1ccff..a02b95819c 100644
--- a/native/Cargo.lock
+++ b/native/Cargo.lock
@@ -675,6 +675,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0ec6fb3fe69024a75fa7e1bfb48aa6cf59706a101658ea01bfd33b2b248a038f"
dependencies = [
"aws-lc-sys",
+ "untrusted 0.7.1",
"zeroize",
]
@@ -2010,7 +2011,7 @@ dependencies = [
"object_store",
"object_store_opendal",
"once_cell",
- "opendal 0.56.0",
+ "opendal",
"parking_lot",
"parquet",
"paste",
@@ -2018,6 +2019,7 @@ dependencies = [
"procfs",
"prost",
"rand 0.10.1",
+ "reqsign-core",
"reqwest 0.12.28",
"serde_json",
"tempfile",
@@ -3530,7 +3532,6 @@ dependencies = [
"tokio",
"tokio-rustls",
"tower-service",
- "webpki-roots",
]
[[package]]
@@ -3583,7 +3584,7 @@ dependencies = [
[[package]]
name = "iceberg"
version = "0.9.0"
-source = "git+https://github.com/apache/iceberg-rust?rev=1ad4bfd#1ad4bfd39319508e79960d16dad1b1cdf965c5f4"
+source = "git+https://github.com/apache/iceberg-rust?rev=83b4595#83b4595e2b5f522974d24d51c8ecbd09a093fa92"
dependencies = [
"aes-gcm",
"anyhow",
@@ -3638,7 +3639,7 @@ dependencies = [
[[package]]
name = "iceberg-storage-opendal"
version = "0.9.0"
-source = "git+https://github.com/apache/iceberg-rust?rev=1ad4bfd#1ad4bfd39319508e79960d16dad1b1cdf965c5f4"
+source = "git+https://github.com/apache/iceberg-rust?rev=83b4595#83b4595e2b5f522974d24d51c8ecbd09a093fa92"
dependencies = [
"anyhow",
"async-trait",
@@ -3646,9 +3647,9 @@ dependencies = [
"cfg-if",
"futures",
"iceberg",
- "opendal 0.55.0",
- "reqsign",
- "reqwest 0.12.28",
+ "opendal",
+ "reqsign-aws-v4",
+ "reqsign-core",
"serde",
"typetag",
"url",
@@ -4044,17 +4045,20 @@ dependencies = [
[[package]]
name = "jsonwebtoken"
-version = "9.3.1"
+version = "10.4.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5a87cc7a48537badeae96744432de36f4be2b4a34a05a5ef32e9dd8a1c169dde"
+checksum = "eba32bfb4ffdeaca3e34431072faf01745c9b26d25504aa7a6cf5684334fc4fc"
dependencies = [
+ "aws-lc-rs",
"base64",
+ "getrandom 0.2.17",
"js-sys",
"pem",
- "ring",
"serde",
"serde_json",
+ "signature",
"simple_asn1",
+ "zeroize",
]
[[package]]
@@ -4585,7 +4589,7 @@ dependencies = [
"futures",
"mea",
"object_store",
- "opendal 0.56.0",
+ "opendal",
"pin-project",
"tokio",
]
@@ -4614,35 +4618,6 @@ version = "0.3.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c08d65885ee38876c4f86fa503fb49d7b507c2b62552df7c70b2fce627e06381"
-[[package]]
-name = "opendal"
-version = "0.55.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d075ab8a203a6ab4bc1bce0a4b9fe486a72bf8b939037f4b78d95386384bc80a"
-dependencies = [
- "anyhow",
- "backon",
- "base64",
- "bytes",
- "crc32c",
- "futures",
- "getrandom 0.2.17",
- "http 1.4.0",
- "http-body 1.0.1",
- "jiff",
- "log",
- "md-5",
- "percent-encoding",
- "quick-xml 0.38.4",
- "reqsign",
- "reqwest 0.12.28",
- "serde",
- "serde_json",
- "tokio",
- "url",
- "uuid",
-]
-
[[package]]
name = "opendal"
version = "0.56.0"
@@ -4655,7 +4630,12 @@ dependencies = [
"opendal-layer-logging",
"opendal-layer-retry",
"opendal-layer-timeout",
+ "opendal-service-azdls",
+ "opendal-service-fs",
+ "opendal-service-gcs",
"opendal-service-hdfs",
+ "opendal-service-oss",
+ "opendal-service-s3",
]
[[package]]
@@ -4729,6 +4709,70 @@ dependencies = [
"tokio",
]
+[[package]]
+name = "opendal-service-azdls"
+version = "0.56.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8f9884c2d8cf8ba2bb077d79c877dac5863ba3bab9e2c9c1e41a2e0491404772"
+dependencies = [
+ "bytes",
+ "http 1.4.0",
+ "log",
+ "opendal-core",
+ "opendal-service-azure-common",
+ "quick-xml 0.38.4",
+ "reqsign-azure-storage",
+ "reqsign-core",
+ "reqsign-file-read-tokio",
+ "serde",
+ "serde_json",
+]
+
+[[package]]
+name = "opendal-service-azure-common"
+version = "0.56.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ffb0e45d6c8dcf66ce2da20e241bcb80e6e540e109a4ff20f318f6c9b4c54e0c"
+dependencies = [
+ "http 1.4.0",
+ "opendal-core",
+]
+
+[[package]]
+name = "opendal-service-fs"
+version = "0.56.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cf0be0417abeeb0053376d816b90fceb9ca98f20dfb54ebf1f2a282729f83663"
+dependencies = [
+ "bytes",
+ "log",
+ "opendal-core",
+ "serde",
+ "tokio",
+ "xattr",
+]
+
+[[package]]
+name = "opendal-service-gcs"
+version = "0.56.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "70a49477a10163431896d106136117f5670717f9c9e49cf6f710528800c6633a"
+dependencies = [
+ "async-trait",
+ "bytes",
+ "http 1.4.0",
+ "log",
+ "opendal-core",
+ "percent-encoding",
+ "quick-xml 0.38.4",
+ "reqsign-core",
+ "reqsign-file-read-tokio",
+ "reqsign-google",
+ "serde",
+ "serde_json",
+ "tokio",
+]
+
[[package]]
name = "opendal-service-hdfs"
version = "0.56.0"
@@ -4744,6 +4788,44 @@ dependencies = [
"tokio",
]
+[[package]]
+name = "opendal-service-oss"
+version = "0.56.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "29c8a917829ad06d21b639558532cb0101fe49b040d946d673a73018683fac05"
+dependencies = [
+ "bytes",
+ "http 1.4.0",
+ "log",
+ "opendal-core",
+ "quick-xml 0.38.4",
+ "reqsign-aliyun-oss",
+ "reqsign-core",
+ "reqsign-file-read-tokio",
+ "serde",
+]
+
+[[package]]
+name = "opendal-service-s3"
+version = "0.56.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9dadddeb9bb50b0d30927dd914c298c4ddca47e4c1cfa7674d311f0cf9b051c8"
+dependencies = [
+ "base64",
+ "bytes",
+ "crc32c",
+ "http 1.4.0",
+ "log",
+ "md-5",
+ "opendal-core",
+ "quick-xml 0.38.4",
+ "reqsign-aws-v4",
+ "reqsign-core",
+ "reqsign-file-read-tokio",
+ "serde",
+ "url",
+]
+
[[package]]
name = "openssl-probe"
version = "0.2.1"
@@ -5288,16 +5370,6 @@ dependencies = [
"memchr",
]
-[[package]]
-name = "quick-xml"
-version = "0.37.5"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "331e97a1af0bf59823e6eadffe373d7b27f485be8748f71471c662c1f269b7fb"
-dependencies = [
- "memchr",
- "serde",
-]
-
[[package]]
name = "quick-xml"
version = "0.38.4"
@@ -5401,7 +5473,6 @@ version = "0.8.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a"
dependencies = [
- "libc",
"rand_chacha 0.3.1",
"rand_core 0.6.4",
]
@@ -5556,35 +5627,63 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "dc897dd8d9e8bd1ed8cdad82b5966c3e0ecae09fb1907d58efaa013543185d0a"
[[package]]
-name = "reqsign"
-version = "0.16.5"
+name = "reqsign-aliyun-oss"
+version = "3.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "43451dbf3590a7590684c25fb8d12ecdcc90ed3ac123433e500447c7d77ed701"
+checksum = "57ac2757f3140aa2e213b554148ae0b52733e624fc6723f0cc6bb3d440176c95"
+dependencies = [
+ "anyhow",
+ "form_urlencoded",
+ "http 1.4.0",
+ "log",
+ "percent-encoding",
+ "reqsign-core",
+ "rust-ini",
+ "serde",
+ "serde_json",
+]
+
+[[package]]
+name = "reqsign-aws-v4"
+version = "3.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "44eaca382e94505a49f1a4849658d153aebf79d9c1a58e5dd3b10361511e9f43"
+dependencies = [
+ "anyhow",
+ "bytes",
+ "form_urlencoded",
+ "http 1.4.0",
+ "log",
+ "percent-encoding",
+ "quick-xml 0.39.2",
+ "reqsign-core",
+ "rust-ini",
+ "serde",
+ "serde_json",
+ "serde_urlencoded",
+ "sha1",
+]
+
+[[package]]
+name = "reqsign-azure-storage"
+version = "3.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7a321980405d596bd34aaf95c4722a3de4128a67fd19e74a81a83aa3fdf082e6"
dependencies = [
"anyhow",
- "async-trait",
"base64",
- "chrono",
+ "bytes",
"form_urlencoded",
- "getrandom 0.2.17",
- "hex",
- "hmac 0.12.1",
- "home",
"http 1.4.0",
"jsonwebtoken",
"log",
- "once_cell",
+ "pem",
"percent-encoding",
- "quick-xml 0.37.5",
- "rand 0.8.6",
- "reqwest 0.12.28",
+ "reqsign-core",
"rsa",
- "rust-ini",
"serde",
"serde_json",
"sha1",
- "sha2 0.10.9",
- "tokio",
]
[[package]]
@@ -5609,6 +5708,37 @@ dependencies = [
"windows-sys 0.61.2",
]
+[[package]]
+name = "reqsign-file-read-tokio"
+version = "3.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e2d89295b3d17abea31851cc8de55d843d89c52132c864963c38d41920613dc5"
+dependencies = [
+ "anyhow",
+ "reqsign-core",
+ "tokio",
+]
+
+[[package]]
+name = "reqsign-google"
+version = "3.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "35cc609b49c69e76ecaceb775a03f792d1ed3e7755ab3548d4534fd801e3242e"
+dependencies = [
+ "form_urlencoded",
+ "http 1.4.0",
+ "jsonwebtoken",
+ "log",
+ "percent-encoding",
+ "reqsign-aws-v4",
+ "reqsign-core",
+ "rsa",
+ "serde",
+ "serde_json",
+ "sha2 0.10.9",
+ "tokio",
+]
+
[[package]]
name = "reqwest"
version = "0.12.28"
@@ -5649,7 +5779,6 @@ dependencies = [
"wasm-bindgen-futures",
"wasm-streams 0.4.2",
"web-sys",
- "webpki-roots",
]
[[package]]
@@ -5709,7 +5838,7 @@ dependencies = [
"cfg-if",
"getrandom 0.2.17",
"libc",
- "untrusted",
+ "untrusted 0.9.0",
"windows-sys 0.52.0",
]
@@ -5880,7 +6009,7 @@ dependencies = [
"aws-lc-rs",
"ring",
"rustls-pki-types",
- "untrusted",
+ "untrusted 0.9.0",
]
[[package]]
@@ -6875,6 +7004,12 @@ version = "0.2.11"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "673aac59facbab8a9007c7f6108d11f63b603f7cabff99fabf650fea5c32b861"
+[[package]]
+name = "untrusted"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a"
+
[[package]]
name = "untrusted"
version = "0.9.0"
@@ -7128,15 +7263,6 @@ dependencies = [
"rustls-pki-types",
]
-[[package]]
-name = "webpki-roots"
-version = "1.0.7"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "52f5ee44c96cf55f1b349600768e3ece3a8f26010c05265ab73f945bb1a2eb9d"
-dependencies = [
- "rustls-pki-types",
-]
-
[[package]]
name = "which"
version = "4.4.2"
@@ -7564,6 +7690,16 @@ version = "0.6.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4"
+[[package]]
+name = "xattr"
+version = "1.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156"
+dependencies = [
+ "libc",
+ "rustix 1.1.4",
+]
+
[[package]]
name = "xmlparser"
version = "0.13.6"
@@ -7639,6 +7775,20 @@ name = "zeroize"
version = "1.8.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "b97154e67e32c85465826e8bcc1c59429aaaf107c1e4a9e53c8d8ccd5eff88d0"
+dependencies = [
+ "zeroize_derive",
+]
+
+[[package]]
+name = "zeroize_derive"
+version = "1.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "85a5b4158499876c763cb03bc4e49185d3cccbabb15b33c627f7884f43db852e"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.117",
+]
[[package]]
name = "zerotrie"
diff --git a/native/Cargo.toml b/native/Cargo.toml
index 9e252a796c..5df04c2ca3 100644
--- a/native/Cargo.toml
+++ b/native/Cargo.toml
@@ -58,8 +58,9 @@ object_store = { version = "0.13.1", features = ["gcp", "azure", "aws", "http"]
url = "2.2"
aws-config = "1.8.17"
aws-credential-types = "1.2.13"
-iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "1ad4bfd" }
-iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "1ad4bfd", features = ["opendal-all"] }
+iceberg = { git = "https://github.com/apache/iceberg-rust", rev = "83b4595" }
+iceberg-storage-opendal = { git = "https://github.com/apache/iceberg-rust", rev = "83b4595", features = ["opendal-all"] }
+reqsign-core = "3"
[profile.release]
debug = true
diff --git a/native/core/Cargo.toml b/native/core/Cargo.toml
index c58d446917..0d3b084ba3 100644
--- a/native/core/Cargo.toml
+++ b/native/core/Cargo.toml
@@ -75,6 +75,7 @@ hdfs-sys = {version = "0.3", optional = true, features = ["hdfs_3_3"]}
opendal = { version = "0.56.0", optional = true, features = ["services-hdfs"] }
iceberg = { workspace = true }
iceberg-storage-opendal = { workspace = true }
+reqsign-core = { workspace = true }
serde_json = "1.0"
uuid = "1.23.0"
diff --git a/native/core/src/cloud/mod.rs b/native/core/src/cloud/mod.rs
new file mode 100644
index 0000000000..634d37b26d
--- /dev/null
+++ b/native/core/src/cloud/mod.rs
@@ -0,0 +1,23 @@
+// 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.
+
+//! Cloud-side helpers shared across Comet's native scan paths.
+//!
+//! Today this hosts the JNI bridge to the JVM `CometS3CredentialDispatcher` SPI, which is reused
+//! by the raw Parquet (`object_store`) path and the Iceberg (`opendal` via `iceberg-rust`) path.
+
+pub mod s3;
diff --git a/native/core/src/cloud/s3/credential_bridge.rs b/native/core/src/cloud/s3/credential_bridge.rs
new file mode 100644
index 0000000000..14e4bc49d9
--- /dev/null
+++ b/native/core/src/cloud/s3/credential_bridge.rs
@@ -0,0 +1,361 @@
+// 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.
+
+//! JNI bridge to the JVM `CometS3CredentialDispatcher` SPI, exposed as
+//! `object_store::CredentialProvider` (raw Parquet path) and `reqsign_core::ProvideCredential`
+//! (Iceberg via `opendal`). See `docs/source/contributor-guide/s3-credential-provider-design.md`.
+
+use crate::execution::operators::ExecutionError;
+use crate::jvm_bridge::{jni_new_global_ref, jni_static_call, JVMClasses};
+use async_trait::async_trait;
+use iceberg_storage_opendal::AwsCredential as IcebergAwsCredential;
+use jni::objects::{Global, JFieldID, JObject, JString, JValue};
+use jni::signature::{Primitive, ReturnType};
+use jni::strings::JNIString;
+use jni::sys::jint;
+use log::warn;
+use object_store::aws::AwsCredential;
+use object_store::CredentialProvider;
+use once_cell::sync::OnceCell;
+use reqsign_core::time::Timestamp;
+use reqsign_core::{
+ Context, Error as ReqsignError, ErrorKind as ReqsignErrorKind,
+ ProvideCredential as IcebergProvideCredential,
+};
+use std::collections::HashMap;
+use std::fmt;
+use std::sync::Arc;
+use std::time::Duration;
+
+/// Cap on opendal's credential cache when the provider does not report an expiry. Prevents the
+/// executor from holding a stale credential for the entire job lifetime.
+const DEFAULT_EXPIRY_WHEN_UNKNOWN: Duration = Duration::from_secs(300);
+
+/// Once-per-process latch for the "missing expiry" warning. Bridges are per-scan, so a per-bridge
+/// latch would re-log on every scan.
+static WARNED_MISSING_EXPIRY: OnceCell<()> = OnceCell::new();
+
+/// Access intent forwarded to the Java SPI. Ordinal must match the JVM `CometS3AccessMode` enum.
+#[derive(Debug, Clone, Copy)]
+pub enum AccessMode {
+ Read = 0,
+ #[allow(dead_code)]
+ Write = 1,
+}
+
+/// Per-scan credential provider that delegates to the JVM SPI via JNI. `handle` is the JVM-side
+/// identity for the `(provider_class, dispatch_key, catalog_properties)` triple returned by
+/// `ensureInitialized`. `bucket_jstr` / `path_jstr` are interned once at construction to avoid
+/// per-call `new_string` allocations on the hot path.
+///
+/// Granularity: although the JVM SPI accepts `(bucket, path)`, neither
+/// `object_store::CredentialProvider::get_credential` nor
+/// `reqsign_core::ProvideCredential::provide_credential` carries a per-request path, so the
+/// effective identity is per-bucket (Parquet) or per-table-location (Iceberg).
+pub struct CometS3CredentialBridge {
+ provider_class: String,
+ dispatch_key: String,
+ bucket: String,
+ path: String,
+ mode: AccessMode,
+ handle: i64,
+ bucket_jstr: Arc<Global<JString<'static>>>,
+ path_jstr: Arc<Global<JString<'static>>>,
+}
+
+impl fmt::Debug for CometS3CredentialBridge {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("CometS3CredentialBridge")
+ .field("provider_class", &self.provider_class)
+ .field("dispatch_key", &self.dispatch_key)
+ .field("handle", &self.handle)
+ .field("bucket", &self.bucket)
+ .field("path", &self.path)
+ .field("mode", &self.mode)
+ .finish()
+ }
+}
+
+impl CometS3CredentialBridge {
+ pub fn new(
+ provider_class: impl Into<String>,
+ dispatch_key: impl Into<String>,
+ bucket: impl Into<String>,
+ path: impl Into<String>,
+ mode: AccessMode,
+ catalog_properties: &HashMap<String, String>,
+ ) -> Result<Self, ExecutionError> {
+ let provider_class = provider_class.into();
+ let dispatch_key = dispatch_key.into();
+ let bucket = bucket.into();
+ let path = path.into();
+
+ let (bucket_jstr, path_jstr) = JVMClasses::with_env(|env| -> Result<_, ExecutionError> {
+ let b = env
+ .new_string(&bucket)
+ .map_err(|e| ExecutionError::GeneralError(format!("new_string(bucket): {e}")))?;
+ let p = env
+ .new_string(&path)
+ .map_err(|e| ExecutionError::GeneralError(format!("new_string(path): {e}")))?;
+ let b_g =
+ Arc::new(jni_new_global_ref!(env, b).map_err(|e| {
+ ExecutionError::GeneralError(format!("global_ref(bucket): {e}"))
+ })?);
+ let p_g = Arc::new(
+ jni_new_global_ref!(env, p)
+ .map_err(|e| ExecutionError::GeneralError(format!("global_ref(path): {e}")))?,
+ );
+ Ok((b_g, p_g))
+ })?;
+
+ let handle = ensure_initialized(&provider_class, &dispatch_key, catalog_properties)?;
+ Ok(Self {
+ provider_class,
+ dispatch_key,
+ bucket,
+ path,
+ mode,
+ handle,
+ bucket_jstr,
+ path_jstr,
+ })
+ }
+
+ fn fetch_raw(&self) -> Result<RawCredentials, ExecutionError> {
+ JVMClasses::with_env(|env| -> Result<RawCredentials, ExecutionError> {
+ let mode = self.mode as jint;
+
+ let creds_obj: JObject = unsafe {
+ jni_static_call!(env,
+ comet_s3_credential_dispatcher.get_credentials_for_path(
+ self.handle,
+ self.bucket_jstr.as_obj(),
+ self.path_jstr.as_obj(),
+ mode
+ ) -> JObject
+ )?
+ };
+ if creds_obj.is_null() {
+ return Err(ExecutionError::GeneralError(
+ "getCredentialsForPath returned null (contract violation)".to_string(),
+ ));
+ }
+
+ let d = &JVMClasses::get().comet_s3_credential_dispatcher;
+ Ok(RawCredentials {
+ access_key_id: read_required_string(
+ env,
+ &creds_obj,
+ d.field_access_key_id,
+ "accessKeyId",
+ )?,
+ secret_access_key: read_required_string(
+ env,
+ &creds_obj,
+ d.field_secret_access_key,
+ "secretAccessKey",
+ )?,
+ session_token: read_optional_string(env, &creds_obj, d.field_session_token)?,
+ expiration_epoch_millis: unsafe {
+ env.get_field_unchecked(
+ &creds_obj,
+ d.field_expiration_epoch_millis,
+ ReturnType::Primitive(Primitive::Long),
+ )
+ }
+ .and_then(|v| v.j())
+ .map_err(|e| {
+ ExecutionError::GeneralError(format!("read expirationEpochMillis: {e}"))
+ })?,
+ })
+ })
+ }
+}
+
+fn ensure_initialized(
+ provider_class: &str,
+ dispatch_key: &str,
+ catalog_properties: &HashMap<String, String>,
+) -> Result<i64, ExecutionError> {
+ JVMClasses::with_env(|env| -> Result<i64, ExecutionError> {
+ let provider_class_jstr = env.new_string(provider_class).map_err(|e| {
+ ExecutionError::GeneralError(format!("new_string(provider_class): {e}"))
+ })?;
+ let dispatch_key_jstr = env
+ .new_string(dispatch_key)
+ .map_err(|e| ExecutionError::GeneralError(format!("new_string(dispatch_key): {e}")))?;
+ let props_obj = build_java_string_map(env, catalog_properties)?;
+
+ let handle: i64 = unsafe {
+ jni_static_call!(env,
+ comet_s3_credential_dispatcher.ensure_initialized(
+ &provider_class_jstr, &dispatch_key_jstr, &props_obj
+ ) -> i64
+ )?
+ };
+ Ok(handle)
+ })
+}
+
+/// Construct a `java.util.HashMap<String,String>` and populate it. Called once per bridge at
+/// construction, so per-call HashMap/put cost stays off the hot path.
+fn build_java_string_map<'a>(
+ env: &mut jni::Env<'a>,
+ map: &HashMap<String, String>,
+) -> Result<JObject<'a>, ExecutionError> {
+ let hashmap_class = env
+ .find_class(JNIString::new("java/util/HashMap"))
+ .map_err(|e| ExecutionError::GeneralError(format!("find_class(HashMap): {e}")))?;
+ let ctor = env
+ .get_method_id(
+ &hashmap_class,
+ jni::jni_str!("<init>"),
+ jni::jni_sig!("(I)V"),
+ )
+ .map_err(|e| ExecutionError::GeneralError(format!("HashMap.<init>(I): {e}")))?;
+ let put = env
+ .get_method_id(
+ &hashmap_class,
+ jni::jni_str!("put"),
+ jni::jni_sig!("(Ljava/lang/Object;Ljava/lang/Object;)Ljava/lang/Object;"),
+ )
+ .map_err(|e| ExecutionError::GeneralError(format!("HashMap.put: {e}")))?;
+ let initial_capacity = JValue::Int(map.len() as jint);
+ let instance =
+ unsafe { env.new_object_unchecked(&hashmap_class, ctor, &[initial_capacity.as_jni()]) }
+ .map_err(|e| ExecutionError::GeneralError(format!("new HashMap(int): {e}")))?;
+
+ for (k, v) in map {
+ let k_jstr = env
+ .new_string(k)
+ .map_err(|e| ExecutionError::GeneralError(format!("new_string(key): {e}")))?;
+ let v_jstr = env
+ .new_string(v)
+ .map_err(|e| ExecutionError::GeneralError(format!("new_string(value): {e}")))?;
+ let prev = unsafe {
+ env.call_method_unchecked(
+ &instance,
+ put,
+ ReturnType::Object,
+ &[
+ JValue::Object(&k_jstr).as_jni(),
+ JValue::Object(&v_jstr).as_jni(),
+ ],
+ )
+ }
+ .map_err(|e| ExecutionError::GeneralError(format!("HashMap.put call: {e}")))?;
+ // Discard return value; Java would have reused the existing key but our maps have no dupes.
+ let _ = prev.l();
+ }
+
+ Ok(instance)
+}
+
+struct RawCredentials {
+ access_key_id: String,
+ secret_access_key: String,
+ session_token: Option<String>,
+ /// Absolute expiry. `0` means the provider did not report one.
+ expiration_epoch_millis: i64,
+}
+
+#[async_trait]
+impl CredentialProvider for CometS3CredentialBridge {
+ type Credential = AwsCredential;
+
+ async fn get_credential(&self) -> object_store::Result<Arc<AwsCredential>> {
+ let raw = self.fetch_raw().map_err(|e| object_store::Error::Generic {
+ store: "S3",
+ source: e.to_string().into(),
+ })?;
+ Ok(Arc::new(AwsCredential {
+ key_id: raw.access_key_id,
+ secret_key: raw.secret_access_key,
+ token: raw.session_token,
+ }))
+ }
+}
+
+impl IcebergProvideCredential for CometS3CredentialBridge {
+ type Credential = IcebergAwsCredential;
+
+ async fn provide_credential(
+ &self,
+ _ctx: &Context,
+ ) -> reqsign_core::Result<Option<Self::Credential>> {
+ let raw = self
+ .fetch_raw()
+ .map_err(|e| ReqsignError::new(ReqsignErrorKind::CredentialInvalid, e.to_string()))?;
+
+ let expires_in = if raw.expiration_epoch_millis > 0 {
+ Some(
+ Timestamp::from_millisecond(raw.expiration_epoch_millis).map_err(|e| {
+ ReqsignError::new(
+ ReqsignErrorKind::CredentialInvalid,
+ format!(
+ "Invalid expirationEpochMillis {}: {e}",
+ raw.expiration_epoch_millis
+ ),
+ )
+ })?,
+ )
+ } else {
+ if WARNED_MISSING_EXPIRY.set(()).is_ok() {
+ warn!(
+ "CometS3CredentialProvider returned credentials without expiration; \
+ defaulting to {}s expiry to bound opendal caching",
+ DEFAULT_EXPIRY_WHEN_UNKNOWN.as_secs()
+ );
+ }
+ Some(Timestamp::now() + DEFAULT_EXPIRY_WHEN_UNKNOWN)
+ };
+
+ Ok(Some(IcebergAwsCredential {
+ access_key_id: raw.access_key_id,
+ secret_access_key: raw.secret_access_key,
+ session_token: raw.session_token,
+ expires_in,
+ }))
+ }
+}
+
+fn read_required_string(
+ env: &mut jni::Env,
+ instance: &JObject,
+ field: JFieldID,
+ name: &str,
+) -> Result<String, ExecutionError> {
+ read_optional_string(env, instance, field)?
+ .ok_or_else(|| ExecutionError::GeneralError(format!("{name} was null")))
+}
+
+fn read_optional_string(
+ env: &mut jni::Env,
+ instance: &JObject,
+ field: JFieldID,
+) -> Result<Option<String>, ExecutionError> {
+ let value = unsafe { env.get_field_unchecked(instance, field, ReturnType::Object) }
+ .and_then(|v| v.l())
+ .map_err(|e| ExecutionError::GeneralError(format!("get_field_unchecked: {e}")))?;
+ if value.is_null() {
+ return Ok(None);
+ }
+ let jstr = unsafe { JString::from_raw(env, value.into_raw()) };
+ jstr.try_to_string(env)
+ .map(Some)
+ .map_err(|e| ExecutionError::GeneralError(format!("try_to_string: {e}")))
+}
diff --git a/native/core/src/cloud/s3/mod.rs b/native/core/src/cloud/s3/mod.rs
new file mode 100644
index 0000000000..054919a691
--- /dev/null
+++ b/native/core/src/cloud/s3/mod.rs
@@ -0,0 +1,18 @@
+// 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.
+
+pub mod credential_bridge;
diff --git a/native/core/src/execution/operators/iceberg_scan.rs b/native/core/src/execution/operators/iceberg_scan.rs
index 55bcbef349..ce1388f78f 100644
--- a/native/core/src/execution/operators/iceberg_scan.rs
+++ b/native/core/src/execution/operators/iceberg_scan.rs
@@ -40,8 +40,10 @@ use datafusion::physical_plan::{
use futures::{Stream, StreamExt, TryStreamExt};
use iceberg::arrow::ScanMetrics;
use iceberg::io::{FileIO, FileIOBuilder, StorageFactory};
+use iceberg_storage_opendal::CustomAwsCredentialLoader;
use iceberg_storage_opendal::OpenDalStorageFactory;
+use crate::cloud::s3::credential_bridge::{AccessMode, CometS3CredentialBridge};
use crate::execution::operators::ExecutionError;
use crate::parquet::parquet_support::SparkParquetOptions;
use crate::parquet::schema_adapter::SparkPhysicalExprAdapterFactory;
@@ -49,10 +51,13 @@ use datafusion_comet_spark_expr::EvalMode;
use datafusion_physical_expr_adapter::{PhysicalExprAdapter, PhysicalExprAdapterFactory};
use iceberg::scan::FileScanTask;
+/// Activation key for the `CometS3CredentialProvider` SPI on the Iceberg path, read from a Spark
+/// catalog's `s3.*` property bag.
+const ICEBERG_PROVIDER_CLASS_PROPERTY: &str = "s3.comet.credential.provider.class";
+
/// Iceberg table scan operator that uses iceberg-rust to read Iceberg tables.
///
/// Executes pre-planned FileScanTasks for efficient parallel scanning.
-#[derive(Debug)]
pub struct IcebergScanExec {
/// Iceberg table metadata location for FileIO initialization
metadata_location: String,
@@ -60,8 +65,13 @@ pub struct IcebergScanExec {
output_schema: SchemaRef,
/// Cached execution plan properties
plan_properties: Arc<PlanProperties>,
- /// Catalog-specific configuration for FileIO
+ /// Catalog-specific configuration for FileIO. Holds the unfiltered FileIO property bag, which
+ /// may contain OAuth tokens, REST `credentials.uri`, and other secrets the credential bridge
+ /// needs. Redacted in `Debug` so plan dumps and tracing do not leak credentials.
catalog_properties: HashMap<String, String>,
+ /// Spark V2 catalog name; forwarded as dispatchKey to the credential bridge. Empty when the
+ /// table has no catalog identity.
+ catalog_name: String,
/// Pre-planned file scan tasks
tasks: Vec<FileScanTask>,
/// Number of data files to read concurrently
@@ -75,6 +85,7 @@ impl IcebergScanExec {
metadata_location: String,
schema: SchemaRef,
catalog_properties: HashMap<String, String>,
+ catalog_name: String,
tasks: Vec<FileScanTask>,
data_file_concurrency_limit: usize,
) -> Result<Self, ExecutionError> {
@@ -88,6 +99,7 @@ impl IcebergScanExec {
output_schema,
plan_properties,
catalog_properties,
+ catalog_name,
tasks,
data_file_concurrency_limit,
metrics,
@@ -154,7 +166,11 @@ impl IcebergScanExec {
context: Arc<TaskContext>,
) -> DFResult<SendableRecordBatchStream> {
let output_schema = Arc::clone(&self.output_schema);
- let file_io = Self::load_file_io(&self.catalog_properties, &self.metadata_location)?;
+ let file_io = Self::load_file_io(
+ &self.catalog_properties,
+ &self.metadata_location,
+ &self.catalog_name,
+ )?;
let batch_size = context.session_config().batch_size();
let metrics = IcebergScanMetrics::new(&self.metrics);
@@ -199,7 +215,11 @@ impl IcebergScanExec {
Ok(Box::pin(wrapped_stream))
}
- fn storage_factory_for(path: &str) -> Result<Arc<dyn StorageFactory>, DataFusionError> {
+ fn storage_factory_for(
+ path: &str,
+ catalog_properties: &HashMap<String, String>,
+ catalog_name: &str,
+ ) -> Result<Arc<dyn StorageFactory>, DataFusionError> {
let scheme = if path.contains("://") {
path.split("://").next().unwrap_or("file")
} else {
@@ -207,9 +227,13 @@ impl IcebergScanExec {
};
match scheme {
"file" => Ok(Arc::new(OpenDalStorageFactory::Fs)),
- "s3" | "s3a" => Ok(Arc::new(OpenDalStorageFactory::S3 {
- customized_credential_load: None,
- })),
+ "s3" | "s3a" => {
+ let customized_credential_load =
+ build_s3_credential_loader(path, catalog_properties, catalog_name);
+ Ok(Arc::new(OpenDalStorageFactory::S3 {
+ customized_credential_load,
+ }))
+ }
"gs" => Ok(Arc::new(OpenDalStorageFactory::Gcs)),
"oss" => Ok(Arc::new(OpenDalStorageFactory::Oss)),
_ => Err(DataFusionError::Execution(format!(
@@ -221,18 +245,67 @@ impl IcebergScanExec {
fn load_file_io(
catalog_properties: &HashMap<String, String>,
metadata_location: &str,
+ catalog_name: &str,
) -> Result<FileIO, DataFusionError> {
- let factory = Self::storage_factory_for(metadata_location)?;
+ let factory =
+ Self::storage_factory_for(metadata_location, catalog_properties, catalog_name)?;
let mut file_io_builder = FileIOBuilder::new(factory);
+ // Narrow to storage-prefix keys before forwarding to iceberg-rust's FileIO. The full
+ // unfiltered bag (catalog URI, OAuth tokens, credentials.uri, tenant-id, etc.) is kept
+ // upstream so CometS3CredentialBridge can read whatever the vendor needs.
for (key, value) in catalog_properties {
- file_io_builder = file_io_builder.with_prop(key, value);
+ if STORAGE_PROPERTY_PREFIXES.iter().any(|p| key.starts_with(p)) {
+ file_io_builder = file_io_builder.with_prop(key, value);
+ }
}
Ok(file_io_builder.build())
}
}
+const STORAGE_PROPERTY_PREFIXES: &[&str] = &["s3.", "gcs.", "adls.", "client."];
+
+/// Wires the configured Comet credential provider into opendal's S3 service, or returns `None`
+/// so opendal falls back to its default credential chain.
+fn build_s3_credential_loader(
+ metadata_location: &str,
+ catalog_properties: &HashMap<String, String>,
+ catalog_name: &str,
+) -> Option<CustomAwsCredentialLoader> {
+ let url = url::Url::parse(metadata_location).ok()?;
+ let bucket = url.host_str()?;
+ let provider_class = catalog_properties
+ .get(ICEBERG_PROVIDER_CLASS_PROPERTY)
+ .map(|s| s.trim())
+ .filter(|s| !s.is_empty())?;
+ // Fall back to the bucket when the table has no catalog identity (e.g. HadoopTables loaded by
+ // raw path).
+ let dispatch_key: &str = if catalog_name.is_empty() {
+ bucket
+ } else {
+ catalog_name
+ };
+ let bridge = CometS3CredentialBridge::new(
+ provider_class,
+ dispatch_key,
+ bucket,
+ url.path(),
+ AccessMode::Read,
+ catalog_properties,
+ );
+ match bridge {
+ Ok(b) => Some(CustomAwsCredentialLoader::new(b)),
+ Err(e) => {
+ log::warn!(
+ "Failed to initialize CometS3CredentialBridge for {provider_class}: {e}; \
+ falling back to default opendal credential chain"
+ );
+ None
+ }
+ }
+}
+
/// Metrics for IcebergScanExec
struct IcebergScanMetrics {
/// Baseline metrics (output rows, elapsed compute time)
@@ -361,6 +434,38 @@ impl DisplayAs for IcebergScanExec {
}
}
+impl fmt::Debug for IcebergScanExec {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ f.debug_struct("IcebergScanExec")
+ .field("metadata_location", &self.metadata_location)
+ .field("catalog_name", &self.catalog_name)
+ .field(
+ "catalog_properties",
+ &RedactedProperties(&self.catalog_properties),
+ )
+ .field("num_tasks", &self.tasks.len())
+ .field(
+ "data_file_concurrency_limit",
+ &self.data_file_concurrency_limit,
+ )
+ .finish_non_exhaustive()
+ }
+}
+
+/// Wraps a property map so `Debug` shows keys but elides values, since the unfiltered FileIO bag
+/// may contain bearer tokens and OAuth secrets.
+struct RedactedProperties<'a>(&'a HashMap<String, String>);
+
+impl fmt::Debug for RedactedProperties<'_> {
+ fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+ let mut m = f.debug_map();
+ for k in self.0.keys() {
+ m.key(k).value(&"<redacted>");
+ }
+ m.finish()
+ }
+}
+
/// Build projection expressions that adapt batches from a file schema to the target schema.
///
/// The returned expressions can be cached and reused across multiple batches
@@ -397,7 +502,7 @@ fn adapt_batch_with_expressions(
return Ok(batch);
}
- // Zero-column projection (e.g. SELECT count(*)) — preserve row count
+ // Zero-column projection (e.g. SELECT count(*)), preserve row count
if projection_exprs.is_empty() {
return Ok(RecordBatch::try_new_with_options(
Arc::clone(target_schema),
diff --git a/native/core/src/execution/planner.rs b/native/core/src/execution/planner.rs
index 542c3d9536..c6160bddd4 100644
--- a/native/core/src/execution/planner.rs
+++ b/native/core/src/execution/planner.rs
@@ -1487,6 +1487,7 @@ impl PhysicalPlanner {
.map(|(k, v)| (k.clone(), v.clone()))
.collect();
let metadata_location = common.metadata_location.clone();
+ 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;
@@ -1494,6 +1495,7 @@ impl PhysicalPlanner {
metadata_location,
required_schema,
catalog_properties,
+ catalog_name,
tasks,
data_file_concurrency_limit,
)?;
diff --git a/native/core/src/lib.rs b/native/core/src/lib.rs
index f4ae0b8834..19a2d774a0 100644
--- a/native/core/src/lib.rs
+++ b/native/core/src/lib.rs
@@ -65,6 +65,7 @@ pub mod jvm_bridge {
use errors::{try_unwrap_or_throw, CometError, CometResult};
+pub mod cloud;
pub mod execution;
pub mod parquet;
// this module is for non release only. Intended for debugging/profiling purposes
diff --git a/native/core/src/parquet/mod.rs b/native/core/src/parquet/mod.rs
index 806800cd59..c17c24a0dd 100644
--- a/native/core/src/parquet/mod.rs
+++ b/native/core/src/parquet/mod.rs
@@ -24,7 +24,7 @@ pub mod schema_adapter;
pub mod util;
mod cast_column;
-mod objectstore;
+pub(crate) mod objectstore;
use std::collections::HashMap;
use std::task::Poll;
diff --git a/native/core/src/parquet/objectstore/s3.rs b/native/core/src/parquet/objectstore/s3.rs
index a427ad8ad5..f9c5f7a885 100644
--- a/native/core/src/parquet/objectstore/s3.rs
+++ b/native/core/src/parquet/objectstore/s3.rs
@@ -20,6 +20,7 @@ use std::collections::HashMap;
use std::sync::OnceLock;
use url::Url;
+use crate::cloud::s3::credential_bridge::{AccessMode, CometS3CredentialBridge};
use crate::execution::jni_api::get_runtime;
use async_trait::async_trait;
use aws_config::{
@@ -78,11 +79,32 @@ pub fn create_store(
source: "Missing bucket name in S3 URL".into(),
})?;
- let credential_provider =
- get_runtime().block_on(build_credential_provider(configs, bucket, min_ttl))?;
- builder = match credential_provider {
- Some(provider) => builder.with_credentials(Arc::new(provider)),
- None => builder.with_skip_signature(true),
+ // Parquet path: catalog_properties is empty; vendors here read from Hadoop conf.
+ let empty_props: HashMap<String, String> = HashMap::new();
+ builder = match lookup_provider_class(configs, bucket) {
+ Some(provider_class) => {
+ // Fail rather than fall back to the default chain, which could resolve to the wrong
+ // identity for a user who explicitly named a provider.
+ let bridge = CometS3CredentialBridge::new(
+ provider_class,
+ bucket,
+ bucket,
+ url.path(),
+ AccessMode::Read,
+ &empty_props,
+ )
+ .map_err(|e| object_store::Error::Generic {
+ store: "S3",
+ source: format!("CometS3CredentialBridge init failed for {bucket}: {e}").into(),
+ })?;
+ builder.with_credentials(Arc::new(bridge))
+ }
+ None => {
+ match get_runtime().block_on(build_credential_provider(configs, bucket, min_ttl))? {
+ Some(provider) => builder.with_credentials(Arc::new(provider)),
+ None => builder.with_skip_signature(true),
+ }
+ }
};
let s3_configs = extract_s3_config_options(configs, bucket);
@@ -287,7 +309,7 @@ fn get_config<'a>(
})
}
-fn get_config_trimmed<'a>(
+pub(super) fn get_config_trimmed<'a>(
configs: &'a HashMap<String, String>,
bucket: &str,
property: &str,
@@ -295,6 +317,17 @@ fn get_config_trimmed<'a>(
get_config(configs, bucket, property).map(|s| s.trim())
}
+/// Activation key (without `fs.s3a.` prefix) naming the vendor `CometS3CredentialProvider` FQCN.
+/// Per-bucket override is honored via [`get_config_trimmed`].
+const PROVIDER_CLASS_PROPERTY: &str = "comet.credential.provider.class";
+
+fn lookup_provider_class<'a>(
+ configs: &'a HashMap<String, String>,
+ bucket: &str,
+) -> Option<&'a str> {
+ get_config_trimmed(configs, bucket, PROVIDER_CLASS_PROPERTY).filter(|s| !s.is_empty())
+}
+
// Hadoop S3A credential provider constants
const HADOOP_IAM_INSTANCE: &str = "org.apache.hadoop.fs.s3a.auth.IAMInstanceCredentialsProvider";
const HADOOP_SIMPLE: &str = "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider";
diff --git a/native/jni-bridge/src/comet_s3_credential_dispatcher.rs b/native/jni-bridge/src/comet_s3_credential_dispatcher.rs
new file mode 100644
index 0000000000..b38ca0348a
--- /dev/null
+++ b/native/jni-bridge/src/comet_s3_credential_dispatcher.rs
@@ -0,0 +1,90 @@
+// 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.
+
+use jni::{
+ errors::Result as JniResult,
+ objects::{JClass, JFieldID, JStaticMethodID},
+ signature::{Primitive, ReturnType},
+ strings::JNIString,
+ Env,
+};
+
+/// JNI handles for the JVM `CometS3CredentialDispatcher` SPI plus the `CometS3Credentials` POJO
+/// whose fields the native bridge reads back.
+pub struct CometS3CredentialDispatcher<'a> {
+ pub class: JClass<'a>,
+ /// Retained so the cached POJO `JFieldID`s stay alive for the executor lifetime.
+ #[allow(dead_code)]
+ pub credentials_class: JClass<'a>,
+ pub method_ensure_initialized: JStaticMethodID,
+ pub method_ensure_initialized_ret: ReturnType,
+ pub method_get_credentials_for_path: JStaticMethodID,
+ pub method_get_credentials_for_path_ret: ReturnType,
+ pub field_access_key_id: JFieldID,
+ pub field_secret_access_key: JFieldID,
+ pub field_session_token: JFieldID,
+ pub field_expiration_epoch_millis: JFieldID,
+}
+
+impl<'a> CometS3CredentialDispatcher<'a> {
+ pub const JVM_CLASS: &'static str = "org/apache/comet/cloud/s3/CometS3CredentialDispatcher";
+ pub const CREDENTIALS_CLASS: &'static str = "org/apache/comet/cloud/s3/CometS3Credentials";
+
+ pub fn new(env: &mut Env<'a>) -> JniResult<CometS3CredentialDispatcher<'a>> {
+ let class = env.find_class(JNIString::new(Self::JVM_CLASS))?;
+ let credentials_class = env.find_class(JNIString::new(Self::CREDENTIALS_CLASS))?;
+
+ Ok(CometS3CredentialDispatcher {
+ method_ensure_initialized: env.get_static_method_id(
+ JNIString::new(Self::JVM_CLASS),
+ jni::jni_str!("ensureInitialized"),
+ jni::jni_sig!("(Ljava/lang/String;Ljava/lang/String;Ljava/util/Map;)J"),
+ )?,
+ method_ensure_initialized_ret: ReturnType::Primitive(Primitive::Long),
+ method_get_credentials_for_path: env.get_static_method_id(
+ JNIString::new(Self::JVM_CLASS),
+ jni::jni_str!("getCredentialsForPath"),
+ jni::jni_sig!(
+ "(JLjava/lang/String;Ljava/lang/String;I)Lorg/apache/comet/cloud/s3/CometS3Credentials;"
+ ),
+ )?,
+ method_get_credentials_for_path_ret: ReturnType::Object,
+ field_access_key_id: env.get_field_id(
+ &credentials_class,
+ jni::jni_str!("accessKeyId"),
+ jni::jni_sig!("Ljava/lang/String;"),
+ )?,
+ field_secret_access_key: env.get_field_id(
+ &credentials_class,
+ jni::jni_str!("secretAccessKey"),
+ jni::jni_sig!("Ljava/lang/String;"),
+ )?,
+ field_session_token: env.get_field_id(
+ &credentials_class,
+ jni::jni_str!("sessionToken"),
+ jni::jni_sig!("Ljava/lang/String;"),
+ )?,
+ field_expiration_epoch_millis: env.get_field_id(
+ &credentials_class,
+ jni::jni_str!("expirationEpochMillis"),
+ jni::jni_sig!("J"),
+ )?,
+ class,
+ credentials_class,
+ })
+ }
+}
diff --git a/native/jni-bridge/src/lib.rs b/native/jni-bridge/src/lib.rs
index d72323c961..490e80d076 100644
--- a/native/jni-bridge/src/lib.rs
+++ b/native/jni-bridge/src/lib.rs
@@ -191,12 +191,14 @@ mod comet_exec;
pub use comet_exec::*;
mod batch_iterator;
mod comet_metric_node;
+mod comet_s3_credential_dispatcher;
mod comet_task_memory_manager;
mod comet_udf_bridge;
mod shuffle_block_iterator;
use batch_iterator::CometBatchIterator;
pub use comet_metric_node::*;
+pub use comet_s3_credential_dispatcher::CometS3CredentialDispatcher;
pub use comet_task_memory_manager::*;
use comet_udf_bridge::CometUdfBridge;
use shuffle_block_iterator::CometShuffleBlockIterator;
@@ -233,6 +235,8 @@ pub struct JVMClasses<'a> {
/// The CometUdfBridge class used to dispatch JVM scalar UDFs.
/// `None` if the class is not on the classpath.
pub comet_udf_bridge: Option<CometUdfBridge<'a>>,
+ /// JNI handles for the CometS3CredentialDispatcher SPI and the CometS3Credentials POJO.
+ pub comet_s3_credential_dispatcher: CometS3CredentialDispatcher<'a>,
}
unsafe impl Send for JVMClasses<'_> {}
@@ -310,6 +314,7 @@ impl JVMClasses<'_> {
}
bridge
},
+ comet_s3_credential_dispatcher: CometS3CredentialDispatcher::new(env).unwrap(),
}
});
}
diff --git a/native/proto/src/proto/operator.proto b/native/proto/src/proto/operator.proto
index ed1684b240..7f50aa928c 100644
--- a/native/proto/src/proto/operator.proto
+++ b/native/proto/src/proto/operator.proto
@@ -205,6 +205,12 @@ message IcebergScanCommon {
// Number of data files to read concurrently within a single task
uint32 data_file_concurrency_limit = 12;
+
+ // Spark V2 catalog name that loaded this table. Forwarded as the dispatchKey to
+ // CometS3CredentialDispatcher.ensureInitialized so two catalogs sharing one provider class get
+ // isolated provider instances. Empty string when the table has no catalog identity (e.g.
+ // HadoopTables loaded by raw path).
+ string catalog_name = 13;
}
message IcebergScan {
diff --git a/pom.xml b/pom.xml
index 7419fecc92..dbd439717f 100644
--- a/pom.xml
+++ b/pom.xml
@@ -83,7 +83,7 @@ under the License.
<semanticdb.version>4.13.6</semanticdb.version>
<slf4j.version>2.0.17</slf4j.version>
<guava.version>33.2.1-jre</guava.version>
- <testcontainers.version>1.21.0</testcontainers.version>
+ <testcontainers.version>1.21.4</testcontainers.version>
<amazon-awssdk-v2.version>2.31.51</amazon-awssdk-v2.version>
<jni.dir>${project.basedir}/../native/target/debug</jni.dir>
<platform>darwin</platform>
@@ -476,6 +476,23 @@ under the License.
<version>${testcontainers.version}</version>
<scope>test</scope>
</dependency>
+ <!--
+ Override docker-java transitive (testcontainers 1.21.x pins 3.4.2, which speaks Docker
+ API 1.32 max). Modern daemons (Colima, OrbStack, recent Docker Desktop) require API
+ >= 1.40. docker-java 3.7.0+ defaults to API 1.44.
+ -->
+ <dependency>
+ <groupId>com.github.docker-java</groupId>
+ <artifactId>docker-java-api</artifactId>
+ <version>3.7.1</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.github.docker-java</groupId>
+ <artifactId>docker-java-transport-zerodep</artifactId>
+ <version>3.7.1</version>
+ <scope>test</scope>
+ </dependency>
<!--
AWS SDK modules for Iceberg REST catalog + S3 tests.
iceberg-spark-runtime treats the AWS SDK as provided scope, so tests
diff --git a/spark/src/main/java/org/apache/comet/cloud/s3/CometS3AccessMode.java b/spark/src/main/java/org/apache/comet/cloud/s3/CometS3AccessMode.java
new file mode 100644
index 0000000000..f6ef294141
--- /dev/null
+++ b/spark/src/main/java/org/apache/comet/cloud/s3/CometS3AccessMode.java
@@ -0,0 +1,28 @@
+/*
+ * 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.cloud.s3;
+
+/** Access intent passed to {@link CometS3CredentialProvider#getCredentialsForPath}. */
+public enum CometS3AccessMode {
+ /** GET / HEAD / LIST. All Comet native scan paths request this today. */
+ READ,
+ /** PUT / POST / DELETE / multipart. Reserved for future native write paths. */
+ WRITE
+}
diff --git a/spark/src/main/java/org/apache/comet/cloud/s3/CometS3CredentialContext.java b/spark/src/main/java/org/apache/comet/cloud/s3/CometS3CredentialContext.java
new file mode 100644
index 0000000000..105ac79ee7
--- /dev/null
+++ b/spark/src/main/java/org/apache/comet/cloud/s3/CometS3CredentialContext.java
@@ -0,0 +1,61 @@
+/*
+ * 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.cloud.s3;
+
+import java.util.Objects;
+
+/**
+ * Per-request context passed to {@link
+ * CometS3CredentialProvider#getCredentialsForPath(CometS3CredentialContext)}. New fields can be
+ * added here without changing the SPI method signature, so vendors compiled against earlier
+ * versions stay binary-compatible.
+ */
+public final class CometS3CredentialContext {
+
+ private final String bucket;
+ private final String path;
+ private final CometS3AccessMode mode;
+
+ public CometS3CredentialContext(String bucket, String path, CometS3AccessMode mode) {
+ this.bucket = Objects.requireNonNull(bucket, "bucket");
+ this.path = Objects.requireNonNull(path, "path");
+ this.mode = Objects.requireNonNull(mode, "mode");
+ }
+
+ /** S3 bucket name (no scheme, no path). */
+ public String getBucket() {
+ return bucket;
+ }
+
+ /** Object key or prefix, leading slash included (matches the URL path component). */
+ public String getPath() {
+ return path;
+ }
+
+ /** Access intent for this request. */
+ public CometS3AccessMode getMode() {
+ return mode;
+ }
+
+ @Override
+ public String toString() {
+ return "CometS3CredentialContext{bucket=" + bucket + ", path=" + path + ", mode=" + mode + "}";
+ }
+}
diff --git a/spark/src/main/java/org/apache/comet/cloud/s3/CometS3CredentialDispatcher.java b/spark/src/main/java/org/apache/comet/cloud/s3/CometS3CredentialDispatcher.java
new file mode 100644
index 0000000000..311b36a64a
--- /dev/null
+++ b/spark/src/main/java/org/apache/comet/cloud/s3/CometS3CredentialDispatcher.java
@@ -0,0 +1,205 @@
+/*
+ * 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.cloud.s3;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.comet.util.ClassLoaders;
+
+/**
+ * JNI entry point that resolves a {@link CometS3CredentialProvider} for native code.
+ *
+ * <p>{@link #ensureInitialized} reflects the named class, runs {@code initialize(Map)} once, and
+ * returns a {@code long} handle. {@link #getCredentialsForPath} takes that handle on every
+ * per-request call. See the design notes in the contributor guide for why the SPI is shaped this
+ * way (keying, multi-tenant isolation, shutdown lifecycle).
+ */
+public final class CometS3CredentialDispatcher {
+
+ private static final Logger LOG = LoggerFactory.getLogger(CometS3CredentialDispatcher.class);
+
+ private static final ConcurrentHashMap<InstanceKey, Long> KEY_TO_HANDLE =
+ new ConcurrentHashMap<>();
+ private static final ConcurrentHashMap<Long, RegisteredProvider> INSTANCES =
+ new ConcurrentHashMap<>();
+ private static final AtomicLong HANDLE_SEQ = new AtomicLong(1L);
+ private static final CometS3AccessMode[] MODES = CometS3AccessMode.values();
+
+ static {
+ Runtime.getRuntime()
+ .addShutdownHook(
+ new Thread(CometS3CredentialDispatcher::closeAll, "comet-s3-credential-shutdown"));
+ }
+
+ private CometS3CredentialDispatcher() {}
+
+ /**
+ * Reflects and initializes the named provider on first call for the {@code (FQCN, dispatchKey,
+ * catalogProperties)} triple, and returns a handle reused by subsequent {@link
+ * #getCredentialsForPath} calls.
+ */
+ public static long ensureInitialized(
+ String providerClassName, String dispatchKey, Map<String, String> catalogProperties) {
+ if (providerClassName == null || providerClassName.isEmpty()) {
+ throw new IllegalArgumentException(
+ "providerClassName is empty; native side should not call without a configured class");
+ }
+ Map<String, String> snapshot =
+ catalogProperties == null
+ ? Collections.emptyMap()
+ : Collections.unmodifiableMap(new HashMap<>(catalogProperties));
+ InstanceKey key =
+ new InstanceKey(providerClassName, dispatchKey == null ? "" : dispatchKey, snapshot);
+ return KEY_TO_HANDLE.computeIfAbsent(
+ key,
+ k -> {
+ CometS3CredentialProvider provider = instantiate(k.providerClassName);
+ provider.initialize(k.catalogProperties);
+ long handle = HANDLE_SEQ.getAndIncrement();
+ INSTANCES.put(handle, new RegisteredProvider(provider, k));
+ return handle;
+ });
+ }
+
+ /**
+ * Invoked by native code on every per-request credential fetch. {@code handle} must be a value
+ * returned by a prior {@link #ensureInitialized} call; otherwise this throws. {@code mode} is the
+ * {@link CometS3AccessMode} ordinal.
+ */
+ public static CometS3Credentials getCredentialsForPath(
+ long handle, String bucket, String path, int mode) throws Exception {
+ if (mode < 0 || mode >= MODES.length) {
+ throw new IllegalArgumentException("Invalid CometS3AccessMode ordinal: " + mode);
+ }
+ RegisteredProvider registered = INSTANCES.get(handle);
+ if (registered == null) {
+ throw new IllegalStateException(
+ "CometS3CredentialProvider handle "
+ + handle
+ + " was not initialized; "
+ + "ensureInitialized must be called before getCredentialsForPath");
+ }
+ CometS3AccessMode accessMode = MODES[mode];
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Fetching credentials via {} (dispatchKey={}, handle={}) for bucket={} path={} mode={}",
+ registered.key.providerClassName,
+ registered.key.dispatchKey,
+ handle,
+ bucket,
+ path,
+ accessMode);
+ }
+ return registered.provider.getCredentialsForPath(
+ new CometS3CredentialContext(bucket, path, accessMode));
+ }
+
+ private static CometS3CredentialProvider instantiate(String providerClassName) {
+ Class<?> clazz;
+ try {
+ clazz = ClassLoaders.loadClass(providerClassName);
+ } catch (ClassNotFoundException e) {
+ throw new IllegalStateException(
+ "CometS3CredentialProvider class not found: "
+ + providerClassName
+ + ". Ensure the vendor JAR is on the executor classpath.",
+ e);
+ }
+ if (!CometS3CredentialProvider.class.isAssignableFrom(clazz)) {
+ throw new IllegalStateException(
+ providerClassName + " does not implement " + CometS3CredentialProvider.class.getName());
+ }
+ try {
+ Object instance = clazz.getDeclaredConstructor().newInstance();
+ LOG.info("Instantiated CometS3CredentialProvider {}", providerClassName);
+ return (CometS3CredentialProvider) instance;
+ } catch (NoSuchMethodException e) {
+ throw new IllegalStateException(
+ providerClassName + " must declare a public no-arg constructor", e);
+ } catch (InstantiationException | IllegalAccessException | InvocationTargetException e) {
+ throw new IllegalStateException(
+ "Failed to instantiate CometS3CredentialProvider " + providerClassName, e);
+ }
+ }
+
+ /** Visible for tests; otherwise invoked from the JVM shutdown hook. */
+ static void closeAll() {
+ for (RegisteredProvider registered : INSTANCES.values()) {
+ try {
+ registered.provider.close();
+ } catch (Throwable t) {
+ LOG.warn(
+ "Exception from {} (dispatchKey={}).close() during shutdown",
+ registered.key.providerClassName,
+ registered.key.dispatchKey,
+ t);
+ }
+ }
+ INSTANCES.clear();
+ KEY_TO_HANDLE.clear();
+ }
+
+ private static final class InstanceKey {
+ final String providerClassName;
+ final String dispatchKey;
+ final Map<String, String> catalogProperties;
+
+ InstanceKey(
+ String providerClassName, String dispatchKey, Map<String, String> catalogProperties) {
+ this.providerClassName = providerClassName;
+ this.dispatchKey = dispatchKey;
+ this.catalogProperties = catalogProperties;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (!(o instanceof InstanceKey)) return false;
+ InstanceKey other = (InstanceKey) o;
+ return providerClassName.equals(other.providerClassName)
+ && dispatchKey.equals(other.dispatchKey)
+ && catalogProperties.equals(other.catalogProperties);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(providerClassName, dispatchKey, catalogProperties);
+ }
+ }
+
+ private static final class RegisteredProvider {
+ final CometS3CredentialProvider provider;
+ final InstanceKey key;
+
+ RegisteredProvider(CometS3CredentialProvider provider, InstanceKey key) {
+ this.provider = provider;
+ this.key = key;
+ }
+ }
+}
diff --git a/spark/src/main/java/org/apache/comet/cloud/s3/CometS3CredentialProvider.java b/spark/src/main/java/org/apache/comet/cloud/s3/CometS3CredentialProvider.java
new file mode 100644
index 0000000000..a77b9f2f01
--- /dev/null
+++ b/spark/src/main/java/org/apache/comet/cloud/s3/CometS3CredentialProvider.java
@@ -0,0 +1,54 @@
+/*
+ * 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.cloud.s3;
+
+import java.util.Map;
+
+/**
+ * SPI for supplying AWS credentials to Comet's native S3 readers, which bypass Hadoop S3A. See the
+ * user guide (operator setup, vendor contract) and the contributor-guide design notes for the
+ * rationale.
+ *
+ * <p>{@link #getCredentialsForPath} may be invoked concurrently from many native worker threads;
+ * implementations must be thread-safe. It returns credentials or throws (no fall-through).
+ */
+public interface CometS3CredentialProvider extends AutoCloseable {
+
+ /**
+ * Called once per Comet-cached instance before any {@link #getCredentialsForPath} call. Must be
+ * cheap and non-blocking. On the Iceberg path the map carries the unfiltered FileIO bag; on the
+ * Parquet path it is empty.
+ *
+ * @param catalogProperties may contain secrets, do not log
+ */
+ default void initialize(Map<String, String> catalogProperties) {}
+
+ /**
+ * @return non-null credentials; {@code null} is a contract violation
+ */
+ CometS3Credentials getCredentialsForPath(CometS3CredentialContext context) throws Exception;
+
+ /**
+ * Invoked from the dispatcher's best-effort JVM shutdown hook. Default no-op suits stateless
+ * providers; override to release HTTP clients, refresh executors, etc.
+ */
+ @Override
+ default void close() throws Exception {}
+}
diff --git a/spark/src/main/java/org/apache/comet/cloud/s3/CometS3Credentials.java b/spark/src/main/java/org/apache/comet/cloud/s3/CometS3Credentials.java
new file mode 100644
index 0000000000..3342f28c24
--- /dev/null
+++ b/spark/src/main/java/org/apache/comet/cloud/s3/CometS3Credentials.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.cloud.s3;
+
+import java.util.Objects;
+
+/**
+ * Credentials returned by a {@link CometS3CredentialProvider}. Field names are read back over JNI
+ * by name and are part of the cross-language contract. {@code sessionToken} is null for non-STS
+ * credentials; {@code expirationEpochMillis} of {@code 0} means "unknown".
+ */
+public final class CometS3Credentials {
+
+ private final String accessKeyId;
+ private final String secretAccessKey;
+ private final String sessionToken;
+ private final long expirationEpochMillis;
+
+ public CometS3Credentials(
+ String accessKeyId, String secretAccessKey, String sessionToken, long expirationEpochMillis) {
+ this.accessKeyId = Objects.requireNonNull(accessKeyId, "accessKeyId");
+ this.secretAccessKey = Objects.requireNonNull(secretAccessKey, "secretAccessKey");
+ this.sessionToken = sessionToken;
+ this.expirationEpochMillis = expirationEpochMillis;
+ }
+
+ public String getAccessKeyId() {
+ return accessKeyId;
+ }
+
+ public String getSecretAccessKey() {
+ return secretAccessKey;
+ }
+
+ public String getSessionToken() {
+ return sessionToken;
+ }
+
+ public long getExpirationEpochMillis() {
+ return expirationEpochMillis;
+ }
+}
diff --git a/spark/src/main/java/org/apache/comet/util/ClassLoaders.java b/spark/src/main/java/org/apache/comet/util/ClassLoaders.java
new file mode 100644
index 0000000000..ee7e620bdc
--- /dev/null
+++ b/spark/src/main/java/org/apache/comet/util/ClassLoaders.java
@@ -0,0 +1,43 @@
+/*
+ * 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.util;
+
+/** ClassLoader helpers shared across Comet modules. */
+public final class ClassLoaders {
+
+ private ClassLoaders() {}
+
+ /**
+ * Loads a class using the thread context ClassLoader if available, falling back to the system
+ * ClassLoader. Spark wires user JARs onto the context ClassLoader, so vendor classes named in
+ * Spark configs are reachable through this path.
+ */
+ public static Class<?> loadClass(String className) throws ClassNotFoundException {
+ ClassLoader classLoader = Thread.currentThread().getContextClassLoader();
+ if (classLoader != null) {
+ // scalastyle:off classforname
+ return Class.forName(className, true, classLoader);
+ // scalastyle:on classforname
+ }
+ // scalastyle:off classforname
+ return Class.forName(className);
+ // scalastyle:on classforname
+ }
+}
diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
index e85eac2c40..2a39369c76 100644
--- a/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergReflection.scala
@@ -20,6 +20,9 @@
package org.apache.comet.iceberg
import org.apache.spark.internal.Logging
+import org.apache.spark.sql.SparkSession
+
+import org.apache.comet.util.ClassLoaders
/**
* Shared reflection utilities for Iceberg operations.
@@ -100,19 +103,7 @@ object IcebergReflection extends Logging {
* @return
* The loaded Class object
*/
- def loadClass(className: String): Class[_] = {
- val classLoader = Thread.currentThread().getContextClassLoader
- if (classLoader != null) {
- // scalastyle:off classforname
- Class.forName(className, true, classLoader)
- // scalastyle:on classforname
- } else {
- // Fallback to default classloader if context classloader is null
- // scalastyle:off classforname
- Class.forName(className)
- // scalastyle:on classforname
- }
- }
+ def loadClass(className: String): Class[_] = ClassLoaders.loadClass(className)
/**
* Searches through class hierarchy to find a method (including protected methods).
@@ -672,6 +663,9 @@ object IcebergReflection extends Logging {
* Mapping from column names to Iceberg field IDs (built from scanSchema)
* @param catalogProperties
* Catalog properties for FileIO (S3 credentials, regions, etc.)
+ * @param catalogName
+ * Spark V2 catalog name forwarded as `dispatchKey` to CometS3CredentialBridge. `None` when the
+ * table has no catalog identity (e.g. HadoopTables loaded by raw path).
*/
case class CometIcebergNativeScanMetadata(
table: Any,
@@ -682,6 +676,7 @@ case class CometIcebergNativeScanMetadata(
tableSchema: Any,
globalFieldIdMapping: Map[String, Int],
catalogProperties: Map[String, String],
+ catalogName: Option[String],
fileFormat: String)
object CometIcebergNativeScanMetadata extends Logging {
@@ -737,7 +732,64 @@ object CometIcebergNativeScanMetadata extends Logging {
tableSchema = tableSchema,
globalFieldIdMapping = globalFieldIdMapping,
catalogProperties = catalogProperties,
+ catalogName = deriveCatalogName(table),
fileFormat = FileFormats.PARQUET)
}
}
+
+ /**
+ * Extracts the Spark V2 catalog name from an Iceberg `Table`. `Table.name()` returns
+ * `catalog.namespace.table` for tables loaded through a catalog; we intersect against the
+ * registered V2 catalogs so a value like `s3.foo` is not mistaken for a catalog `s3`. Returns
+ * `None` for HadoopTables loaded by raw path or when reflection fails.
+ */
+ private[iceberg] def deriveCatalogName(table: Any): Option[String] =
+ deriveCatalogName(table, registeredCatalogNames _)
+
+ /**
+ * Test seam that lets tests inject a fixed catalog set without bootstrapping a SparkSession.
+ */
+ private[iceberg] def deriveCatalogName(
+ table: Any,
+ knownCatalogNames: () => Iterable[String]): Option[String] = {
+ if (table == null) return None
+ invokeTableName(table).flatMap { name =>
+ if (name.isEmpty || name == "null") {
+ None
+ } else {
+ knownCatalogNames()
+ .find(c => name == c || name.startsWith(c + "."))
+ .orElse {
+ val idx = name.indexOf('.')
+ if (idx > 0) Some(name.substring(0, idx)) else None
+ }
+ }
+ }
+ }
+
+ private def invokeTableName(table: Any): Option[String] = {
+ try {
+ table.getClass.getMethod("name").invoke(table) match {
+ case s: String => Some(s)
+ case other if other != null => Some(other.toString)
+ case null => None
+ }
+ } catch {
+ case e: Exception =>
+ logWarning(
+ s"Iceberg reflection: Table.name() not callable on ${table.getClass.getName}. " +
+ "Native S3 credential dispatch will fall back to bucket-keyed isolation: " +
+ s"${e.getMessage}")
+ None
+ }
+ }
+
+ private def registeredCatalogNames(): Iterable[String] =
+ try {
+ SparkSession.active.sessionState.catalogManager.listCatalogs(None)
+ } catch {
+ case e: Exception =>
+ logDebug(s"Could not list V2 catalogs from SparkSession: ${e.getMessage}")
+ Nil
+ }
}
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 e6b049aae2..7601fa1c6b 100644
--- a/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
+++ b/spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala
@@ -367,12 +367,13 @@ case class CometScanRule(session: SparkSession)
val hadoopDerivedProperties =
CometIcebergNativeScan.hadoopToIcebergS3Properties(hadoopS3Options)
- // Extract vended credentials from FileIO (REST catalog credential vending).
- // FileIO properties take precedence over Hadoop-derived properties because
- // they contain per-table credentials vended by the REST catalog.
+ // Forward the full FileIO property bag (including credentials.uri, OAuth tokens,
+ // tenant-id, etc.) so a CometS3CredentialProvider can see everything LoadTableResponse
+ // returned. The storage-prefix narrowing happens native-side just before
+ // FileIOBuilder.with_prop, since iceberg-rust's FileIO is the only consumer that
+ // requires the narrowed view.
val fileIOProperties = tableOpt
.flatMap(IcebergReflection.getFileIOProperties)
- .map(CometIcebergNativeScan.filterStorageProperties)
.getOrElse(Map.empty)
val catalogProperties = hadoopDerivedProperties ++ fileIOProperties
diff --git a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
index b0de978af8..8c88c19f1f 100644
--- a/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
+++ b/spark/src/main/scala/org/apache/comet/serde/operator/CometIcebergNativeScan.scala
@@ -500,21 +500,6 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
}
}
- /** Storage-related property prefixes passed through to native FileIO. */
- private val storagePropertyPrefixes =
- Seq("s3.", "gcs.", "adls.", "client.")
-
- /**
- * Filters a properties map to only include storage-related keys. FileIO.properties() may
- * contain catalog URIs, bearer tokens, and other non-storage settings that should not be passed
- * to the native FileIO builder.
- */
- def filterStorageProperties(props: Map[String, String]): Map[String, String] = {
- props.filter { case (key, _) =>
- storagePropertyPrefixes.exists(prefix => key.startsWith(prefix))
- }
- }
-
/**
* Transforms Hadoop S3A configuration keys to Iceberg FileIO property keys.
*
@@ -771,6 +756,7 @@ object CometIcebergNativeScan extends CometOperatorSerde[CometBatchScanExec] wit
commonBuilder.setMetadataLocation(metadata.metadataLocation)
commonBuilder.setDataFileConcurrencyLimit(
CometConf.COMET_ICEBERG_DATA_FILE_CONCURRENCY_LIMIT.get())
+ metadata.catalogName.foreach(commonBuilder.setCatalogName)
metadata.catalogProperties.foreach { case (key, value) =>
commonBuilder.putCatalogProperties(key, value)
}
diff --git a/spark/src/test/java/org/apache/comet/cloud/s3/CometS3CredentialDispatcherTest.java b/spark/src/test/java/org/apache/comet/cloud/s3/CometS3CredentialDispatcherTest.java
new file mode 100644
index 0000000000..9601497f1b
--- /dev/null
+++ b/spark/src/test/java/org/apache/comet/cloud/s3/CometS3CredentialDispatcherTest.java
@@ -0,0 +1,270 @@
+/*
+ * 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.cloud.s3;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertSame;
+import static org.junit.Assert.assertThrows;
+import static org.junit.Assert.assertTrue;
+
+public class CometS3CredentialDispatcherTest {
+
+ private static final String TEST_PROVIDER = TestCometS3CredentialProvider.class.getName();
+ private static final String DK = "test-dispatch-key";
+ private static final int READ = CometS3AccessMode.READ.ordinal();
+ private static final int WRITE = CometS3AccessMode.WRITE.ordinal();
+
+ @Before
+ public void resetTestProvider() {
+ // Each test starts with an empty dispatcher cache so providers and counters from prior tests
+ // do not leak through closeAll() / initCount / closeCount assertions.
+ CometS3CredentialDispatcher.closeAll();
+ TestCometS3CredentialProvider.reset();
+ }
+
+ private long init() {
+ return CometS3CredentialDispatcher.ensureInitialized(TEST_PROVIDER, DK, Collections.emptyMap());
+ }
+
+ @Test
+ public void getCredentialsRoundTripsThroughProvider() throws Exception {
+ long handle = init();
+ CometS3Credentials creds =
+ CometS3CredentialDispatcher.getCredentialsForPath(
+ handle, "my-bucket", "path/to/object", READ);
+
+ assertNotNull(creds);
+ assertEquals("AKIATEST", creds.getAccessKeyId());
+ assertEquals("secret", creds.getSecretAccessKey());
+ assertEquals("session-tok", creds.getSessionToken());
+ assertEquals(0L, creds.getExpirationEpochMillis());
+
+ assertEquals(1, TestCometS3CredentialProvider.callCount.get());
+ assertEquals("my-bucket", TestCometS3CredentialProvider.lastBucket);
+ assertEquals("path/to/object", TestCometS3CredentialProvider.lastPath);
+ assertEquals(CometS3AccessMode.READ, TestCometS3CredentialProvider.lastMode);
+ }
+
+ @Test
+ public void writeModeIsForwarded() throws Exception {
+ long handle = init();
+ CometS3CredentialDispatcher.getCredentialsForPath(handle, "b", "k", WRITE);
+ assertEquals(CometS3AccessMode.WRITE, TestCometS3CredentialProvider.lastMode);
+ }
+
+ @Test
+ public void unknownModeRejected() {
+ long handle = init();
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> CometS3CredentialDispatcher.getCredentialsForPath(handle, "b", "k", 99));
+ }
+
+ @Test
+ public void emptyClassNameRejected() {
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> CometS3CredentialDispatcher.ensureInitialized("", DK, Collections.emptyMap()));
+ assertThrows(
+ IllegalArgumentException.class,
+ () -> CometS3CredentialDispatcher.ensureInitialized(null, DK, Collections.emptyMap()));
+ }
+
+ @Test
+ public void missingClassReportsActionableError() {
+ Exception thrown =
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ CometS3CredentialDispatcher.ensureInitialized(
+ "com.does.not.Exist", DK, Collections.emptyMap()));
+ assertTrue(thrown.getMessage().contains("not found"));
+ }
+
+ @Test
+ public void classNotImplementingInterfaceRejected() {
+ Exception thrown =
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ CometS3CredentialDispatcher.ensureInitialized(
+ NotACredentialProvider.class.getName(), DK, Collections.emptyMap()));
+ assertTrue(thrown.getMessage().contains("does not implement"));
+ }
+
+ @Test
+ public void classWithoutNoArgCtorRejected() {
+ Exception thrown =
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ CometS3CredentialDispatcher.ensureInitialized(
+ NoNoArgCtorProvider.class.getName(), DK, Collections.emptyMap()));
+ assertTrue(thrown.getMessage().contains("no-arg constructor"));
+ }
+
+ @Test
+ public void getWithoutEnsureInitializedThrows() {
+ Exception thrown =
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ CometS3CredentialDispatcher.getCredentialsForPath(Long.MAX_VALUE, "b", "k", READ));
+ assertTrue(thrown.getMessage().contains("not initialized"));
+ }
+
+ @Test
+ public void initializeCalledExactlyOncePerKey() {
+ Map<String, String> props = new HashMap<>();
+ props.put("tenant-id", "T1");
+
+ long h1 = CometS3CredentialDispatcher.ensureInitialized(TEST_PROVIDER, "cat-a", props);
+ long h2 = CometS3CredentialDispatcher.ensureInitialized(TEST_PROVIDER, "cat-a", props);
+ long h3 = CometS3CredentialDispatcher.ensureInitialized(TEST_PROVIDER, "cat-a", props);
+
+ assertEquals(1, TestCometS3CredentialProvider.initCount.get());
+ assertEquals(h1, h2);
+ assertEquals(h1, h3);
+ }
+
+ @Test
+ public void distinctDispatchKeysIsolateInstances() throws Exception {
+ Map<String, String> propsA = new HashMap<>();
+ propsA.put("tenant-id", "T-A");
+ Map<String, String> propsB = new HashMap<>();
+ propsB.put("tenant-id", "T-B");
+
+ long ha = CometS3CredentialDispatcher.ensureInitialized(TEST_PROVIDER, "iso-cat-a", propsA);
+ long hb = CometS3CredentialDispatcher.ensureInitialized(TEST_PROVIDER, "iso-cat-b", propsB);
+
+ assertNotEquals(ha, hb);
+ assertEquals(2, TestCometS3CredentialProvider.initCount.get());
+
+ CometS3CredentialDispatcher.getCredentialsForPath(ha, "b", "k", READ);
+ assertEquals("T-A", TestCometS3CredentialProvider.lastTenantSeen);
+
+ CometS3CredentialDispatcher.getCredentialsForPath(hb, "b", "k", READ);
+ assertEquals("T-B", TestCometS3CredentialProvider.lastTenantSeen);
+ }
+
+ /** Multi-tenant collision: same FQCN and dispatchKey, different catalogProperties; isolated. */
+ @Test
+ public void distinctCatalogPropertiesIsolateInstances() throws Exception {
+ Map<String, String> propsA = new HashMap<>();
+ propsA.put("tenant-id", "T-A");
+ propsA.put("credentials.uri", "https://rest-a.example/credentials");
+ Map<String, String> propsB = new HashMap<>();
+ propsB.put("tenant-id", "T-B");
+ propsB.put("credentials.uri", "https://rest-b.example/credentials");
+
+ long ha = CometS3CredentialDispatcher.ensureInitialized(TEST_PROVIDER, "shared-cat", propsA);
+ long hb = CometS3CredentialDispatcher.ensureInitialized(TEST_PROVIDER, "shared-cat", propsB);
+
+ assertNotEquals(ha, hb);
+ assertEquals(2, TestCometS3CredentialProvider.initCount.get());
+
+ CometS3CredentialDispatcher.getCredentialsForPath(ha, "b", "k", READ);
+ assertEquals("T-A", TestCometS3CredentialProvider.lastTenantSeen);
+
+ CometS3CredentialDispatcher.getCredentialsForPath(hb, "b", "k", READ);
+ assertEquals("T-B", TestCometS3CredentialProvider.lastTenantSeen);
+ }
+
+ @Test
+ public void providerExceptionsPropagate() {
+ long handle = init();
+ IllegalStateException boom = new IllegalStateException("simulated STS failure");
+ TestCometS3CredentialProvider.throwOnNext = boom;
+
+ Exception thrown =
+ assertThrows(
+ Exception.class,
+ () -> CometS3CredentialDispatcher.getCredentialsForPath(handle, "b", "k", READ));
+ assertSame(boom, thrown);
+ }
+
+ @Test
+ public void nullSessionTokenAllowed() throws Exception {
+ long handle = init();
+ TestCometS3CredentialProvider.nextResult = new CometS3Credentials("AKIA", "sec", null, 0L);
+
+ CometS3Credentials creds =
+ CometS3CredentialDispatcher.getCredentialsForPath(handle, "b", "k", READ);
+
+ assertNull(creds.getSessionToken());
+ }
+
+ @Test
+ public void providerReceivesEachCallSeparately() throws Exception {
+ long handle = init();
+ CometS3CredentialDispatcher.getCredentialsForPath(handle, "b1", "k1", READ);
+ CometS3CredentialDispatcher.getCredentialsForPath(handle, "b2", "k2", READ);
+ CometS3CredentialDispatcher.getCredentialsForPath(handle, "b3", "k3", READ);
+
+ assertEquals(3, TestCometS3CredentialProvider.callCount.get());
+ assertEquals("b3", TestCometS3CredentialProvider.lastBucket);
+ assertEquals("k3", TestCometS3CredentialProvider.lastPath);
+ }
+
+ @Test
+ public void closeAllInvokesEveryProvider() {
+ Map<String, String> propsA = new HashMap<>();
+ propsA.put("tenant-id", "T-A");
+ Map<String, String> propsB = new HashMap<>();
+ propsB.put("tenant-id", "T-B");
+
+ CometS3CredentialDispatcher.ensureInitialized(TEST_PROVIDER, "close-a", propsA);
+ CometS3CredentialDispatcher.ensureInitialized(TEST_PROVIDER, "close-b", propsB);
+
+ CometS3CredentialDispatcher.closeAll();
+
+ assertEquals(2, TestCometS3CredentialProvider.closeCount.get());
+
+ long fresh = CometS3CredentialDispatcher.ensureInitialized(TEST_PROVIDER, "close-a", propsA);
+ assertEquals(3, TestCometS3CredentialProvider.initCount.get());
+ assertNotNull(fresh);
+ }
+
+ /** A failing vendor close() must not block other providers from being closed. */
+ @Test
+ public void closeAllSwallowsProviderExceptions() {
+ CometS3CredentialDispatcher.ensureInitialized(
+ TEST_PROVIDER, "close-throws-a", Collections.emptyMap());
+ CometS3CredentialDispatcher.ensureInitialized(
+ TEST_PROVIDER, "close-throws-b", Collections.emptyMap());
+
+ TestCometS3CredentialProvider.throwOnClose = new IllegalStateException("simulated cleanup");
+
+ CometS3CredentialDispatcher.closeAll();
+
+ // Both close() invocations ran even though one (the first to be visited) threw.
+ assertEquals(2, TestCometS3CredentialProvider.closeCount.get());
+ }
+}
diff --git a/spark/src/test/java/org/apache/comet/cloud/s3/MinioCometS3CredentialProvider.java b/spark/src/test/java/org/apache/comet/cloud/s3/MinioCometS3CredentialProvider.java
new file mode 100644
index 0000000000..2b4f5e8c86
--- /dev/null
+++ b/spark/src/test/java/org/apache/comet/cloud/s3/MinioCometS3CredentialProvider.java
@@ -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.
+ */
+
+package org.apache.comet.cloud.s3;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+/**
+ * Test-only {@link CometS3CredentialProvider} named via {@code
+ * spark.hadoop.fs.s3a.comet.credential.provider.class} (and the per-catalog Iceberg form). Returns
+ * Minio's static credentials once {@link #installCredentials} has been called and counts every
+ * invocation so suites can assert the bridge was actually used.
+ */
+public final class MinioCometS3CredentialProvider implements CometS3CredentialProvider {
+
+ /**
+ * Property key used by tests to tag a per-catalog provider instance. The dispatcher key alone is
+ * not visible to {@code initialize(Map)}, so suites that want to identify which catalog's
+ * instance they are inspecting put a discriminator under this key on the catalog config.
+ */
+ public static final String TEST_INSTANCE_TAG = "comet.test-instance-tag";
+
+ private static final AtomicReference<Credentials> CREDS = new AtomicReference<>();
+ private static final AtomicInteger CALL_COUNT = new AtomicInteger(0);
+ private static final AtomicInteger INIT_COUNT = new AtomicInteger(0);
+ private static final AtomicReference<String> LAST_BUCKET = new AtomicReference<>();
+ private static final AtomicReference<String> LAST_PATH = new AtomicReference<>();
+ private static final AtomicReference<Map<String, String>> LAST_INIT_PROPS =
+ new AtomicReference<>();
+
+ /**
+ * Captures one entry per `initialize(Map)` invocation, keyed by the value of {@link
+ * #TEST_INSTANCE_TAG} in the supplied map. Lets multi-catalog tests look up the per-instance init
+ * bag without relying on the most-recent-wins {@link #LAST_INIT_PROPS}.
+ */
+ private static final ConcurrentHashMap<String, Map<String, String>> INIT_BY_TAG =
+ new ConcurrentHashMap<>();
+
+ public static void installCredentials(String accessKeyId, String secretAccessKey) {
+ CREDS.set(new Credentials(accessKeyId, secretAccessKey));
+ }
+
+ public static int callCount() {
+ return CALL_COUNT.get();
+ }
+
+ public static int initCount() {
+ return INIT_COUNT.get();
+ }
+
+ public static String lastBucket() {
+ return LAST_BUCKET.get();
+ }
+
+ public static String lastPath() {
+ return LAST_PATH.get();
+ }
+
+ public static Map<String, String> lastInitProperties() {
+ return LAST_INIT_PROPS.get();
+ }
+
+ public static Map<String, String> initPropertiesForTag(String tag) {
+ return INIT_BY_TAG.get(tag);
+ }
+
+ public static void resetCounters() {
+ CALL_COUNT.set(0);
+ INIT_COUNT.set(0);
+ LAST_BUCKET.set(null);
+ LAST_PATH.set(null);
+ LAST_INIT_PROPS.set(null);
+ INIT_BY_TAG.clear();
+ }
+
+ @Override
+ public void initialize(Map<String, String> catalogProperties) {
+ INIT_COUNT.incrementAndGet();
+ Map<String, String> snapshot =
+ Collections.unmodifiableMap(new java.util.HashMap<>(catalogProperties));
+ LAST_INIT_PROPS.set(snapshot);
+ String tag = catalogProperties.get(TEST_INSTANCE_TAG);
+ if (tag != null) {
+ INIT_BY_TAG.put(tag, snapshot);
+ }
+ }
+
+ @Override
+ public CometS3Credentials getCredentialsForPath(CometS3CredentialContext context) {
+ CALL_COUNT.incrementAndGet();
+ LAST_BUCKET.set(context.getBucket());
+ LAST_PATH.set(context.getPath());
+ Credentials c = CREDS.get();
+ if (c == null) {
+ throw new IllegalStateException(
+ "MinioCometS3CredentialProvider used before installCredentials() was called");
+ }
+ return new CometS3Credentials(c.accessKeyId, c.secretAccessKey, null, 0L);
+ }
+
+ private static final class Credentials {
+ final String accessKeyId;
+ final String secretAccessKey;
+
+ Credentials(String accessKeyId, String secretAccessKey) {
+ this.accessKeyId = accessKeyId;
+ this.secretAccessKey = secretAccessKey;
+ }
+ }
+}
diff --git a/spark/src/test/java/org/apache/comet/cloud/s3/NoNoArgCtorProvider.java b/spark/src/test/java/org/apache/comet/cloud/s3/NoNoArgCtorProvider.java
new file mode 100644
index 0000000000..7f3d7e2dbc
--- /dev/null
+++ b/spark/src/test/java/org/apache/comet/cloud/s3/NoNoArgCtorProvider.java
@@ -0,0 +1,32 @@
+/*
+ * 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.cloud.s3;
+
+/** Helper for the dispatcher's "missing public no-arg constructor" rejection path. */
+public class NoNoArgCtorProvider implements CometS3CredentialProvider {
+
+ @SuppressWarnings("unused")
+ public NoNoArgCtorProvider(String required) {}
+
+ @Override
+ public CometS3Credentials getCredentialsForPath(CometS3CredentialContext context) {
+ throw new UnsupportedOperationException("never instantiable");
+ }
+}
diff --git a/spark/src/test/java/org/apache/comet/cloud/s3/NotACredentialProvider.java b/spark/src/test/java/org/apache/comet/cloud/s3/NotACredentialProvider.java
new file mode 100644
index 0000000000..ebe292f677
--- /dev/null
+++ b/spark/src/test/java/org/apache/comet/cloud/s3/NotACredentialProvider.java
@@ -0,0 +1,25 @@
+/*
+ * 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.cloud.s3;
+
+/** Helper for the dispatcher's "does not implement interface" rejection path. */
+public class NotACredentialProvider {
+ public NotACredentialProvider() {}
+}
diff --git a/spark/src/test/java/org/apache/comet/cloud/s3/TestCometS3CredentialProvider.java b/spark/src/test/java/org/apache/comet/cloud/s3/TestCometS3CredentialProvider.java
new file mode 100644
index 0000000000..77ce86c836
--- /dev/null
+++ b/spark/src/test/java/org/apache/comet/cloud/s3/TestCometS3CredentialProvider.java
@@ -0,0 +1,88 @@
+/*
+ * 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.cloud.s3;
+
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * Test-only provider instantiated by the dispatcher when its FQCN is passed across JNI. State is
+ * static because the dispatcher caches one instance per (FQCN, dispatchKey) for the JVM lifetime.
+ */
+public class TestCometS3CredentialProvider implements CometS3CredentialProvider {
+
+ static final AtomicInteger initCount = new AtomicInteger(0);
+ static final AtomicInteger callCount = new AtomicInteger(0);
+ static final AtomicInteger closeCount = new AtomicInteger(0);
+ static volatile String lastBucket;
+ static volatile String lastPath;
+ static volatile CometS3AccessMode lastMode;
+ static volatile String lastTenantSeen;
+ static volatile RuntimeException throwOnNext;
+ static volatile Exception throwOnClose;
+ static volatile CometS3Credentials nextResult =
+ new CometS3Credentials("AKIATEST", "secret", "session-tok", 0L);
+
+ private volatile String tenantId;
+
+ static void reset() {
+ initCount.set(0);
+ callCount.set(0);
+ closeCount.set(0);
+ lastBucket = null;
+ lastPath = null;
+ lastMode = null;
+ lastTenantSeen = null;
+ throwOnNext = null;
+ throwOnClose = null;
+ nextResult = new CometS3Credentials("AKIATEST", "secret", "session-tok", 0L);
+ }
+
+ @Override
+ public void initialize(Map<String, String> catalogProperties) {
+ initCount.incrementAndGet();
+ this.tenantId = catalogProperties.get("tenant-id");
+ }
+
+ @Override
+ public CometS3Credentials getCredentialsForPath(CometS3CredentialContext context) {
+ callCount.incrementAndGet();
+ lastBucket = context.getBucket();
+ lastPath = context.getPath();
+ lastMode = context.getMode();
+ lastTenantSeen = tenantId;
+ RuntimeException toThrow = throwOnNext;
+ if (toThrow != null) {
+ throwOnNext = null;
+ throw toThrow;
+ }
+ return nextResult;
+ }
+
+ @Override
+ public void close() throws Exception {
+ closeCount.incrementAndGet();
+ Exception toThrow = throwOnClose;
+ if (toThrow != null) {
+ throwOnClose = null;
+ throw toThrow;
+ }
+ }
+}
diff --git a/spark/src/test/scala/org/apache/comet/CometS3TestBase.scala b/spark/src/test/scala/org/apache/comet/CometS3TestBase.scala
index a8e1a58eac..467a360483 100644
--- a/spark/src/test/scala/org/apache/comet/CometS3TestBase.scala
+++ b/spark/src/test/scala/org/apache/comet/CometS3TestBase.scala
@@ -66,6 +66,26 @@ trait CometS3TestBase extends CometTestBase {
conf.set("spark.hadoop.fs.s3a.path.style.access", "true")
}
+ protected def icebergAvailable: Boolean = {
+ try {
+ Class.forName("org.apache.iceberg.catalog.Catalog")
+ true
+ } catch {
+ case _: ClassNotFoundException => false
+ }
+ }
+
+ /**
+ * Apply the S3 properties Comet's native Iceberg reader requires on the given catalog.
+ * iceberg-rust / opendal disables region auto-detection when a custom credential loader is
+ * wired in, so the region/endpoint/path-style triple has to be set explicitly.
+ */
+ protected def applyS3CatalogProps(conf: SparkConf, catalogName: String): Unit = {
+ conf.set(s"spark.sql.catalog.$catalogName.s3.endpoint", minioContainer.getS3URL)
+ conf.set(s"spark.sql.catalog.$catalogName.s3.region", "us-east-1")
+ conf.set(s"spark.sql.catalog.$catalogName.s3.path-style-access", "true")
+ }
+
protected def createBucketIfNotExists(bucketName: String): Unit = {
val credentials = AwsBasicCredentials.create(userName, password)
val s3Client = S3Client
diff --git a/spark/src/test/scala/org/apache/comet/IcebergReadFromS3Suite.scala b/spark/src/test/scala/org/apache/comet/IcebergReadFromS3Suite.scala
index c1c90adfa3..4b15f1c978 100644
--- a/spark/src/test/scala/org/apache/comet/IcebergReadFromS3Suite.scala
+++ b/spark/src/test/scala/org/apache/comet/IcebergReadFromS3Suite.scala
@@ -29,21 +29,13 @@ class IcebergReadFromS3Suite extends CometS3TestBase with RESTCatalogHelper {
override protected val testBucketName = "test-iceberg-bucket"
- private def icebergAvailable: Boolean = {
- try {
- Class.forName("org.apache.iceberg.catalog.Catalog")
- true
- } catch {
- case _: ClassNotFoundException => false
- }
- }
-
override protected def sparkConf: SparkConf = {
val conf = super.sparkConf
conf.set("spark.sql.catalog.s3_catalog", "org.apache.iceberg.spark.SparkCatalog")
conf.set("spark.sql.catalog.s3_catalog.type", "hadoop")
conf.set("spark.sql.catalog.s3_catalog.warehouse", s"s3a://$testBucketName/warehouse")
+ applyS3CatalogProps(conf, "s3_catalog")
conf.set(CometConf.COMET_ENABLED.key, "true")
conf.set(CometConf.COMET_EXEC_ENABLED.key, "true")
@@ -237,6 +229,7 @@ class IcebergReadFromS3Suite extends CometS3TestBase with RESTCatalogHelper {
"s3.access-key-id" -> "WRONG_ACCESS_KEY",
"s3.secret-access-key" -> "WRONG_SECRET_KEY",
"s3.endpoint" -> minioContainer.getS3URL,
+ "s3.region" -> "us-east-1",
"s3.path-style-access" -> "true")
val warehouse = s"s3a://$testBucketName/warehouse-bad-creds"
@@ -269,6 +262,7 @@ class IcebergReadFromS3Suite extends CometS3TestBase with RESTCatalogHelper {
"s3.access-key-id" -> userName,
"s3.secret-access-key" -> password,
"s3.endpoint" -> minioContainer.getS3URL,
+ "s3.region" -> "us-east-1",
"s3.path-style-access" -> "true")
val warehouse = s"s3a://$testBucketName/warehouse-vending"
diff --git a/spark/src/test/scala/org/apache/comet/cloud/s3/CometS3CredentialBridgeSuite.scala b/spark/src/test/scala/org/apache/comet/cloud/s3/CometS3CredentialBridgeSuite.scala
new file mode 100644
index 0000000000..172d339901
--- /dev/null
+++ b/spark/src/test/scala/org/apache/comet/cloud/s3/CometS3CredentialBridgeSuite.scala
@@ -0,0 +1,237 @@
+/*
+ * 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.cloud.s3
+
+import org.apache.spark.SparkConf
+import org.apache.spark.sql.SaveMode
+import org.apache.spark.sql.comet.{CometIcebergNativeScanExec, CometNativeScanExec, CometScanExec}
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.functions.{col, sum}
+
+import org.apache.comet.{CometConf, CometS3TestBase}
+import org.apache.comet.iceberg.RESTCatalogHelper
+
+/**
+ * End-to-end test that exercises [[CometS3CredentialDispatcher]] and the Rust JNI bridge against
+ * a real S3 server (Minio). Asserts the test SPI was actually invoked rather than the default AWS
+ * credential chain.
+ */
+class CometS3CredentialBridgeSuite
+ extends CometS3TestBase
+ with AdaptiveSparkPlanHelper
+ with RESTCatalogHelper {
+
+ override protected val testBucketName = "bridge-test-bucket"
+
+ override protected def sparkConf: SparkConf = {
+ val conf = super.sparkConf
+ val providerClassName = classOf[MinioCometS3CredentialProvider].getName
+ // Activate the bridge for the Parquet (object_store) path via the Hadoop S3A namespace.
+ conf.set("spark.hadoop.fs.s3a.comet.credential.provider.class", providerClassName)
+ // Activate the bridge for the Iceberg (opendal) path via the per-catalog s3 namespace.
+ conf.set("spark.sql.catalog.s3_catalog", "org.apache.iceberg.spark.SparkCatalog")
+ conf.set("spark.sql.catalog.s3_catalog.type", "hadoop")
+ conf.set("spark.sql.catalog.s3_catalog.warehouse", s"s3a://$testBucketName/warehouse")
+ conf.set("spark.sql.catalog.s3_catalog.s3.comet.credential.provider.class", providerClassName)
+ applyS3CatalogProps(conf, "s3_catalog")
+ conf.set(CometConf.COMET_ICEBERG_NATIVE_ENABLED.key, "true")
+ conf
+ }
+
+ override def beforeAll(): Unit = {
+ super.beforeAll()
+ MinioCometS3CredentialProvider.installCredentials(userName, password)
+ }
+
+ private def assertHasCometParquetScan(plan: SparkPlan): Unit = {
+ val scans = collect(plan) {
+ case p: CometScanExec => p
+ case p: CometNativeScanExec => p
+ }
+ assert(scans.nonEmpty, s"Expected at least one Comet Parquet scan in plan:\n$plan")
+ }
+
+ private def assertHasCometIcebergScan(plan: SparkPlan): Unit = {
+ val scans = collect(plan) { case p: CometIcebergNativeScanExec => p }
+ assert(scans.nonEmpty, s"Expected at least one CometIcebergNativeScanExec in plan:\n$plan")
+ }
+
+ test("Parquet read on S3 routes credentials through CometS3CredentialProvider") {
+ val testFilePath = s"s3a://$testBucketName/data/bridge-parquet.parquet"
+ val rowCount = 1000L
+ spark.range(0, rowCount).write.format("parquet").mode(SaveMode.Overwrite).save(testFilePath)
+ val expectedSum = (0L until rowCount).sum
+
+ MinioCometS3CredentialProvider.resetCounters()
+ val df = spark.read.format("parquet").load(testFilePath).agg(sum(col("id")))
+ assertHasCometParquetScan(df.queryExecution.executedPlan)
+ assert(df.first().getLong(0) == expectedSum)
+
+ assert(
+ MinioCometS3CredentialProvider.callCount() > 0,
+ "Bridge was not invoked during Comet Parquet read")
+ assert(
+ MinioCometS3CredentialProvider.lastBucket() == testBucketName,
+ s"Bridge received unexpected bucket: ${MinioCometS3CredentialProvider.lastBucket()}")
+ }
+
+ test("Iceberg read on S3 routes credentials through CometS3CredentialProvider") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ spark.sql("""
+ CREATE TABLE s3_catalog.db.bridge_iceberg (
+ id INT,
+ name STRING
+ ) USING iceberg
+ """)
+
+ spark.sql("""
+ INSERT INTO s3_catalog.db.bridge_iceberg
+ VALUES (1, 'a'), (2, 'b'), (3, 'c')
+ """)
+
+ MinioCometS3CredentialProvider.resetCounters()
+ val df = spark.sql("SELECT * FROM s3_catalog.db.bridge_iceberg ORDER BY id")
+ assertHasCometIcebergScan(df.queryExecution.executedPlan)
+ assert(df.count() == 3)
+
+ assert(
+ MinioCometS3CredentialProvider.callCount() > 0,
+ "Bridge was not invoked during Comet Iceberg read")
+
+ spark.sql("DROP TABLE s3_catalog.db.bridge_iceberg")
+ }
+
+ test("REST catalog vended creds reach CometS3CredentialProvider.initialize via JNI") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ // Mix storage-prefixed creds (which the old filter would have allowed through) with a
+ // non-storage-prefixed sentinel key. The sentinel only reaches initialize(Map) if Comet
+ // forwards the unfiltered FileIO property bag end-to-end.
+ val vendedCreds = Map(
+ "s3.access-key-id" -> userName,
+ "s3.secret-access-key" -> password,
+ "s3.endpoint" -> minioContainer.getS3URL,
+ "s3.region" -> "us-east-1",
+ "s3.path-style-access" -> "true",
+ "comet.test-tenant-id" -> "tenant-A")
+ val warehouse = s"s3a://$testBucketName/warehouse-rest-spi"
+ val providerClassName = classOf[MinioCometS3CredentialProvider].getName
+
+ withRESTCatalog(vendedCredentials = vendedCreds, warehouseLocation = Some(warehouse)) {
+ (restUri, _, _) =>
+ withSQLConf(
+ "spark.sql.catalog.rest_spi" -> "org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.rest_spi.catalog-impl" -> "org.apache.iceberg.rest.RESTCatalog",
+ "spark.sql.catalog.rest_spi.uri" -> restUri,
+ "spark.sql.catalog.rest_spi.warehouse" -> warehouse,
+ "spark.sql.catalog.rest_spi.s3.comet.credential.provider.class" -> providerClassName) {
+
+ spark.sql("CREATE NAMESPACE rest_spi.db")
+ spark.sql("CREATE TABLE rest_spi.db.tbl (id INT) USING iceberg")
+ spark.sql("INSERT INTO rest_spi.db.tbl VALUES (1), (2), (3)")
+
+ MinioCometS3CredentialProvider.resetCounters()
+ val df = spark.sql("SELECT * FROM rest_spi.db.tbl ORDER BY id")
+ assertHasCometIcebergScan(df.queryExecution.executedPlan)
+ assert(df.count() == 3)
+
+ assert(
+ MinioCometS3CredentialProvider.callCount() > 0,
+ "Bridge was not invoked during REST catalog read")
+
+ val captured = MinioCometS3CredentialProvider.lastInitProperties()
+ assert(captured != null, "initialize(Map) was never called")
+ assert(
+ captured.containsKey("comet.test-tenant-id"),
+ s"Unfiltered FileIO key did not reach initialize(Map). Captured keys: ${captured
+ .keySet()}")
+ assert(captured.get("comet.test-tenant-id") == "tenant-A")
+
+ spark.sql("DROP TABLE rest_spi.db.tbl")
+ spark.sql("DROP NAMESPACE rest_spi.db")
+ }
+ }
+ }
+
+ test(
+ "two catalogs sharing one provider FQCN get isolated CometS3CredentialProvider instances") {
+ assume(icebergAvailable, "Iceberg not available in classpath")
+
+ val providerClassName = classOf[MinioCometS3CredentialProvider].getName
+ val s3Endpoint = minioContainer.getS3URL
+
+ withSQLConf(
+ "spark.sql.catalog.iso_a" -> "org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.iso_a.type" -> "hadoop",
+ "spark.sql.catalog.iso_a.warehouse" -> s"s3a://$testBucketName/iso-warehouse-a",
+ "spark.sql.catalog.iso_a.s3.comet.credential.provider.class" -> providerClassName,
+ "spark.sql.catalog.iso_a.s3.endpoint" -> s3Endpoint,
+ "spark.sql.catalog.iso_a.s3.region" -> "us-east-1",
+ "spark.sql.catalog.iso_a.s3.path-style-access" -> "true",
+ s"spark.sql.catalog.iso_a.${MinioCometS3CredentialProvider.TEST_INSTANCE_TAG}" -> "tag-A",
+ "spark.sql.catalog.iso_b" -> "org.apache.iceberg.spark.SparkCatalog",
+ "spark.sql.catalog.iso_b.type" -> "hadoop",
+ "spark.sql.catalog.iso_b.warehouse" -> s"s3a://$testBucketName/iso-warehouse-b",
+ "spark.sql.catalog.iso_b.s3.comet.credential.provider.class" -> providerClassName,
+ "spark.sql.catalog.iso_b.s3.endpoint" -> s3Endpoint,
+ "spark.sql.catalog.iso_b.s3.region" -> "us-east-1",
+ "spark.sql.catalog.iso_b.s3.path-style-access" -> "true",
+ s"spark.sql.catalog.iso_b.${MinioCometS3CredentialProvider.TEST_INSTANCE_TAG}" -> "tag-B") {
+
+ MinioCometS3CredentialProvider.resetCounters()
+
+ spark.sql("CREATE TABLE iso_a.db.t (id INT) USING iceberg")
+ spark.sql("INSERT INTO iso_a.db.t VALUES (1), (2)")
+ spark.sql("CREATE TABLE iso_b.db.t (id INT) USING iceberg")
+ spark.sql("INSERT INTO iso_b.db.t VALUES (10), (20), (30)")
+
+ val dfA = spark.sql("SELECT * FROM iso_a.db.t ORDER BY id")
+ assertHasCometIcebergScan(dfA.queryExecution.executedPlan)
+ assert(dfA.count() == 2)
+
+ val dfB = spark.sql("SELECT * FROM iso_b.db.t ORDER BY id")
+ assertHasCometIcebergScan(dfB.queryExecution.executedPlan)
+ assert(dfB.count() == 3)
+
+ // Two distinct dispatchKeys (catalog names "iso_a" and "iso_b") must yield two separate
+ // initialize(Map) calls even though the provider FQCN is shared.
+ assert(
+ MinioCometS3CredentialProvider.initCount() >= 2,
+ "Expected at least 2 initialize() calls across two catalogs, got " +
+ s"${MinioCometS3CredentialProvider.initCount()}")
+
+ val initA = MinioCometS3CredentialProvider.initPropertiesForTag("tag-A")
+ val initB = MinioCometS3CredentialProvider.initPropertiesForTag("tag-B")
+ assert(initA != null, "initialize(Map) for catalog iso_a never landed in INIT_BY_TAG")
+ assert(initB != null, "initialize(Map) for catalog iso_b never landed in INIT_BY_TAG")
+ assert(
+ initA.get(MinioCometS3CredentialProvider.TEST_INSTANCE_TAG) == "tag-A",
+ s"iso_a init map tagged wrong: ${initA.get(MinioCometS3CredentialProvider.TEST_INSTANCE_TAG)}")
+ assert(
+ initB.get(MinioCometS3CredentialProvider.TEST_INSTANCE_TAG) == "tag-B",
+ s"iso_b init map tagged wrong: ${initB.get(MinioCometS3CredentialProvider.TEST_INSTANCE_TAG)}")
+
+ spark.sql("DROP TABLE iso_a.db.t")
+ spark.sql("DROP TABLE iso_b.db.t")
+ }
+ }
+}
diff --git a/spark/src/test/spark-4.x/java/org/apache/comet/cloud/s3/IcebergRESTVendedS3Provider.java b/spark/src/test/spark-4.x/java/org/apache/comet/cloud/s3/IcebergRESTVendedS3Provider.java
new file mode 100644
index 0000000000..0d5cde70d0
--- /dev/null
+++ b/spark/src/test/spark-4.x/java/org/apache/comet/cloud/s3/IcebergRESTVendedS3Provider.java
@@ -0,0 +1,72 @@
+/*
+ * 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.cloud.s3;
+
+import java.util.Map;
+
+import org.apache.iceberg.aws.s3.VendedCredentialsProvider;
+
+import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
+
+/**
+ * Example implementation of {@link CometS3CredentialProvider} for Iceberg REST catalogs that vend
+ * AWS credentials. Wraps Iceberg's {@code VendedCredentialsProvider} so caching and
+ * refresh-near-expiry come from its {@code CachedSupplier}; Comet adds only the JNI shape and the
+ * one-shot {@code initialize} call.
+ *
+ * <p>Test scope only, to keep iceberg-aws and AWS SDK v2 off Comet's runtime classpath. Production
+ * users should copy this into their own jar.
+ *
+ * <p>Spark 4.x build only (Iceberg 1.9.0+ is required for the in-properties caching short-circuit
+ * in {@code VendedCredentialsProvider}; earlier pins issue an HTTP GET against {@code
+ * credentials.uri} on every refresh).
+ *
+ * <p>Activation: set {@code spark.sql.catalog.<cat>.s3.comet.credential.provider.class =
+ * org.apache.comet.cloud.s3.IcebergRESTVendedS3Provider}. Comet calls {@link #initialize} once per
+ * catalog with the unfiltered FileIO property bag, which carries {@code credentials.uri} and
+ * {@code uri} as required by {@code VendedCredentialsProvider.create}.
+ */
+public final class IcebergRESTVendedS3Provider implements CometS3CredentialProvider {
+
+ private volatile VendedCredentialsProvider provider;
+
+ @Override
+ public void initialize(Map<String, String> catalogProperties) {
+ this.provider = VendedCredentialsProvider.create(catalogProperties);
+ }
+
+ @Override
+ public CometS3Credentials getCredentialsForPath(CometS3CredentialContext context) {
+ VendedCredentialsProvider p = provider;
+ if (p == null) {
+ throw new IllegalStateException(
+ "IcebergRESTVendedS3Provider used before initialize(Map) was called; "
+ + "Comet should always invoke initialize before getCredentialsForPath");
+ }
+ AwsCredentials c = p.resolveCredentials();
+ String sessionToken =
+ (c instanceof AwsSessionCredentials) ? ((AwsSessionCredentials) c).sessionToken() : null;
+ // Expiration is owned by VendedCredentialsProvider's CachedSupplier; we publish 0 so the
+ // native bridge applies its conservative floor to `opendal`'s cache while the inner
+ // CachedSupplier handles refresh on its own schedule.
+ return new CometS3Credentials(c.accessKeyId(), c.secretAccessKey(), sessionToken, 0L);
+ }
+}
diff --git a/spark/src/test/spark-4.x/java/org/apache/comet/cloud/s3/IcebergRESTVendedS3ProviderTest.java b/spark/src/test/spark-4.x/java/org/apache/comet/cloud/s3/IcebergRESTVendedS3ProviderTest.java
new file mode 100644
index 0000000000..b49ca74012
--- /dev/null
+++ b/spark/src/test/spark-4.x/java/org/apache/comet/cloud/s3/IcebergRESTVendedS3ProviderTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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.cloud.s3;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertThrows;
+
+/**
+ * Unit tests for {@link IcebergRESTVendedS3Provider}. Feeds the wrapped {@code
+ * VendedCredentialsProvider} a fully-populated {@code s3.*} bag so its {@code CachedSupplier}
+ * short-circuits without calling the REST {@code credentials.uri} endpoint. Refresh-via-REST
+ * coverage lives upstream in Iceberg.
+ */
+public class IcebergRESTVendedS3ProviderTest {
+
+ private static Map<String, String> staticVendedProps() {
+ Map<String, String> props = new HashMap<>();
+ // Required by VendedCredentialsProvider.create; not actually contacted in this test.
+ props.put("uri", "http://localhost:0/catalog");
+ props.put("credentials.uri", "http://localhost:0/credentials");
+ // Required four keys for the in-properties short-circuit path.
+ props.put("s3.access-key-id", "AKIA_TEST");
+ props.put("s3.secret-access-key", "secret_TEST");
+ props.put("s3.session-token", "token_TEST");
+ props.put(
+ "s3.session-token-expires-at-ms", Long.toString(System.currentTimeMillis() + 3_600_000L));
+ return props;
+ }
+
+ @Test
+ public void initializeThenGetReturnsVendedCredentials() {
+ IcebergRESTVendedS3Provider p = new IcebergRESTVendedS3Provider();
+ p.initialize(staticVendedProps());
+
+ CometS3Credentials c =
+ p.getCredentialsForPath(
+ new CometS3CredentialContext("bucket", "/k", CometS3AccessMode.READ));
+
+ assertEquals("AKIA_TEST", c.getAccessKeyId());
+ assertEquals("secret_TEST", c.getSecretAccessKey());
+ assertEquals("token_TEST", c.getSessionToken());
+ // Wrapper publishes 0 since VendedCredentialsProvider's CachedSupplier owns the expiry.
+ assertEquals(0L, c.getExpirationEpochMillis());
+ }
+
+ @Test
+ public void getBeforeInitializeThrows() {
+ IcebergRESTVendedS3Provider p = new IcebergRESTVendedS3Provider();
+ assertThrows(
+ IllegalStateException.class,
+ () ->
+ p.getCredentialsForPath(
+ new CometS3CredentialContext("bucket", "/k", CometS3AccessMode.READ)));
+ }
+
+ @Test
+ public void multipleCallsServedByCache() {
+ IcebergRESTVendedS3Provider p = new IcebergRESTVendedS3Provider();
+ p.initialize(staticVendedProps());
+
+ CometS3Credentials a =
+ p.getCredentialsForPath(new CometS3CredentialContext("b", "/k", CometS3AccessMode.READ));
+ CometS3Credentials b =
+ p.getCredentialsForPath(new CometS3CredentialContext("b", "/k2", CometS3AccessMode.READ));
+ CometS3Credentials c =
+ p.getCredentialsForPath(new CometS3CredentialContext("b", "/k3", CometS3AccessMode.READ));
+
+ // CachedSupplier hands out the same identity-equal credential until expiry; we assert by
+ // value so the test does not depend on AWS SDK internal caching semantics.
+ assertEquals(a.getAccessKeyId(), b.getAccessKeyId());
+ assertEquals(a.getAccessKeyId(), c.getAccessKeyId());
+ assertEquals(a.getSecretAccessKey(), b.getSecretAccessKey());
+ assertEquals(a.getSessionToken(), b.getSessionToken());
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org