andygrove commented on code in PR #6031:
URL: https://github.com/apache/datafusion-comet/pull/6031#discussion_r4053635741
##########
native/core/src/cloud/s3/credential_bridge.rs:
##########
@@ -358,3 +390,66 @@ fn read_optional_string(
.map(Some)
.map_err(|e| ExecutionError::GeneralError(format!("try_to_string:
{e}")))
}
+
+/// Decode a `java.util.List<String>` into `Vec<String>`, treating null list,
null elements, and
+/// non-string elements as end-of-list / skip respectively. Only called from
+/// [`CometS3CredentialBridge::fetch_policy_locations`] where the JVM
dispatcher normalizes null
+/// returns to `Collections.emptyList()`, but the extra null-check keeps the
helper self-contained.
+fn read_java_string_list(
Review Comment:
Every other JNI call in this file goes through `jni_static_call!`, which
runs `check_exception` for you. This helper calls `call_method_unchecked`
directly, so if a vendor hands back a lazily evaluated `List` that throws from
`size()` or `get(i)`, the exception stays pending and the next JNI call on that
thread misbehaves. Could it use the checked path instead?
Separately, line 448 casts each element straight through `JString::from_raw`
with no type check, while the doc comment above says non-string elements are
skipped. Worth making the code match the comment, since a vendor returning a
`List<Object>` would currently be undefined behavior rather than a skipped
element.
##########
spark/src/test/scala/org/apache/comet/CometPublicApiSuite.scala:
##########
@@ -44,7 +44,8 @@ class CometPublicApiSuite extends AnyFunSuite {
"org.apache.comet.cloud.s3.CometS3AccessMode",
"org.apache.comet.cloud.s3.CometS3CredentialContext",
"org.apache.comet.cloud.s3.CometS3CredentialProvider",
- "org.apache.comet.cloud.s3.CometS3Credentials")
+ "org.apache.comet.cloud.s3.CometS3Credentials",
+ "org.apache.comet.cloud.s3.CometS3ScopedCredentialProvider")
Review Comment:
`docs/source/about/versioning_policy.md` enumerates the public API as well,
and it still lists only the four pre-existing types. The comment at the top of
this suite asks for the doc and this list to change together, so could
`CometS3ScopedCredentialProvider` be added there too? It would also help to say
what adding a method to the sub-interface later would mean for compatibility,
since the existing text covers that question for `CometS3CredentialProvider`.
##########
spark/src/test/scala/org/apache/comet/cloud/s3/CometS3CredentialBridgeSuite.scala:
##########
@@ -229,4 +229,86 @@ class CometS3CredentialBridgeSuite
spark.sql("DROP TABLE iso_b.db.t")
}
}
+
+ // ---------------------------------------------------------------------
+ // Scope-aware object_store cache scenarios
+ //
+ // The Rust unit tests in native/core/src/parquet/objectstore/retry.rs cover
the retry-on-403
+ // wrapper end-to-end (pass-through, rebuild-then-retry, second-403
propagation, idempotency,
+ // error surface, 401-not-retried, Send+Sync, Arc<Mutex> composition). Minio
does not enforce
+ // per-prefix denial out of the box, so the full
"overreport-then-403-recover" path is proven
+ // in Rust rather than here. The two IT scenarios below exercise the
Java/JNI + native-cache
+ // plumbing: the scope hint reaches Rust and the cache preserves scope
granularity.
+ // ---------------------------------------------------------------------
+
+ test("scoped provider: two reads inside the same scope share one
object_store entry") {
Review Comment:
Beyond the URI-shaped hint and the callback assertion sunchao flagged,
there's a third thing working against this test. The native `object_store`
cache is a process-wide static and `MinioCometS3CredentialProvider` defaults to
an empty hint, so the earlier tests in this suite have already seeded a
catchall entry under the same bucket key. `find_matching_scope` matches that at
effective length 0, so the scoped entry never gets exercised even once the
prefix format is fixed.
Would using a bucket the other tests don't touch be enough here, or is there
a way to reset the native cache between tests?
##########
docs/source/user-guide/latest/s3-credential-providers.md:
##########
@@ -201,6 +201,33 @@ public CometS3Credentials
getCredentialsForPath(CometS3CredentialContext ctx) th
}
```
+### Scope hints via `CometS3ScopedCredentialProvider`
+
+The base `CometS3CredentialProvider` gives Comet one credential per (bucket,
path) call. Vendors whose credentials are inherently *scoped* — a single STS
session covers `s3://bucket/prefix-A/**` but not `s3://bucket/prefix-B/**` —
can implement the opt-in sub-interface `CometS3ScopedCredentialProvider` to let
Comet keep multiple scoped stores side-by-side in a single bucket instead of
caching a single session and 403-ing when it is asked to serve a path outside
its scope.
+
+```java
+package org.apache.comet.cloud.s3;
+
+public interface CometS3ScopedCredentialProvider extends
CometS3CredentialProvider {
+ /**
+ * Advisory list of path prefixes (or absolute s3://... URIs) the
credential
+ * returned by {@link #getCredentialsForPath(CometS3CredentialContext)} for
+ * the same {@code context} is valid against. An empty list means
"unknown /
+ * catchall"; Comet then falls back to the pre-existing
single-entry-per-bucket
+ * behavior.
+ */
+ java.util.List<String> getPolicyLocationsFor(CometS3CredentialContext
context);
+}
+```
+
+Comet uses the hint to key the native `object_store` registry as `(bucket,
config_hash, backend, scope_prefixes)` instead of the base three-tuple, so two
prefixes with disjoint scopes each get their own store. **The hint is
advisory.** S3 itself is authoritative: if the vendor overreports a scope and
S3 returns 403 anyway, the native cache transparently rebuilds the store under
a widened (catchall) scope, retries the operation once, and continues. A second
403 propagates to Spark as a real error.
Review Comment:
Two things on this paragraph. The registry isn't keyed as `(bucket,
config_hash, backend, scope_prefixes)` -- the design doc has a whole section
explaining why that shape was rejected, and the code keeps the three-tuple with
the prefixes hanging off the entry. And the rebuild doesn't widen to a
catchall; it re-fires the SPI and uses whatever prefixes come back.
Together with the point sunchao raised about base-interface providers being
wrapped despite this page saying their retry path stays inactive, that's three
claims on this page that need reconciling with the final behavior. The backward
compatibility promise rests on the third one, so I'd like that one to be
accurate in particular.
##########
native/core/src/parquet/parquet_support.rs:
##########
@@ -832,18 +939,96 @@ pub(crate) fn prepare_object_store_with_configs(
let (store, path): (Box<dyn ObjectStore>, Path) = if
is_hdfs_scheme {
create_hdfs_object_store(&url)
} else if scheme == "s3" {
- objectstore::s3::create_store(&url, object_store_configs,
Duration::from_secs(300))
+ objectstore::s3::create_store_with_bridge(
+ &url,
+ object_store_configs,
+ bridge_opt.clone(),
+ Duration::from_secs(300),
+ )
} else if is_azure_scheme(scheme) {
objectstore::azure::create_store(&url, object_store_configs)
} else {
parse_url(&url)
}
.map_err(|e| ExecutionError::GeneralError(e.to_string()))?;
- let store: Arc<dyn ObjectStore> = Arc::from(store);
- // Insert into cache
+ let raw_store: Arc<dyn ObjectStore> = Arc::from(store);
+
+ // Wrap SPI-backed stores in the 403-retry safety net. Non-SPI
paths keep their
+ // plain store so we don't add overhead for the base credential
chain (which is
+ // already correct without the wrapper).
+ let store: Arc<dyn ObjectStore> = if bridge_opt.is_some() {
+ let cache_key_for_rebuild = cache_key.clone();
+ let url_for_rebuild = url.clone();
+ let configs_for_rebuild: HashMap<String, String> =
object_store_configs.clone();
+ let rebuild: RebuildFn = Arc::new(move |failing_path:
Option<&Path>| {
+ // Fresh bridge → fresh SPI dispatch → fresh credentials
scoped for the
+ // actual failing request. We rebind the bridge's baked-in
path to the
+ // 403'd location (falling back to the URL's path when the
retry site did
+ // not supply one) so `fetch_policy_locations` returns the
vendor's scope
+ // for *this* request rather than whatever the initial
construction picked.
+ let failing_path_str = failing_path.map(|p|
format!("/{p}"));
+ let rebuilt_bridge =
objectstore::s3::try_construct_bridge_with_path(
+ &url_for_rebuild,
+ &configs_for_rebuild,
+ failing_path_str.as_deref(),
+ )
+ .map_err(|e| object_store::Error::Generic {
+ store: "S3",
+ source: format!("rebuild bridge failed: {e}").into(),
+ })?;
+ // Query the vendor for the fresh session's scope hint.
Errors normalize
+ // to an empty prefix list (catchall for the new entry
only) — matching
+ // the fallback the initial builder uses on the same call.
+ let fresh_prefixes = if let Some(bridge) =
rebuilt_bridge.as_ref() {
+ bridge.fetch_policy_locations().unwrap_or_else(|e| {
+ debug!("fetch_policy_locations on rebuild failed:
{e}");
+ Vec::new()
+ })
+ } else {
+ Vec::new()
+ };
+ let (rebuilt_raw, _path) =
objectstore::s3::create_store_with_bridge(
+ &url_for_rebuild,
+ &configs_for_rebuild,
+ rebuilt_bridge,
+ Duration::from_secs(300),
+ )?;
+ let rebuilt: Arc<dyn ObjectStore> = Arc::from(rebuilt_raw);
+ // Append the new scope entry alongside any pre-existing
ones. The old
+ // entry stays: the vendor may still legitimately serve
its original scope
+ // even though the failing path fell outside it.
Longest-prefix match on
+ // the lookup side ensures each request routes to the
narrowest applicable
+ // session.
+ if let Ok(mut cache) = object_store_cache().write() {
+ let entries =
cache.entry(cache_key_for_rebuild.clone()).or_default();
+ // Deduplicate: a concurrent rebuild may have already
inserted an
+ // entry with the same prefixes.
+ if !entries.iter().any(|e| e.prefixes ==
fresh_prefixes) {
Review Comment:
This dedup check means the append that the whole recovery story depends on
usually doesn't happen. In the overreporting case the vendor returns the same
over-broad prefix on the rebuild as it did on the initial fetch, so
`fresh_prefixes` equals the existing entry's prefixes, the check fires, and
nothing gets pushed. The wrapper is left silently rebound to the new store with
no entry recording it, which is the state sunchao describes on the retry
wrapper. Should the dedup key on something that tells the two sessions apart
rather than on the prefix list?
--
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]
