[SPARK-58428][SQL] Fix optimizer hang in DSv2 expression pushdown

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]) {

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]].