[SPARK-57512][SQL] Test V2 aggregate/group-by pushdown unfolds surviving RuntimeReplaceable

Add a translateAggregation regression test confirming the V2ExpressionBuilder fallback
also covers aggregate arguments and group-by expressions -- both route through
generateExpression via PushableExpression. Verified it fails without the fallback.

Co-authored-by: Isaac

Author: cloud-fan

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

@@ -21,11 +21,14 @@ import org.apache.spark.SparkConf
 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.aggregate.Min
 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.execution.TestRuntimeReplaceable
+import org.apache.spark.sql.execution.datasources.DataSourceStrategy
 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}
@@ -1040,6 +1043,28 @@ class DataSourceV2StrategySuite extends SharedSparkSession {
     }
   }
 
+  test("SPARK-57512: aggregate and group-by pushdown unfold surviving RuntimeReplaceable") {
+    val attrInt = $"cint".int
+    // A surviving RuntimeReplaceable whose `replacement` is `cint + 1`. It appears both as the MIN
+    // argument and as the group-by expression -- both route through `V2ExpressionBuilder`.
+    def wrapped: Expression = TestRuntimeReplaceable(attrInt, Literal(1))
+    def unfolded: Expression = Add(attrInt, Literal(1))
+
+    val actual = DataSourceStrategy.translateAggregation(
+      Seq(Min(wrapped).toAggregateExpression()), Seq(wrapped))
+    // Translating the already-unfolded equivalent gives the reference result.
+    val expected = DataSourceStrategy.translateAggregation(
+      Seq(Min(unfolded).toAggregateExpression()), Seq(unfolded))
+
+    assert(expected.isDefined)
+    assert(actual.isDefined,
+      "aggregate/group-by containing a surviving RuntimeReplaceable should translate")
+    assert(actual.get.groupByExpressions().map(_.describe()).toSeq ==
+      expected.get.groupByExpressions().map(_.describe()).toSeq)
+    assert(actual.get.aggregateExpressions().map(_.describe()).toSeq ==
+      expected.get.aggregateExpressions().map(_.describe()).toSeq)
+  }
+
   /**
    * Translate the given Catalyst [[Expression]] into data source V2 [[Predicate]]
    * then verify against the given [[Predicate]].