spark-test commented on code in PR #59681:
URL: https://github.com/apache/spark/pull/59681#discussion_r4348346563
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala:
##########
@@ -2796,6 +2796,10 @@ class Analyzer(
resolveSubQueries(j, j)
case tvf: UnresolvedTableValuedFunction =>
resolveSubQueries(tvf, tvf)
+ // An inline table has no output, so its subqueries cannot reference its
columns.
+ case t: UnresolvedInlineTable
Review Comment:
**Non-blocking (P2):** This case makes scalar subqueries in VALUES lists
resolvable when the conf is on, but `CheckAnalysis` still has the unconditional
`case u: UnresolvedInlineTable if
unresolvedInlineTableContainsScalarSubquery(u)` (CheckAnalysis.scala:495). It
matches any unresolved inline table containing a `ScalarSubquery`, resolved or
not, and runs inside `foreachUp` before the generic `case operator` branch that
reports `UNRESOLVED_COLUMN`. It also runs before the subquery plan is checked,
which is where a missing table would be reported.
So with the default conf, `SELECT * FROM VALUES ((SELECT 1), unknown_col)`
and `SELECT * FROM VALUES ((SELECT x FROM missing_table))` (and the same in
`INSERT ... VALUES`) fail with `SCALAR_SUBQUERY_IN_VALUES`, i.e. "scalar
subqueries in VALUES are unsupported". That contradicts the new docs and points
users away from the actual typo. Applying that check only when
`INLINE_TABLE_NON_FOLDABLE_EXPRESSIONS_ENABLED` is false would let the generic
errors surface and keep the conf-off behavior unchanged.
**Verification:**
- **Regression:** With the feature conf on, a VALUES list that has a
resolved scalar subquery plus an unknown column, and a VALUES list whose scalar
subquery reads a missing table, report the unresolved column and the missing
table respectively instead of the unsupported-scalar-subquery error.
- **Compatibility:** With the feature conf off, scalar subqueries in VALUES
lists still fail with the unsupported-scalar-subquery error at the same query
context as before.
##########
sql/core/src/test/scala/org/apache/spark/sql/InlineTableNonFoldableExpressionsSuite.scala:
##########
@@ -0,0 +1,301 @@
+/*
+ * 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") {
+ checkAnswer(sql("SELECT * FROM VALUES ((SELECT 1) + (SELECT 2), 'a')"),
Row(3, "a"))
+ withTable("t") {
+ sql("CREATE TABLE t(id INT) USING parquet")
+ sql("INSERT INTO t VALUES (1), (2)")
+ sql("INSERT INTO t VALUES ((SELECT max(id) FROM t) + 1)")
+ checkAnswer(sql("SELECT * FROM t"), Seq(Row(1), Row(2), Row(3)))
+ }
+ }
+
+ test("uncorrelated scalar subqueries with CURRENT_LIKE expressions") {
+ // The CURRENT_LIKE expressions inside and next to the subqueries share
the value of the rest
+ // of the query.
+ checkAnswer(
+ sql("SELECT d = current_date() FROM VALUES ((SELECT current_date())) AS
t(d)"),
+ Row(true))
+ checkAnswer(
+ sql(
+ """SELECT d = date_add(current_date(), 1)
+ |FROM VALUES (date_add(current_date(), (SELECT 1))) AS t(d)
+ |""".stripMargin),
+ Row(true))
+ withTable("t") {
+ sql("CREATE TABLE t(id INT, ts TIMESTAMP) USING parquet")
+ sql("INSERT INTO t VALUES (1, TIMESTAMP'2000-01-01 00:00:00')")
+ sql(
+ """INSERT INTO t VALUES
+ | ((SELECT max(id) FROM t WHERE ts < current_timestamp()) + 1,
current_timestamp())
+ |""".stripMargin)
+ checkAnswer(sql("SELECT id FROM t"), Seq(Row(1), Row(2)))
+ }
+ }
+
+ test("correlated references and subqueries other than scalar subqueries are
rejected") {
+ withTempView("t") {
+ sql("CREATE TEMPORARY VIEW t AS SELECT * FROM VALUES (1) AS t(a)")
+ val query = "SELECT * FROM t, LATERAL (VALUES (t.a, rand())) AS v(b, r)"
+ checkError(
+ exception = intercept[AnalysisException](sql(query)),
+ condition =
"INVALID_INLINE_TABLE.CANNOT_EVALUATE_EXPRESSION_IN_INLINE_TABLE",
Review Comment:
**Nit (P3):** This test covers `LATERAL (VALUES (t.a, ...))`, `EXISTS` and
`IN`, but not a correlated scalar subquery. The `case s: ScalarSubquery =>
s.isCorrelated` branch of `isRuntimeEvaluable` is checked only with a
hand-built `ScalarSubquery(correlatedPlan, outerAttrs = ...)` in
`ResolveInlineTablesSuite`.
The real analyzer may never produce that shape. The new `ResolveSubquery`
case calls `resolveSubQueries(t, t)`, and `resolveSubQuery` runs the subquery
under `AnalysisContext.withOuterPlan(outer)`. That makes the inline table,
which has no children or output, the only outer plan, and `resolveOuterRef`
only looks at `outerPlan.resolveChildren`. So in `SELECT * FROM t, LATERAL
(VALUES ((SELECT t.a), rand()))`, `t.a` likely stays unresolved, `isCorrelated`
is never reached, and users get a different error, possibly the stale
`SCALAR_SUBQUERY_IN_VALUES`. I haven't run it to confirm which. Adding this
query to the test would show the actual error and pin it to a clear one.
**Verification:**
- **Regression:** A correlated scalar subquery in a VALUES list, analyzed by
the real analyzer, is rejected with a clear error that names the correlated
reference. If the outer column is not resolved because the inline table is used
as the subquery's outer plan, the test makes that visible, and the error is
fixed or documented accordingly.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/EvaluateUnresolvedInlineTable.scala:
##########
@@ -85,18 +120,19 @@ object EvaluateUnresolvedInlineTable extends SQLConfHelper
}
/**
- * Validates that all inline table data are valid expressions that can be
evaluated
- * (in this they must be foldable).
- * Note that nondeterministic expressions are not supported since they are
not foldable.
- * Exception are CURRENT_LIKE expressions, which are replaced by a literal
in later stages.
+ * Validates that all inline table data are valid expressions that can be
evaluated: during the
+ * analysis phase (see [[canEvalDuringAnalysis]]), or, if
+ * [[SQLConf.INLINE_TABLE_NON_FOLDABLE_EXPRESSIONS_ENABLED]] is true, when
the query runs (see
+ * [[isRuntimeEvaluable]]).
* This is package visible for unit testing.
*/
def validateInputEvaluable(table: UnresolvedInlineTable): Unit = {
+ val allowRuntimeEvaluation =
+ conf.getConf(SQLConf.INLINE_TABLE_NON_FOLDABLE_EXPRESSIONS_ENABLED)
table.rows.foreach { row =>
row.foreach { e =>
- if (e.containsPattern(CURRENT_LIKE)) {
- // Do nothing.
- } else if (!e.resolved || !trimAliases(prepareForEval(e)).foldable) {
+ if (!canEvalDuringAnalysis(e) &&
Review Comment:
**Nit (P3):** `canEvalDuringAnalysis` is true for any cell that contains a
CURRENT_LIKE expression and no subquery, so `isRuntimeEvaluable` never runs for
such cells. When another cell forces the `Expand` path, `toExpand` keeps the
cell as a runtime projection because it contains CURRENT_LIKE.
For `SELECT * FROM VALUES (max(current_date()), rand())`, the aggregate ends
up in the `Expand`. CheckAnalysis then fails through
`PlanHelper.specialExpressionsInUnsupportedOperator` with
`UNSUPPORTED_EXPR_FOR_OPERATOR` ("The query operator `Expand` contains one or
more unsupported expressions"), naming an operator the user never wrote. The PR
description says such cells fail with
`INVALID_INLINE_TABLE.CANNOT_EVALUATE_EXPRESSION_IN_INLINE_TABLE`, and before
this PR this query did fail with that error (on `rand()`). Window and generator
functions over `current_date()` take the same route, and this also makes the
CURRENT_LIKE-leaf carve-out in `isRuntimeEvaluable` unreachable. When the table
goes to the `Expand` path, also requiring `isRuntimeEvaluable` for non-foldable
CURRENT_LIKE cells would fix this without touching lists that are fully
evaluable during analysis.
**Verification:**
- **Regression:** An aggregate, a window function and a generator applied to
a CURRENT_LIKE expression, each in a VALUES list that also contains a
nondeterministic cell, are rejected with the inline-table cannot-evaluate error
instead of an operator-level error.
- **Compatibility:** Lists that were valid before, including a deterministic
Scala UDF over current_user() and current_date() next to arithmetic, keep
identical analyzed plans and results with the feature conf on and off, and
current_timestamp() next to rand() still yields one timestamp per query.
--
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]
