This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 59d325a7b58d [SPARK-57617][SQL] Refine BIN BY error conditions and
SQLSTATEs
59d325a7b58d is described below
commit 59d325a7b58d6a5c589ed5ca012fb883bad89631
Author: Nikolina Vraneš <[email protected]>
AuthorDate: Tue Jun 23 10:11:03 2026 -0700
[SPARK-57617][SQL] Refine BIN BY error conditions and SQLSTATEs
### What changes were proposed in this pull request?
Follow-up to the `BIN BY` operator (SPARK-57133) that refines its
analysis-time error conditions:
1. Correct SQLSTATEs that did not match their semantics:
`BIN_BY_ALIGN_TO_TYPE_MISMATCH` and `BIN_BY_RANGE_TYPE_MISMATCH` to `42K09`,
`BIN_BY_DUPLICATE_DISTRIBUTE_COLUMN` to `42713`, `BIN_BY_INVALID_ALIGN_TO` to
`42K08`, `BIN_BY_REQUIRES_TOP_LEVEL_COLUMN` to `0A000`.
2. Rename `BIN_BY_DISTRIBUTE_TYPE_MISMATCH` to
`BIN_BY_INVALID_DISTRIBUTE_COLUMN_TYPE` (an absolute required-type check, not a
two-operand mismatch).
3. Split the `BIN_BY_INVALID_BIN_WIDTH` catch-all into
`BIN_BY_INVALID_BIN_WIDTH` (fold failure), `BIN_BY_NON_POSITIVE_BIN_WIDTH`, and
`BIN_BY_INVALID_BIN_WIDTH_TYPE`, so each failure carries the right SQLSTATE.
4. Replace `BIN_BY_COLUMN_NOT_FOUND` with the standard
`UNRESOLVED_COLUMN.WITH_SUGGESTION`.
### Why are the changes needed?
The conditions were added with the operator and some carried placeholder
SQLSTATEs, conflated distinct failures, or duplicated existing standard errors.
This aligns them with the rest of Spark SQL before the operator ships.
### Does this PR introduce _any_ user-facing change?
Yes, but only within unreleased `master`. `BIN BY` is new on `master`
(5.0.0), gated off by default, and unreleased, so no released version is
affected.
### How was this patch tested?
`ResolveBinBySuite` and `SparkThrowableSuite` (validates
`error-conditions.json`).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56674 from vranes/bin-by-error-code-cleanup.
Authored-by: Nikolina Vraneš <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 8578a26af3002143471f6007c74c2564a1e0ca85)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 38 +++++++++++++---------
.../spark/sql/catalyst/analysis/ResolveBinBy.scala | 17 +++++-----
.../spark/sql/errors/QueryCompilationErrors.scala | 22 ++++++++-----
.../sql/catalyst/analysis/ResolveBinBySuite.scala | 27 +++++++++------
4 files changed, 61 insertions(+), 43 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index 3ccaa3bd94bd..1cdb48100c6a 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -357,35 +357,35 @@
"message" : [
"The ALIGN TO expression type <originType> must match the range column
type <rangeType>."
],
- "sqlState" : "42804"
+ "sqlState" : "42K09"
},
- "BIN_BY_COLUMN_NOT_FOUND" : {
+ "BIN_BY_DUPLICATE_DISTRIBUTE_COLUMN" : {
"message" : [
- "The column <columnName> referenced in BIN BY was not found in the input
relation."
+ "Column <columnName> appears multiple times in the BIN BY DISTRIBUTE
UNIFORM clause."
],
- "sqlState" : "42703"
+ "sqlState" : "42713"
},
- "BIN_BY_DISTRIBUTE_TYPE_MISMATCH" : {
+ "BIN_BY_INVALID_ALIGN_TO" : {
"message" : [
- "The DISTRIBUTE UNIFORM columns in BIN BY must be FLOAT or DOUBLE.
Column <columnName> has type <columnType>."
+ "The ALIGN TO expression could not be evaluated to a constant value.
Got: <expr>."
],
- "sqlState" : "42804"
+ "sqlState" : "42K08"
},
- "BIN_BY_DUPLICATE_DISTRIBUTE_COLUMN" : {
+ "BIN_BY_INVALID_BIN_WIDTH" : {
"message" : [
- "Column <columnName> appears multiple times in the BIN BY DISTRIBUTE
UNIFORM clause."
+ "The BIN WIDTH expression could not be evaluated to a constant value.
Got: <expr>."
],
- "sqlState" : "42701"
+ "sqlState" : "42K08"
},
- "BIN_BY_INVALID_ALIGN_TO" : {
+ "BIN_BY_INVALID_BIN_WIDTH_TYPE" : {
"message" : [
- "The ALIGN TO expression could not be evaluated to a constant value.
Got: <expr>."
+ "The BIN WIDTH expression must be of type \"INTERVAL DAY TO SECOND\",
but got <inputType>."
],
"sqlState" : "42K09"
},
- "BIN_BY_INVALID_BIN_WIDTH" : {
+ "BIN_BY_INVALID_DISTRIBUTE_COLUMN_TYPE" : {
"message" : [
- "The BIN WIDTH expression must be a positive DAY-TIME INTERVAL. Got:
<expr>."
+ "The DISTRIBUTE UNIFORM columns in BIN BY must be FLOAT or DOUBLE.
Column <columnName> has type <columnType>."
],
"sqlState" : "42K09"
},
@@ -401,6 +401,12 @@
],
"sqlState" : "42601"
},
+ "BIN_BY_NON_POSITIVE_BIN_WIDTH" : {
+ "message" : [
+ "The BIN WIDTH expression must be positive. Got: <expr>."
+ ],
+ "sqlState" : "42816"
+ },
"BIN_BY_NULL_ARGUMENT" : {
"message" : [
"The <clause> expression in BIN BY must not be NULL."
@@ -411,13 +417,13 @@
"message" : [
"The range columns in BIN BY must be TIMESTAMP or TIMESTAMP_NTZ. Column
<columnName> has type <columnType>. Both range columns must have the same type."
],
- "sqlState" : "42804"
+ "sqlState" : "42K09"
},
"BIN_BY_REQUIRES_TOP_LEVEL_COLUMN" : {
"message" : [
"BIN BY requires a top-level column, but <columnName> is a nested or
computed field. Alias it to a top-level column in a select before BIN BY."
],
- "sqlState" : "42K09"
+ "sqlState" : "0A000"
},
"CALL_ON_STREAMING_DATASET_UNSUPPORTED" : {
"message" : [
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
index 1f7d48787d78..9018cf9a9be8 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
@@ -48,9 +48,7 @@ object ResolveBinBy extends Rule[LogicalPlan] {
// `binWidthExpr` must be fully resolved before validation can proceed. The
optional
// `originExpr` must also be resolved when present; an absent `ALIGN TO`
clause defaults to
// `1970-01-01 00:00:00` and bypasses user-facing foldability and type
checks. The
- // range/distribute column references can stay as `UnresolvedAttribute`;
`resolveColumn` below
- // converts a missing name to `BIN_BY_COLUMN_NOT_FOUND` rather than letting
the default
- // resolution pass leave them unresolved indefinitely.
+ // range/distribute column references can stay as `UnresolvedAttribute`.
private def readyToResolve(b: UnresolvedBinBy): Boolean = {
b.childrenResolved && b.binWidthExpr.resolved &&
b.originExpr.forall(_.resolved)
}
@@ -87,11 +85,11 @@ object ResolveBinBy extends Rule[LogicalPlan] {
throw QueryCompilationErrors.binByNullArgumentError("BIN WIDTH")
}
if (v.asInstanceOf[Long] <= 0L) {
- throw
QueryCompilationErrors.binByInvalidBinWidthError(b.binWidthExpr)
+ throw
QueryCompilationErrors.binByNonPositiveBinWidthError(b.binWidthExpr)
}
v.asInstanceOf[Long]
case _ =>
- throw QueryCompilationErrors.binByInvalidBinWidthError(b.binWidthExpr)
+ throw
QueryCompilationErrors.binByInvalidBinWidthTypeError(b.binWidthExpr)
}
val sessionZone = SQLConf.get.sessionLocalTimeZone
@@ -132,7 +130,7 @@ object ResolveBinBy extends Rule[LogicalPlan] {
attr.dataType match {
case _: FloatType | _: DoubleType => // ok
case other =>
- throw
QueryCompilationErrors.binByDistributeTypeMismatchError(attr.name, other)
+ throw
QueryCompilationErrors.binByInvalidDistributeColumnTypeError(attr.name, other)
}
}
val seen = mutable.HashSet.empty[ExprId]
@@ -163,12 +161,13 @@ object ResolveBinBy extends Rule[LogicalPlan] {
// Still unresolved here means genuinely absent: ResolveReferences
rewrites resolvable refs.
child.resolve(u.nameParts, resolver) match {
case Some(a: Attribute) => a
- case _ => throw QueryCompilationErrors.binByColumnNotFoundError(u.name)
+ case _ =>
+ throw QueryCompilationErrors.unresolvedColumnError(u.name,
child.output.map(_.name))
}
case a: Attribute => a
case ne: NamedExpression if ne.resolved =>
- // A nested ref (e.g. `struct.field`) is resolved to an
Alias(GetStructField) by
- // ResolveReferences; the column exists, so this is distinct from
BIN_BY_COLUMN_NOT_FOUND.
+ // A resolved non-Attribute here is a nested field (e.g. `struct.field`)
that
+ // ResolveReferences wrapped in an Alias; BIN BY only accepts top-level
columns.
throw QueryCompilationErrors.binByRequiresTopLevelColumnError(ne.name)
case other =>
// Genuinely unexpected; the grammar restricts these positions to
multipartIdentifier.
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 fc06fbf19bb1..06dd971af80d 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
@@ -307,12 +307,6 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
"rangeType" -> toSQLType(rangeType)))
}
- def binByColumnNotFoundError(columnName: String): Throwable = {
- new AnalysisException(
- errorClass = "BIN_BY_COLUMN_NOT_FOUND",
- messageParameters = Map("columnName" -> toSQLId(columnName)))
- }
-
def binByDisabledError(): Throwable = {
new AnalysisException(
errorClass = "UNSUPPORTED_FEATURE.BIN_BY",
@@ -325,11 +319,11 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
messageParameters = Map("columnName" -> toSQLId(columnName)))
}
- def binByDistributeTypeMismatchError(
+ def binByInvalidDistributeColumnTypeError(
columnName: String,
columnType: DataType): Throwable = {
new AnalysisException(
- errorClass = "BIN_BY_DISTRIBUTE_TYPE_MISMATCH",
+ errorClass = "BIN_BY_INVALID_DISTRIBUTE_COLUMN_TYPE",
messageParameters = Map(
"columnName" -> toSQLId(columnName),
"columnType" -> toSQLType(columnType)))
@@ -353,6 +347,18 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
messageParameters = Map("expr" -> toSQLExpr(expr)))
}
+ def binByNonPositiveBinWidthError(expr: Expression): Throwable = {
+ new AnalysisException(
+ errorClass = "BIN_BY_NON_POSITIVE_BIN_WIDTH",
+ messageParameters = Map("expr" -> toSQLExpr(expr)))
+ }
+
+ def binByInvalidBinWidthTypeError(expr: Expression): Throwable = {
+ new AnalysisException(
+ errorClass = "BIN_BY_INVALID_BIN_WIDTH_TYPE",
+ messageParameters = Map("inputType" -> toSQLType(expr.dataType)))
+ }
+
def binByMissingDistributeError(): Throwable = {
new AnalysisException(
errorClass = "BIN_BY_MISSING_DISTRIBUTE",
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
index 5b182cb7c38c..2fbf8aee8dc4 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
@@ -126,17 +126,21 @@ class ResolveBinBySuite extends AnalysisTest {
}
test("rejects invalid BIN WIDTH") {
- def invalid(w: Expression): Unit =
- expectError(unresolved(binWidth = w), "BIN_BY_INVALID_BIN_WIDTH")
- invalid(Literal(1, YearMonthIntervalType())) // year-month
- invalid(Literal(0L, DayTimeIntervalType())) // zero
- invalid(Literal(-1L, DayTimeIntervalType())) // negative
+ expectError(
+ unresolved(binWidth = Literal(1, YearMonthIntervalType())),
"BIN_BY_INVALID_BIN_WIDTH_TYPE")
+ expectError(
+ unresolved(binWidth = Literal(0L, DayTimeIntervalType())),
"BIN_BY_NON_POSITIVE_BIN_WIDTH")
+ expectError(
+ unresolved(binWidth = Literal(-1L, DayTimeIntervalType())),
"BIN_BY_NON_POSITIVE_BIN_WIDTH")
// Null width is rejected as a null argument, distinct from a non-positive
width.
expectError(
unresolved(binWidth = Literal(null, DayTimeIntervalType())),
"BIN_BY_NULL_ARGUMENT")
// A foldable CAST that throws on eval (ANSI) surfaces cleanly, not as a
raw exception.
withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
- invalid(Cast(Literal.create("not an interval", StringType),
DayTimeIntervalType()))
+ expectError(
+ unresolved(binWidth = Cast(Literal.create("not an interval",
StringType),
+ DayTimeIntervalType())),
+ "BIN_BY_INVALID_BIN_WIDTH")
}
// Non-foldable is rejected before evaluation.
val widthAttr = AttributeReference("w", DayTimeIntervalType(), nullable =
true)()
@@ -236,11 +240,13 @@ class ResolveBinBySuite extends AnalysisTest {
assert(resolved2.rangeEnd.exprId == t2End.exprId)
}
- test("rejects unresolvable column references with BIN_BY_COLUMN_NOT_FOUND") {
+ test("rejects unresolvable column references with UNRESOLVED_COLUMN") {
expectError(
- unresolved(rangeStart = UnresolvedAttribute("nonexistent")),
"BIN_BY_COLUMN_NOT_FOUND")
+ unresolved(rangeStart = UnresolvedAttribute("nonexistent")),
+ "UNRESOLVED_COLUMN.WITH_SUGGESTION")
expectError(
- unresolved(distribute = Seq(UnresolvedAttribute("nonexistent"))),
"BIN_BY_COLUMN_NOT_FOUND")
+ unresolved(distribute = Seq(UnresolvedAttribute("nonexistent"))),
+ "UNRESOLVED_COLUMN.WITH_SUGGESTION")
}
test("rejects nested column refs through the full analyzer
(RULE_ORDERING_DEPENDENCIES)") {
@@ -291,7 +297,8 @@ class ResolveBinBySuite extends AnalysisTest {
// Integral, DECIMAL, DT INTERVAL, and other non-float/double types are
rejected;
// users CAST to DOUBLE in an upstream projection.
Seq(intCol, decimalCol, intervalCol, label).foreach { c =>
- expectError(unresolved(child = child, distribute = Seq(c)),
"BIN_BY_DISTRIBUTE_TYPE_MISMATCH")
+ expectError(
+ unresolved(child = child, distribute = Seq(c)),
"BIN_BY_INVALID_DISTRIBUTE_COLUMN_TYPE")
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
