snmvaughan opened a new pull request, #6031:
URL: https://github.com/apache/datafusion-comet/pull/6031

   ## Which issue does this PR close?
   
   Closes / refs: internal tracker `rdar://187805096` — "Comet native 
`object_store` cache serves stale scoped STS session across bucket prefixes → 
403".
   
   <!-- The bug does not have a public GitHub issue yet. If reviewers want one 
open, please advise. -->
   
   ## Rationale for this change
   
   Vendors that mint scoped STS sessions per S3 prefix 
(`s3://bucket/prefix-A/**` covered by one credential, `s3://bucket/prefix-B/**` 
by another) currently 403 on the second prefix under Comet's Parquet native 
scan.
   
   Root cause is entirely inside Comet native:
   
   * `parquet_support::prepare_object_store_with_configs` caches
     `Arc<dyn ObjectStore>` per `(scheme://bucket, config_hash, hdfs_backend)`
     — one entry per bucket.
   * `CometS3CredentialProvider::getCredentialsForPath` is fired only at
     construction time, so the very first path read on a bucket binds a
     particular scoped STS session into the cached store forever.
   * Any subsequent read on the same bucket that lies outside that session's
     scope reuses the store, sends the wrong credential to S3, and 403s.
   
   The fix is two-pronged and stays inside Comet (no vendor code changes 
required to preserve today's behavior):
   
   1. **Advisory scope hint on the read path** via a new opt-in sub-interface
      `CometS3ScopedCredentialProvider` that lets a vendor return the prefix
      list a credential is valid for. The native cache keys entries by
      `(bucket, config_hash, backend)` + `Vec<ScopeEntry>` so a single bucket
      can carry disjoint scoped stores side-by-side, and the read path picks
      the entry whose covering prefix is the *longest* match for the request
      (ties break by insertion order; catchall matches with effective length
      0 as the fallback of last resort).
   
   2. **Correctness safety net** via a thin `RetryOn403ObjectStore` decorator
      that transparently rebuilds the store on the first 403 and retries the
      operation once. A second 403 propagates. The rebuild closure receives
      the `Path` of the failing request, constructs a fresh bridge bound to
      that path, re-fires the SPI (so `getPolicyLocationsFor` reports the
      vendor's scope for the *actual* failing request rather than whatever
      was baked in at initial construction), and appends the new
      `ScopeEntry` to the cache alongside any pre-existing entries. Nested
      scoped stores on the same bucket then coexist and route deterministically
      under longest-prefix match. This handles both vendor-side overreporting
      (a one-round-trip performance cost) and the legitimate case where a
      scope was accurate at plan time but expired mid-scan.
   
   **Why longest-prefix matching is required.** The `append-on-rebuild`
   semantics above are what make scopes overlap in the first place: after a
   403-driven rebuild, the cache holds both the pre-existing (possibly
   broader) `ScopeEntry` and the newly-appended narrower one on the same
   bucket. A subsequent request whose path falls under the narrower entry's
   prefix is also covered by the broader entry's prefix — so a naive
   first-match or insertion-order lookup would keep routing that request back
   to the older, wrong-scoped store and re-trigger the same 403 on every
   read. Longest-prefix match resolves the overlap by always picking the
   narrowest applicable entry; the catchall (empty prefix list, effective
   length 0) stays available strictly as the fallback of last resort for
   paths outside every hinted scope. Insertion order is used only as a
   tiebreak between entries that cover the request path at exactly the same
   prefix length.
   
   Base-interface providers (not implementing the `Scoped` sub-interface) see 
zero behavior change: the dispatcher returns an empty prefix list, the native 
side treats that as "catchall", the cache degrades to single-entry-per-bucket, 
and the 403-retry wrapper stays inactive.
   
   ## What changes are included in this PR?
   
   Three logical commits:
   
   **(1) `feat: add CometS3ScopedCredentialProvider opt-in scope-hint 
sub-interface`**
   
   * `spark/.../CometS3ScopedCredentialProvider.java` — new `@Public` 
sub-interface with a single `getPolicyLocationsFor(CometS3CredentialContext)` 
method.
   * `spark/.../CometS3CredentialDispatcher.java` — new 
`getPolicyLocationsFor(long, String, String, int)` static entry that resolves 
by handle, does an `instanceof` check, dispatches, and normalizes null → empty 
list.
   * `spark/.../CometS3ScopedCredentialProviderTest.java` — JUnit unit tests 
covering base vs scoped dispatch, null normalization, exception propagation, 
handle validation.
   * `spark/.../MinioCometS3CredentialProvider.java` — implements the 
sub-interface with an installable prefix list, so scope-aware IT scenarios can 
drive the same test provider.
   * `spark/.../TestCometS3ScopedCredentialProvider.java` — deterministic 
in-process test fixture (no Minio) used by the JUnit tests above.
   * `CometPublicApiSuite` — pins `CometS3ScopedCredentialProvider` to the 
`@Public` set.
   
   **(2) `feat: add scope-hint JNI bridge + RetryOn403ObjectStore correctness 
net`**
   
   * `native/jni-bridge/src/comet_s3_credential_dispatcher.rs` — new 
`method_get_policy_locations_for` static-method ID and return-type descriptor 
matching `(JLjava/lang/String;Ljava/lang/String;I)Ljava/util/List;`.
   * `native/core/src/cloud/s3/credential_bridge.rs` — new 
`fetch_policy_locations(bucket, path, mode)` that reads a 
`java.util.List<String>` back from the dispatcher and normalizes null / 
base-interface / empty results all to an empty vec. Module doc rewired for 
scope-hint semantics.
   * `native/core/src/parquet/objectstore/retry.rs` — new 
`RetryOn403ObjectStore` wrapper (Send + Sync + `ObjectStore`). Retry scope is 
deliberately narrow: `put_opts`, `get_opts`, `get_ranges`, 
`list_with_delimiter`, `copy_opts`, `rename_opts`. Streams and multipart put 
pass through. Only `Error::PermissionDenied` triggers retry; `Unauthenticated` 
(401) is treated as permanent. The `RebuildFn` closure takes `Option<&Path>` so 
the wrapper can thread the failing request's location into rebuild — retry 
sites pass the operation's read-side location (source for copy/rename). 9 unit 
tests cover pass-through, rebuild-then-retry, second-403 propagation, rebuild 
idempotency across ops, rebuild-error surface, 401-not-retried, per-path 
threading, `Send + Sync` bound, and composition with `Arc<Mutex<...>>`.
   
   **(3) `feat: scope-aware object_store cache with rebuild-on-403 wiring`**
   
   * `native/core/src/parquet/objectstore/s3.rs` — factored builder into 
`try_construct_bridge(url, configs)` + `create_store_with_bridge(url, configs, 
bridge, min_ttl)`; deleted the monolithic `create_store`. Added 
`try_construct_bridge_with_path(url, configs, override_path)` sibling so the 
rebuild path can rebind the bridge's baked-in path to the 403'd location before 
re-firing the SPI.
   * `native/core/src/parquet/parquet_support.rs` — cache value changed from 
`Arc<dyn ObjectStore>` to `Vec<ScopeEntry>`. New helpers: `path_covered`, 
`longest_covering_prefix_len`, `find_matching_scope` (longest-prefix match with 
insertion-order tie-break; catchall matches with effective length 0 as fallback 
of last resort). Rewrote `prepare_object_store_with_configs` to fetch scope 
hints per miss, wrap the store in `RetryOn403ObjectStore` when a bridge is 
present, and *append* a new `ScopeEntry` on rebuild alongside any pre-existing 
entries — the closure re-fires the SPI with the failing path so the vendor's 
returned prefixes reflect the actual failing request, and disjoint scoped 
stores on the same bucket coexist across 403 recoveries. 4 new 
`find_matching_scope` tests + 1 helper test + 3 migrated seed tests.
   * `docs/source/user-guide/latest/s3-credential-providers.md` — new "Scope 
hints via `CometS3ScopedCredentialProvider`" section for operators/vendors.
   * `docs/source/contributor-guide/s3-credential-provider-design.md` — new 
"Scope-aware `object_store` registry" section for maintainers.
   * `spark/.../CometS3CredentialBridgeSuite.scala` — two new Docker-tagged 
(Minio) scenarios: same-scope-share and disjoint-scope-fork. The full 
overreport-then-403-recover path is proven by the Rust unit tests since Minio 
does not enforce per-prefix denial in the test base.
   
   **Per-path rebuild threading.** The rebuild closure receives the `Path` of 
the request that triggered the 403 and rebinds the fresh bridge to that path 
via `try_construct_bridge_with_path`, so `getPolicyLocationsFor` is 
re-consulted against the *actual* failing request context. The returned 
prefixes become the new `ScopeEntry`'s scope, and longest-prefix lookup on 
subsequent requests routes each to the narrowest applicable session — 
preserving scope granularity across 403 recoveries without invalidating the 
pre-existing entry the vendor may still legitimately serve.
   
   ## Are these changes tested?
   
   **Rust — unit**
   
   * `parquet::objectstore::retry` — 8 tests. All green.
   * `parquet::objectstore::s3` — pre-existing 26 tests + updated 
`test_create_store`. All green.
   * `parquet::parquet_support::tests` — 13 pre-existing + 3 new scope-helper 
tests + 3 updated seed tests. All 16 green.
   * `parquet::objectstore::*` module (all sub-modules) — 60 tests. All green.
   
   Command: `JAVA_HOME=... DYLD_LIBRARY_PATH=$JAVA_HOME/lib/server cargo test 
--offline -p datafusion-comet --lib parquet::objectstore 
parquet::parquet_support`.
   
   **Java — unit**
   
   * `CometS3ScopedCredentialProviderTest` — JUnit tests exercising the 
dispatcher's `getPolicyLocationsFor` entry against 
`TestCometS3ScopedCredentialProvider` and a base-interface-only stub.
   
   **Scala — Docker-tagged IT**
   
   * `CometS3CredentialBridgeSuite`:
       - `scoped provider: two reads inside the same scope share one 
object_store entry`
       - `scoped provider: reads under disjoint scopes get separate 
object_store entries`
   
   These require Minio. Run locally with 
`-Dsuites=org.apache.comet.cloud.s3.CometS3CredentialBridgeSuite` under the 
`testcontainers-docker` profile.
   
   ## Are there any user-facing changes?
   
   Yes — additive only. Vendors gain an opt-in 
`CometS3ScopedCredentialProvider` sub-interface. See 
`docs/source/user-guide/latest/s3-credential-providers.md#scope-hints-via-comets3scopedcredentialprovider`
 for the operator/vendor contract.
   
   Existing `CometS3CredentialProvider` implementations require no changes and 
see identical runtime behavior — cache keys, provider lifecycle, dispatch 
mechanics all preserved for base-interface providers.
   
   ## Non-goals
   
   * No change to Iceberg's `opendal`/`reqsign-core` credential path. The 
`object_store` cache defect this PR fixes is Parquet-native only; the Iceberg 
path uses `expires_in` refresh and does not exhibit the same failure mode.
   * No new tuning knob for the 403-retry wrapper. Retry-once is hardcoded — 
the alternative (bounded-retry with configurable count) would risk masking real 
permission failures.
   * No cross-bucket sharing of scope hints. Each `(scheme://bucket, 
config_hash, backend)` key maintains its own `Vec<ScopeEntry>` independently.
   
   ## Discovery discrepancy vs. original design notes
   
   The pre-existing `ObjectStoreCacheKey` is `(String, u64, bool)` — the third 
element is `hdfs_backend`, not a 2-tuple as an earlier draft plan assumed. The 
`Vec<ScopeEntry>` shape sits on the value side, so the outer key is unchanged. 
All three existing seed-based tests were migrated in this PR.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: [email protected]

For queries about this service, please contact Infrastructure at:
[email protected]


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to