This is an automated email from the ASF dual-hosted git repository. taiyang-li pushed a commit to branch fake_add_bolt_backend in repository https://gitbox.apache.org/repos/asf/gluten.git
commit ee2f50bad6f033894867f52bebff17bbf91194a9 Author: liyang.127 <[email protected]> AuthorDate: Wed Jul 1 15:39:27 2026 +0800 [GLUTEN][CORE] Share additional tier-2 test suites Move ArrowEvalPythonExecSuite and UDFPartialProjectSuite into backends-core, and switch their plan assertions to reflective lookup so the shared tests no longer depend on backend-only test-time symbols. --- .../python/ArrowEvalPythonExecSuite.scala | 29 ++++++++- .../gluten/expression/UDFPartialProjectSuite.scala | 69 +++++++++++++++------- 2 files changed, 74 insertions(+), 24 deletions(-) diff --git a/backends-velox/src/test/scala/org/apache/gluten/execution/python/ArrowEvalPythonExecSuite.scala b/backends-core/src/test/scala/org/apache/gluten/execution/python/ArrowEvalPythonExecSuite.scala similarity index 82% rename from backends-velox/src/test/scala/org/apache/gluten/execution/python/ArrowEvalPythonExecSuite.scala rename to backends-core/src/test/scala/org/apache/gluten/execution/python/ArrowEvalPythonExecSuite.scala index 7747bd2193..808b82040b 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/execution/python/ArrowEvalPythonExecSuite.scala +++ b/backends-core/src/test/scala/org/apache/gluten/execution/python/ArrowEvalPythonExecSuite.scala @@ -19,13 +19,16 @@ package org.apache.gluten.execution.python import org.apache.gluten.execution.WholeStageTransformerSuite import org.apache.spark.SparkConf -import org.apache.spark.api.python.ColumnarArrowEvalPythonExec +import org.apache.spark.sql.DataFrame import org.apache.spark.sql.IntegratedUDFTestUtils import org.apache.spark.sql.types.{DataType, LongType, StringType} import org.apache.spark.util.SparkVersionUtil class ArrowEvalPythonExecSuite extends WholeStageTransformerSuite { + private val columnarArrowEvalPythonExecClassName = + "org.apache.spark.api.python.ColumnarArrowEvalPythonExec" + import IntegratedUDFTestUtils._ import testImplicits.localSeqToDatasetHolder import testImplicits.newProductEncoder @@ -61,7 +64,7 @@ class ArrowEvalPythonExecSuite extends WholeStageTransformerSuite { ).toDF("a", "p_a") val df2 = base.select("a").withColumn("p_a", pyarrowTestUDFString(base("a"))) - checkSparkPlan[ColumnarArrowEvalPythonExec](df2) + checkColumnarArrowEvalPythonExec(df2) checkAnswer(df2, expected) } @@ -83,7 +86,7 @@ class ArrowEvalPythonExecSuite extends WholeStageTransformerSuite { val df = base.withColumn("p_a", pyarrowTestUDFString(base("a"))).withColumn("d_b", base("b") * 2) - checkSparkPlan[ColumnarArrowEvalPythonExec](df) + checkColumnarArrowEvalPythonExec(df) checkAnswer(df, expected) } @@ -123,4 +126,24 @@ class ArrowEvalPythonExecSuite extends WholeStageTransformerSuite { TestScalarPandasUDF(name) } } + + private def checkColumnarArrowEvalPythonExec(df: DataFrame): Unit = { + val planClass = loadPlanClass(columnarArrowEvalPythonExecClassName) + val executedPlan = getExecutedPlan(df) + assert( + executedPlan.exists(planClass.isInstance), + s"Expect ${planClass.getSimpleName} exists in executedPlan:\n ${executedPlan.last}" + ) + } + + private def loadPlanClass(className: String): Class[_] = { + try { + // scalastyle:off classforname + Class.forName(className) + // scalastyle:on classforname + } catch { + case e: ClassNotFoundException => + fail(s"$className is not available on current backend classpath", e) + } + } } diff --git a/backends-velox/src/test/scala/org/apache/gluten/expression/UDFPartialProjectSuite.scala b/backends-core/src/test/scala/org/apache/gluten/expression/UDFPartialProjectSuite.scala similarity index 83% rename from backends-velox/src/test/scala/org/apache/gluten/expression/UDFPartialProjectSuite.scala rename to backends-core/src/test/scala/org/apache/gluten/expression/UDFPartialProjectSuite.scala index 41cd9ac186..edb3808b2f 100644 --- a/backends-velox/src/test/scala/org/apache/gluten/expression/UDFPartialProjectSuite.scala +++ b/backends-core/src/test/scala/org/apache/gluten/expression/UDFPartialProjectSuite.scala @@ -17,9 +17,10 @@ package org.apache.gluten.expression import org.apache.gluten.config.GlutenConfig -import org.apache.gluten.execution.{ColumnarPartialProjectExec, WholeStageTransformerSuite} +import org.apache.gluten.execution.WholeStageTransformerSuite import org.apache.spark.SparkConf +import org.apache.spark.sql.DataFrame import org.apache.spark.sql.catalyst.optimizer.{ConstantFolding, NullPropagation} import org.apache.spark.sql.execution.ProjectExec import org.apache.spark.sql.functions.udf @@ -31,6 +32,9 @@ case class MyStruct(a: Long, b: Array[Long]) case class MyStructWithNullValue(a: Option[Long], b: Array[Long]) class UDFPartialProjectSuite extends WholeStageTransformerSuite { + private val columnarPartialProjectExecClassName = + "org.apache.gluten.execution.ColumnarPartialProjectExec" + disableFallbackCheck override protected val resourcePath: String = "/tpch-data-parquet" override protected val fileFormat: String = "parquet" @@ -69,7 +73,7 @@ class UDFPartialProjectSuite extends WholeStageTransformerSuite { ignore("test plus_one") { runQueryAndCompare("SELECT sum(plus_one(cast(l_orderkey as long))) from lineitem") { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } @@ -78,20 +82,20 @@ class UDFPartialProjectSuite extends WholeStageTransformerSuite { "select plus_one(" + "(select plus_one(count(*)) from (values (1)) t0(inner_c))) as col " + "from (values (2),(3)) t1(outer_c)") { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } ignore("test plus_one with column used twice") { runQueryAndCompare( "SELECT sum(plus_one(cast(l_orderkey as long)) + hash(l_orderkey)) from lineitem") { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } ignore("test plus_one without cast") { runQueryAndCompare("SELECT sum(plus_one(l_orderkey) + hash(l_orderkey)) from lineitem") { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } @@ -100,7 +104,7 @@ class UDFPartialProjectSuite extends WholeStageTransformerSuite { "SELECT sum(plus_one(cast(l_orderkey as long)) + hash(l_partkey))" + "from lineitem " + "where l_orderkey < 3") { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } @@ -110,25 +114,23 @@ class UDFPartialProjectSuite extends WholeStageTransformerSuite { | select plus_one(l_orderkey) as col1, l_partkey from lineitem |)""".stripMargin runQueryAndCompare(sql) { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } val df = spark.sql(sql) - assert(df.queryExecution.executedPlan.collect { - case p: ColumnarPartialProjectExec => p - }.size == 2) + assert(countColumnarPartialProjectExec(df) == 2) } test("test plus_one with many columns in project") { runQueryAndCompare("SELECT plus_one(cast(l_orderkey as long)), hash(l_partkey) from lineitem") { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } ignore("test function no argument") { runQueryAndCompare("""SELECT no_argument(), l_orderkey | from lineitem limit 100""".stripMargin) { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } @@ -138,14 +140,14 @@ class UDFPartialProjectSuite extends WholeStageTransformerSuite { df.collect() assert( df.queryExecution.executedPlan - .find(p => p.isInstanceOf[ColumnarPartialProjectExec]) + .find(loadPlanClass(columnarPartialProjectExecClassName).isInstance) .isEmpty) } test("udf in agg simple") { runQueryAndCompare("""select sum(hash(plus_one(l_extendedprice)) + hash(l_orderkey) ) as revenue | from lineitem""".stripMargin) { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } @@ -153,13 +155,13 @@ class UDFPartialProjectSuite extends WholeStageTransformerSuite { runQueryAndCompare("""select sum(hash(plus_one(l_extendedprice)) * l_discount | + hash(l_orderkey) + hash(l_comment)) as revenue | from lineitem""".stripMargin) { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } test("test concat with string") { runQueryAndCompare("SELECT concat_concat(l_comment), hash(l_partkey) from lineitem") { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } @@ -175,7 +177,7 @@ class UDFPartialProjectSuite extends WholeStageTransformerSuite { | GROUP BY l_partkey |) |""".stripMargin) { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } @@ -196,7 +198,7 @@ class UDFPartialProjectSuite extends WholeStageTransformerSuite { | FROM lineitem |) |""".stripMargin) { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } @@ -215,7 +217,7 @@ class UDFPartialProjectSuite extends WholeStageTransformerSuite { | FROM lineitem |) |""".stripMargin) { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } } // only SparkVersion >= 3.4 support columnar native writer @@ -241,7 +243,7 @@ class UDFPartialProjectSuite extends WholeStageTransformerSuite { |""".stripMargin) { if (enableNativeScanAndWriter.toBoolean) { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec } else { checkSparkPlan[ProjectExec] } @@ -270,7 +272,32 @@ class UDFPartialProjectSuite extends WholeStageTransformerSuite { | FROM lineitem |) |""".stripMargin) { - checkGlutenPlan[ColumnarPartialProjectExec] + checkColumnarPartialProjectExec + } + } + + private def checkColumnarPartialProjectExec(df: DataFrame): Unit = { + val planClass = loadPlanClass(columnarPartialProjectExecClassName) + val executedPlan = getExecutedPlan(df) + assert( + executedPlan.exists(planClass.isInstance), + s"Expect ${planClass.getSimpleName} exists in executedPlan:\n ${executedPlan.last}" + ) + } + + private def countColumnarPartialProjectExec(df: DataFrame): Int = { + val planClass = loadPlanClass(columnarPartialProjectExecClassName) + getExecutedPlan(df).count(planClass.isInstance) + } + + private def loadPlanClass(className: String): Class[_] = { + try { + // scalastyle:off classforname + Class.forName(className) + // scalastyle:on classforname + } catch { + case e: ClassNotFoundException => + fail(s"$className is not available on current backend classpath", e) } } } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
