LuciferYang commented on code in PR #7476:
URL: https://github.com/apache/kyuubi/pull/7476#discussion_r3809717861
##########
extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/HiveCatalogSuite.scala:
##########
@@ -554,4 +553,40 @@ class HiveCatalogSuite extends KyuubiHiveTest {
}
}
}
+
+ test("KyuubiParquetScan and KyuubiOrcScan hadoopConf snapshots at catalog
first use") {
+ // Unlike Spark's built-in ScanBuilders rebuilt per query planning from
+ // the current sqlConf, KSHC clones a `lazy val` snapshot taken at the
+ // catalog's first use, so confs SET after that snapshot cannot reach
+ // readers that consult the Hadoop `Configuration` directly.
+ val fieldIdRead = SQLConf.PARQUET_FIELD_ID_READ_ENABLED.key
+ val fieldIdIgnoreMissing = SQLConf.IGNORE_MISSING_PARQUET_FIELD_ID.key
+
+ Seq("orc", "parquet").foreach { provider =>
+ withSparkSession() { spark =>
+ // `catalog` is built in `beforeEach`, its `hadoopConf` snapshot is
+ // already frozen by the time this test runs, so any conf we SET below
+ // is "mid-session" from its perspective.
+ val props = new util.HashMap[String, String]()
+ props.put(TableCatalog.PROP_PROVIDER, provider)
+
+ try {
+ val table = catalog.createTable(testIdent, schema,
Array.empty[Transform], props)
+ spark.sessionState.conf.setConfString(fieldIdRead, "true")
+ spark.sessionState.conf.setConfString(fieldIdIgnoreMissing, "true")
+
+ val hadoopConf = table.asInstanceOf[HiveTable]
+ .newScanBuilder(CaseInsensitiveStringMap.empty()).build() match {
+ case s: KyuubiParquetScan => s.hadoopConf
+ case s: KyuubiOrcScan => s.hadoopConf
+ case other => fail(s"unexpected scan type:
${other.getClass.getName}")
+ }
+ assert(hadoopConf.get(fieldIdRead) != "true")
Review Comment:
`assert(hadoopConf.get(fieldIdRead) != "true")` is trivially true:
`Configuration.get` returns `null` for an absent key, and `null != "true"`
always holds, so the assertion cannot tell "never arrived" from "arrived with a
different value". `== null` is the assertion you actually mean, and it
documents the snapshot semantics on the spot.
It would also pass with an empty `hadoopConf`, so a positive assertion on a
catalog option would close that — but it has to use the lowercased key
(`javax.jdo.option.connectionurl`), since `catalogOptions` is a
`CaseInsensitiveStringMap` and the overlay writes lowercase; the camelCase
spelling also arrives from the SparkConf via `getAllConfs`, so asserting that
one proves nothing about the overlay.
The test also cements "a conf SET after the snapshot cannot reach the
reader" as a contract, which does not hold for Spark's own builder on the same
path. Honoring per-catalog settings does not require freezing session confs:
add a non-cached `newScanHadoopConf()` on `HiveTableCatalog` (the same steps as
today's `hadoopConf`, token signature included), have both ScanBuilders clone
from it, and leave the catalog's own `lazy val` alone. Could you confirm the
freeze is deliberate, or switch to the overlay and flip this assertion with it?
##########
extensions/spark/kyuubi-spark-connector-hive/src/test/scala/org/apache/kyuubi/spark/connector/hive/DynamicPartitionPruningSuite.scala:
##########
@@ -86,19 +90,51 @@ class DynamicPartitionPruningSuite extends KyuubiHiveTest {
| WHERE d.tag = 'target'
|""".stripMargin
+ val df = spark.sql(sql)
checkAnswer(
- spark.sql(sql),
+ df,
Seq(
Row(3, "c", "2026-05-01"),
Row(4, "d", "2026-05-01")))
// DPP being actually applied is observable as a
`DynamicPruningExpression`
// injected into `BatchScanExec.runtimeFilters`.
- val exec = findBatchScanExec(spark, sql, fact.split('.').last)
+ val exec = findBatchScanExec(df.queryExecution.executedPlan,
fact.split('.').last)
val hasDpp =
exec.runtimeFilters.exists(_.isInstanceOf[DynamicPruningExpression])
assert(hasDpp == enabled)
+
+ val planned = exec.scan.toBatch.planInputPartitions().length
+ plannedPartitions(enabled) = planned
+
+ exec.scan match {
+ case _: KyuubiOrcScan | _: KyuubiParquetScan =>
+ assert(exec.scan.columnarSupportMode() ==
Scan.ColumnarSupportMode.SUPPORTED)
Review Comment:
For the two Kyuubi scans the `columnarSupportMode()` assertions only cover
the `SUPPORTED` side (the `HiveScan` one asserts `UNSUPPORTED`, but that is a
constant override and pins no logic). Two things are untested: the
`UNSUPPORTED` branch when the vectorized reader is off, and the deliberate
divergence spelled out in the scaladoc — with every partition pruned by DPP,
Spark's default evaluates to false while this still returns `SUPPORTED`. CI
would catch neither logic rewritten to ignore the conf nor a future "cleanup"
of that divergence.
Both are worth adding, with two things to watch: the `UNSUPPORTED` case
needs `spark.sql.parquet.enableVectorizedReader` and
`spark.sql.orc.enableVectorizedReader` turned off separately, or `wholeStage`,
the one input both branches read; and the all-pruned case has to assert
`planInputPartitions().isEmpty` first, since `columnarSupportMode()` never
consults the partition list and the assertion would otherwise just repeat the
existing one.
--
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]
