andygrove commented on code in PR #5314:
URL: https://github.com/apache/datafusion-comet/pull/5314#discussion_r3906737015
##########
spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala:
##########
@@ -42,43 +44,157 @@ 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 the first path segment
for the authorityless
+ * `blob:///bucket/key` form (matching the native rewrite that promotes it
into the host). The
+ * path-segment fallback fires only for S3-family schemes (`s3`, `s3a`,
`s3n`, and the opted-in
+ * `s3CompliantSchemes`). For any other scheme an absent authority means no
bucket, so a local
+ * Hadoop-catalog `file:///tmp/warehouse/...` returns None rather than the
surprising `tmp`.
+ * Returns None when neither source applies.
+ */
+ def bucketForUri(uri: URI, s3CompliantSchemes: Set[String]): Option[String]
= {
+ Option(uri.getAuthority)
+ .map(_.trim)
+ .filter(_.nonEmpty)
+ .orElse {
+ val scheme =
Option(uri.getScheme).map(_.toLowerCase(Locale.ROOT)).getOrElse("")
+ if (isS3FamilyScheme(scheme, s3CompliantSchemes)) {
+ Option(uri.getPath)
+ .map(_.stripPrefix("/"))
+ .map(_.takeWhile(_ != '/'))
+ .filter(_.nonEmpty)
+ } else {
+ None
+ }
+ }
+ }
+
+ // s3/s3a/s3n and any opted-in alias share the authorityless path-promotion
semantics above.
+ private def isS3FamilyScheme(scheme: String, s3CompliantSchemes:
Set[String]): Boolean =
+ scheme == "s3" || scheme == "s3a" || scheme == "s3n" ||
s3CompliantSchemes.contains(scheme)
+
+ /**
+ * Translate vendor-style `fs.<scheme>.<authority>.<property>` keys into the
`fs.s3a.*` shape
+ * object_store's AmazonS3Builder reads, for a configured S3-compliant
`scheme`. Recognized
+ * properties are `vendorPropertyToS3aSuffix`; unknown ones are dropped.
+ *
+ * Keys land at the per-bucket `fs.s3a.bucket.<bucket>.*` scope (for the
`default` authority the
+ * bucket is `defaultBucket`, promoted from the URL path) -- the scope
native `get_config`
+ * checks first. The authority is matched greedily so dotted bucket names
survive.
+ *
+ * An `endpoint` also synthesizes `path.style.access=true` as a soft
default, suppressed by an
+ * explicit per-bucket setting or vendor `pathStyleAccess`. Applied AFTER
the plain `fs.s3a.*`
+ * pass so real vendor keys override conflicting `fs.s3a.*` (the
403-misdirect note below).
+ */
+ private def translateVendorKeys(
+ hadoopConf: Configuration,
+ scheme: String,
+ defaultBucket: Option[String]): Map[String, String] = {
+ import scala.jdk.CollectionConverters._
+ // `fs.<scheme>.<authority>.<property>`; scheme is regex-quoted (may
contain `.`/`+`/`-`).
+ val vendorKeyPattern = ("^fs\\." + Pattern.quote(scheme) +
"\\.(.+)\\.([^.]+)$").r
+ val out = scala.collection.mutable.Map[String, String]()
+ hadoopConf.iterator().asScala.foreach { entry =>
+ entry.getKey match {
+ case vendorKeyPattern(authority, property) =>
+ vendorPropertyToS3aSuffix.get(property).foreach { suffix =>
+ val bucket =
+ if (authority == vendorDefaultAuthority) defaultBucket else
Some(authority)
+ val scope = bucket.map(b =>
s"fs.s3a.bucket.$b").getOrElse("fs.s3a")
+ out(s"$scope.$suffix") = entry.getValue
Review Comment:
There is no precedence rule here between an authority of `default` and an
explicit authority that resolves to the same bucket. Both write
`fs.s3a.bucket.<bucket>.<suffix>` through this plain assignment, so whichever
one `hadoopConf.iterator()` happens to yield last wins.
With `fs.blob.default.endpoint=https://DEFAULT.example` and
`fs.blob.<bucket>.endpoint=https://EXPLICIT.example` both set, I get:
```
blob://mybucket/data/part-0.parquet -> fs.s3a.bucket.mybucket.endpoint =
https://EXPLICIT.example
blob://warehouse/data/part-0.parquet -> fs.s3a.bucket.warehouse.endpoint =
https://DEFAULT.example
blob://lake/data/part-0.parquet -> fs.s3a.bucket.lake.endpoint =
https://DEFAULT.example
blob://data/data/part-0.parquet -> fs.s3a.bucket.data.endpoint =
https://EXPLICIT.example
```
Same config, same code path, different answer depending on where the bucket
name lands in the `Configuration` hash order. Across 40 synthetic bucket names
it split 30 explicit to 10 default.
This is reachable from the shape the new docs recommend. `fs.<s>.default.*`
is the documented spelling for the authorityless form, so anyone with both URI
forms in play, or with `fs.blob.default.*` shipped in `core-site.xml` and
`fs.blob.<bucket>.*` set at the job level, gets a coin flip on which endpoint
and credentials are used. The symptom would be a 403 or a read against the
wrong endpoint, with nothing in the config to explain why one bucket works and
another does not.
Could the two be ordered, so `default`-authority keys are folded in first
and an explicit authority overwrites them? Worth a `NativeConfigSuite` case
pinning explicit-beats-default too. The existing `blob:// default authority
overrides per-bucket key` test passes only because `mybucket` happens to land
on the winning side, so it would not catch a regression here.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -262,13 +262,18 @@ case class CometScanRule(session: SparkSession)
s.split(",").map(_.trim.toLowerCase(Locale.ROOT)).filter(_.nonEmpty).toSet
Review Comment:
This PR adds `NativeConfig.parseSchemeSet` and documents it as the shared
spelling so the JVM gate and the native rewrite parse identically, but
`libhdfsSchemes` four lines above still inlines the byte-identical
`split(",").map(_.trim.toLowerCase(Locale.ROOT)).filter(_.nonEmpty).toSet`.
Could it call the helper too?
```scala
val libhdfsSchemes: Set[String] = COMET_LIBHDFS_SCHEMES.get() match {
case Some(s) => NativeConfig.parseSchemeSet(s)
case None => Set("hdfs")
}
```
Keeps the `hdfs` default and leaves one parse to maintain instead of two
adjacent ones.
##########
spark/src/test/scala/org/apache/comet/IcebergReadFromS3Suite.scala:
##########
@@ -37,6 +37,22 @@ class IcebergReadFromS3Suite extends CometS3TestBase with
RESTCatalogHelper {
conf.set("spark.sql.catalog.s3_catalog.warehouse",
s"s3a://$testBucketName/warehouse")
applyS3CatalogProps(conf, "s3_catalog")
+ // blob:// variant: `blob` is an opt-in S3-compliant alias, backed by an
S3A-derived FileSystem
+ // so Iceberg reads/writes blob://<bucket>/... against the same MinIO.
Exercises the
+ // delete-matching invariant (metadata/delete paths normalized to s3://,
data_file_path raw).
Review Comment:
This comment and the one at line 246 both describe the design the head
commit removed. Here it says "metadata/delete paths normalized to s3://,
data_file_path raw", and line 246 says "the delete file's own location is
normalized to s3://. If either half is wrong, deletes leak". After `413b47f2`
nothing on the Iceberg path is normalized, and removing that asymmetry was the
whole point of the change. A reader landing here later will go looking for a
normalizer that no longer exists.
I would not usually push on wording, but this is the same drift I flagged
last round and it recurred inside the commit that fixed the previous instance.
It also matters more here than elsewhere: `dev/ci/check-suites.py` ignores this
suite, so no CI run will ever surface the next one.
Two smaller ones in the same family while you are in here.
`is_s3_family_scheme` at
`native/core/src/execution/operators/iceberg_scan.rs:418` says it "Mirrors the
Scala `NativeConfig.isS3FamilyScheme` gate", but the Scala one also matches
`s3n` and the Rust one does not. And `CometScanRule.scala:517` says the alias
key "never leaks" because `load_file_io` filters `catalog_properties`, which is
true of `FileIO` but not of `CometS3CredentialBridge::new`, which is handed the
unfiltered map and forwards it to the user's credential provider.
##########
native/core/src/parquet/objectstore/s3_blob_fs_support.rs:
##########
@@ -0,0 +1,207 @@
+// 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-compatible filesystem alias handling.
+//!
+//! Operators mark extra schemes as `s3://` aliases via
`fs.comet.s3Compliant.schemes`
+//! (comma-separated, case-insensitive, opt-in). They route to
`object_store::AmazonS3`, but
+//! `ObjectStoreScheme::parse` does not recognize them.
+//!
+//! On the Parquet scan path, [`normalize_object_store_url`] rewrites those
aliases (and `s3a`) to
+//! `s3://bucket/key` so `prepare_object_store_with_configs`'s `scheme ==
"s3"` dispatch fires. Its
+//! callers consume only `url.scheme()`/`url.path()`, so re-serialization
through `url::Url` is
+//! harmless.
+//!
+//! The Iceberg path does NOT rewrite URLs. iceberg-rust matches
positional/equality deletes by an
+//! exact string comparison of a delete file's recorded `file_path` against
the `data_file_path`
+//! Comet supplies, so any scheme rewrite there would desync the two and
silently drop deletes.
+//! Instead `IcebergScanExec::storage_factory_for` routes an alias scheme to
its S3 backend directly
+//! (via [`is_s3_compliant_alias_scheme`]), and iceberg-storage-opendal's
scheme-agnostic S3
+//! operator opens the raw `blob://...` path as-is (it derives the bucket from
`Url::host_str` and
+//! the key prefix from the path's own scheme).
+
+use std::collections::HashMap;
+
+use url::Url;
+
+use crate::execution::operators::ExecutionError;
+use crate::parquet::parquet_support::is_hdfs_scheme;
+
+/// Rewrites `s3a` and the configured s3-compliant aliases to
`s3://bucket/key`, promoting a missing
+/// authority into the host. Non-alias schemes are returned unchanged. `s3a`
routed through libhdfs
+/// (`fs.comet.libhdfs.schemes`) is left alone.
+pub(crate) fn normalize_object_store_url(
+ url_str: &str,
+ object_store_configs: &HashMap<String, String>,
+) -> Result<Url, ExecutionError> {
+ let url = Url::parse(url_str)
+ .map_err(|e| ExecutionError::GeneralError(format!("Error parsing URL
{url_str}: {e}")))?;
+ if is_hdfs_scheme(&url, object_store_configs) {
+ return Ok(url);
+ }
+ // Callers consume only `url.scheme()`/`url.path()`, so re-serialization
through `url::Url` is
+ // harmless here (the Iceberg path avoids this function precisely because
it must preserve raw
+ // bytes; it routes aliases by scheme in `storage_factory_for` instead).
+ let scheme = url.scheme();
+ if scheme != "s3a" && !is_s3_compliant_alias_scheme(scheme,
object_store_configs) {
+ return Ok(url);
+ }
+ rewrite_alias_to_s3(url)
+}
+
+/// True if `scheme` is a configured s3-compliant alias
(`fs.comet.s3Compliant.schemes`,
+/// comma-separated, case-insensitive; empty/unset means none). These route to
`AmazonS3` but are
+/// not recognized by `ObjectStoreScheme::parse`. `s3a` is excluded --
object_store knows it and
+/// callers special-case it. Shared by the Parquet normalizer above and the
Iceberg storage-factory
+/// gate (`IcebergScanExec::storage_factory_for`) so both admit the same
schemes.
+pub(crate) fn is_s3_compliant_alias_scheme(
+ scheme: &str,
+ object_store_configs: &HashMap<String, String>,
+) -> bool {
+ const COMET_S3_COMPLIANT_SCHEMES_KEY: &str =
"fs.comet.s3Compliant.schemes";
+ match object_store_configs.get(COMET_S3_COMPLIANT_SCHEMES_KEY) {
+ Some(schemes) => schemes
+ .split(',')
+ .any(|s| s.trim().eq_ignore_ascii_case(scheme)),
+ None => false,
+ }
+}
+
+/// Rewrites a parsed alias URL (`s3a` or a configured alias) to
`s3://bucket/key`. When it has no
+/// authority (empty-authority `blob:///bucket/key` or opaque
`blob:/bucket/key`, both host=None)
+/// the first path segment is promoted into the host. Defensive: object-store
URLs normally have one.
+fn rewrite_alias_to_s3(mut url: Url) -> Result<Url, ExecutionError> {
+ let original = url.scheme().to_string();
+ let needs_host_promotion = url.host_str().is_none();
+ url.set_scheme("s3").map_err(|_| {
Review Comment:
`Url::set_scheme` refuses transitions between the URL spec's special schemes
(`file`, `http`, `https`, `ftp`, `ws`, `wss`) and non-special ones, so listing
any of those in `fs.comet.s3Compliant.schemes` does not just fail to help. It
makes every URL of that scheme return `Err` from `normalize_object_store_url`,
and since both `prepare_object_store_with_configs` and `get_partitioned_files`
propagate it, the query fails instead of falling back. I confirmed it for all
six:
```
fs.comet.s3Compliant.schemes = file
normalize_object_store_url("file:///tmp/x") -> Err("Could not convert scheme
from file to s3")
```
A stray `file` in the list would take down every local Parquet scan in the
session, and `https` is the one someone might plausibly reach for, since these
services are usually described to users by their HTTPS endpoint.
Since the list is user-typed and the failure is a hard error rather than a
fallback, would it be worth skipping those schemes in
`is_s3_compliant_alias_scheme` the way `s3a` is already excluded, or validating
the list once in `CometConf` with a message that says why?
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -490,7 +499,9 @@ case class CometScanRule(session: SparkSession)
val hadoopS3Options =
NativeConfig.extractObjectStoreOptions(hadoopConf, effectiveUri)
val hadoopDerivedProperties =
-
CometIcebergNativeScan.hadoopToIcebergS3Properties(hadoopS3Options)
+ CometIcebergNativeScan.hadoopToIcebergS3Properties(
+ hadoopS3Options,
+ NativeConfig.bucketForUri(effectiveUri, s3CompliantSchemes))
Review Comment:
`effectiveUri` is the only string `storage_factory_for` and `load_file_io`
dispatch on, and it never goes through a scheme gate. It is used here and at
line 499, then handed to `CometIcebergNativeScanMetadata.extract` unchecked.
`isIcebergReadableScheme` is only applied to data-file and delete-file paths
inside `validateIcebergFileScanTasks`.
The comment at line 1094 explains why that gate exists: object_store
recognizes schemes iceberg-rust's storage factory cannot build, and admitting
them "turns a clean JVM fallback into a native runtime `Unsupported storage
scheme` error". That reasoning applies to the metadata location at least as
much as to the data files, since it is the value the factory actually matches
on. A table whose data files pass the gate but whose metadata location uses a
scheme outside `{file, s3, s3a, gs, oss}` plus the configured aliases is
claimed by the planner and then dies at execution, which is the exact outcome
the narrowing was meant to prevent. It is also the half of the earlier `Match
Iceberg admission to its actual storage factory` review point that narrowing
the task-path allowlist did not cover.
Would adding `isIcebergReadableScheme(effectiveUri, s3CompliantSchemes)`
alongside the existing checks, with a fallback reason, close it? A schemeless
local-catalog path already returns true there, so the local Hadoop-catalog case
keeps working.
--
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]
