You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/04 06:44:09 UTC
(spark) branch branch-4.3 updated: [SPARK-58423][SQL] Raise MISSING_ATTRIBUTES for missing input after single-pass hidden-output insertion
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.3
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.3 by this push:
new fa72e2126c70 [SPARK-58423][SQL] Raise MISSING_ATTRIBUTES for missing input after single-pass hidden-output insertion
fa72e2126c70 is described below
commit fa72e2126c70ac220e56c1e3f275f11948527f6a
Author: Nikolina Vraneš <ni...@databricks.com>
AuthorDate: Tue Aug 4 14:43:13 2026 +0800
[SPARK-58423][SQL] Raise MISSING_ATTRIBUTES for missing input after single-pass hidden-output insertion
### What changes were proposed in this pull request?
A clause like `ORDER BY` / `WHERE` / `HAVING` / `QUALIFY` / `DISTRIBUTE BY` can reference a column not in its operator's output. The single-pass resolver resolves such a column from hidden output and appends it to the `Project` / `Aggregate` / `Window` below (in `ResolvesNameByHiddenOutput.expandOperatorsOutputList`) so the clause can reference it.
That append never checked whether the child actually produces the column. When the child drops it (for example `PIVOT` / `UNPIVOT`), the operator ends up referencing an attribute its child does not output. The fixed-point analyzer rejects this with a clean `MISSING_ATTRIBUTES` error, but the single-pass resolver hit a failed assert in the resolution validator and reported `INTERNAL_ERROR`.
This PR runs a missing-input check right after `expandOperatorsOutputList` builds the operator, so the single-pass resolver raises the same `MISSING_ATTRIBUTES` error. The error is built by a shared `QueryCompilationErrors.missingAttributesError` used by both the single-pass call sites and the fixed-point `CheckAnalysis`.
### Why are the changes needed?
An invalid query surfaced as `INTERNAL_ERROR` under the single-pass resolver instead of the clean `MISSING_ATTRIBUTES` the fixed-point analyzer already reports. This aligns the two.
### Does this PR introduce _any_ user-facing change?
Yes. For the affected invalid queries, the single-pass resolver now reports `MISSING_ATTRIBUTES` instead of `INTERNAL_ERROR`. Valid queries are unaffected.
### How was this patch tested?
New tests in `DataFramePivotSuite`, `DatasetUnpivotSuite`, and `BinBySuite` assert `MISSING_ATTRIBUTES` for a sort on a column dropped or re-output by `PIVOT` / `UNPIVOT` / `BIN BY`. A new `DataFrameAnalyzerTestGapsSuite` test confirms the check does not fire on a valid query. Existing `AnalysisErrorSuite` coverage confirms the shared error message is unchanged.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic)
Closes #57637 from vranes/missing-attributes-at-hidden-output-insertion.
Authored-by: Nikolina Vraneš <ni...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
(cherry picked from commit b14a42610e51587c74d687c98e4b9960d9644659)
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../sql/catalyst/analysis/CheckAnalysis.scala | 28 +++-------
.../sql/catalyst/analysis/resolver/Resolver.scala | 63 +++++++++-------------
.../resolver/ResolvesNameByHiddenOutput.scala | 17 +++++-
.../spark/sql/errors/QueryCompilationErrors.scala | 32 +++++++++++
.../scala/org/apache/spark/sql/BinBySuite.scala | 33 ++++++++++++
.../org/apache/spark/sql/DataFramePivotSuite.scala | 41 ++++++++++++++
.../org/apache/spark/sql/DatasetUnpivotSuite.scala | 45 ++++++++++++++++
.../resolver/DataFrameAnalyzerTestGapsSuite.scala | 8 +++
8 files changed, 206 insertions(+), 61 deletions(-)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
index 1d78d455084a..f4f9a0296a06 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/CheckAnalysis.scala
@@ -974,33 +974,17 @@ trait CheckAnalysis extends LookupCatalog with QueryErrorsBase with PlanToString
operator match {
case o if o.children.nonEmpty && o.missingInput.nonEmpty =>
- val missingAttributes = o.missingInput.map(attr => toSQLExpr(attr)).mkString(", ")
- val input = o.inputSet.map(attr => toSQLExpr(attr)).mkString(", ")
-
val resolver = plan.conf.resolver
val attrsWithSameName = o.missingInput.filter { missing =>
o.inputSet.exists(input => resolver(missing.name, input.name))
}
- if (attrsWithSameName.nonEmpty) {
- val sameNames = attrsWithSameName.map(attr => toSQLExpr(attr)).mkString(", ")
- o.failAnalysis(
- errorClass = "MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_APPEAR_IN_OPERATION",
- messageParameters = Map(
- "missingAttributes" -> missingAttributes,
- "input" -> input,
- "operator" -> operator.simpleString(SQLConf.get.maxToStringFields),
- "operation" -> sameNames
- ))
- } else {
- o.failAnalysis(
- errorClass = "MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_MISSING_FROM_INPUT",
- messageParameters = Map(
- "missingAttributes" -> missingAttributes,
- "input" -> input,
- "operator" -> operator.simpleString(SQLConf.get.maxToStringFields)
- ))
- }
+ throw QueryCompilationErrors.missingAttributesError(
+ operator = o,
+ missingInput = o.missingInput,
+ input = o.inputSet,
+ attributesWithSameName = attrsWithSameName
+ )
case p @ Project(projectList, _) =>
checkForUnspecifiedWindow(projectList)
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/Resolver.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/Resolver.scala
index 5d9a029b7fb6..d46227719f85 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/Resolver.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/Resolver.scala
@@ -55,7 +55,6 @@ import org.apache.spark.sql.catalyst.trees.CurrentOrigin
import org.apache.spark.sql.catalyst.util.EvaluateUnresolvedInlineTable
import org.apache.spark.sql.connector.catalog.CatalogManager
import org.apache.spark.sql.errors.{QueryCompilationErrors, QueryErrorsBase}
-import org.apache.spark.sql.internal.SQLConf
/**
* The Resolver implements a single-pass bottom-up analysis algorithm in the Catalyst.
@@ -810,7 +809,7 @@ class Resolver(
if (operator.children.nonEmpty) {
val missingInput = operator.missingInput
if (missingInput.nonEmpty) {
- throwMissingAttributesError(operator, missingInput)
+ Resolver.throwMissingAttributesError(operator, missingInput)
}
}
}
@@ -831,42 +830,6 @@ class Resolver(
}
}
- private def throwMissingAttributesError(
- operator: LogicalPlan,
- missingInput: AttributeSet): Nothing = {
- val inputSet = operator.inputSet
-
- val inputAttributesByName = new IdentifierMap[Attribute]
- for (attribute <- inputSet) {
- inputAttributesByName.put(attribute.name, attribute)
- }
-
- val attributesWithSameName = missingInput.filter { missingAttribute =>
- inputAttributesByName.contains(missingAttribute.name)
- }
-
- if (attributesWithSameName.nonEmpty) {
- operator.failAnalysis(
- errorClass = "MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_APPEAR_IN_OPERATION",
- messageParameters = Map(
- "missingAttributes" -> makeCommaSeparatedExpressionString(missingInput.toSeq),
- "input" -> makeCommaSeparatedExpressionString(inputSet.toSeq),
- "operator" -> operator.simpleString(SQLConf.get.maxToStringFields),
- "operation" -> makeCommaSeparatedExpressionString(attributesWithSameName.toSeq)
- )
- )
- } else {
- operator.failAnalysis(
- errorClass = "MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_MISSING_FROM_INPUT",
- messageParameters = Map(
- "missingAttributes" -> makeCommaSeparatedExpressionString(missingInput.toSeq),
- "input" -> makeCommaSeparatedExpressionString(inputSet.toSeq),
- "operator" -> operator.simpleString(SQLConf.get.maxToStringFields)
- )
- )
- }
- }
-
private def throwSinglePassFailedToResolveOperator(operator: LogicalPlan): Nothing =
throw SparkException.internalError(
msg = s"Failed to resolve operator in single-pass: $operator",
@@ -893,6 +856,30 @@ class Resolver(
object Resolver {
+ /**
+ * Fails with `MISSING_ATTRIBUTES` because `operator` references attributes its child does not
+ * produce.
+ */
+ def throwMissingAttributesError(
+ operator: LogicalPlan,
+ missingInput: AttributeSet): Nothing = {
+ val inputAttributesByName = new IdentifierMap[Attribute]
+ for (attribute <- operator.inputSet) {
+ inputAttributesByName.put(attribute.name, attribute)
+ }
+
+ val attributesWithSameName = missingInput.filter { missingAttribute =>
+ inputAttributesByName.contains(missingAttribute.name)
+ }.toSeq
+
+ throw QueryCompilationErrors.missingAttributesError(
+ operator = operator,
+ missingInput = missingInput.toSeq,
+ input = operator.inputSet.toSeq,
+ attributesWithSameName = attributesWithSameName
+ )
+ }
+
/**
* Create a new instance of the [[RelationResolution]].
*/
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolvesNameByHiddenOutput.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolvesNameByHiddenOutput.scala
index 6faecbe83281..0b06eaa9be45 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolvesNameByHiddenOutput.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/resolver/ResolvesNameByHiddenOutput.scala
@@ -284,7 +284,7 @@ trait ResolvesNameByHiddenOutput extends SQLConfHelper {
val (metadataCols, nonMetadataCols) =
operatorOutput.partition(_.toAttribute.qualifiedAccessOnly)
- operator match {
+ val expandedOperator = operator match {
case aggregate: Aggregate =>
val newAggregateList = nonMetadataCols ++ filteredMissingExpressions ++ metadataCols
aggregate.copy(aggregateExpressions = newAggregateList)
@@ -306,11 +306,26 @@ trait ResolvesNameByHiddenOutput extends SQLConfHelper {
project.copy(projectList = newProjectList, child = expandedChild)
}
+
+ checkMissingInput(expandedOperator)
+
+ expandedOperator
} else {
operator
}
}
+ /**
+ * Rejects an expanded `operator` whose child does not produce an appended hidden-output
+ * expression, e.g. an ORDER BY on a column an operator re-outputs under a fresh id.
+ */
+ private def checkMissingInput(operator: LogicalPlan): Unit = {
+ val missingInput = operator.missingInput
+ if (missingInput.nonEmpty) {
+ Resolver.throwMissingAttributesError(operator, missingInput)
+ }
+ }
+
private def filterMissingExpressions(
operatorOutput: Seq[NamedExpression],
missingExpressions: Seq[NamedExpression]): Seq[NamedExpression] = {
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
index 65b5f2b9e3a2..e98c7966b894 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryCompilationErrors.scala
@@ -4835,6 +4835,38 @@ private[sql] object QueryCompilationErrors extends QueryErrorsBase with Compilat
)
}
+ def missingAttributesError(
+ operator: LogicalPlan,
+ missingInput: Iterable[Attribute],
+ input: Iterable[Attribute],
+ attributesWithSameName: Iterable[Attribute]): Throwable = {
+ val missingAttributes = missingInput.map(toSQLExpr).mkString(", ")
+ val inputAttributes = input.map(toSQLExpr).mkString(", ")
+ val operatorString = operator.simpleString(SQLConf.get.maxToStringFields)
+ if (attributesWithSameName.nonEmpty) {
+ new AnalysisException(
+ errorClass = "MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_APPEAR_IN_OPERATION",
+ messageParameters = Map(
+ "missingAttributes" -> missingAttributes,
+ "input" -> inputAttributes,
+ "operator" -> operatorString,
+ "operation" -> attributesWithSameName.map(toSQLExpr).mkString(", ")
+ ),
+ origin = operator.origin
+ )
+ } else {
+ new AnalysisException(
+ errorClass = "MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_MISSING_FROM_INPUT",
+ messageParameters = Map(
+ "missingAttributes" -> missingAttributes,
+ "input" -> inputAttributes,
+ "operator" -> operatorString
+ ),
+ origin = operator.origin
+ )
+ }
+ }
+
def resolutionValidationError(cause: Throwable, plan: LogicalPlan): Throwable = {
new ExtendedAnalysisException(
new AnalysisException(
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
index 7ec687e3d953..12867a5786a3 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
@@ -460,4 +460,37 @@ class BinBySuite extends QueryTest with SharedSparkSession {
Row(2, null, null, null, null)))
}
}
+
+ test("ORDER BY on a column BIN BY re-outputs without its qualifier is rejected") {
+ // BIN BY re-outputs its input columns without the original qualifier, so `ORDER BY t.value`
+ // cannot resolve against them; hidden-output insertion appends the column to an operator whose
+ // child does not produce it, and analysis fails with MISSING_ATTRIBUTES.
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key -> "false") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01 00:05:00', 100.0D)
+ | AS t(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end
+ | BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00'
+ | DISTRIBUTE UNIFORM (value))
+ |ORDER BY t.value""".stripMargin)
+ },
+ condition = "MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_APPEAR_IN_OPERATION",
+ parameters = Map(
+ "missingAttributes" -> "\"value\"",
+ "input" -> ("\"ts_start\", \"ts_end\", \"value\", \"bin_start\", \"bin_end\", " +
+ "\"bin_distribute_ratio\""),
+ "operator" -> "!Sort \\[value#\\d+ ASC NULLS FIRST\\], true",
+ "operation" -> "\"value\""),
+ matchPVals = true,
+ queryContext =
+ Array(ExpectedContext(fragment = "ORDER BY t.value", start = 271, stop = 286)))
+ }
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFramePivotSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DataFramePivotSuite.scala
index 270d74cf5ef5..81e69fd09d97 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFramePivotSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFramePivotSuite.scala
@@ -476,4 +476,45 @@ class DataFramePivotSuite extends SharedSparkSession {
Row(10, 30))
}
}
+
+ test("ORDER BY on a column dropped by PIVOT fails name resolution") {
+ // PIVOT drops the pivoted measure column, so `t.v` fails name resolution with
+ // UNRESOLVED_COLUMN: the dropped column is never appended to a lower operator.
+ withSQLConf(SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key -> "false") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.sql(
+ """SELECT * FROM VALUES (1, 1, 100), (2, 2, 200) AS t(id, m, v)
+ |PIVOT (SUM(v) FOR m IN (1, 2))
+ |ORDER BY t.v""".stripMargin)
+ },
+ condition = "UNRESOLVED_COLUMN.WITH_SUGGESTION",
+ parameters = Map(
+ "objectName" -> "`t`.`v`",
+ "proposal" -> "`1`, `2`, `t`.`id`"),
+ queryContext =
+ Array(ExpectedContext(fragment = "t.v", start = 101, stop = 103)))
+ }
+ }
+
+ test("DataFrame sort on a column dropped by PIVOT is rejected") {
+ // The Column reference carries the dropped column's id, so unlike the SQL case above it
+ // reaches hidden-output insertion rather than failing name resolution, and analysis fails
+ // with MISSING_ATTRIBUTES.
+ withSQLConf(SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key -> "false") {
+ val df = Seq((1, 1, 100), (2, 2, 200)).toDF("id", "m", "v")
+ val pivoted = df.groupBy("id").pivot("m", Seq(1, 2)).agg(sum($"v"))
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ pivoted.sort(df("v"))
+ },
+ condition = "MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_MISSING_FROM_INPUT",
+ parameters = Map(
+ "missingAttributes" -> "\"v\"",
+ "input" -> "\"id\", \"1\", \"2\"",
+ "operator" -> "!Sort \\[v#\\d+ ASC NULLS FIRST\\], true"),
+ matchPVals = true)
+ }
+ }
}
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DatasetUnpivotSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/DatasetUnpivotSuite.scala
index f83eef5fb0ce..97ab31c3405a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DatasetUnpivotSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DatasetUnpivotSuite.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql
import org.apache.spark.sql.functions.{length, struct, sum}
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
import org.apache.spark.sql.types._
import org.apache.spark.util.ArrayImplicits._
@@ -666,6 +667,50 @@ class DatasetUnpivotSuite extends SharedSparkSession {
}
}
+ test("ORDER BY on a column dropped by UNPIVOT is rejected") {
+ // UNPIVOT drops the unpivoted value columns, so `ORDER BY t.a` cannot resolve against them;
+ // hidden-output insertion appends the column to an operator whose child does not produce it,
+ // and analysis fails with MISSING_ATTRIBUTES.
+ withSQLConf(SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key -> "false") {
+ checkError(
+ exception = intercept[AnalysisException] {
+ spark.sql(
+ """SELECT * FROM VALUES (1, 10, 20) AS t(id, a, b)
+ |UNPIVOT (val FOR name IN (a, b))
+ |ORDER BY t.a""".stripMargin)
+ },
+ condition = "MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_MISSING_FROM_INPUT",
+ parameters = Map(
+ "missingAttributes" -> "\"a\"",
+ "input" -> "\"id\", \"name\", \"val\"",
+ "operator" -> "!Sort \\[a#\\d+ ASC NULLS FIRST\\], true"),
+ matchPVals = true,
+ queryContext =
+ Array(ExpectedContext(fragment = "ORDER BY t.a", start = 81, stop = 92)))
+ }
+ }
+
+ test("DataFrame sort on a column dropped by UNPIVOT is rejected") {
+ // The Column reference carries the dropped column's id, so it reaches hidden-output insertion
+ // rather than failing name resolution as a bare name would, and analysis fails with
+ // MISSING_ATTRIBUTES.
+ withSQLConf(SQLConf.ANALYZER_DUAL_RUN_LEGACY_AND_SINGLE_PASS_RESOLVER.key -> "false") {
+ val df = Seq((1, 10, 20)).toDF("id", "a", "b")
+ val unpivoted = df.unpivot(Array($"id"), Array($"a", $"b"), "name", "val")
+
+ checkError(
+ exception = intercept[AnalysisException] {
+ unpivoted.sort(df("a"))
+ },
+ condition = "MISSING_ATTRIBUTES.RESOLVED_ATTRIBUTE_MISSING_FROM_INPUT",
+ parameters = Map(
+ "missingAttributes" -> "\"a\"",
+ "input" -> "\"id\", \"name\", \"val\"",
+ "operator" -> "!Sort \\[a#\\d+ ASC NULLS FIRST\\], true"),
+ matchPVals = true)
+ }
+ }
+
}
case class WideData(id: Int, str1: String, str2: String, int1: Option[Int], long1: Option[Long])
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/DataFrameAnalyzerTestGapsSuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/DataFrameAnalyzerTestGapsSuite.scala
index 001ec8f9cb0e..a533d1525b4a 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/DataFrameAnalyzerTestGapsSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/analysis/resolver/DataFrameAnalyzerTestGapsSuite.scala
@@ -75,4 +75,12 @@ class DataFrameAnalyzerTestGapsSuite extends SharedSparkSession {
Seq(Row(4), Row(6), Row(8))
)
}
+
+ test("Stacked orderBy resolving hidden columns across levels") {
+ // Each orderBy resolves a dropped column from hidden output, widening an intermediate Project.
+ // The missing-input check runs on each such Project; a valid query must not trip it.
+ val table = Seq((1, 2, 3), (4, 5, 6)).toDF("col1", "col2", "col3")
+ val query = table.select($"col1").orderBy($"col2").orderBy($"col1").orderBy($"col2")
+ checkAnswer(query, Seq(Row(1), Row(4)))
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org