Copilot commented on code in PR #13109:
URL: https://github.com/apache/gluten/pull/13109#discussion_r4104933268
##########
gluten-substrait/src/main/scala/org/apache/gluten/execution/BatchScanExecTransformer.scala:
##########
@@ -80,6 +81,13 @@ case class BatchScanExecTransformer(
this.copy(pushDownFilters = Some(filters))
}
+ override def supportsMapKeyPruning: Boolean = true
+
+ override def withRequiredMapSubfields(
+ subfields: Map[String, Seq[SubfieldPath]]): BatchScanExecTransformerBase
= {
+ this.copy(requiredMapSubfields = subfields)
Review Comment:
`requiredMapSubfields` is dropped by `doCanonicalize`: this `copy` does not
pass the new field, so a BatchScan with `m["a"]` and one with `m["b"]` both
canonicalize with an empty declaration. Spark uses canonicalized plans for
exchange reuse, which can then reuse a scan/exchange materialized with the
wrong key set and return nulls for the other branch. Preserve
`requiredMapSubfields` in this copy so canonicalization and the custom equality
remain consistent.
##########
gluten-delta/src/test/scala/org/apache/gluten/execution/DeltaSuite.scala:
##########
@@ -632,7 +661,62 @@ abstract class DeltaSuite extends
WholeStageTransformerSuite {
}
}
- test("deletion vector on partitioned table") {
+ testWithMinSparkVersion("map-key pruning on delta scan with and without
deletion vector", "3.4") {
+ withTempPath {
+ p =>
+ val path = p.getCanonicalPath
+ spark
+ .range(0, 2000)
+ .selectExpr(
+ "id",
+ "map('a', named_struct('s', concat('v', cast(id % 7 as string)),
't', id), " +
+ "'b', named_struct('s', 'x', 't', id * 2), " +
+ "'c', named_struct('s', 'y', 't', id + 1)) as m"
+ )
+ .coalesce(1)
+ .write
+ .format("delta")
+ .save(path)
+ val sql = s"SELECT count(*), sum(m['a'].t), max(m['a'].s) " +
+ s"FROM delta.`$path` WHERE m['a'].s = 'v3'"
+ val pruningFlag =
"spark.gluten.sql.columnar.backend.velox.scanMapKeyPruningEnabled"
+
+ def scanSubfields(df: DataFrame): Map[String, Seq[String]] =
+ getExecutedPlan(df)
+ .collect { case scan: DeltaScanTransformer =>
scan.requiredMapSubfields }
+ .flatten
+ .toMap
+ .map { case (col, paths) => col -> paths.map(_.toString) }
+
+ withSQLConf(pruningFlag -> "true", "spark.sql.adaptive.enabled" ->
"false") {
+ runQueryAndCompare(sql) {
+ df =>
+ val subfields = scanSubfields(df)
+ assert(subfields.contains("m"), s"expected pruning on m, got
$subfields")
Review Comment:
This test is in the shared `DeltaSuite`, which is inherited by both
`VeloxDeltaSuite` and `BoltDeltaSuite`. The map-key pruning rule and Velox
config are only installed for Velox, so Bolt will reach this assertion with an
empty `requiredMapSubfields` and fail; move the test to a Velox-only suite or
gate the Velox-specific assertions/backend setup.
--
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]
