From 40106935854d25fcfb0f4d7737145bf574b2fca4 Mon Sep 17 00:00:00 2001
From: Abhinav Battu <abhinavbattu00@gmail.com>
Date: Thu, 13 Aug 2026 00:19:41 +0530
Subject: [PATCH] [SPARK-58428][SQL] Fix optimizer hang in DSv2 expression
 pushdown

`V2ExpressionBuilder.generateExpression` constant folds a foldable expression and
recurses on the result, assuming it is now a literal. `ConstantFolding` returns the
expression unchanged when its evaluation failed inside a conditional branch, so the
recursion never makes progress. The call is in tail position, so this hangs instead
of overflowing the stack.

Only recurse when folding actually changed the expression.
---
 .../catalyst/util/V2ExpressionBuilder.scala   |  8 +++++++-
 .../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 3fb20dcd6420d..d3cd4c3956bf9 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 4af5a32515349..a0ca66183fdf4 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]].