sunchao commented on code in PR #5314:
URL: https://github.com/apache/datafusion-comet/pull/5314#discussion_r3916471214
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -1018,30 +1060,48 @@ case class CometScanTypeChecker() extends
DataTypeSupport with CometTypeShim {
object CometScanRule extends Logging {
- // Per-scheme memo of `NativeBase.isObjectStoreSchemeSupported`. The answer
depends only on the
- // URL scheme, so we cache by scheme and never re-cross the JNI boundary for
a repeated scheme.
+ // Memo of `NativeBase.isObjectStoreSchemeSupported`, keyed by (scheme,
has-authority) because
+ // object_store's parser keys on that pair (host-based stores need a host,
file/memory need
+ // none). Caching by scheme alone lets an authorityless URL poison the whole
scheme's answer.
private val schemeSupportCache =
- new ConcurrentHashMap[String, JBoolean]()
+ new ConcurrentHashMap[(String, Boolean), JBoolean]()
+
+ /**
+ * Opt-in S3-compliant alias schemes from `fs.comet.s3Compliant.schemes`
(comma-separated,
+ * trimmed, lowercased; empty/missing => none). The native gate no longer
claims these, so the
+ * opt-in is decided here where the Hadoop config is available.
+ */
+ private[rules] def resolveS3CompliantSchemes(hadoopConf: Configuration):
Set[String] =
+ NativeConfig.parseSchemeSet(hadoopConf.get(COMET_S3_COMPLIANT_SCHEMES_KEY))
/**
- * True when Comet's native object_store layer recognizes this URI's scheme
(so the scan is
- * natively readable). Delegates to the native layer -- the source of truth
-- instead of a
- * hardcoded scheme list. On any failure to consult native (e.g. the library
isn't loaded on
- * this JVM, or predates this method) we assume the scheme IS supported: the
scheme gate is an
- * early-fallback optimization, and a build without a working native library
can't run Comet's
- * native scan anyway, so declining here would only over-restrict.
+ * True when Comet's native Parquet scan can read this URI's scheme:
object_store recognizes it
+ * (asked of the native layer, the source of truth) OR it is an opt-in
S3-compliant alias. If
+ * native can't be consulted (library not loaded), assume supported -- the
gate is only an
+ * early-fallback optimization and such a build can't run the native scan
anyway.
*/
- private[rules] def isNativelyReadableScheme(uri: URI): Boolean = {
+ private[rules] def isNativelyReadableScheme(
+ uri: URI,
+ s3CompliantSchemes: Set[String]): Boolean = {
val scheme = uri.getScheme
if (scheme == null) return true
+ val lower = scheme.toLowerCase(Locale.ROOT)
+ if (s3CompliantSchemes.contains(lower)) return true
Review Comment:
**[P1] Preserve bucket identity when admitting multi-bucket alias scans**
With `fs.comet.s3Compliant.schemes=blob`, this now admits a Parquet read
spanning two buckets, but native planning still selects the first file's object
store for the entire `FilePartition` and strips the authority from every file's
object key.
I reproduced this against MinIO with two distinct, equal-size Parquet files
at the same key, asserted one native scan and one partition containing both
input URLs:
```text
blob://comet-review-first/review-multibucket/same-key.parquet -> 111
blob://comet-review-second/review-multibucket/same-key.parquet -> 222
Spark: [111, 222]
Comet: [111, 111]
```
Restoring the previous alias-admission behavior on this head makes the same
query fall back and return `[111, 222]`. The single-store limitation already
exists for plain S3, but this PR newly exposes alias reads that previously fell
back. Preserve resolved storage/bucket identity when grouping files, or decline
mixed-store alias scans.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -490,7 +498,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:
**[P2] Do not apply metadata-bucket settings to every Iceberg file**
`effectiveUri` selects the metadata bucket, so this promotes that bucket's
overrides into the global properties used by one native `FileIO` for all
data/delete files. Iceberg supports a separate data location through
`write.data.path`.
For a `HadoopFileIO` table with metadata under `s3://metadata/...`, data
under `s3://data/...`, working global `fs.s3a.*` data-store settings, and
`fs.s3a.bucket.metadata.*` overrides, the base translation sends a data request
to the data endpoint with the data credentials. This revision instead sends the
same request to the metadata endpoint with metadata credentials. An actual HTTP
probe through the pinned FileIO returned success before the change and 404
afterward.
This trigger applies when FileIO properties do not supply overriding `s3.*`
settings, and also affects ordinary S3 scans with aliases disabled. Evidence is
compiled exact base/head configuration helpers plus FileIO requests, not a full
Spark query. Resolve settings for the actual data/delete locations, or fall
back when the scan cannot share one configuration.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -1018,30 +1060,48 @@ case class CometScanTypeChecker() extends
DataTypeSupport with CometTypeShim {
object CometScanRule extends Logging {
- // Per-scheme memo of `NativeBase.isObjectStoreSchemeSupported`. The answer
depends only on the
- // URL scheme, so we cache by scheme and never re-cross the JNI boundary for
a repeated scheme.
+ // Memo of `NativeBase.isObjectStoreSchemeSupported`, keyed by (scheme,
has-authority) because
+ // object_store's parser keys on that pair (host-based stores need a host,
file/memory need
+ // none). Caching by scheme alone lets an authorityless URL poison the whole
scheme's answer.
private val schemeSupportCache =
- new ConcurrentHashMap[String, JBoolean]()
+ new ConcurrentHashMap[(String, Boolean), JBoolean]()
+
+ /**
+ * Opt-in S3-compliant alias schemes from `fs.comet.s3Compliant.schemes`
(comma-separated,
+ * trimmed, lowercased; empty/missing => none). The native gate no longer
claims these, so the
+ * opt-in is decided here where the Hadoop config is available.
+ */
+ private[rules] def resolveS3CompliantSchemes(hadoopConf: Configuration):
Set[String] =
+ NativeConfig.parseSchemeSet(hadoopConf.get(COMET_S3_COMPLIANT_SCHEMES_KEY))
/**
- * True when Comet's native object_store layer recognizes this URI's scheme
(so the scan is
- * natively readable). Delegates to the native layer -- the source of truth
-- instead of a
- * hardcoded scheme list. On any failure to consult native (e.g. the library
isn't loaded on
- * this JVM, or predates this method) we assume the scheme IS supported: the
scheme gate is an
- * early-fallback optimization, and a build without a working native library
can't run Comet's
- * native scan anyway, so declining here would only over-restrict.
+ * True when Comet's native Parquet scan can read this URI's scheme:
object_store recognizes it
+ * (asked of the native layer, the source of truth) OR it is an opt-in
S3-compliant alias. If
+ * native can't be consulted (library not loaded), assume supported -- the
gate is only an
+ * early-fallback optimization and such a build can't run the native scan
anyway.
*/
- private[rules] def isNativelyReadableScheme(uri: URI): Boolean = {
+ private[rules] def isNativelyReadableScheme(
+ uri: URI,
+ s3CompliantSchemes: Set[String]): Boolean = {
val scheme = uri.getScheme
if (scheme == null) return true
+ val lower = scheme.toLowerCase(Locale.ROOT)
+ if (s3CompliantSchemes.contains(lower)) return true
+ val hasAuthority = uri.getRawAuthority != null
schemeSupportCache
.computeIfAbsent(
- scheme.toLowerCase(Locale.ROOT),
- _ =>
- try
JBoolean.valueOf(NativeBase.isObjectStoreSchemeSupported(uri.toString))
+ (lower, hasAuthority),
+ _ => {
+ // Probe a scheme(+fixed dummy host) URL, never the caller's
authority/path: object_store
+ // keys on (scheme, host-presence) so we cache/probe by that pair. A
fixed host stops an
+ // authorityless URL poisoning the authority-bearing form; dropping
the real path avoids a
+ // spurious `false` from chars object_store rejects in a `Path`
(e.g. Iceberg's `p=%0A`).
+ val probe = if (hasAuthority) s"$lower://comet-probe-host/" else
s"$lower:///"
+ try JBoolean.valueOf(NativeBase.isObjectStoreSchemeSupported(probe))
Review Comment:
**[P2] Preserve actual-path validation before selecting native Parquet**
The dummy probe loses a path-specific rejection that native execution still
enforces. A local Parquet directory whose name contains an actual newline is
readable by Spark; its URI contains `%0A`. The exact native probe returns
`false` for that URI and `true` for `file:///`.
In a full Spark reproduction, Spark read `[0, 1, 2]`, while this head
selected `CometNativeScanExec` and failed with `Generic URL error` /
`Encountered illegal character sequence` when opening the real path. Restoring
the previous admission decisions on this head, in a fresh test JVM, restored
successful Spark fallback.
This is an actual newline in a Parquet directory name, distinct from
Iceberg's literal `p=%0A` path that motivated the earlier cache fix. Keep
scheme-capability caching path-independent, but validate the actual Parquet
path separately before claiming native execution.
##########
spark/src/main/scala/org/apache/comet/rules/CometScanRule.scala:
##########
@@ -1054,13 +1114,52 @@ object CometScanRule extends Logging {
val SKIP_COMET_SCAN_TAG:
org.apache.spark.sql.catalyst.trees.TreeNodeTag[Unit] =
org.apache.spark.sql.catalyst.trees.TreeNodeTag[Unit]("comet.skipCometScan")
+ /**
+ * Schemes Comet's native Iceberg scan can actually open, mirroring the
match arms in
+ *
`native/core/src/execution/operators/iceberg_scan.rs::storage_factory_for`.
Deliberately NOT
+ * delegated to `isNativelyReadableScheme`: object_store recognizes schemes
(http/https, azure,
+ * memory) that iceberg-rust's OpenDAL storage factory cannot build, and
admitting them here
+ * turns a clean JVM fallback into a native runtime "Unsupported storage
scheme" error. Add here
+ * what you add to `storage_factory_for` (currently Aliyun `oss` and GCS
`gs`). S3-compliant
+ * aliases like `blob` are opt-in via `fs.comet.s3Compliant.schemes` (see
+ * `isIcebergReadableScheme`), not hardcoded, since the native planner opens
them via S3.
+ */
+ private val icebergReadableSchemes: Set[String] =
+ Set("file", "s3", "s3a", "gs", "oss")
+
+ /** Effective Iceberg-readable schemes: built-in allowlist plus opt-in
S3-compliant aliases. */
+ private[rules] def icebergSupportedSchemes(s3CompliantSchemes: Set[String]):
Set[String] =
+ icebergReadableSchemes ++ s3CompliantSchemes
+
+ /** "Supported schemes: ..." suffix shared by the Iceberg scheme-fallback
messages. */
+ private[rules] def icebergSupportedSchemesMessage(s3CompliantSchemes:
Set[String]): String = {
+ val schemes =
icebergSupportedSchemes(s3CompliantSchemes).toSeq.sorted.mkString(", ")
+ s"Supported schemes: $schemes"
+ }
+
+ /**
+ * Scheme gate for the Iceberg scan path: admit schemes iceberg-rust's
storage factory can build
+ * (`icebergReadableSchemes`) or opt-in S3-compliant aliases
(`fs.comet.s3Compliant.schemes`),
+ * plus schemeless local paths (which route to its LocalFs backend).
+ */
+ private[rules] def isIcebergReadableScheme(
+ uri: URI,
+ s3CompliantSchemes: Set[String]): Boolean = {
+ val scheme = uri.getScheme
+ if (scheme == null) return true
+ val lower = scheme.toLowerCase(Locale.ROOT)
+ icebergReadableSchemes.contains(lower) ||
s3CompliantSchemes.contains(lower)
Review Comment:
**[P2] Reject authorityless Iceberg alias file paths**
The scheme-only gate admits an actual data/delete location such as
`blob:///bucket/key.parquet`, but Iceberg preserves that raw string and the
selected S3 backend requires a URL host. Promoting the bucket while extracting
Scala configuration does not change the transport path.
Using the pinned Iceberg/OpenDAL FileIO and a local HTTP success control,
`new_input(path).metadata()` succeeds for `blob://bucket/key.parquet` and
returns `DataInvalid: ... missing bucket` for `blob:///bucket/key.parquet`.
This is a FileIO component reproduction; an authorityless metadata location
alone is not sufficient, since it selects the factory without necessarily being
opened.
Parquet explicitly promotes the authorityless form, so its support cannot be
assumed for the raw Iceberg route. Decline these data/delete paths before
native admission, or adapt only their transport location while preserving the
original logical identity used for delete matching.
--
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]
