dongjoon-hyun commented on code in PR #57437:
URL: https://github.com/apache/spark/pull/57437#discussion_r3635760462
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/dynamicpruning/PlanDynamicPruningFilters.scala:
##########
@@ -46,47 +46,86 @@ case class PlanDynamicPruningFilters(sparkSession:
SparkSession) extends Rule[Sp
HashedRelationBroadcastMode(packedKeys)
}
+ private def reusableBroadcast(
Review Comment:
This tightens the *existing* direct reuse path, independently of the new
feature: the old code only checked `left.sameResult(sparkPlan)`, while the new
code additionally requires broadcast mode equality and excludes
`isNullAwareAntiJoin` joins.
With the default
`spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly=true`, an edge
case that previously got DPP (at the cost of planning an extra broadcast that
`ReuseExchange` couldn't dedup, e.g. when key order differs or the only
matching join is a null-aware anti join) now falls back to
`Literal.TrueLiteral`, i.e. loses DPP entirely.
Making the non-AQE path consistent with the AQE path (which already compares
the whole exchange including its mode) is arguably an improvement, but it is a
semantic change to released behavior that ships silently inside this feature
PR. Could you either call this out explicitly in the PR description (ideally
with a test pinning the new behavior), or split it into a separate commit/PR so
it can be evaluated on its own?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/ReusableBroadcastValueProjection.scala:
##########
@@ -0,0 +1,126 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.catalyst.optimizer
+
+import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.catalyst.plans.Inner
+import org.apache.spark.sql.catalyst.plans.logical._
+import org.apache.spark.sql.catalyst.trees.TreePattern.{CTE,
DYNAMIC_PRUNING_SUBQUERY}
+import org.apache.spark.sql.types.{DateType, StringType, TimestampNTZType,
TimestampType}
+
+/**
+ * Finds an existing broadcast whose stored rows provide a safe superset of a
pruning domain.
+ */
+private[sql] object ReusableBroadcastValueProjection extends PredicateHelper {
+
+ private def isSafeValueExpression(expression: Expression): Boolean = {
Review Comment:
The `isSafeValueExpression` whitelist seems inconsistent with the motivating
example in the PR description. The `Attribute` case requires
`UnsafeRow.isFixedLength(attribute.dataType)`, which rejects variable-length
types like `StringType` — yet the example in the description projects a string
column (`category`) from the broadcast. As written, that example query would
not benefit from this optimization.
Also, the `DateFormatClass` case hardcodes the `'yyyy-MM-dd'` format string,
which looks tailored to the test queries rather than a principled rule.
Could you clarify:
1. Why is fixed-length required for plain attribute projections? Output size
is already bounded by `maxOutputBytes`, so it doesn't seem to be a memory
concern, and correctness shouldn't depend on the value width.
2. If the restriction is intentional, please update the PR description so
the example matches what the whitelist actually supports; if not, consider
generalizing the whitelist (at least plain string attributes) and documenting
the criteria for extending it.
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/subquery.scala:
##########
@@ -127,23 +127,33 @@ case class InSubqueryExec(
override def nullable: Boolean = child.nullable
override def toString: String = s"$child IN ${plan.name}"
- override def withNewPlan(plan: BaseSubqueryExec): InSubqueryExec = copy(plan
= plan)
+ override def withNewPlan(plan: BaseSubqueryExec): InSubqueryExec =
+ copy(plan = plan, resultBroadcast = null, result = null)
final override def nodePatternsInternal(): Seq[TreePattern] =
Seq(IN_SUBQUERY_EXEC)
def updateResult(): Unit = {
- val rows = plan.executeCollect()
- result = if (plan.output.length > 1) {
+ val (rows, unavailable) =
ProjectedBroadcastValueSubqueryExec.resultOf(plan) match {
+ case Some(BroadcastValueResult.Available(values)) => (values, false)
+ case Some(BroadcastValueResult.Unavailable) =>
(Array.empty[InternalRow], true)
+ case None => (plan.executeCollect(), false)
+ }
+ result = if (unavailable) {
Review Comment:
Since `InSubqueryExec` is also used a general `IN` subquery, shall we
`assert(isDynamicPruning)` here?
```suggestion
result = if (unavailable) {
assert(isDynamicPruning,
"An unavailable broadcast value projection result is only allowed
for dynamic " +
"pruning, where the filter can safely fail open to true.")
```
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/ProjectedBroadcastValueSubqueryExec.scala:
##########
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution
+
+import java.util.concurrent.{Future => JFuture}
+
+import scala.concurrent.duration.Duration
+
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, Cast, Expression, NamedExpression, UnsafeRow}
+import org.apache.spark.sql.catalyst.plans.QueryPlan
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.BroadcastValueResult.{Available,
Unavailable}
+import org.apache.spark.sql.execution.joins.HashedRelation
+import org.apache.spark.sql.execution.metric.SQLMetrics
+import org.apache.spark.util.ThreadUtils
+
+/** Collects pruning values from the full rows of an already required hash
broadcast. */
+case class ProjectedBroadcastValueSubqueryExec(
+ name: String,
+ valueExpression: Expression,
+ child: SparkPlan) extends BaseSubqueryExec with UnaryExecNode {
+
+ override def output: Seq[Attribute] = {
+ val outputName = valueExpression match {
+ case named: NamedExpression => named.name
+ case Cast(named: NamedExpression, _, _, _) => named.name
+ case _ => "key"
+ }
+ Seq(AttributeReference(outputName, valueExpression.dataType,
valueExpression.nullable)())
+ }
+
+ override lazy val metrics = Map(
+ "numInputRows" -> SQLMetrics.createMetric(sparkContext, "number of input
rows"),
+ "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output
rows"),
+ "dataSize" -> SQLMetrics.createMetric(sparkContext, "data size (bytes)"),
+ "projectionDisabled" -> SQLMetrics.createMetric(sparkContext, "projection
disabled"),
+ "projectionErrors" -> SQLMetrics.createMetric(sparkContext, "projection
errors"),
+ "collectTime" -> SQLMetrics.createMetric(sparkContext, "time to collect
(ms)"))
+
+ override def doCanonicalize(): SparkPlan = {
+ ProjectedBroadcastValueSubqueryExec(
+ "dpp",
+ QueryPlan.normalizeExpressions(valueExpression, child.output),
+ child.canonicalized)
+ }
+
+ @transient
+ private lazy val relationFuture:
JFuture[BroadcastValueResult[Array[InternalRow]]] = {
+ val executionId =
sparkContext.getLocalProperty(SQLExecution.EXECUTION_ID_KEY)
+
SQLExecution.withThreadLocalCaptured[BroadcastValueResult[Array[InternalRow]]](
+ session, SubqueryBroadcastExec.executionContext) {
+ SQLExecution.withExecutionId(session, executionId) {
+ val beforeCollect = System.nanoTime()
+ val broadcast = child.executeBroadcast[HashedRelation]()
+ val limits = BroadcastValueProjectionLimits(
+ maxInputRows =
conf.dynamicPartitionPruningBroadcastProjectionMaxRows.toLong,
+ maxOutputBytes =
conf.dynamicPartitionPruningBroadcastProjectionMaxBytes,
+ maxSourceBytes =
conf.dynamicPartitionPruningBroadcastProjectionMaxSourceBytes)
+ val result = BroadcastValueProjector.collectExactValueDomain(
+ broadcast,
+ child,
+ valueExpression,
+ limits,
+ onInputRow = () => longMetric("numInputRows") += 1,
+ onError = () => longMetric("projectionErrors") += 1)
+
+ longMetric("collectTime") += (System.nanoTime() - beforeCollect) /
1000000
+ result match {
+ case Available(rows) =>
+ longMetric("numOutputRows") += rows.length
+ longMetric("dataSize") +=
+
rows.iterator.map(_.asInstanceOf[UnsafeRow].getSizeInBytes.toLong).sum
+ case Unavailable =>
+ longMetric("projectionDisabled") += 1
+ }
+ SQLMetrics.postDriverMetricUpdates(sparkContext, executionId,
metrics.values.toSeq)
+ result
+ }
+ }
+ }
+
+ override protected def doPrepare(): Unit = {
+ relationFuture
+ }
+
+ override protected def doExecute(): RDD[InternalRow] = {
+ throw QueryExecutionErrors.executeCodePathUnsupportedError(
+ "ProjectedBroadcastValueSubqueryExec")
+ }
+
+ override def executeCollect(): Array[InternalRow] = executeCollectResult()
match {
+ case Available(rows) => rows
+ case Unavailable => Array.empty
Review Comment:
`executeCollect()` maps `Unavailable` to an **empty array**. The whole
design rests on "an unavailable domain is never interpreted as an empty
domain", but that invariant is only upheld because
`InSubqueryExec.updateResult` goes through
`ProjectedBroadcastValueSubqueryExec.resultOf`. Any future caller that invokes
`executeCollect()` directly would read "empty domain" and prune *every*
partition — a silent correctness bug. Consider throwing from `executeCollect()`
instead (like `doExecute()` does), or at minimum adding a prominent comment
stating that callers must use `resultOf`/`executeCollectResult()`?
--
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]
