dwsmith1983 commented on code in PR #6838:
URL: https://github.com/apache/datafusion-comet/pull/6838#discussion_r4266268921
##########
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:
Split the helper so the table is written once: `writeParquet(prefix)` writes
it at one shared path, and `applyScanRule(path, conf, readOptions)` plans the
read and applies the rule. The `fake://` and `hdfs://` tests now use both, and
the `dir` parameter is gone. The three new tests are one table-driven test over
(conf, reader options, claimed), and each case's name is in its failure
messages.
--
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]
