peter-toth commented on PR #59916: URL: https://github.com/apache/spark/pull/59916#issuecomment-6586702257
@dongjoon-hyun, thanks for the thorough review! The fixes are in [2420b70](https://github.com/apache/spark/pull/59916/commits/2420b70b2288e0cfced46a4f2d4eb293f15f40da): - 1 and 3: a key whose DPP filters are all dropped gets no bloom filter if its creation side is not deterministic, or if the key on the application side is not an attribute. The same problems without this PR are SPARK-60808 and SPARK-60809. - 6: in a subquery, every DPP filter counts as kept. The rule decides on such keys in its run on the whole plan. - 2: your SPARK-60796 fixed `prunePartitions` in the meantime, so this PR needs nothing for it. - The rest is answered in the threads. The temporary check from the description is attached below as a patch. It prints a line for every DPP filter that physical planning keeps on a key that also has a bloom filter. <details><summary>The temporary check</summary> Apply it on the PR head with `git apply`, run the suites that the description lists, and grep the output for `DPPBLOOM-REDUNDANT`. ```diff diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/PlanAdaptiveDynamicPruningFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/PlanAdaptiveDynamicPruningFilters.scala index 9374758d473..ab14eb56a4c 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/PlanAdaptiveDynamicPruningFilters.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/adaptive/PlanAdaptiveDynamicPruningFilters.scala @@ -112,6 +112,9 @@ case class PlanAdaptiveDynamicPruningFilters( } } + if (reusedBroadcast.isDefined || !onlyInBroadcast) { + org.apache.spark.sql.execution.dynamicpruning.DppBloomDebug.check(plan, value, "AQE") + } reusedBroadcast match { case Some(broadcastValues) => DynamicPruningExpression(InSubqueryExec(value, broadcastValues, exprId)) diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PlanDynamicPruningFilters.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PlanDynamicPruningFilters.scala index ca7407eb2a8..6040d3a94a4 100644 --- a/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PlanDynamicPruningFilters.scala +++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PlanDynamicPruningFilters.scala @@ -117,6 +117,9 @@ case class PlanDynamicPruningFilters(sparkSession: SparkSession) extends Rule[Sp } } + if (reusedBroadcast.isDefined || !onlyInBroadcast) { + DppBloomDebug.check(plan, value, "nonAQE") + } reusedBroadcast match { case Some(broadcastValues) => DynamicPruningExpression(InSubqueryExec(value, broadcastValues, exprId)) @@ -133,3 +136,18 @@ case class PlanDynamicPruningFilters(sparkSession: SparkSession) extends Rule[Sp } } } + +object DppBloomDebug { + def check(plan: SparkPlan, value: Expression, kind: String): Unit = { + val ids = value.references.map(_.exprId).toSet + val hasBloom = plan.exists(_.expressions.exists(_.exists { + case org.apache.spark.sql.catalyst.expressions.BloomFilterMightContain(_, + org.apache.spark.sql.catalyst.expressions.XxHash64(Seq(v), _)) => + v.references.exists(a => ids.contains(a.exprId)) + case _ => false + })) + // scalastyle:off println + if (hasBloom) println(s"DPPBLOOM-REDUNDANT[$kind] value=$value") + // scalastyle:on println + } +} ``` </details> -- 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]
