peter-toth commented on code in PR #57437:
URL: https://github.com/apache/spark/pull/57437#discussion_r3690061638
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/v2/PushDownUtils.scala:
##########
@@ -434,6 +434,9 @@ object PushDownUtils extends Logging {
runtimeFilters: Seq[Expression],
partitionFields: Seq[PartitionPredicateField]):
Seq[PartitionPredicateImpl] = {
val catalystExprs = runtimeFilters.flatMap {
+ case DynamicPruningExpression(in: InSubqueryExec) if
in.isResultUnavailable =>
+ None
+ case DynamicPruningExpression(Literal.TrueLiteral) => None
Review Comment:
**Finding 5.** This line changes behaviour on the pre-existing, default-on
iterative V2 pushdown path (SPARK-55596), independently of the new feature.
(The `isResultUnavailable` case above it is essential — without it
`translateRuntimeFilterV2` would hit `values().getOrElse { throw internalError
}`. And the `case TrueLiteral => None` you added in `translateRuntimeFilterV2`
is behaviour-neutral, since base's `case other` already returned `None` after a
warning. This one is different: it actually changes what gets pushed.)
`DynamicPruningExpression(Literal.TrueLiteral)` is what
`PlanDynamicPruningFilters`/`PlanAdaptiveDynamicPruningFilters` emit whenever
`onlyInBroadcast` holds and no broadcast can be reused — the default with
`spark.sql.optimizer.dynamicPartitionPruning.reuseBroadcastOnly=true`. On base
it survives all the way into a pushed predicate:
- `pushRuntimeFilters` keeps it as a candidate: `translateRuntimeFilterV2`
returns `None` for it, and `TrueLiteral.references` is empty so
`f.references.subsetOf(filterAttrs)` is trivially true;
- `createRuntimePartitionPredicates` maps it to `Some(TrueLiteral)` through
`case DynamicPruningExpression(e)`;
- in `createPartitionPredicates`, `getPartitionFiltersAndDataFilters` first
puts it in `dataFilters` (it requires `references.nonEmpty`), but then
`extractPredicatesWithinOutputSet` pulls it straight back as an
`extraPartitionFilter` — `predicates.scala:266-271` returns `Some(other)` when
`other.references.subsetOf(outputSet)`, which an empty reference set always
satisfies;
- `isPushablePartitionFilter` accepts it (deterministic, no subquery, no
`PythonUDF`), and `PartitionPredicateImpl.apply` returns `Some` because it only
rejects *unmatched* references (`PartitionPredicateImpl.scala:105-117`).
So today a degenerate DPP filter pushes a `PartitionPredicate` wrapping
`true` down to the connector and sets `partPredicatesPushed = true`, which
makes `replanWithRuntimeFilters` re-run `scan.toBatch.planInputPartitions()`
for no gain. Dropping it is a real improvement — but it is a silent change to
released, default-on behaviour, with no test and nothing in the description:
the same shape as the direct-reuse tightening @dongjoon-hyun asked you to
separate out.
`DataSourceV2EnhancedRuntimePartitionFilterSuite` already has everything
needed to pin it — a DPP query with `reuseBroadcastOnly=true` and no reusable
broadcast, asserting `assertPushedPartitionPredicates(df, expectedCount = 0)`.
Otherwise, please call it out in the description, or split it out (arguably
under its own ticket, since it is a SPARK-55596 bug rather than part of
SPARK-58265).
##########
sql/core/src/main/scala/org/apache/spark/sql/execution/ProjectedBroadcastValueSubqueryExec.scala:
##########
@@ -0,0 +1,136 @@
+/*
+ * 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.SparkException
+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(
Review Comment:
**Finding 4.** This new `BaseSubqueryExec` subtype needs to be registered in
`SQLLastAttemptAccumulator`, which matches the subtype set exhaustively and
deliberately bails out on anything it doesn't know
(`SQLLastAttemptAccumulator.scala:365-384`, untouched by this PR):
```scala
case sl: BaseSubqueryExec => sl match {
case s: SubqueryExec => scopeIds(s.child)
case _: SubqueryBroadcastExec =>
// Used by DPP filter only, not part of main flow of query execution.
Nil
case _: SubqueryAdaptiveBroadcastExec =>
// Used by DPP filter only.
Nil
case r: ReusedSubqueryExec => recurse(r.child)
case p =>
// Bail out if future unknown implementation is encountered.
bailOutReason = Some(s"Unsupported BaseSubqueryExec:
${p.getClass.getName}")
Nil
}
```
`extractStageRDDScopes` ends with `p.subqueries.flatMap(recurse)`, and
`AdaptiveSparkPlanHelper.flatMap` applies the function to the root node first
(`foreach(p)` calls `f(p)` before descending), so the
`ProjectedBroadcastValueSubqueryExec` itself falls into `case p` and sets
`bailOutReason`. Once that is set, `extractStageRDDScopes` returns `Left(...)`
and `lastAttemptValueForQueryExecution` — and `lastAttemptValueForDataset`
through it — returns `None` for the **whole query**, with an "Unable to extract
RDD scopes from query execution plan" warning.
`unexpectedLastAttemptMetricOperation` is called with `invalidate = false` and
no exception, so it only logs: it fails quietly and no existing test notices —
`SQLLastAttemptMetricPlanShapesSuite` covers `lastAttemptValueForDataset` per
plan shape but has no DPP shape.
Both DPP siblings are already listed there, so the existing
`SubqueryBroadcastExec`/`SubqueryAdaptiveBroadcastExec` paths keep working and
only the new node breaks it. Fix is the same shape as its neighbours:
```scala
case _: ProjectedBroadcastValueSubqueryExec =>
// Used by DPP filter only.
Nil
```
--
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]
