comphead commented on code in PR #6838:
URL: https://github.com/apache/datafusion-comet/pull/6838#discussion_r4265451755
##########
spark/src/main/scala/org/apache/comet/objectstore/NativeConfig.scala:
##########
@@ -123,6 +123,15 @@ object NativeConfig {
}
}
+ /**
+ * The schemes native reads through libhdfs: the `fs.comet.libhdfs.schemes`
list, or only `hdfs`
+ * when it is unset or blank, as native `is_hdfs_scheme` decides.
+ */
+ def resolveLibhdfsSchemes(hadoopConf: Configuration): Set[String] = {
Review Comment:
This is public, but its siblings `parseSchemeSet` and
`resolveS3CompliantSchemes` are `private[comet]`, and `DeltaScanSupport`
already calls `resolveS3CompliantSchemes` from `org.apache.comet.contrib.delta`
under that same modifier. Only `resolveS3CompliantSchemesAsJava` is public, and
that is because a Java caller needs the forwarder. Could this be
`private[comet]` so it does not become API?
##########
contrib/delta-spark/src/test/scala/org/apache/comet/contrib/delta/DeltaScanContribSuite.scala:
##########
@@ -252,11 +254,71 @@ class DeltaScanContribSuite extends CometDeltaTestBase {
assert(DeltaScanSupport.s3CompliantAliasSchemeReason(notOptedIn,
Seq(blobUri)).isEmpty)
}
- test("libhdfsSchemes parses the list exactly like core's scan gate (trim,
lowercase, blanks)") {
- withSQLConf(CometConf.COMET_LIBHDFS_SCHEMES.key -> " HDFS , viewfs ,, ") {
- assert(DeltaScanSupport.libhdfsSchemes == Set("hdfs", "viewfs"))
+ test("scheme gate claims a scheme listed in the Hadoop conf's
libhdfs.schemes") {
Review Comment:
Each of these three tests writes a fresh Delta table and starts a new plan,
and they repeat the same three cases as the core suite. The delta suites run in
a separate job, so I would write the table once and plan the three variants in
one test. The `assert(!isNativelyReadableScheme(fake))` precondition is
repeated in two tests and is already the subject of the core suite's `fake://`
test. The padded, mixed-case value `" FAKE , hdfs ,, "` re-tests parsing that
`NativeConfigSuite` already pins for `resolveLibhdfsSchemes`, so a plain
`fake,hdfs` would keep this test about the gate reading the Hadoop conf.
##########
spark/src/test/scala/org/apache/comet/rules/CometScanSchemeFallbackSuite.scala:
##########
@@ -389,4 +389,68 @@ class CometScanSchemeFallbackSuite extends CometTestBase
with CometIcebergTestBa
assert(sparkScans.isEmpty, s"expected no leftover Spark
FileSourceScanExec:\n$transformed")
}
}
+
+ test("native scan claims a scheme listed in the Hadoop conf's
libhdfs.schemes") {
+ // The bare key set in SQLConf reaches the scan's Hadoop conf, where
native reads the list.
+ val transformed = applyScanRuleToFakeScan(
+ "libhdfs-hadoop",
+ Seq(CometConf.COMET_LIBHDFS_SCHEMES_KEY -> "fake,hdfs"))
+ val cometScans = transformed.collect { case s: CometScanExec => s }
+ assert(
+ cometScans.size == 1,
+ s"`fake` is in the Hadoop conf's libhdfs list; Comet must claim the
scan:\n$transformed")
+ }
+
+ test("runtime SET of the spark.hadoop. key does not claim a scheme native
can't route") {
+ // newHadoopConf copies SQLConf keys with their `spark.hadoop.` prefix, so
the list never
+ // reaches the Hadoop conf native reads and `fake` is still not routed
through libhdfs.
+ val transformed = applyScanRuleToFakeScan(
+ "libhdfs-sqlconf",
+ Seq(CometConf.COMET_LIBHDFS_SCHEMES.key -> "fake,hdfs"))
+ val cometScans = transformed.collect { case s: CometScanExec => s }
+ assert(
+ cometScans.isEmpty,
+ s"native would not route `fake` through libhdfs; the scan must fall
back:\n$transformed")
+ val reasons = new ExtendedExplainInfo().getFallbackReasons(transformed)
+ assert(
+ reasons.exists(_.contains("Unsupported filesystem schemes: fake")),
+ s"the scan must be declined for its scheme, got reasons: $reasons")
+ }
+
+ test("native scan claims a scheme listed only in a reader option's
libhdfs.schemes") {
+ // A reader option reaches the scan's Hadoop conf but never SQLConf.
+ val transformed = applyScanRuleToFakeScan(
+ "libhdfs-option",
+ readOptions = Map(CometConf.COMET_LIBHDFS_SCHEMES_KEY -> "fake,hdfs"))
+ val cometScans = transformed.collect { case s: CometScanExec => s }
+ assert(
+ cometScans.size == 1,
+ s"`fake` is in the scan's libhdfs list; Comet must claim the
scan:\n$transformed")
+ }
+
+ /**
+ * Writes a Parquet table under `fake://`, plans a read of it with
`readOptions` and Comet off,
+ * and applies `CometScanRule` to that plan with the native scan enabled and
`libhdfsConf` set.
+ */
+ private def applyScanRuleToFakeScan(
Review Comment:
This helper does the same write, plan, `CometScanRule` sequence that `native
scan declines a filesystem scheme object_store can't read (fake://)` and the
`hdfs://` unset test in this file still spell out inline. The existing fake
test is exactly `applyScanRuleToFakeScan("data")` with no conf, so it can use
the helper, and the `hdfs://` one only needs the path prefix as a parameter.
Also, the three new tests differ only in (conf, reader options, expected
claim), and each rewrites the same 10-row Parquet table. One table-driven test
over those three cases would write once. The `dir` parameter is not needed
either, since `SaveMode.Overwrite` already allows one shared path.
--
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]
