andygrove commented on code in PR #5314:
URL: https://github.com/apache/datafusion-comet/pull/5314#discussion_r3897661236
##########
spark/src/main/java/org/apache/comet/parquet/CometFileKeyUnwrapper.java:
##########
@@ -92,6 +93,12 @@
*/
public class CometFileKeyUnwrapper {
+ // S3-family schemes that all address the same logical filesystem and must
fold to a single
+ // canonical cache key. Kept in sync with the native alias handling
+ // (native/core/src/parquet/objectstore/s3_blob_fs_support.rs and the
s3a/s3n aliases object_store
+ // recognizes). The `:` disambiguates them, so order is irrelevant.
+ private static final String[] S3_ALIAS_SCHEMES = {"s3", "s3n", "s3a",
"blob"};
Review Comment:
This list is hardcoded to `blob`, but the scheme set is configurable and the
docs suggest `blob,minio,r2`. If someone sets
`fs.comet.s3Compliant.schemes=minio` and reads encrypted Parquet, I think the
read fails.
The put side caches under `minio://bucket/key`, since `CometExecIterator`
passes `relation.inputFiles` straight through and nothing here matches. The get
side arrives as `s3://bucket/key`, because `parquet_exec.rs` sets `uri_base`
from the already-normalized `ObjectStoreUrl`, and that folds to
`s3a://bucket/key`. The two keys never meet, so it comes out as `Failed to find
DecryptionKeyRetriever`.
`storeDecryptionKeyRetriever` already receives the `Configuration` and
already caches it in the `conf` field, so the configured set is available at
exactly the right moment. Could the alias list be read from
`fs.comet.s3Compliant.schemes` there and cached for the get side, instead of
the static array? `TestCometFileKeyUnwrapper` only covers `blob` today, so a
`minio://` case alongside it would keep this from drifting again.
##########
native/core/src/execution/operators/iceberg_scan.rs:
##########
@@ -255,6 +255,9 @@ impl IcebergScanExec {
};
match scheme {
"file" => Ok(Arc::new(OpenDalStorageFactory::Fs)),
+ // blob:// never reaches here: the planner normalizes it to s3://
via
Review Comment:
Does the Iceberg path need the rewrite at all?
The S3 backend derives both the bucket and the key prefix from the path's
own scheme: `s3_config_build` takes the bucket from `Url::host_str` without
checking the scheme, and `create_operator` builds `format!("{}://{}/",
url.scheme(), op_info.name())`. That is exactly why a raw `blob://` data file
path works, as the comment on `data_file_path` says. Which means the only
reason `metadata_location` gets normalized upstream is to land in this match
arm. As far as I can tell it is never opened, only used here, in
`build_s3_credential_loader` for the host and path, and in the `is_s3_family`
region check below.
If this arm matched configured aliases directly, metadata, delete files and
data files could all stay raw, and `normalize_object_store_url_string` could go
away entirely, taking both of its invariants with it, along with the asymmetry
where the data file path must stay raw but the delete file's own location must
not. That asymmetry is the part I would most like to see shrink, because it is
exactly the kind of thing a future change breaks silently: get it wrong and
deleted rows come back, with no error.
The one thing you would give up is the single-slash `blob:/bucket/key` form,
which the tests already describe as defensive rather than a shape Iceberg
emits. `is_s3_family` below would need to consult the alias set too. Am I
missing a reason the rewrite has to happen upstream in the planner?
##########
docs/source/user-guide/latest/datasources.md:
##########
@@ -216,6 +216,62 @@ Beyond credential providers, Comet's Parquet scan supports
additional S3 configu
All configuration options support bucket-specific overrides using the pattern
`fs.s3a.bucket.{bucket-name}.{option}`.
+### S3-Compliant Filesystem Schemes
+
+Some environments front an S3-compatible service (MinIO, Ceph RGW, Cloudflare
R2, Wasabi, and
+similar) with a vendor-branded Hadoop filesystem client that registers its own
URL scheme, for
+example `blob://`, instead of `s3://` or `s3a://`. Comet can treat such
schemes as aliases for
+`s3://` so the native Parquet and Iceberg scans read them directly, without
the caller rewriting
+URLs.
+
+This is opt-in and disabled by default. Enable it by listing the schemes to
treat as S3-compliant
+aliases in `spark.hadoop.fs.comet.s3Compliant.schemes` (Hadoop key
+`fs.comet.s3Compliant.schemes`), a comma-separated, case-insensitive list.
This mirrors the
+existing `fs.comet.libhdfs.schemes` config.
+
+```shell
+--conf spark.hadoop.fs.comet.s3Compliant.schemes=blob
+```
+
+Multiple schemes can be listed together:
+
+```shell
+--conf spark.hadoop.fs.comet.s3Compliant.schemes=blob,minio,r2
+```
+
+With no configuration, Comet claims none of these aliases, so for example a
`blob://` path falls
+back to Spark unchanged. The empty default is deliberate: short scheme names
like `blob` are not
+unique to S3-compatible storage. Azure Blob Storage is the clearest example,
so claiming `blob://`
+unconditionally would risk misrouting paths that were never meant for Comet's
S3 client. Only add a
+scheme here if you intend Comet to treat it as S3-compatible.
+
+For each scheme `<s>` listed in `fs.comet.s3Compliant.schemes`, Comet also
reads vendor-style,
+per-authority Hadoop keys of the form `fs.<s>.<authority>.<property>` (the
authority is typically
+the bucket or account name from the URL) and translates them into the
`fs.s3a.*` surface described
+above. The recognized vendor-style properties and their `fs.s3a.*` targets are:
+
+| Vendor property (`fs.<s>.<authority>.<property>`) | Translated `fs.s3a.*`
suffix |
+| ------------------------------------------------- |
---------------------------- |
+| `awsAccessKeyId` | `access.key`
|
+| `awsSecretAccessKey` | `secret.key`
|
+| `awsSessionToken` | `session.token`
|
+| `endpoint` | `endpoint`
|
+| `region` | `endpoint.region`
|
+| `pathStyleAccess` | `path.style.access`
|
+
+Unrecognized `fs.<s>.<authority>.*` properties are ignored.
Review Comment:
The `fs.<s>.default.<property>` case is implemented and tested, where an
authority of literally `default` resolves to the bucket taken from the URL
path, but it is not mentioned here. Since that is the form an authorityless
`blob:///bucket/key` URI produces, it seems worth a sentence so users know the
key exists.
##########
spark/src/main/scala/org/apache/comet/CometConf.scala:
##########
@@ -878,6 +878,19 @@ object CometConf extends ShimCometConf {
.stringConf
.createOptional
+ val COMET_S3_COMPLIANT_SCHEMES_KEY = "fs.comet.s3Compliant.schemes"
+
+ val COMET_S3_COMPLIANT_SCHEMES: OptionalConfigEntry[String] =
Review Comment:
This entry is never read anywhere. Every consumer goes through
`hadoopConf.get(COMET_S3_COMPLIANT_SCHEMES_KEY)` instead, which is different
from `COMET_LIBHDFS_SCHEMES`, where `CometScanRule` reads the entry with
`.get()`.
The practical consequence is that
`spark.conf.set("spark.hadoop.fs.comet.s3Compliant.schemes", ...)` after the
session has started is silently ignored: `SessionState.newHadoopConf` copies
SQLConf keys with the `spark.hadoop.` prefix still attached, so
`fs.comet.s3Compliant.schemes` is never set in the Hadoop conf. Only SparkConf
at startup or core-site.xml works, which is fine but not what a user would
guess from seeing it in configs.md.
Could this either read the entry the way the libhdfs gate does, or say in
the doc string that it has to be set before the session starts?
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -654,24 +667,36 @@ case class CometScanRule(session: SparkSession)
false
}
+ // Opt-in S3-compliant alias schemes (e.g. `blob`) from
`fs.comet.s3Compliant.schemes`,
+ // resolved once for this scan (used by the file scheme gate below and
its fallback
+ // message).
+ val icebergS3CompliantSchemes =
+
CometScanRule.resolveS3CompliantSchemes(session.sessionState.newHadoopConf())
Review Comment:
Minor: this builds a fresh `Configuration` per Iceberg scan node just to
read one key, and `transformV2Scan` already builds another one at line 464
inside the `metadataOpt` block. Could one be hoisted to the top of the method
and shared?
##########
spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala:
##########
@@ -42,43 +44,144 @@ object NativeConfig {
"abfs" -> Seq("fs.azure.", "fs.abfs."),
"abfss" -> Seq("fs.azure.", "fs.abfss.", "fs.abfs."))
+ // Some alias filesystems report the literal authority "default" when the
URI has none (e.g.
+ // `scheme:///bucket/key`); the real bucket is then promoted from the URL
path. Keys under this
+ // authority map to the per-bucket `fs.s3a.bucket.<resolved-bucket>.*`
scope, NOT global
+ // `fs.s3a.*`, because native `get_config` checks per-bucket before global.
+ private val vendorDefaultAuthority = "default"
+
+ // Recognized vendor properties -> `fs.s3a` suffix. Mirrors
CometIcebergNativeScan's target set.
+ // Unknown properties are dropped.
+ private val vendorPropertyToS3aSuffix = Map(
+ "awsAccessKeyId" -> "access.key",
+ "awsSecretAccessKey" -> "secret.key",
+ "awsSessionToken" -> "session.token",
+ "endpoint" -> "endpoint",
+ "region" -> "endpoint.region",
+ "pathStyleAccess" -> "path.style.access")
+
+ // Comma-separated scheme list -> trimmed, lowercased set
(case-insensitive). Shared with
+ // CometScanRule's scheme gate so the JVM admit-decision and native rewrite
parse identically.
+ private[comet] def parseSchemeSet(raw: String): Set[String] =
+ Option(raw)
+
.map(_.split(",").iterator.map(_.trim.toLowerCase(Locale.ROOT)).filter(_.nonEmpty).toSet)
+ .getOrElse(Set.empty)
+
+ // True when the user pinned path-style at this bucket scope. Suppresses the
synthesized soft
+ // default so an explicit per-bucket setting (including `false`, the escape
hatch) survives. A
+ // global `fs.s3a.path.style.access` is intentionally ignored: it may target
other s3a workloads
+ // or be an ambient cluster default. Per-bucket synth wins over global in
native anyway.
+ private def userSetPathStyle(hadoopConf: Configuration, bucket:
Option[String]): Boolean =
+ bucket.exists(b => hadoopConf.get(s"fs.s3a.bucket.$b.path.style.access")
!= null)
+
+ /**
+ * The S3 bucket a URI addresses: its authority, or -- for the authorityless
+ * `blob:///bucket/key` form -- the first path segment (matching the native
rewrite that
+ * promotes it into the host). None when neither is present (e.g. a
schemeless root path).
+ */
+ def bucketForUri(uri: URI): Option[String] = {
Review Comment:
For a local Hadoop catalog this returns something odd:
`file:///tmp/warehouse/db/t/metadata/v1.metadata.json` has no authority, so the
first path segment wins and the bucket comes back as `Some("tmp")`. Every
local-catalog Iceberg table then calls `hadoopToIcebergS3Properties` with a
target bucket of `tmp`.
Harmless today, since nobody has `fs.s3a.bucket.tmp.*` configured, but it is
a surprising value to hand to a function whose entire job is choosing which
bucket's credentials get promoted to global. Worth returning `None` when the
scheme is not S3-family?
--
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]
