gengliangwang commented on code in PR #59681:
URL: https://github.com/apache/spark/pull/59681#discussion_r4342418421
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/EvaluateUnresolvedInlineTable.scala:
##########
@@ -61,6 +71,29 @@ object EvaluateUnresolvedInlineTable extends SQLConfHelper
if (earlyEvalPossible) EvalInlineTables.eval(table) else table
}
+ /**
+ * Turns an inline table with expressions that are evaluated when the query
runs into an
+ * [[Expand]] over a [[OneRowRelation]], which produces one output row per
projection. The
+ * foldable expressions are evaluated now, except CURRENT_LIKE expressions,
which are replaced
+ * with literals at the end of the analysis phase.
+ */
+ private def toExpand(table: ResolvedInlineTable): Expand = {
+ val projections = table.rows.map(_.map {
+ case e if !e.containsPattern(CURRENT_LIKE) &&
trimAliases(prepareForEval(e)).foldable =>
+ Literal(EvalInlineTables.evalExpression(table, e), e.dataType)
+ case e => e
+ })
+ Expand(projections, table.output, OneRowRelation())
+ }
+
+ /**
+ * Returns whether the expression can be evaluated during the analysis
phase: it must be
+ * foldable, or contain CURRENT_LIKE expressions, which are replaced by
literals at the end of
+ * the analysis phase. Inline tables with only such expressions are
evaluated during analysis.
+ */
+ private def canEvalDuringAnalysis(e: Expression): Boolean =
+ e.containsPattern(CURRENT_LIKE) || (e.resolved &&
trimAliases(prepareForEval(e)).foldable)
Review Comment:
**Non-blocking (P2):** With the conf on, the `containsPattern(CURRENT_LIKE)`
short-circuit also matches resolved subqueries.
`PlanExpression.treePatternBits` merges in the subquery plan's bits, so
`(SELECT max(ts) FROM t WHERE ts < current_timestamp())` or `current_date() -
(SELECT max(d) FROM t)` counts as CURRENT_LIKE. When every cell passes this
check, `evaluate` takes the `earlyEvalIfPossible` path and keeps a
`ResolvedInlineTable`. In the Finish Analysis batch, `EvalInlineTables` then
calls `eval()` on the `Unevaluable` `ScalarSubquery`, and the query fails with
`INVALID_INLINE_TABLE.FAILED_SQL_EXPRESSION_EVALUATION`.
So `SELECT * FROM VALUES ((SELECT current_date()))` and `INSERT INTO t
VALUES ((SELECT max(id) FROM t WHERE ts < current_timestamp()) + 1)` fail,
although the docs and the conf say uncorrelated scalar subqueries are evaluated
when the query runs. The same query succeeds once any other cell is
non-foldable, e.g. after adding a `rand()` column. The same short-circuit lets
an `EXISTS` subquery that mentions a CURRENT_LIKE function bypass the intended
`CANNOT_EVALUATE_EXPRESSION_IN_INLINE_TABLE` rejection. Excluding cells that
contain a subquery expression from the CURRENT_LIKE branch would route them
through `isRuntimeEvaluable` and the Expand path. Lists that were valid before
this PR would be unaffected, since none of them can contain a subquery.
See **Shared repair plan 1** in the review body.
##########
sql/core/src/test/scala/org/apache/spark/sql/InlineTableNonFoldableExpressionsSuite.scala:
##########
@@ -0,0 +1,273 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql
+
+import java.util.UUID
+
+import org.apache.spark.{SparkRuntimeException, TaskContext}
+import org.apache.spark.sql.IntegratedUDFTestUtils.{registerTestUDF,
shouldTestPythonUDFs, TestPythonUDF}
+import org.apache.spark.sql.execution.{ExpandExec, LocalTableScanExec}
+import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
+import org.apache.spark.sql.functions.udf
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.sql.types.{DoubleType, LongType, StructField,
StructType}
+
+/**
+ * End-to-end tests for VALUES lists with expressions that are not foldable,
which are evaluated
+ * by an `Expand` over a `OneRowRelation` when the query runs.
+ */
+class InlineTableNonFoldableExpressionsSuite
+ extends QueryTest with SharedSparkSession with AdaptiveSparkPlanHelper {
+ import testImplicits._
+
+ private def expandExecs(df: DataFrame): Seq[ExpandExec] =
+ collect(df.queryExecution.executedPlan) { case e: ExpandExec => e }
+
+ private def contextOf(query: String, fragment: String): ExpectedContext = {
+ val start = query.indexOf(fragment)
+ assert(start >= 0, s"'$fragment' is not part of '$query'")
+ ExpectedContext(fragment, start, start + fragment.length - 1)
+ }
+
+ test("non-foldable expressions are evaluated on executors") {
+ withUserDefinedFunction("in_task" -> true) {
+ spark.udf.register("in_task", udf(() => TaskContext.get() !=
null).asNondeterministic())
+ val df = sql("SELECT * FROM VALUES (1, in_task()), (2, in_task()) AS
t(id, in_task)")
+ checkAnswer(df, Seq(Row(1, true), Row(2, true)))
+ val execs = expandExecs(df)
+ assert(execs.size == 1)
+ // All the rows are evaluated in a single task.
+ assert(execs.head.execute().getNumPartitions == 1)
+ }
+ }
+
+ test("VALUES lists with only foldable expressions still become local
relations") {
+ val df = sql("SELECT * FROM VALUES (1, 'a'), (2 + 1, upper('b')) AS t(id,
s)")
+ checkAnswer(df, Seq(Row(1, "a"), Row(3, "B")))
+ assert(expandExecs(df).isEmpty)
+ assert(collect(df.queryExecution.executedPlan) { case l:
LocalTableScanExec => l }.size == 1)
+ }
+
+ test("VALUES lists that were supported before are evaluated during analysis
as before") {
+ withUserDefinedFunction("on_driver" -> true) {
+ spark.udf.register("on_driver", udf((s: String) => s != null &&
TaskContext.get() == null))
+ val queries = Seq(
+ "SELECT * FROM VALUES (1, 'a'), (2 + 1, upper('b')) AS t(a, b)",
+ "SELECT * FROM VALUES (current_date(), 1 + 1) AS t(a, b)",
+ // The UDF is not foldable, but it was supported because its input is
a CURRENT_LIKE
+ // expression. It is evaluated on the driver at the end of the
analysis phase.
+ "SELECT * FROM VALUES (on_driver(current_user())) AS t(a)")
+ queries.foreach { query =>
+ val df = sql(query)
+ val legacy = withSQLConf(
+ SQLConf.INLINE_TABLE_NON_FOLDABLE_EXPRESSIONS_ENABLED.key ->
"false") {
+ sql(query)
+ }
+ comparePlans(
+ df.queryExecution.analyzed, legacy.queryExecution.analyzed,
checkAnalysis = false)
+ checkAnswer(df, legacy)
+ assert(expandExecs(df).isEmpty)
+ }
+ checkAnswer(sql(queries.last), Row(true))
+ }
+
+ // The expressions of inline tables with CURRENT_LIKE expressions are
evaluated at the end of
+ // the analysis phase, so evaluation errors are reported when the query is
optimized.
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
+ val df = sql("SELECT * FROM VALUES (current_date(), 1 / 0) AS t(a, b)")
+ val e = intercept[AnalysisException](df.collect())
+ assert(e.getCondition ==
"INVALID_INLINE_TABLE.FAILED_SQL_EXPRESSION_EVALUATION")
+ }
+ }
+
+ test("nondeterministic expressions") {
+ // Every expression has its own seed.
+ val uuids = sql("SELECT * FROM VALUES (uuid()), (uuid()),
(uuid())").collect()
+ assert(uuids.map(_.getString(0)).distinct.length == 3)
+
+ // Each row is evaluated like the only row of a one-row relation.
+ checkAnswer(
+ sql("SELECT * FROM VALUES (rand(5)), (rand(5))"),
+ sql("SELECT rand(5) UNION ALL SELECT rand(5)"))
+
+ // The seeds are chosen during analysis, so executing the plan again
returns the same values.
+ val df = sql("SELECT * FROM VALUES (rand(), 1), (rand(), 2)")
+ checkAnswer(df, df.collect().toSeq)
+ }
+
+ test("equal nondeterministic expressions are evaluated separately for every
row") {
+ withUserDefinedFunction("random_uuid" -> true) {
+ spark.udf.register("random_uuid", udf(() =>
UUID.randomUUID().toString).asNondeterministic())
+ Seq(true, false).foreach { codegen =>
+ withSQLConf(SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key ->
codegen.toString) {
+ val rows = sql("SELECT * FROM VALUES (random_uuid()),
(random_uuid())").collect()
+ assert(rows.map(_.getString(0)).distinct.length == 2)
+ }
+ }
+ }
+ }
+
+ test("CURRENT_LIKE expressions are evaluated once per query") {
+ val df = sql(
+ """SELECT count(DISTINCT ts), count(DISTINCT r)
+ |FROM VALUES (current_timestamp(), rand()), (current_timestamp(),
rand()) AS t(ts, r)
+ |""".stripMargin)
+ checkAnswer(df, Row(1, 2))
+ }
+
+ test("type coercion and nullability") {
+ val df = sql("SELECT * FROM VALUES (1, rand()), (2L, NULL) AS t(a, b)")
+ assert(df.schema == StructType(Seq(
+ StructField("a", LongType, nullable = false),
+ StructField("b", DoubleType, nullable = true))))
+ checkAnswer(df.select($"a", $"b".isNull), Seq(Row(1L, false), Row(2L,
true)))
+ }
+
+ test("uncorrelated scalar subqueries") {
Review Comment:
**Nit (P3):** These subquery cases contain no CURRENT_LIKE expression. The
CURRENT_LIKE tests (`CURRENT_LIKE expressions are evaluated once per query`
here, and `CURRENT_LIKE expressions next to non-foldable expressions` in
`ResolveInlineTablesSuite`) always pair it with `rand()`, which forces the
Expand path. A cell like `(SELECT current_date())`, or `(SELECT max(id) FROM t
WHERE ts < current_timestamp())` inside `INSERT ... VALUES`, currently fails
during evaluation at the end of analysis, and every test still passes. Please
add the combined cases for SELECT and INSERT, plus an `EXISTS` subquery that
mentions a CURRENT_LIKE function, which should be rejected with
`CANNOT_EVALUATE_EXPRESSION_IN_INLINE_TABLE`.
See **Shared repair plan 1** in the review body.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
