This is an automated email from the ASF dual-hosted git repository.
gengliangwang pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new 33ccc3907a5a [SPARK-58428][SQL] Fix optimizer hang in DSv2 expression
pushdown
33ccc3907a5a is described below
commit 33ccc3907a5a88f8889a39d99599efd54e13c1ce
Author: AbhinavBattu <[email protected]>
AuthorDate: Sat Aug 22 15:39:32 2026 -0700
[SPARK-58428][SQL] Fix optimizer hang in DSv2 expression pushdown
### What changes were proposed in this pull request?
`V2ExpressionBuilder.generateExpression` handles a context independent
foldable
expression by constant folding it and recursing on the result, so that the
folded
value is translated by the `Literal` case:
```scala
case _ if expr.contextIndependentFoldable && <datasourceV2ExprFolding> =>
val constantExpr = ConstantFolding.constantFolding(expr)
generateExpression(constantExpr, isPredicate)
```
That assumes constant folding returns a literal. It does not.
`ConstantFolding`
returns an expression unchanged when it carries the `FAILED_TO_EVALUATE`
tag, which
is set when evaluation failed inside a conditional branch so that the error
is not
raised at planning time for a branch that may never be reached. The
recursive call
then re-enters the same case with the same expression and never makes
progress.
Because the call is in tail position it loops instead of raising
`StackOverflowError`, so the symptom is a query that never returns.
This PR recurses only when folding actually changed the expression.
This follows SPARK-50380, which made `ReorderAssociativeOperator` stop
assuming that
a foldable expression folds to a literal.
The guard is on the folding result rather than on the tag, so it also
covers any
future case where `constantFolding` returns its input unchanged.
### Why are the changes needed?
With default settings, a pushdown eligible query against a DSv2 source
hangs when a
filter contains an expression that fails to evaluate inside a conditional
branch:
```sql
SET
spark.sql.catalog.d=org.apache.spark.sql.execution.datasources.v2.jdbc.JDBCTableCatalog;
SET spark.sql.catalog.d.url=jdbc:derby:memory:v2loopdb;
SET spark.sql.catalog.d.driver=org.apache.derby.jdbc.EmbeddedDriver;
SET spark.sql.catalog.d.create=true;
CREATE NAMESPACE IF NOT EXISTS d.test;
CREATE TABLE d.test.t (c INT);
SELECT count(*) FROM d.test.t WHERE c = 1; -- control: 0
SELECT * FROM d.test.t WHERE coalesce(c, 1 div 0) = 1; -- never returns
```
Four conditions have to hold, which is why this was not hit earlier:
- ANSI mode, so `1 div 0` throws while the planner folds it
- the failing expression inside a conditional, so the error is deferred and
tagged
rather than raised
- a DSv2 source that pushes predicates, since `V2ExpressionBuilder` is only
used on
that path
- `spark.sql.optimizer.datasourceV2ExprFolding`, which defaults to true and
was
added in 4.1.0
Setting `spark.sql.optimizer.datasourceV2ExprFolding` to false avoids the
hang.
### Does this PR introduce _any_ user-facing change?
Yes. A query of the shape above previously hung and now completes. Such an
expression is no longer translated, so the predicate is not pushed to the
source and
is evaluated by Spark instead. Query results are unchanged.
### How was this patch tested?
Added a test to `DataSourceV2StrategySuite`, next to the existing
`datasourceV2ExprFolding` test. It builds `coalesce(c, 1 div 0) = 1`,
asserts that
constant folding tagged the failing branch, and then asserts that
translation
produces no V2 expression. The tag assertion is there so that the test
cannot pass
for the wrong reason if the tagging behaviour changes.
Without this change the test does not terminate. The recursion is in tail
position
and does not block, so it cannot be interrupted by a time limit, which is
why the
test asserts the result instead of using `failAfter`. This matches existing
tests
for similar issues, for example "SPARK-48843: Prevent infinite loop with
BindParameters" in `ParametersSuite`.
```
build/sbt 'sql/testOnly *DataSourceV2StrategySuite *JDBCV2Suite'
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5)
Closes #57974 from AbhinavBattu/SPARK-58428.
Authored-by: AbhinavBattu <[email protected]>
Signed-off-by: Gengliang Wang <[email protected]>
(cherry picked from commit 247748230419b2729ad753063e8df5a40647b4dc)
Signed-off-by: Gengliang Wang <[email protected]>
---
.../sql/catalyst/util/V2ExpressionBuilder.scala | 8 +++++++-
.../datasources/v2/DataSourceV2StrategySuite.scala | 20 ++++++++++++++++++++
2 files changed, 27 insertions(+), 1 deletion(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
index 3fb20dcd6420..d3cd4c3956bf 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala
@@ -98,8 +98,14 @@ class V2ExpressionBuilder(e: Expression, isPredicate:
Boolean = false) extends L
&&
SQLConf.get.getConfByKeyStrict[Boolean]("spark.sql.optimizer.datasourceV2ExprFolding")
=>
// If the expression is context independent foldable, we can convert it
to a literal.
// This is useful for increasing the coverage of V2 expressions.
+ // Folding returns the expression unchanged when it failed to evaluate
inside a conditional
+ // branch, and recursing on an unchanged expression would loop forever.
val constantExpr = ConstantFolding.constantFolding(expr)
- generateExpression(constantExpr, isPredicate)
+ if (constantExpr.fastEquals(expr)) {
+ None
+ } else {
+ generateExpression(constantExpr, isPredicate)
+ }
case col @ ColumnOrField(nameParts) =>
val ref = FieldReference(nameParts)
if (isPredicate && col.dataType.isInstanceOf[BooleanType]) {
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2StrategySuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2StrategySuite.scala
index 4af5a3251534..a0ca66183fdf 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2StrategySuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2StrategySuite.scala
@@ -22,6 +22,7 @@ import
org.apache.spark.sql.catalyst.analysis.UnresolvedAttribute
import org.apache.spark.sql.catalyst.dsl.expressions._
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.variant.VariantGet
+import org.apache.spark.sql.catalyst.optimizer.ConstantFolding
import org.apache.spark.sql.catalyst.util.V2ExpressionBuilder
import org.apache.spark.sql.connector.expressions.{Expression => V2Expression,
FieldReference, GeneralScalarExpression, LiteralValue, VariantGet =>
V2VariantGet}
import org.apache.spark.sql.connector.expressions.filter.{AlwaysFalse,
AlwaysTrue, And => V2And, Not => V2Not, Or => V2Or, Predicate}
@@ -1034,6 +1035,25 @@ class DataSourceV2StrategySuite extends
SharedSparkSession {
}
}
+ test("SPARK-58428: translating an expression that failed to evaluate does
not loop forever") {
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ // `coalesce(c, 1 div 0) = 1`. Constant folding defers the divide by
zero error because the
+ // failing expression sits in a conditional branch, so it is tagged
FAILED_TO_EVALUATE and
+ // left as is. `div` returns BIGINT, so `c` is LONG to keep the
`coalesce` inputs equal.
+ val c = AttributeReference("c", LongType)()
+ val predicate =
+ EqualTo(Coalesce(Seq(c, IntegralDivide(Literal(1), Literal(0)))),
Literal(1L))
+ val folded = ConstantFolding.constantFolding(predicate)
+ assert(
+ folded.exists(_.containsTag(ConstantFolding.FAILED_TO_EVALUATE)),
+ "expected the divide by zero branch to be tagged FAILED_TO_EVALUATE")
+
+ // Translating such an expression used to recurse forever. Note that a
regression hangs
+ // this test instead of failing it, as the recursion is in tail position.
+ assert(new V2ExpressionBuilder(folded, isPredicate =
true).build().isEmpty)
+ }
+ }
+
/**
* Translate the given Catalyst [[Expression]] into data source V2
[[Predicate]]
* then verify against the given [[Predicate]].
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
