sunchao commented on code in PR #6753:
URL: https://github.com/apache/datafusion-comet/pull/6753#discussion_r4214819433
##########
spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala:
##########
@@ -197,6 +253,46 @@ object NativeConfig {
* The result feeds object_store's parse_url_opts natively.
*/
def extractObjectStoreOptions(hadoopConf: Configuration, uri: URI):
Map[String, String] = {
+ val (options, aliasOptions) =
objectStoreOptionsAndAliasOptions(hadoopConf, uri)
+ options ++ aliasOptions
+ }
+
+ /**
+ * The object store options for a scan of `uris`:
`extractObjectStoreOptions` of one URI per
+ * scheme, and of one per bucket for an opted-in alias. Each scheme forwards
keys under its own
+ * prefixes. An alias bucket keeps only the settings translated for that
bucket, applied after
+ * every scheme's own keys so that they win over a raw
`fs.s3a.bucket.<bucket>.*` key.
+ */
+ def extractObjectStoreOptions(
+ hadoopConf: Configuration,
+ uris: Iterable[URI]): Map[String, String] = {
+ val s3CompliantSchemes = resolveS3CompliantSchemes(hadoopConf)
+ val representatives = scala.collection.mutable.LinkedHashMap[(String,
Option[String]), URI]()
+ uris.foreach { uri =>
+ val scheme = lowerScheme(uri).getOrElse("file")
+ val aliasBucket =
+ if (s3CompliantSchemes.contains(scheme)) bucketForUri(uri,
s3CompliantSchemes) else None
+ representatives.getOrElseUpdate((scheme, aliasBucket), uri)
+ }
+ val parts = representatives.values.toSeq.map { uri =>
+ val (options, aliasOptions) =
objectStoreOptionsAndAliasOptions(hadoopConf, uri)
Review Comment:
[P2] Honor libhdfs routing before translating alias options. With
`fs.comet.s3Compliant.schemes=blob` and `fs.comet.libhdfs.schemes=hdfs,blob`, a
scan over `s3a://bucket/...` and `blob://bucket/...` passes the fallback
checks. However, this call translates `fs.blob.default.endpoint` into
`fs.s3a.bucket.bucket.endpoint`, which overrides the native S3 store’s own
`fs.s3a.endpoint`. The S3 files are consequently fetched from the alias
endpoint, causing failed reads or wrong rows when that endpoint contains the
same key. This regresses previously safe Parquet scans with separate file
partitions and an S3 file first. Please translate alias options only when that
scheme actually routes to native S3.
Evidence: Compiled the exact-head helper and the base extractor from
f80042f78715e5a009352f1582f6c246a5f96131. Using synthetic endpoints
`fs.s3a.endpoint=http://127.0.0.1:19001` and
`fs.blob.default.endpoint=http://127.0.0.1:19002`, the store keys were
`s3://bucket` and `blob://bucket (libhdfs)`, with fallback=None. Base
first-file extraction selected endpoint 19001. Head union extraction selected
endpoint 19002 in both URI orders. The exact native `get_config` function
independently confirmed that the translated per-bucket value wins. No network
requests were performed. Probe source and output:
`/tmp/comet6753-e909-review-gnwf8u52/probe/AliasBackendProbe.scala` and
`/tmp/comet6753-e909-review-gnwf8u52/base-head-options.log`.
--
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]
