This is an automated email from the ASF dual-hosted git repository.
MaxGekk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 9448bf930689 [SPARK-57476][SQL] Extract BIN BY validation to lay
groundwork for single-pass resolver
9448bf930689 is described below
commit 9448bf93068958a8d076d7e13fd79d0bc83ae45b
Author: Nikolina Vraneš <[email protected]>
AuthorDate: Fri Jun 26 16:14:04 2026 +0200
[SPARK-57476][SQL] Extract BIN BY validation to lay groundwork for
single-pass resolver
### What changes were proposed in this pull request?
Extract the BIN BY validation and interval/origin folding out of the
`ResolveBinBy` rule into a new neutral
`BinByResolution.validateAndComputeParameters`, and have `ResolveBinBy`
delegate to it (rule behavior unchanged).
Two BIN BY error messages are also refined to match what the single-pass
resolver produces, so the legacy and single-pass analyzers stay consistent:
- `BIN_BY_REQUIRES_TOP_LEVEL_COLUMN` reports the full nested reference
(e.g. `"outer.ts_start"`) instead of only the leaf identifier.
- `UNRESOLVED_COLUMN.WITH_SUGGESTION` orders its suggestion list by
similarity to the missing name instead of by output order.
This PR does not add the single-pass resolver itself; that lands
separately. `BinByResolution` is the shared helper it will reuse.
### Why are the changes needed?
The validation is shared by the fixed-point `ResolveBinBy` and the upcoming
single-pass `BinByResolver`; extracting it into a neutral object avoids
duplicating the checks. Aligning the two error messages keeps both analyzers
emitting the same message for the same query.
### Does this PR introduce _any_ user-facing change?
Yes, to the two BIN BY error messages above. The operator is gated off by
default via `spark.sql.binByRelationOperator.enabled`. For example, a nested
range column now yields `"outer.ts_start"` rather than `` `ts_start` ``.
### How was this patch tested?
Updated `ResolveBinBySuite` (nested-column expectation; new case asserting
the similarity-ordered suggestion list) and `BinBySuite` (NTZ inputs plus a
custom `ALIGN TO` with renamed outputs). Both pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
Closes #56545 from vranes/bin-by-single-pass-analyzer.
Authored-by: Nikolina Vraneš <[email protected]>
Signed-off-by: Max Gekk <[email protected]>
---
.../sql/catalyst/analysis/BinByResolution.scala | 157 +++++++++++++++++++++
.../spark/sql/catalyst/analysis/ResolveBinBy.scala | 118 +++-------------
.../spark/sql/errors/QueryCompilationErrors.scala | 4 +-
.../sql/catalyst/analysis/ResolveBinBySuite.scala | 12 +-
.../scala/org/apache/spark/sql/BinBySuite.scala | 25 ++++
5 files changed, 218 insertions(+), 98 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/BinByResolution.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/BinByResolution.scala
new file mode 100644
index 000000000000..890df8eb83fc
--- /dev/null
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/BinByResolution.scala
@@ -0,0 +1,157 @@
+/*
+ * 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.catalyst.analysis
+
+import scala.collection.mutable
+import scala.util.control.NonFatal
+
+import org.apache.spark.sql.catalyst.expressions.{Attribute, EmptyRow,
Expression, ExprId}
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.errors.QueryCompilationErrors
+import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.types.{
+ AnyTimestampType,
+ DataType,
+ DayTimeIntervalType,
+ DoubleType,
+ FloatType,
+ TimestampType
+}
+
+/**
+ * Folded, validated BIN BY parameters produced by
+ * [[BinByResolution.validateAndComputeParameters]].
+ *
+ * @param binWidthMicros bin width in microseconds, folded from the BIN WIDTH
interval
+ * @param originMicros ALIGN TO origin in microseconds; epoch (session-zone
epoch for LTZ) when the
+ * ALIGN TO clause is omitted
+ * @param rangeType data type of the RANGE columns (`TIMESTAMP` or
`TIMESTAMP_NTZ`)
+ * @param timeZoneId session time zone, set iff `rangeType` is `TIMESTAMP`
(LTZ)
+ */
+case class ResolvedBinByParameters(
+ binWidthMicros: Long,
+ originMicros: Long,
+ rangeType: DataType,
+ timeZoneId: Option[String])
+
+/**
+ * Shared BIN BY validation and folding for the [[ResolveBinBy]] rule and the
single-pass
+ * `BinByResolver`: validates types/foldability and folds bin width and origin
to microseconds.
+ */
+object BinByResolution {
+
+ /**
+ * Validates the BIN BY range, bin width, align-to, and distribute inputs
and folds the bin width
+ * and origin to microseconds, returning the [[ResolvedBinByParameters]].
+ */
+ def validateAndComputeParameters(
+ rangeStart: Attribute,
+ rangeEnd: Attribute,
+ distributeAttributes: Seq[Attribute],
+ binWidthExpr: Expression,
+ originExpr: Option[Expression]): ResolvedBinByParameters = {
+ val rangeType = rangeStart.dataType
+
+ if (!AnyTimestampType.acceptsType(rangeType)) {
+ throw
QueryCompilationErrors.binByRangeTypeMismatchError(rangeStart.name, rangeType)
+ }
+ if (rangeEnd.dataType != rangeType) {
+ throw QueryCompilationErrors.binByRangeTypeMismatchError(rangeEnd.name,
rangeEnd.dataType)
+ }
+
+ if (!binWidthExpr.foldable) {
+ throw QueryCompilationErrors.binByNonFoldableInputError("BIN WIDTH",
binWidthExpr)
+ }
+
+ val binWidthMicros: Long = binWidthExpr.dataType match {
+ case _: DayTimeIntervalType =>
+ val v = try {
+ binWidthExpr.eval(EmptyRow)
+ } catch {
+ case NonFatal(_) =>
+ throw
QueryCompilationErrors.binByInvalidBinWidthError(binWidthExpr)
+ }
+ if (v == null) {
+ throw QueryCompilationErrors.binByNullArgumentError("BIN WIDTH")
+ }
+ if (v.asInstanceOf[Long] <= 0L) {
+ throw
QueryCompilationErrors.binByNonPositiveBinWidthError(binWidthExpr)
+ }
+ v.asInstanceOf[Long]
+ case _ =>
+ throw
QueryCompilationErrors.binByInvalidBinWidthTypeError(binWidthExpr)
+ }
+
+ val sessionZone = SQLConf.get.sessionLocalTimeZone
+ val isLTZ = rangeType.isInstanceOf[TimestampType]
+
+ // `ALIGN TO` is optional. When omitted, default the resolved plan's
origin to
+ // `1970-01-01 00:00:00` in the session zone for `TIMESTAMP` (LTZ) and
epoch for
+ // `TIMESTAMP_NTZ`.
+ val originMicros: Long = originExpr match {
+ case Some(o) =>
+ if (!o.foldable) {
+ throw QueryCompilationErrors.binByNonFoldableInputError("ALIGN TO",
o)
+ }
+ if (o.dataType != rangeType) {
+ throw
QueryCompilationErrors.binByAlignToTypeMismatchError(o.dataType, rangeType)
+ }
+ val v = try {
+ o.eval(EmptyRow)
+ } catch {
+ case NonFatal(_) =>
+ throw QueryCompilationErrors.binByInvalidAlignToError(o)
+ }
+
+ if (v == null) {
+ throw QueryCompilationErrors.binByNullArgumentError("ALIGN TO")
+ }
+
+ v.asInstanceOf[Long]
+ case None if isLTZ =>
+ DateTimeUtils.daysToMicros(0, DateTimeUtils.getZoneId(sessionZone))
+ case None =>
+ 0L
+ }
+
+ if (distributeAttributes.isEmpty) {
+ throw QueryCompilationErrors.binByMissingDistributeError()
+ }
+
+ distributeAttributes.foreach { attr =>
+ attr.dataType match {
+ case _: FloatType | _: DoubleType =>
+ case other =>
+ throw
QueryCompilationErrors.binByInvalidDistributeColumnTypeError(attr.name, other)
+ }
+ }
+ val seen = mutable.HashSet.empty[ExprId]
+ distributeAttributes.foreach { attr =>
+ if (!seen.add(attr.exprId)) {
+ throw
QueryCompilationErrors.binByDuplicateDistributeColumnError(attr.name)
+ }
+ }
+
+ ResolvedBinByParameters(
+ binWidthMicros = binWidthMicros,
+ originMicros = originMicros,
+ rangeType = rangeType,
+ timeZoneId = if (isLTZ) Some(sessionZone) else None
+ )
+ }
+}
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 9018cf9a9be8..9cd225628d64 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
@@ -17,18 +17,14 @@
package org.apache.spark.sql.catalyst.analysis
-import scala.collection.mutable
-import scala.util.control.NonFatal
-
import org.apache.spark.SparkException
-import org.apache.spark.sql.catalyst.expressions.{Attribute, EmptyRow,
Expression, ExprId, NamedExpression}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, Expression}
import org.apache.spark.sql.catalyst.plans.logical.{BinBy, LogicalPlan,
UnresolvedBinBy}
import org.apache.spark.sql.catalyst.rules.Rule
import org.apache.spark.sql.catalyst.trees.TreePattern.UNRESOLVED_BIN_BY
-import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.catalyst.util.StringUtils
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{AnyTimestampType, DayTimeIntervalType,
DoubleType, FloatType, TimestampType}
/**
* Resolves [[UnresolvedBinBy]] into [[BinBy]]: looks up column references
against the child's
@@ -59,98 +55,27 @@ object ResolveBinBy extends Rule[LogicalPlan] {
val rangeStart = resolveColumn(b.rangeStartCol, child, resolver)
val rangeEnd = resolveColumn(b.rangeEndCol, child, resolver)
- val distributeAttrs = b.distributeColumns.map(c => resolveColumn(c, child,
resolver))
-
- val rangeType = rangeStart.dataType
- if (!AnyTimestampType.acceptsType(rangeType)) {
- throw
QueryCompilationErrors.binByRangeTypeMismatchError(rangeStart.name, rangeType)
- }
- if (rangeEnd.dataType != rangeType) {
- throw QueryCompilationErrors.binByRangeTypeMismatchError(rangeEnd.name,
rangeEnd.dataType)
- }
-
- if (!b.binWidthExpr.foldable) {
- throw QueryCompilationErrors.binByNonFoldableInputError("BIN WIDTH",
b.binWidthExpr)
- }
- // Fold the bin width to micros.
- val binWidthMicros: Long = b.binWidthExpr.dataType match {
- case _: DayTimeIntervalType =>
- val v = try {
- b.binWidthExpr.eval(EmptyRow)
- } catch {
- case NonFatal(_) =>
- throw
QueryCompilationErrors.binByInvalidBinWidthError(b.binWidthExpr)
- }
- if (v == null) {
- throw QueryCompilationErrors.binByNullArgumentError("BIN WIDTH")
- }
- if (v.asInstanceOf[Long] <= 0L) {
- throw
QueryCompilationErrors.binByNonPositiveBinWidthError(b.binWidthExpr)
- }
- v.asInstanceOf[Long]
- case _ =>
- throw
QueryCompilationErrors.binByInvalidBinWidthTypeError(b.binWidthExpr)
- }
-
- val sessionZone = SQLConf.get.sessionLocalTimeZone
- val isLTZ = rangeType.isInstanceOf[TimestampType]
-
- // `ALIGN TO` is optional. When omitted, default the resolved plan's
origin to
- // `1970-01-01 00:00:00` in the session zone for `TIMESTAMP` (LTZ) and
epoch for
- // `TIMESTAMP_NTZ`.
- val originMicros: Long = b.originExpr match {
- case Some(o) =>
- if (!o.foldable) {
- throw QueryCompilationErrors.binByNonFoldableInputError("ALIGN TO",
o)
- }
- if (o.dataType != rangeType) {
- throw
QueryCompilationErrors.binByAlignToTypeMismatchError(o.dataType, rangeType)
- }
- // Fold the origin to micros.
- val v = try {
- o.eval(EmptyRow)
- } catch {
- case NonFatal(_) =>
- throw QueryCompilationErrors.binByInvalidAlignToError(o)
- }
- if (v == null) {
- throw QueryCompilationErrors.binByNullArgumentError("ALIGN TO")
- }
- v.asInstanceOf[Long]
- case None if isLTZ =>
- DateTimeUtils.daysToMicros(0, DateTimeUtils.getZoneId(sessionZone))
- case None =>
- 0L
- }
+ val distributeAttributes = b.distributeColumns.map(c => resolveColumn(c,
child, resolver))
- if (distributeAttrs.isEmpty) {
- throw QueryCompilationErrors.binByMissingDistributeError()
- }
- distributeAttrs.foreach { attr =>
- attr.dataType match {
- case _: FloatType | _: DoubleType => // ok
- case other =>
- throw
QueryCompilationErrors.binByInvalidDistributeColumnTypeError(attr.name, other)
- }
- }
- val seen = mutable.HashSet.empty[ExprId]
- distributeAttrs.foreach { attr =>
- if (!seen.add(attr.exprId)) {
- throw
QueryCompilationErrors.binByDuplicateDistributeColumnError(attr.name)
- }
- }
+ val parameters = BinByResolution.validateAndComputeParameters(
+ rangeStart = rangeStart,
+ rangeEnd = rangeEnd,
+ distributeAttributes = distributeAttributes,
+ binWidthExpr = b.binWidthExpr,
+ originExpr = b.originExpr)
- val appendedAttributes = BinBy.appendedAttributesWithAliases(rangeType,
b.outputAliases)
+ val appendedAttributes =
+ BinBy.appendedAttributesWithAliases(parameters.rangeType,
b.outputAliases)
BinBy(
- binWidthMicros = binWidthMicros,
+ binWidthMicros = parameters.binWidthMicros,
rangeStart = rangeStart,
rangeEnd = rangeEnd,
- originMicros = originMicros,
- distributeColumns = distributeAttrs,
+ originMicros = parameters.originMicros,
+ distributeColumns = distributeAttributes,
appendedAttributes = appendedAttributes,
child = child,
- timeZoneId = if (isLTZ) Some(sessionZone) else None)
+ timeZoneId = parameters.timeZoneId)
}
private def resolveColumn(
@@ -162,13 +87,16 @@ object ResolveBinBy extends Rule[LogicalPlan] {
child.resolve(u.nameParts, resolver) match {
case Some(a: Attribute) => a
case _ =>
- throw QueryCompilationErrors.unresolvedColumnError(u.name,
child.output.map(_.name))
+ throw QueryCompilationErrors.unresolvedColumnError(
+ u.name,
+ proposal = StringUtils.orderSuggestedIdentifiersBySimilarity(
+ u.name,
+ candidates = child.output.map(attribute => attribute.qualifier
:+ attribute.name)))
}
case a: Attribute => a
- case ne: NamedExpression if ne.resolved =>
- // 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 alias: Alias if alias.resolved =>
+ // A resolved nested ref (e.g. `struct.field`) exists but isn't a
top-level column.
+ throw
QueryCompilationErrors.binByRequiresTopLevelColumnError(alias.child)
case other =>
// Genuinely unexpected; the grammar restricts these positions to
multipartIdentifier.
throw SparkException.internalError(
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 25b750fa319e..f26c32a07059 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
@@ -313,10 +313,10 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
messageParameters = Map.empty)
}
- def binByRequiresTopLevelColumnError(columnName: String): Throwable = {
+ def binByRequiresTopLevelColumnError(reference: Expression): Throwable = {
new AnalysisException(
errorClass = "BIN_BY_REQUIRES_TOP_LEVEL_COLUMN",
- messageParameters = Map("columnName" -> toSQLId(columnName)))
+ messageParameters = Map("columnName" -> toSQLExpr(reference)))
}
def binByInvalidDistributeColumnTypeError(
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 2fbf8aee8dc4..65f093a72171 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
@@ -249,6 +249,16 @@ class ResolveBinBySuite extends AnalysisTest {
"UNRESOLVED_COLUMN.WITH_SUGGESTION")
}
+ test("UNRESOLVED_COLUMN suggestions are ordered by similarity to the missing
name") {
+ // `valeu` is one edit away from `value`, so `value` is suggested first
and the remaining
+ // columns follow by edit distance.
+ val ex = intercept[SparkThrowable](
+ ResolveBinBy.apply(unresolved(rangeStart =
UnresolvedAttribute("valeu"))))
+ assert(ex.getCondition == "UNRESOLVED_COLUMN.WITH_SUGGESTION")
+ assert(ex.getMessageParameters.get("objectName") == "`valeu`")
+ assert(ex.getMessageParameters.get("proposal") == "`value`, `label`,
`ts_end`, `ts_start`")
+ }
+
test("rejects nested column refs through the full analyzer
(RULE_ORDERING_DEPENDENCIES)") {
val structCol = AttributeReference(
"outer",
@@ -268,7 +278,7 @@ class ResolveBinBySuite extends AnalysisTest {
assertAnalysisErrorCondition(
plan,
"BIN_BY_REQUIRES_TOP_LEVEL_COLUMN",
- Map("columnName" -> "`ts_start`"))
+ Map("columnName" -> "\"outer.ts_start\""))
}
test("rejects non-timestamp or mismatched RANGE columns") {
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 8ec0a1a44a8c..a00649c1f1cd 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
@@ -71,4 +71,29 @@ class BinBySuite extends QueryTest with SharedSparkSession {
parameters = Map.empty[String, String])
}
}
+
+ test("BIN BY analyzes NTZ inputs and a custom ALIGN TO with renamed
outputs") {
+ withSQLConf(SQLConf.BIN_BY_ENABLED.key -> "true") {
+ // NTZ inputs default the origin to epoch (LTZ defaults to the
session-zone epoch).
+ spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP_NTZ'2024-01-01 00:00:00', TIMESTAMP_NTZ'2024-01-01
01:00:00', 1.0D)
+ | AS t(ts_start, ts_end, value)
+ |BIN BY (RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | DISTRIBUTE UNIFORM (value))
+ |""".stripMargin).queryExecution.assertAnalyzed()
+
+ // Custom ALIGN TO origin with renamed output columns.
+ spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP'2024-01-01 00:00:00', TIMESTAMP'2024-01-01 02:00:00',
10.0D, 5.0D)
+ | AS t(ts_start, ts_end, a, b)
+ |BIN BY (RANGE ts_start TO ts_end BIN WIDTH INTERVAL '1' HOUR
+ | ALIGN TO TIMESTAMP'2024-01-01 00:30:00'
+ | DISTRIBUTE UNIFORM (a, b)
+ | BIN_START AS w_start BIN_END AS w_end
+ | BIN_DISTRIBUTE_RATIO AS frac)
+ |""".stripMargin).queryExecution.assertAnalyzed()
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
