peter-toth commented on code in PR #59916:
URL: https://github.com/apache/spark/pull/59916#discussion_r4484822245
##########
sql/core/src/test/scala/org/apache/spark/sql/InjectRuntimeFilterSuite.scala:
##########
@@ -302,6 +304,66 @@ class InjectRuntimeFilterSuite extends SharedSparkSession
checkWithAndWithoutFeatureEnabled(query, shouldReplace = false)
}
+ // Runs `f` with the thresholds of the DPP tests, `reuseBroadcastOnly` on,
and `confs`, with AQE
+ // on and off unless `confs` sets it.
+ def withDppTestConf(confs: (String, String)*)(f: => Unit): Unit = {
+ val adaptiveModes = confs.collectFirst {
+ case (SQLConf.ADAPTIVE_EXECUTION_ENABLED.key, adaptive) => Seq(adaptive)
+ }.getOrElse(Seq("true", "false"))
+ adaptiveModes.foreach { adaptive =>
+ withSQLConf(Seq(
+
SQLConf.RUNTIME_BLOOM_FILTER_APPLICATION_SIDE_SCAN_SIZE_THRESHOLD.key -> "3000",
+ SQLConf.AUTO_BROADCASTJOIN_THRESHOLD.key -> "2000",
+ SQLConf.DYNAMIC_PARTITION_PRUNING_REUSE_BROADCAST_ONLY.key -> "true",
+ SQLConf.ADAPTIVE_EXECUTION_ENABLED.key -> adaptive) ++ confs: _*) {
+ withClue(s"adaptive=$adaptive: ")(f)
+ }
+ }
+ }
+
+ // Runs `query` and returns the DPP filters that physical planning left on
the file source scans:
+ // the class of the subquery that each filter reads, or the filter itself if
it reads none.
+ def dynamicPruningFilters(query: String): Seq[Any] = {
+ // Keep the scans in the final plan even if they return no rows.
+ withSQLConf(
+ SQLConf.ADAPTIVE_OPTIMIZER_EXCLUDED_RULES.key ->
AQEPropagateEmptyRelation.ruleName) {
+ val df = sql(query)
+ df.collect()
+ collect(df.queryExecution.executedPlan) {
+ case scan: FileSourceScanExec => scan.partitionFilters.collect {
+ case DynamicPruningExpression(in: InSubqueryExec) => in.plan.getClass
+ case DynamicPruningExpression(filter) => filter
+ }
+ }.flatten
+ }
+ }
+
+ // Asserts that `query` gets a bloom filter, and that physical planning
plans its DPP filter as
+ // `true`.
+ def assertRewroteWithBloomFilterForDroppedDpp(query: String): Unit = {
+ assertRewroteWithBloomFilter(query)
+ assert(dynamicPruningFilters(query) == Seq(Literal.TrueLiteral))
+ }
+
+ // The keys that the bloom filters of `query` check.
+ def bloomFilterKeys(query: String): Seq[Expression] =
+ sql(query).queryExecution.optimizedPlan.flatMap(_.expressions)
+ .flatMap(_.collect { case BloomFilterMightContain(_, XxHash64(Seq(key),
_)) => key })
+
+ // The names of the columns that the bloom filters of `query` check.
+ def bloomFilterKeyNames(query: String): Seq[String] =
+ bloomFilterKeys(query).flatMap(_.references.map(_.name))
+
+ // Asserts that `query` gets `numBloomFilters` bloom filters without DPP, so
that a test that sees
+ // fewer with DPP shows that the rule kept them out. In the tests for a key
or a creation side
+ // that is not safe, that is master's bloom filter for a key without a DPP
filter, which can fail
+ // the query or drop rows there: SPARK-58486 (a collated key), SPARK-60808
(a creation side that
Review Comment:
Added SPARK-60842.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,101 @@ object InjectRuntimeFilter extends Rule[LogicalPlan]
with PredicateHelper with J
}
}
- // This checks if there is already a DPP filter, as this rule is called just
after DPP.
- @tailrec
- private def hasDynamicPruningSubquery(
- left: LogicalPlan,
- right: LogicalPlan,
- leftKey: Expression,
- rightKey: Expression): Boolean = {
- (left, right) match {
- case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan), _) =>
- pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan,
right, leftKey, rightKey)
- case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan)) =>
- pruningKey.fastEquals(rightKey) ||
- hasDynamicPruningSubquery(left, plan, leftKey, rightKey)
- case _ => false
+ // The plans that a broadcast hash join of `plan` could broadcast, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // unless the statistics change after this rule. It takes each side without
the DPP filters on top
+ // of it, since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // Whether evaluating `e` cannot fail, also on rows that the join does not
read. It asks
+ // `ExprUtils.canEvaluateUnconditionally`, after it replaces two wrappers
that cannot fail either,
+ // over a child that cannot fail: a numeric up-cast, and the normalization
that
+ // `NormalizeFloatingNumbers` puts around a floating point join key. Each is
checked against its
+ // own child before anything is replaced, and its stand-in keeps its type.
An `onlyInBroadcast`
+ // DPP filter counts too, since it finds no broadcast in the bloom filter's
subquery and is `true`
+ // there.
+ private def cannotFail(e: Expression): Boolean = e match {
+ case dpp: DynamicPruningSubquery => dpp.onlyInBroadcast
+ case _ => ExprUtils.canEvaluateUnconditionally(e.transformDown {
+ case c: Cast if isNumericUpCast(c) && cannotFail(c.child) =>
+ AttributeReference("upCast", c.dataType, c.nullable)()
+ case n @ KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) if
cannotFail(child) =>
+ AttributeReference("normalized", n.dataType, n.nullable)()
+ })
+ }
+
+ // `canUpCast` also takes `DECIMAL(19,0)` to `BIGINT`, which can overflow.
So a decimal to an
+ // integral type does not count.
+ private def isNumericUpCast(c: Cast): Boolean =
+ c.child.dataType.isInstanceOf[NumericType] &&
c.dataType.isInstanceOf[NumericType] &&
+ Cast.canUpCast(c.child.dataType, c.dataType) &&
+ !(c.child.dataType.isInstanceOf[DecimalType] &&
c.dataType.isInstanceOf[IntegralType])
+
+ // Whether the bloom filter's subquery can evaluate `creationSide` again,
apart from the join. The
+ // plan it reads must give the same rows again (SPARK-60808). So it must
read no CTE, whose
+ // reference hides the determinism of its plan, and a cache only if Spark
can tell that scanning
+ // it again gives the same rows. It must not read a file metadata column
either, such as
+ // `_metadata.file_block_start`, which depends on how the scan splits its
files. The subquery can
+ // split them differently. Its filters, and the projections that the key and
the filters need,
+ // must pass `cannotFail`, which takes only expressions that are
deterministic in the subquery
+ // and do not fail there. A part of the join side outside that plan can only
drop rows from the
+ // join, so the bloom filter still holds every key that the join sees.
+ private def canEvaluateAgain(creationSide: FilterCreationSide): Boolean = {
+ val evaluated =
+ ColumnPruning(Project(Seq(Alias(creationSide.key, "key")()),
creationSide.plan))
+ !evaluated.exists { node =>
+ node.references.exists {
+ case FileSourceMetadataAttribute(_) => true
Review Comment:
I kept the wider scope, and the Known limits now say that any `_metadata`
read keeps the bloom filter out.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:
##########
@@ -379,23 +378,101 @@ object InjectRuntimeFilter extends Rule[LogicalPlan]
with PredicateHelper with J
}
}
- // This checks if there is already a DPP filter, as this rule is called just
after DPP.
- @tailrec
- private def hasDynamicPruningSubquery(
- left: LogicalPlan,
- right: LogicalPlan,
- leftKey: Expression,
- rightKey: Expression): Boolean = {
- (left, right) match {
- case (Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan), _) =>
- pruningKey.fastEquals(leftKey) || hasDynamicPruningSubquery(plan,
right, leftKey, rightKey)
- case (_, Filter(DynamicPruningSubquery(pruningKey, _, _, _, _, _, _),
plan)) =>
- pruningKey.fastEquals(rightKey) ||
- hasDynamicPruningSubquery(left, plan, leftKey, rightKey)
- case _ => false
+ // The plans that a broadcast hash join of `plan` could broadcast, as
+ // `getBroadcastHashJoinBuildSide` picks them, but no null-aware anti join
under AQE. The list can
+ // hold more plans than physical planning broadcasts, and no fewer that a
DPP filter can reuse,
+ // unless the statistics change after this rule. It takes each side without
the DPP filters on top
+ // of it, since cleanup can still drop them.
+ private def broadcastablePlans(plan: LogicalPlan): Seq[LogicalPlan] =
plan.flatMap {
+ // AQE plans a DPP filter as `true` even if a null-aware anti join
broadcasts its build plan.
+ // Without AQE, physical planning keeps such a filter, though the filter
builds its own
+ // broadcast.
+ case ExtractSingleColumnNullAwareAntiJoin(_, _) if
conf.adaptiveExecutionEnabled => Nil
+ case join: Join =>
+ getBroadcastHashJoinBuildSide(join, conf).map {
+ case BuildLeft => withoutDppFilters(join.left)
+ case BuildRight => withoutDppFilters(join.right)
+ }
+ case _ => Nil
+ }
+
+ // `plan` without the DPP filters on top of it that `drop` picks, by default
all of them.
+ private def withoutDppFilters(
+ plan: LogicalPlan,
+ drop: DynamicPruningSubquery => Boolean = _ => true): LogicalPlan = plan
match {
+ case f @ Filter(dpp: DynamicPruningSubquery, child) =>
+ val newChild = withoutDppFilters(child, drop)
+ if (drop(dpp)) newChild else f.withNewChildren(Seq(newChild))
+ case _ => plan
+ }
+
+ // Whether evaluating `e` cannot fail, also on rows that the join does not
read. It asks
+ // `ExprUtils.canEvaluateUnconditionally`, after it replaces two wrappers
that cannot fail either,
+ // over a child that cannot fail: a numeric up-cast, and the normalization
that
+ // `NormalizeFloatingNumbers` puts around a floating point join key. Each is
checked against its
+ // own child before anything is replaced, and its stand-in keeps its type.
An `onlyInBroadcast`
+ // DPP filter counts too, since it finds no broadcast in the bloom filter's
subquery and is `true`
+ // there.
+ private def cannotFail(e: Expression): Boolean = e match {
+ case dpp: DynamicPruningSubquery => dpp.onlyInBroadcast
+ case _ => ExprUtils.canEvaluateUnconditionally(e.transformDown {
+ case c: Cast if isNumericUpCast(c) && cannotFail(c.child) =>
+ AttributeReference("upCast", c.dataType, c.nullable)()
+ case n @ KnownFloatingPointNormalized(NormalizeNaNAndZero(child)) if
cannotFail(child) =>
+ AttributeReference("normalized", n.dataType, n.nullable)()
+ })
+ }
+
+ // `canUpCast` also takes `DECIMAL(19,0)` to `BIGINT`, which can overflow.
So a decimal to an
+ // integral type does not count.
+ private def isNumericUpCast(c: Cast): Boolean =
+ c.child.dataType.isInstanceOf[NumericType] &&
c.dataType.isInstanceOf[NumericType] &&
+ Cast.canUpCast(c.child.dataType, c.dataType) &&
+ !(c.child.dataType.isInstanceOf[DecimalType] &&
c.dataType.isInstanceOf[IntegralType])
+
+ // Whether the bloom filter's subquery can evaluate `creationSide` again,
apart from the join. The
+ // plan it reads must give the same rows again (SPARK-60808). So it must
read no CTE, whose
+ // reference hides the determinism of its plan, and a cache only if Spark
can tell that scanning
+ // it again gives the same rows. It must not read a file metadata column
either, such as
+ // `_metadata.file_block_start`, which depends on how the scan splits its
files. The subquery can
+ // split them differently. Its filters, and the projections that the key and
the filters need,
+ // must pass `cannotFail`, which takes only expressions that are
deterministic in the subquery
+ // and do not fail there. A part of the join side outside that plan can only
drop rows from the
+ // join, so the bloom filter still holds every key that the join sees.
+ private def canEvaluateAgain(creationSide: FilterCreationSide): Boolean = {
+ val evaluated =
+ ColumnPruning(Project(Seq(Alias(creationSide.key, "key")()),
creationSide.plan))
+ !evaluated.exists { node =>
Review Comment:
Done.
--
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]
