[SPARK-57512][SQL] Unfold surviving RuntimeReplaceable at predicate-pushdown boundaries

Generalize the cached-batch fix: a surviving RuntimeReplaceable must be unfolded to
its replacement wherever a predicate leaves Spark's expression-evaluation engine for
a structure-interpreting consumer. In addition to CachedBatchSerializer.buildFilter,
this covers data source filter pushdown -- V1 DataSourceStrategy.translateLeafNodeFilter
and V2 V2ExpressionBuilder. V2 unfolds only as a fallback so explicit native handling
(e.g. AES_ENCRYPT) still wins; the V1 leaf unfold preserves the translatedFilterToExpr
mapping by keeping the original expression in the recursive caller. Internal Spark
evaluation handles RuntimeReplaceable via delegation + MaterializeRuntimeReplaceable,
so it needs no unfold.

Co-authored-by: Isaac

Author: cloud-fan

sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/V2ExpressionBuilder.scala

@@ -376,6 +376,11 @@ class V2ExpressionBuilder(e: Expression, isPredicate: Boolean = false) extends L
       } else {
         None
       }
+    // A surviving `RuntimeReplaceable` (`eagerReplace = false`) that no explicit case above pushes
+    // natively is a Spark-internal optimizer node the connector cannot understand. Fall back to its
+    // concrete `replacement` so the lowered form can still be pushed -- same boundary rationale as
+    // `DataSourceStrategy.translateLeafNodeFilter` (V1) and `CachedBatchSerializer.buildFilter`.
+    case r: RuntimeReplaceable => generateExpression(r.replacement, isPredicate)
     case _ => None
   }
 

sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategy.scala

@@ -631,9 +631,16 @@ object DataSourceStrategy
     }
   }
 
+  // A surviving `RuntimeReplaceable` (`eagerReplace = false`) is a Spark-internal optimizer node a
+  // data source cannot understand, so pushdown must translate its concrete `replacement`. Unfold at
+  // this boundary -- where the predicate leaves Spark's evaluation engine for an external system --
+  // mirroring `V2ExpressionBuilder` (V2) and `CachedBatchSerializer.buildFilter` (cache). The
+  // recursive caller keeps the original expression for the `translatedFilterToExpr` mapping, so
+  // only this leaf translation sees the unfolded form.
   private def translateLeafNodeFilter(
       predicate: Expression,
-      pushableColumn: PushableColumnBase): Option[Filter] = predicate match {
+      pushableColumn: PushableColumnBase): Option[Filter] = {
+    RuntimeReplaceable.unfold(predicate) match {
     case expressions.EqualTo(e @ pushableColumn(name), Literal(v, t)) =>
       Some(collationAwareFilter(sources.EqualTo(name, convertToScala(v, t)), e.dataType))
     case expressions.EqualTo(Literal(v, t), e @ pushableColumn(name)) =>
@@ -699,6 +706,7 @@ object DataSourceStrategy
       Some(sources.EqualTo(name, true))
 
     case _ => None
+    }
   }
 
   /**

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/DataSourceStrategySuite.scala

@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.datasources
 
 import org.apache.spark.sql.catalyst.dsl.expressions._
 import org.apache.spark.sql.catalyst.expressions._
+import org.apache.spark.sql.execution.TestPredicateRuntimeReplaceable
 import org.apache.spark.sql.sources
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}
@@ -96,6 +97,11 @@ class DataSourceStrategySuite extends SharedSparkSession {
     testTranslateFilter(GreaterThan(attrInt, 1), Some(sources.GreaterThan(intColName, 1)))
     testTranslateFilter(GreaterThan(1, attrInt), Some(sources.LessThan(intColName, 1)))
 
+    // SPARK-57512: a surviving RuntimeReplaceable is unfolded so its replacement (here a
+    // `GreaterThan`) can be translated and pushed, instead of being left unrecognized.
+    testTranslateFilter(TestPredicateRuntimeReplaceable(attrInt, Literal(1)),
+      Some(sources.GreaterThan(intColName, 1)))
+
     testTranslateFilter(LessThan(attrInt, 1), Some(sources.LessThan(intColName, 1)))
     testTranslateFilter(LessThan(1, attrInt), Some(sources.GreaterThan(intColName, 1)))
 

sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/v2/DataSourceV2StrategySuite.scala

@@ -25,6 +25,7 @@ import org.apache.spark.sql.catalyst.expressions.variant.VariantGet
 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}
+import org.apache.spark.sql.execution.TestPredicateRuntimeReplaceable
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.test.SharedSparkSession
 import org.apache.spark.sql.types.{BooleanType, DoubleType, IntegerType, LongType, StringType, StructField, StructType, TimestampType, VariantType}
@@ -110,6 +111,11 @@ class DataSourceV2StrategySuite extends SharedSparkSession {
       testTranslateFilter(GreaterThan(1, attrInt),
         Some(new Predicate("<", Array(FieldReference(intColName), LiteralValue(1, IntegerType)))))
 
+      // SPARK-57512: a surviving RuntimeReplaceable is unfolded so its replacement (here a
+      // `GreaterThan`) can be translated and pushed, instead of being left unrecognized.
+      testTranslateFilter(TestPredicateRuntimeReplaceable(attrInt, Literal(1)),
+        Some(new Predicate(">", Array(FieldReference(intColName), LiteralValue(1, IntegerType)))))
+
       testTranslateFilter(LessThan(attrInt, 1),
         Some(new Predicate("<", Array(FieldReference(intColName), LiteralValue(1, IntegerType)))))
       testTranslateFilter(LessThan(1, attrInt),