This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun 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 191f657b64a9 Revert "[SPARK-57133][SQL] Add BIN BY relation operator
parsing and resolution"
191f657b64a9 is described below
commit 191f657b64a9a57f3ae5060641a4326ad48f6a59
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Tue Jun 9 15:28:10 2026 -0700
Revert "[SPARK-57133][SQL] Add BIN BY relation operator parsing and
resolution"
### What changes were proposed in this pull request?
This reverts commit 761afcb676189a4ef58439441da03ba73aed1e21 to recover CI.
We can land 761afcb676189a4ef58439441da03ba73aed1e21 back after merging the
following PR first.
- https://github.com/apache/spark/pull/56406
Please see the CI failure details in the above PR.
### Why are the changes needed?
To recover the CIs. The original PR is not a breaking change, but it was a
CI-breaking commit unfortunately.
### Does this PR introduce _any_ user-facing change?
No, this is not released yet.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #56413 from dongjoon-hyun/SPARK-57133.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
(cherry picked from commit af39d9580efd675aecf8965b443bb404fabd1281)
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../src/main/resources/error/error-conditions.json | 71 -----
docs/sql-ref-ansi-compliance.md | 7 -
python/pyspark/sql/tvf.py | 8 +-
.../spark/sql/catalyst/parser/SqlBaseLexer.g4 | 7 -
.../spark/sql/catalyst/parser/SqlBaseParser.g4 | 32 +--
.../spark/sql/catalyst/analysis/Analyzer.scala | 1 -
.../catalyst/analysis/DeduplicateRelations.scala | 15 --
.../spark/sql/catalyst/analysis/ResolveBinBy.scala | 176 -------------
.../spark/sql/catalyst/parser/AstBuilder.scala | 44 +---
.../plans/logical/basicLogicalOperators.scala | 119 ---------
.../sql/catalyst/rules/RuleIdCollection.scala | 1 -
.../spark/sql/catalyst/trees/TreePatterns.scala | 2 -
.../spark/sql/errors/QueryCompilationErrors.scala | 82 ------
.../sql/catalyst/analysis/ResolveBinBySuite.scala | 291 ---------------------
.../sql/catalyst/parser/PlanParserSuite.scala | 168 ------------
.../jdbc/SparkConnectDatabaseMetaDataSuite.scala | 2 +-
.../spark/sql/execution/SparkStrategies.scala | 2 -
.../sql-tests/results/keywords-enforced.sql.out | 7 -
.../resources/sql-tests/results/keywords.sql.out | 7 -
.../sql-tests/results/nonansi/keywords.sql.out | 7 -
.../scala/org/apache/spark/sql/BinBySuite.scala | 51 ----
.../ThriftServerWithSparkContextSuite.scala | 2 +-
22 files changed, 10 insertions(+), 1092 deletions(-)
diff --git a/common/utils/src/main/resources/error/error-conditions.json
b/common/utils/src/main/resources/error/error-conditions.json
index c46a36d09b19..bf5b3092511f 100644
--- a/common/utils/src/main/resources/error/error-conditions.json
+++ b/common/utils/src/main/resources/error/error-conditions.json
@@ -353,72 +353,6 @@
],
"sqlState" : "22003"
},
- "BIN_BY_ALIGN_TO_TYPE_MISMATCH" : {
- "message" : [
- "The ALIGN TO expression type <originType> must match the range column
type <rangeType>."
- ],
- "sqlState" : "42804"
- },
- "BIN_BY_COLUMN_NOT_FOUND" : {
- "message" : [
- "The column <columnName> referenced in BIN BY was not found in the input
relation."
- ],
- "sqlState" : "42703"
- },
- "BIN_BY_DISTRIBUTE_TYPE_MISMATCH" : {
- "message" : [
- "The DISTRIBUTE UNIFORM columns in BIN BY must be numeric or DAY-TIME
INTERVAL. Column <columnName> has type <columnType>."
- ],
- "sqlState" : "42804"
- },
- "BIN_BY_DUPLICATE_DISTRIBUTE_COLUMN" : {
- "message" : [
- "Column <columnName> appears multiple times in the BIN BY DISTRIBUTE
UNIFORM clause."
- ],
- "sqlState" : "42701"
- },
- "BIN_BY_INVALID_ALIGN_TO" : {
- "message" : [
- "The ALIGN TO expression could not be evaluated to a constant value.
Got: <expr>."
- ],
- "sqlState" : "42K09"
- },
- "BIN_BY_INVALID_BIN_WIDTH" : {
- "message" : [
- "The BIN WIDTH expression must be a positive DAY-TIME INTERVAL. Got:
<expr>."
- ],
- "sqlState" : "42K09"
- },
- "BIN_BY_INVALID_RANGE" : {
- "message" : [
- "The input row to BIN BY has range_start (<rangeStart>) greater than
range_end (<rangeEnd>). BIN BY requires range_start <= range_end."
- ],
- "sqlState" : "22023"
- },
- "BIN_BY_MISSING_DISTRIBUTE" : {
- "message" : [
- "BIN BY requires at least one column in DISTRIBUTE UNIFORM (...)."
- ],
- "sqlState" : "42601"
- },
- "BIN_BY_NULL_ARGUMENT" : {
- "message" : [
- "The <clause> expression in BIN BY must not be NULL."
- ],
- "sqlState" : "22004"
- },
- "BIN_BY_RANGE_TYPE_MISMATCH" : {
- "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"
- },
- "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"
- },
"CALL_ON_STREAMING_DATASET_UNSUPPORTED" : {
"message" : [
"The method <methodName> can not be called on streaming
Dataset/DataFrame."
@@ -8087,11 +8021,6 @@
"The ANALYZE TABLE command does not support views."
]
},
- "BIN_BY" : {
- "message" : [
- "Physical execution of BIN BY is not yet implemented."
- ]
- },
"CATALOG_INTERFACE_METHOD" : {
"message" : [
"Catalog API <methodName> is not supported by <catalogClass>."
diff --git a/docs/sql-ref-ansi-compliance.md b/docs/sql-ref-ansi-compliance.md
index bd449524895c..4f21b7b4b3c7 100644
--- a/docs/sql-ref-ansi-compliance.md
+++ b/docs/sql-ref-ansi-compliance.md
@@ -412,7 +412,6 @@ Below is a list of all the keywords in Spark SQL.
|ADD|non-reserved|non-reserved|non-reserved|
|AFTER|non-reserved|non-reserved|non-reserved|
|AGGREGATE|non-reserved|non-reserved|non-reserved|
-|ALIGN|non-reserved|non-reserved|non-reserved|
|ALL|reserved|non-reserved|reserved|
|ALTER|non-reserved|non-reserved|reserved|
|ALWAYS|non-reserved|non-reserved|non-reserved|
@@ -434,12 +433,8 @@ Below is a list of all the keywords in Spark SQL.
|BERNOULLI|non-reserved|non-reserved|non-reserved|
|BETWEEN|non-reserved|non-reserved|reserved|
|BIGINT|non-reserved|non-reserved|reserved|
-|BIN|non-reserved|non-reserved|non-reserved|
|BINARY|non-reserved|non-reserved|reserved|
|BINDING|non-reserved|non-reserved|non-reserved|
-|BIN_DISTRIBUTE_RATIO|non-reserved|non-reserved|non-reserved|
-|BIN_END|non-reserved|non-reserved|non-reserved|
-|BIN_START|non-reserved|non-reserved|non-reserved|
|BOOLEAN|non-reserved|non-reserved|reserved|
|BOTH|reserved|non-reserved|reserved|
|BUCKET|non-reserved|non-reserved|non-reserved|
@@ -806,7 +801,6 @@ Below is a list of all the keywords in Spark SQL.
|UNARCHIVE|non-reserved|non-reserved|non-reserved|
|UNBOUNDED|non-reserved|non-reserved|non-reserved|
|UNCACHE|non-reserved|non-reserved|non-reserved|
-|UNIFORM|non-reserved|non-reserved|non-reserved|
|UNION|reserved|strict-non-reserved|reserved|
|UNIQUE|reserved|non-reserved|reserved|
|UNKNOWN|reserved|non-reserved|reserved|
@@ -834,7 +828,6 @@ Below is a list of all the keywords in Spark SQL.
|WHEN|reserved|non-reserved|reserved|
|WHERE|reserved|non-reserved|reserved|
|WHILE|non-reserved|non-reserved|non-reserved|
-|WIDTH|non-reserved|non-reserved|non-reserved|
|WINDOW|non-reserved|non-reserved|reserved|
|WITH|reserved|non-reserved|reserved|
|WITHIN|reserved|non-reserved|reserved|
diff --git a/python/pyspark/sql/tvf.py b/python/pyspark/sql/tvf.py
index 3f20fc3501bb..97d8ce59913b 100644
--- a/python/pyspark/sql/tvf.py
+++ b/python/pyspark/sql/tvf.py
@@ -551,11 +551,11 @@ class TableValuedFunction:
Examples
--------
>>> spark.tvf.sql_keywords().show()
- +----------+--------+
- | keyword|reserved|
- +----------+--------+
+ +-------------+--------+
+ | keyword|reserved|
+ +-------------+--------+
...
- +----------+--------+...
+ +-------------+--------+...
"""
return self._fn("sql_keywords")
diff --git
a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4
b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4
index bf759e501db9..af71f441012c 100644
---
a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4
+++
b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseLexer.g4
@@ -131,7 +131,6 @@ BANG: '!';
ADD: 'ADD';
AFTER: 'AFTER';
AGGREGATE: 'AGGREGATE';
-ALIGN: 'ALIGN';
ALL: 'ALL';
ALTER: 'ALTER';
ALWAYS: 'ALWAYS';
@@ -153,10 +152,6 @@ BEGIN: 'BEGIN';
BERNOULLI: 'BERNOULLI';
BETWEEN: 'BETWEEN';
BIGINT: 'BIGINT';
-BIN: 'BIN';
-BIN_DISTRIBUTE_RATIO: 'BIN_DISTRIBUTE_RATIO';
-BIN_END: 'BIN_END';
-BIN_START: 'BIN_START';
BINARY: 'BINARY';
BINDING: 'BINDING';
BOOLEAN: 'BOOLEAN';
@@ -524,7 +519,6 @@ TYPE: 'TYPE';
UNARCHIVE: 'UNARCHIVE';
UNBOUNDED: 'UNBOUNDED';
UNCACHE: 'UNCACHE';
-UNIFORM: 'UNIFORM';
UNION: 'UNION';
UNIQUE: 'UNIQUE';
UNKNOWN: 'UNKNOWN';
@@ -552,7 +546,6 @@ WEEKS: 'WEEKS';
WHEN: 'WHEN';
WHERE: 'WHERE';
WHILE: 'WHILE';
-WIDTH: 'WIDTH';
WINDOW: 'WINDOW';
WITH: 'WITH';
WITHIN: 'WITHIN';
diff --git
a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4
b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4
index c6617eb495df..744c472b2017 100644
---
a/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4
+++
b/sql/api/src/main/antlr4/org/apache/spark/sql/catalyst/parser/SqlBaseParser.g4
@@ -78,7 +78,7 @@ options { tokenVocab = SqlBaseLexer; }
la == ANTI || la == JOIN || la == UNION || la == EXCEPT ||
la == SETMINUS || la == INTERSECT || la == ORDER || la == CLUSTER ||
la == DISTRIBUTE || la == SORT || la == LIMIT || la == OFFSET ||
- la == AGGREGATE || la == WINDOW || la == LATERAL || la == BIN;
+ la == AGGREGATE || la == WINDOW || la == LATERAL;
}
}
@@ -1029,20 +1029,6 @@ unpivotAlias
: AS? errorCapturingIdentifier
;
-binByClause
- : BIN BY LEFT_PAREN
- RANGE rangeStart=multipartIdentifier TO rangeEnd=multipartIdentifier
- BIN WIDTH binWidth=expression
- (ALIGN TO origin=expression)?
- DISTRIBUTE UNIFORM LEFT_PAREN
- distributeCol+=multipartIdentifier
- (COMMA distributeCol+=multipartIdentifier)* RIGHT_PAREN
- (BIN_START AS binStartAlias=errorCapturingIdentifier)?
- (BIN_END AS binEndAlias=errorCapturingIdentifier)?
- (BIN_DISTRIBUTE_RATIO AS binRatioAlias=errorCapturingIdentifier)?
- RIGHT_PAREN (AS? tblAlias=errorCapturingIdentifier)?
- ;
-
lateralView
: LATERAL VIEW (OUTER)? qualifiedName LEFT_PAREN (expression (COMMA
expression)*)? RIGHT_PAREN tblName=identifier (AS? colName+=identifier (COMMA
colName+=identifier)*)?
;
@@ -1064,7 +1050,6 @@ relationExtension
: joinRelation
| pivotClause
| unpivotClause
- | binByClause
;
joinRelation
@@ -1921,7 +1906,6 @@ operatorPipeRightSide
// messages in the event that both are present (this is not allowed).
| pivotClause unpivotClause?
| unpivotClause pivotClause?
- | binByClause
| sample
| joinRelation
| operator=(UNION | EXCEPT | SETMINUS | INTERSECT) setQuantifier?
right=queryPrimary
@@ -1953,7 +1937,6 @@ ansiNonReserved
: ADD
| AFTER
| AGGREGATE
- | ALIGN
| ALTER
| ALWAYS
| ANALYZE
@@ -1970,10 +1953,6 @@ ansiNonReserved
| BERNOULLI
| BETWEEN
| BIGINT
- | BIN
- | BIN_DISTRIBUTE_RATIO
- | BIN_END
- | BIN_START
| BINARY
| BINARY_HEX
| BINDING
@@ -2275,7 +2254,6 @@ ansiNonReserved
| UNARCHIVE
| UNBOUNDED
| UNCACHE
- | UNIFORM
| UNLOCK
| UNPIVOT
| UNSET
@@ -2296,7 +2274,6 @@ ansiNonReserved
| WEEKS
| WHILE
| WATERMARK
- | WIDTH
| WINDOW
| WITHOUT
| YEAR
@@ -2338,7 +2315,6 @@ nonReserved
: ADD
| AFTER
| AGGREGATE
- | ALIGN
| ALL
| ALTER
| ALWAYS
@@ -2359,10 +2335,6 @@ nonReserved
| BERNOULLI
| BETWEEN
| BIGINT
- | BIN
- | BIN_DISTRIBUTE_RATIO
- | BIN_END
- | BIN_START
| BINARY
| BINARY_HEX
| BINDING
@@ -2718,7 +2690,6 @@ nonReserved
| UNARCHIVE
| UNBOUNDED
| UNCACHE
- | UNIFORM
| UNIQUE
| UNKNOWN
| UNLOCK
@@ -2744,7 +2715,6 @@ nonReserved
| WHILE
| WHEN
| WHERE
- | WIDTH
| WINDOW
| WITH
| WITHIN
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
index a97e0fac6ce5..48de2e347fea 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/Analyzer.scala
@@ -527,7 +527,6 @@ class Analyzer(
ResolveGroupingAnalytics ::
ResolvePivot ::
ResolveUnpivot ::
- ResolveBinBy ::
ResolveOrdinalInOrderByAndGroupBy ::
ExtractGenerator ::
ResolveGenerate ::
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
index 57aede9805d7..ec2ba4f69221 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/DeduplicateRelations.scala
@@ -197,14 +197,6 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
_.generatorOutput.map(_.exprId.id), newGenerate =>
newGenerate.copy(generatorOutput =
newGenerate.generatorOutput.map(_.newInstance())))
- case b: BinBy =>
- deduplicateAndRenew[BinBy](
- existingRelations,
- b,
- _.producedAttributes.map(_.exprId.id).toSeq,
- newBinBy => newBinBy.copy(appendedAttributes =
- newBinBy.appendedAttributes.map(_.newInstance())))
-
case e: Expand =>
deduplicateAndRenew[Expand](
existingRelations,
@@ -467,13 +459,6 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
newVersion.copyTagsFrom(oldVersion)
Seq((oldVersion, newVersion))
- case oldVersion: BinBy
- if
oldVersion.producedAttributes.intersect(conflictingAttributes).nonEmpty =>
- val newVersion = oldVersion.copy(
- appendedAttributes =
oldVersion.appendedAttributes.map(_.newInstance()))
- newVersion.copyTagsFrom(oldVersion)
- Seq((oldVersion, newVersion))
-
case oldVersion: Expand
if
oldVersion.producedAttributes.intersect(conflictingAttributes).nonEmpty =>
val producedAttributes = oldVersion.producedAttributes
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
deleted file mode 100644
index fc10020f3c7c..000000000000
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
+++ /dev/null
@@ -1,176 +0,0 @@
-/*
- * 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.SparkException
-import org.apache.spark.sql.catalyst.expressions.{Attribute, EmptyRow,
Expression, ExprId, NamedExpression}
-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.errors.QueryCompilationErrors
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types.{AnyTimestampType, DayTimeIntervalType,
NumericType, TimestampType}
-
-/**
- * Resolves [[UnresolvedBinBy]] into [[BinBy]]: looks up column references
against the child's
- * output, validates types and foldability, and captures the session local
time zone for the
- * physical execution to use.
- */
-object ResolveBinBy extends Rule[LogicalPlan] {
-
- override def apply(plan: LogicalPlan): LogicalPlan =
plan.resolveOperatorsWithPruning(
- _.containsPattern(UNRESOLVED_BIN_BY), ruleId) {
- case b: UnresolvedBinBy if !readyToResolve(b) => b
- case b: UnresolvedBinBy => resolve(b)
- }
-
- // `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.
- private def readyToResolve(b: UnresolvedBinBy): Boolean = {
- b.childrenResolved && b.binWidthExpr.resolved &&
b.originExpr.forall(_.resolved)
- }
-
- private def resolve(b: UnresolvedBinBy): LogicalPlan = {
- val resolver = SQLConf.get.resolver
- val child = b.child
-
- 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.binByInvalidBinWidthError(b.binWidthExpr)
- }
- v.asInstanceOf[Long]
- case _ =>
- throw QueryCompilationErrors.binByInvalidBinWidthError(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
- }
-
- if (distributeAttrs.isEmpty) {
- throw QueryCompilationErrors.binByMissingDistributeError()
- }
- distributeAttrs.foreach { attr =>
- attr.dataType match {
- case _: NumericType | _: DayTimeIntervalType => // ok
- case other =>
- throw
QueryCompilationErrors.binByDistributeTypeMismatchError(attr.name, other)
- }
- }
- val seen = mutable.HashSet.empty[ExprId]
- distributeAttrs.foreach { attr =>
- if (!seen.add(attr.exprId)) {
- throw
QueryCompilationErrors.binByDuplicateDistributeColumnError(attr.name)
- }
- }
-
- val appendedAttributes = BinBy.appendedAttributesWithAliases(rangeType,
b.outputAliases)
-
- BinBy(
- binWidthMicros = binWidthMicros,
- rangeStart = rangeStart,
- rangeEnd = rangeEnd,
- originMicros = originMicros,
- distributeColumns = distributeAttrs,
- appendedAttributes = appendedAttributes,
- child = child,
- timeZoneId = if (isLTZ) Some(sessionZone) else None)
- }
-
- private def resolveColumn(
- expr: Expression,
- child: LogicalPlan,
- resolver: Resolver): Attribute = expr match {
- case u: UnresolvedAttribute =>
- // 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 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.
- throw QueryCompilationErrors.binByRequiresTopLevelColumnError(ne.name)
- case other =>
- // Genuinely unexpected; the grammar restricts these positions to
multipartIdentifier.
- throw SparkException.internalError(
- s"Unexpected expression in BIN BY column position: $other")
- }
-}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
index bb7446c236cf..1e33212ba12d 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/parser/AstBuilder.scala
@@ -2303,40 +2303,6 @@ class AstBuilder extends DataTypeAstBuilder
(exprs, alias)
}
- /**
- * Add an [[UnresolvedBinBy]] to a logical plan; the analyzer rewrites it to
[[BinBy]] during
- * analysis.
- */
- private def withBinBy(
- ctx: BinByClauseContext,
- query: LogicalPlan): LogicalPlan = withOrigin(ctx) {
- val binWidth = expression(ctx.binWidth)
- val originExpr = Option(ctx.origin).map(expression)
- val rangeStart =
UnresolvedAttribute(visitMultipartIdentifier(ctx.rangeStart))
- val rangeEnd = UnresolvedAttribute(visitMultipartIdentifier(ctx.rangeEnd))
- val distributeColumns = ctx.distributeCol.asScala
- .map(c => UnresolvedAttribute(visitMultipartIdentifier(c))).toSeq
- val outputAliases = BinByOutputAliases(
- binStart = Option(ctx.binStartAlias).map(getIdentifierText),
- binEnd = Option(ctx.binEndAlias).map(getIdentifierText),
- binRatio = Option(ctx.binRatioAlias).map(getIdentifierText))
-
- val binBy = UnresolvedBinBy(
- binWidthExpr = binWidth,
- rangeStartCol = rangeStart,
- rangeEndCol = rangeEnd,
- originExpr = originExpr,
- distributeColumns = distributeColumns,
- outputAliases = outputAliases,
- child = query)
-
- if (ctx.tblAlias != null) {
- SubqueryAlias(getIdentifierText(ctx.tblAlias), binBy)
- } else {
- binBy
- }
- }
-
/**
* Add a [[Generate]] (Lateral View) to a logical plan.
*/
@@ -2400,11 +2366,9 @@ class AstBuilder extends DataTypeAstBuilder
withJoinRelation(extension.joinRelation(), left)
} else if (extension.pivotClause() != null) {
withPivot(extension.pivotClause(), left)
- } else if (extension.unpivotClause() != null) {
- withUnpivot(extension.unpivotClause(), left)
} else {
- assert(extension.binByClause() != null)
- withBinBy(extension.binByClause(), left)
+ assert(extension.unpivotClause() != null)
+ withUnpivot(extension.unpivotClause(), left)
}
}
}
@@ -7471,8 +7435,6 @@ class AstBuilder extends DataTypeAstBuilder
throw
QueryParsingErrors.unpivotWithPivotInFromClauseNotAllowedError(ctx)
}
withUnpivot(c, left)
- }.getOrElse(Option(ctx.binByClause()).map { c =>
- withBinBy(c, left)
}.getOrElse(Option(ctx.sample).map { c =>
withSample(c, left)
}.getOrElse(Option(ctx.joinRelation()).map { c =>
@@ -7484,7 +7446,7 @@ class AstBuilder extends DataTypeAstBuilder
withQueryResultClauses(c, PipeOperator(left), forPipeOperators = true)
}.getOrElse(
visitOperatorPipeAggregate(ctx, left)
- )))))))))))))
+ ))))))))))))
}
private def visitOperatorPipeSet(
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
index cd06b75c7f2c..d73a83e7356c 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/basicLogicalOperators.scala
@@ -17,7 +17,6 @@
package org.apache.spark.sql.catalyst.plans.logical
-import org.apache.spark.SparkException
import org.apache.spark.sql.catalyst.{AliasIdentifier, InternalRow,
SQLConfHelper}
import org.apache.spark.sql.catalyst.analysis.{Analyzer, AnsiTypeCoercion,
MultiInstanceRelation, Resolver, TypeCoercion, TypeCoercionBase,
UnresolvedUnaryNode, WidenStatefulOpNullability}
import org.apache.spark.sql.catalyst.catalog.{CatalogStorageFormat,
CatalogTable}
@@ -1736,124 +1735,6 @@ case class Unpivot(
}
-/**
- * Optional user-supplied names for the three columns appended by [[BinBy]]. A
`None` falls back
- * to the default name (`bin_start`, `bin_end`, `bin_distribute_ratio`) when
the output schema is
- * constructed during analysis.
- */
-case class BinByOutputAliases(
- binStart: Option[String] = None,
- binEnd: Option[String] = None,
- binRatio: Option[String] = None) {
- def effectiveBinStart: String = binStart.getOrElse("bin_start")
- def effectiveBinEnd: String = binEnd.getOrElse("bin_end")
- def effectiveBinRatio: String = binRatio.getOrElse("bin_distribute_ratio")
-}
-
-object BinByOutputAliases {
- val empty: BinByOutputAliases = BinByOutputAliases()
-}
-
-/**
- * Unresolved counterpart of [[BinBy]] produced by the parser. `ResolveBinBy`
enforces type and
- * foldability constraints and rewrites this into a resolved [[BinBy]].
- *
- * @param binWidthExpr Bin-width expression (DAY-TIME INTERVAL).
- * @param rangeStartCol Reference to the row's measurement-window start
column.
- * @param rangeEndCol Reference to the row's measurement-window end
column.
- * @param originExpr Optional alignment anchor expression (`ALIGN TO`
clause). `None`
- * when the clause is omitted; `ResolveBinBy`
defaults the resolved
- * plan's origin to `1970-01-01 00:00:00`.
- * @param distributeColumns Columns whose values are proportionally
redistributed across sub-rows.
- * @param outputAliases Optional renames for the three appended output
columns.
- * @param child Input relation.
- */
-case class UnresolvedBinBy(
- binWidthExpr: Expression,
- rangeStartCol: Expression,
- rangeEndCol: Expression,
- originExpr: Option[Expression],
- distributeColumns: Seq[Expression],
- outputAliases: BinByOutputAliases,
- child: LogicalPlan) extends UnresolvedUnaryNode {
-
- final override val nodePatterns: Seq[TreePattern] = Seq(UNRESOLVED_BIN_BY)
-
- override protected def withNewChildInternal(newChild: LogicalPlan):
UnresolvedBinBy =
- copy(child = newChild)
-}
-
-/**
- * Aligns range-typed rows to fixed-width bin boundaries by splitting any row
whose
- * `[range_start, range_end)` crosses a boundary and proportionally
redistributing values in
- * `distributeColumns` across the resulting sub-ranges. Emits one or more
output rows per input row
- * plus three appended columns with default names `bin_start`, `bin_end`,
`bin_distribute_ratio`.
- * The names come from the resolved `appendedAttributes`.
- *
- * Bin boundaries align to `originMicros + k * binWidthMicros` for integer `k`.
- * For `TIMESTAMP` (LTZ) inputs the boundary arithmetic uses civil-time in the
session zone for
- * multi-day bins; sub-day LTZ bins and `TIMESTAMP_NTZ` bins use UTC
microsecond arithmetic.
- *
- * Construction invariants (bin width is positive day-time, range columns are
timestamps and share
- * a type, origin matches the range column type, distribute columns are
numeric and distinct) are
- * enforced by `ResolveBinBy`. Code paths that construct `BinBy` outside the
analyzer are
- * responsible for upholding the same invariants; only the
`timeZoneId`-vs-range-type pairing is
- * checked in the constructor itself (raises `INTERNAL_ERROR`). The three
appended-column names
- * are not checked for collisions with `child.output`; a colliding name
surfaces as
- * `AMBIGUOUS_REFERENCE` when the output is referenced downstream rather than
at the BIN BY clause
- * itself.
- *
- * @param binWidthMicros Bin width in microseconds: the folded value of
the day-time interval
- * `BIN WIDTH` expression. Always positive.
- * @param rangeStart Resolved attribute holding each row's
window-start timestamp.
- * @param rangeEnd Resolved attribute holding each row's window-end
timestamp.
- * @param originMicros Alignment anchor in microseconds since the
epoch: the folded value of
- * `ALIGN TO`, or the type-specific default when
the clause is omitted.
- * @param distributeColumns Resolved columns to proportionally redistribute.
- * @param appendedAttributes The three output attributes appended after
`child.output`. Provided
- * by the analyzer; held in the case class so the
attribute `ExprId`s
- * remain stable across `.output` calls and
`withNewChildInternal`.
- * @param child Input relation.
- * @param timeZoneId Captured session local time zone for LTZ inputs;
`None` for NTZ.
- * Required when `rangeStart.dataType` is
`TimestampType`; must be
- * `None` when it is `TimestampNTZType`.
- */
-case class BinBy(
- binWidthMicros: Long,
- rangeStart: Attribute,
- rangeEnd: Attribute,
- originMicros: Long,
- distributeColumns: Seq[Attribute],
- appendedAttributes: Seq[Attribute],
- child: LogicalPlan,
- timeZoneId: Option[String])
- extends UnaryNode {
-
- if (timeZoneId.isDefined != rangeStart.dataType.isInstanceOf[TimestampType])
{
- throw SparkException.internalError(
- s"timeZoneId must be set iff rangeStart is TIMESTAMP (LTZ); got
rangeStart.dataType=" +
- s"${rangeStart.dataType}, timeZoneId=$timeZoneId")
- }
-
- override def output: Seq[Attribute] = child.output ++ appendedAttributes
-
- override def producedAttributes: AttributeSet =
AttributeSet(appendedAttributes)
-
- final override val nodePatterns: Seq[TreePattern] = Seq(BIN_BY)
-
- override protected def withNewChildInternal(newChild: LogicalPlan): BinBy =
- copy(child = newChild)
-}
-
-object BinBy {
- def appendedAttributesWithAliases(
- rangeType: DataType,
- aliases: BinByOutputAliases): Seq[Attribute] = Seq(
- AttributeReference(aliases.effectiveBinStart, rangeType, nullable =
true)(),
- AttributeReference(aliases.effectiveBinEnd, rangeType, nullable = true)(),
- AttributeReference(aliases.effectiveBinRatio, DoubleType, nullable =
true)())
-}
-
/**
* A logical plan node for creating a logical limit, which is split into two
separate logical nodes:
* a [[LocalLimit]], which is a partition local limit, followed by a
[[GlobalLimit]].
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
index 12fcf3ed9aa3..36fdd4706016 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/rules/RuleIdCollection.scala
@@ -100,7 +100,6 @@ object RuleIdCollection {
"org.apache.spark.sql.catalyst.analysis.ResolveRowLevelCommandAssignments" ::
"org.apache.spark.sql.catalyst.analysis.ResolveCursors" ::
"org.apache.spark.sql.catalyst.analysis.ResolveFetchCursor" ::
- "org.apache.spark.sql.catalyst.analysis.ResolveBinBy" ::
"org.apache.spark.sql.catalyst.analysis.ResolveSetVariable" ::
"org.apache.spark.sql.catalyst.analysis.ResolveTableConstraints" ::
"org.apache.spark.sql.catalyst.analysis.ResolveTableSpec" ::
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
index 79decd1c2e57..66bf9dcc4ce1 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/trees/TreePatterns.scala
@@ -136,7 +136,6 @@ object TreePattern extends Enumeration {
val AGGREGATE: Value = Value
val APPEND_COLUMNS: Value = Value
val AS_OF_JOIN: Value = Value
- val BIN_BY: Value = Value
val COLLECT_METRICS: Value = Value
val COMMAND: Value = Value
val CTE: Value = Value
@@ -194,7 +193,6 @@ object TreePattern extends Enumeration {
// Unresolved Plan patterns (Alphabetically ordered)
val NAMED_STREAMING_RELATION: Value = Value
val PLAN_WITH_UNRESOLVED_IDENTIFIER: Value = Value
- val UNRESOLVED_BIN_BY: Value = Value
val UNRESOLVED_EVENT_TIME_WATERMARK: Value = Value
val UNRESOLVED_HAVING: Value = Value
val UNRESOLVED_HINT: Value = Value
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 78e2e212969f..d5a9cc723bc3 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
@@ -297,88 +297,6 @@ private[sql] object QueryCompilationErrors extends
QueryErrorsBase with Compilat
messageParameters = Map("types" -> dataTypes.mkString(", ")))
}
- def binByAlignToTypeMismatchError(
- originType: DataType,
- rangeType: DataType): Throwable = {
- new AnalysisException(
- errorClass = "BIN_BY_ALIGN_TO_TYPE_MISMATCH",
- messageParameters = Map(
- "originType" -> toSQLType(originType),
- "rangeType" -> toSQLType(rangeType)))
- }
-
- def binByColumnNotFoundError(columnName: String): Throwable = {
- new AnalysisException(
- errorClass = "BIN_BY_COLUMN_NOT_FOUND",
- messageParameters = Map("columnName" -> toSQLId(columnName)))
- }
-
- def binByRequiresTopLevelColumnError(columnName: String): Throwable = {
- new AnalysisException(
- errorClass = "BIN_BY_REQUIRES_TOP_LEVEL_COLUMN",
- messageParameters = Map("columnName" -> toSQLId(columnName)))
- }
-
- def binByDistributeTypeMismatchError(
- columnName: String,
- columnType: DataType): Throwable = {
- new AnalysisException(
- errorClass = "BIN_BY_DISTRIBUTE_TYPE_MISMATCH",
- messageParameters = Map(
- "columnName" -> toSQLId(columnName),
- "columnType" -> toSQLType(columnType)))
- }
-
- def binByDuplicateDistributeColumnError(columnName: String): Throwable = {
- new AnalysisException(
- errorClass = "BIN_BY_DUPLICATE_DISTRIBUTE_COLUMN",
- messageParameters = Map("columnName" -> toSQLId(columnName)))
- }
-
- def binByInvalidAlignToError(expr: Expression): Throwable = {
- new AnalysisException(
- errorClass = "BIN_BY_INVALID_ALIGN_TO",
- messageParameters = Map("expr" -> toSQLExpr(expr)))
- }
-
- def binByInvalidBinWidthError(expr: Expression): Throwable = {
- new AnalysisException(
- errorClass = "BIN_BY_INVALID_BIN_WIDTH",
- messageParameters = Map("expr" -> toSQLExpr(expr)))
- }
-
- def binByMissingDistributeError(): Throwable = {
- new AnalysisException(
- errorClass = "BIN_BY_MISSING_DISTRIBUTE",
- messageParameters = Map.empty)
- }
-
- def binByNonFoldableInputError(inputName: String, expr: Expression):
Throwable = {
- new AnalysisException(
- errorClass = "DATATYPE_MISMATCH.NON_FOLDABLE_INPUT",
- messageParameters = Map(
- "sqlExpr" -> toSQLExpr(expr),
- "inputName" -> toSQLId(inputName),
- "inputType" -> toSQLType(expr.dataType),
- "inputExpr" -> toSQLExpr(expr)))
- }
-
- def binByNullArgumentError(clause: String): Throwable = {
- new AnalysisException(
- errorClass = "BIN_BY_NULL_ARGUMENT",
- messageParameters = Map("clause" -> clause))
- }
-
- def binByRangeTypeMismatchError(
- columnName: String,
- columnType: DataType): Throwable = {
- new AnalysisException(
- errorClass = "BIN_BY_RANGE_TYPE_MISMATCH",
- messageParameters = Map(
- "columnName" -> toSQLId(columnName),
- "columnType" -> toSQLType(columnType)))
- }
-
def unsupportedIfNotExistsError(tableName: String): Throwable = {
new AnalysisException(
errorClass = "UNSUPPORTED_FEATURE.INSERT_PARTITION_SPEC_IF_NOT_EXISTS",
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
deleted file mode 100644
index f2c3b07da22c..000000000000
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
+++ /dev/null
@@ -1,291 +0,0 @@
-/*
- * 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 org.apache.spark.SparkThrowable
-import org.apache.spark.sql.catalyst.QueryPlanningTracker
-import org.apache.spark.sql.catalyst.dsl.expressions._
-import org.apache.spark.sql.catalyst.expressions.{AttributeReference, Cast,
Expression, Literal}
-import org.apache.spark.sql.catalyst.plans.Inner
-import org.apache.spark.sql.catalyst.plans.logical.{BinBy, BinByOutputAliases,
Join, JoinHint, LocalRelation, LogicalPlan, SubqueryAlias, UnresolvedBinBy}
-import org.apache.spark.sql.internal.SQLConf
-import org.apache.spark.sql.types._
-
-class ResolveBinBySuite extends AnalysisTest {
-
- private val tsStart = $"ts_start".timestamp
- private val tsEnd = $"ts_end".timestamp
- private val tsStartNtz = $"ts_start".timestampNTZ
- private val tsEndNtz = $"ts_end".timestampNTZ
- private val value = $"value".long
- private val label = $"label".string
-
- private val ltzChild: LogicalPlan = LocalRelation(tsStart, tsEnd, value,
label)
- private val ntzChild: LogicalPlan = LocalRelation(tsStartNtz, tsEndNtz,
value)
-
- private val fiveMinutes: Expression = Literal(5L * 60L * 1000000L,
DayTimeIntervalType())
- private val ltzOrigin: Expression = Literal(0L, TimestampType)
- private val ntzOrigin: Expression = Literal(0L, TimestampNTZType)
-
- private def unresolved(
- child: LogicalPlan = ltzChild,
- binWidth: Expression = fiveMinutes,
- rangeStart: Expression = tsStart,
- rangeEnd: Expression = tsEnd,
- originExpr: Option[Expression] = Some(ltzOrigin),
- distribute: Seq[Expression] = Seq(value),
- aliases: BinByOutputAliases = BinByOutputAliases.empty): UnresolvedBinBy
=
- UnresolvedBinBy(binWidth, rangeStart, rangeEnd, originExpr, distribute,
aliases, child)
-
- private def expectError(u: UnresolvedBinBy, condition: String): Unit = {
- val ex = intercept[SparkThrowable](ResolveBinBy.apply(u))
- assert(ex.getCondition == condition,
- s"expected condition '$condition' but got '${ex.getCondition}'")
- }
-
- test("user-supplied foldable ALIGN TO of matching type is folded to micros")
{
- val origin = Some(Literal(123456L, TimestampType))
- val resolved = ResolveBinBy.apply(unresolved(originExpr = origin))
- assert(resolved.asInstanceOf[BinBy].originMicros == 123456L)
- }
-
- test("omitted ALIGN TO fills the default origin per type and session zone") {
- // LTZ, UTC: epoch.
- withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
- val utc = ResolveBinBy.apply(unresolved(originExpr =
None)).asInstanceOf[BinBy]
- assert(utc.originMicros == 0L)
- }
-
- // LTZ, non-UTC: shifted by the zone offset. LA is UTC-8 on 1970-01-01, so
local
- // midnight is UTC 08:00 = 28800000000 micros.
- withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
- val la = ResolveBinBy.apply(unresolved(originExpr =
None)).asInstanceOf[BinBy]
- assert(la.originMicros == 8L * 3600L * 1000000L)
- }
-
- // NTZ: epoch, zone-independent.
- val ntz = ResolveBinBy.apply(
- unresolved(child = ntzChild, rangeStart = tsStartNtz, rangeEnd =
tsEndNtz,
- originExpr = None)).asInstanceOf[BinBy]
- assert(ntz.originMicros == 0L)
- }
-
- test("rejects invalid ALIGN TO") {
- // Type must match the range columns, both directions.
- expectError(unresolved(originExpr = Some(ntzOrigin)),
"BIN_BY_ALIGN_TO_TYPE_MISMATCH")
- expectError(
- unresolved(child = ntzChild, rangeStart = tsStartNtz, rangeEnd =
tsEndNtz,
- originExpr = Some(ltzOrigin)),
- "BIN_BY_ALIGN_TO_TYPE_MISMATCH")
- // Must be foldable.
- val originAttr = AttributeReference("o", TimestampType, nullable = true)()
- val childWithOrigin = LocalRelation(tsStart, tsEnd, value, originAttr)
- expectError(
- unresolved(child = childWithOrigin, originExpr = Some(originAttr)),
- "DATATYPE_MISMATCH.NON_FOLDABLE_INPUT")
- // A foldable but null origin is rejected as a null argument.
- expectError(
- unresolved(originExpr = Some(Literal(null, TimestampType))),
"BIN_BY_NULL_ARGUMENT")
- // A foldable CAST that throws on eval (ANSI) surfaces cleanly, not as a
raw exception.
- // TimestampType cast needs a resolved time zone, mirroring the
post-ResolveTimeZone state.
- withSQLConf(SQLConf.ANSI_ENABLED.key -> "true") {
- expectError(
- unresolved(originExpr =
- Some(Cast(Literal.create("not a ts", StringType), TimestampType,
Some("UTC")))),
- "BIN_BY_INVALID_ALIGN_TO")
- }
- }
-
- 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
- // 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()))
- }
- // Non-foldable is rejected before evaluation.
- val widthAttr = AttributeReference("w", DayTimeIntervalType(), nullable =
true)()
- val childWithWidth = LocalRelation(tsStart, tsEnd, value, widthAttr)
- expectError(
- unresolved(child = childWithWidth, binWidth = widthAttr),
- "DATATYPE_MISMATCH.NON_FOLDABLE_INPUT")
- }
-
- test("captures the session zone for LTZ inputs and none for NTZ") {
- // LTZ: columns resolve and a zone is captured.
- val ltz = ResolveBinBy.apply(unresolved()).asInstanceOf[BinBy]
- assert(ltz.rangeStart == tsStart)
- assert(ltz.rangeEnd == tsEnd)
- assert(ltz.distributeColumns == Seq(value))
- assert(ltz.timeZoneId.isDefined)
-
- // The configured zone is the one captured.
- withSQLConf(SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
- val la = ResolveBinBy.apply(unresolved()).asInstanceOf[BinBy]
- assert(la.timeZoneId.contains("America/Los_Angeles"))
- }
-
- // NTZ: no zone captured.
- val ntz = ResolveBinBy.apply(
- unresolved(child = ntzChild, rangeStart = tsStartNtz, rangeEnd =
tsEndNtz,
- originExpr = Some(ntzOrigin))).asInstanceOf[BinBy]
- assert(ntz.timeZoneId.isEmpty)
- }
-
- test("resolved BinBy output schema appends three columns, honoring renames")
{
- // Default names and types.
- val default = ResolveBinBy.apply(unresolved()).asInstanceOf[BinBy]
- assert(default.output.length == ltzChild.output.length + 3)
- val appended = default.output.takeRight(3)
- assert(appended.map(_.name) == Seq("bin_start", "bin_end",
"bin_distribute_ratio"))
- assert(appended.map(_.dataType) == Seq(TimestampType, TimestampType,
DoubleType))
-
- // All three renamed.
- val full = ResolveBinBy.apply(
- unresolved(aliases = BinByOutputAliases(Some("ws"), Some("we"),
Some("frac"))))
- .asInstanceOf[BinBy]
- assert(full.output.takeRight(3).map(_.name) == Seq("ws", "we", "frac"))
-
- // Partial rename; the other two keep defaults.
- val partial = ResolveBinBy.apply(
- unresolved(aliases = BinByOutputAliases(binStart =
Some("ws")))).asInstanceOf[BinBy]
- assert(partial.output.takeRight(3).map(_.name) == Seq("ws", "bin_end",
"bin_distribute_ratio"))
- }
-
- test("resolves UnresolvedAttribute references against child output") {
- val tsStartU = UnresolvedAttribute("ts_start")
- val tsEndU = UnresolvedAttribute("ts_end")
- val valueU = UnresolvedAttribute("value")
- val result = ResolveBinBy.apply(
- unresolved(rangeStart = tsStartU, rangeEnd = tsEndU, distribute =
Seq(valueU)))
- val bi = result.asInstanceOf[BinBy]
- assert(bi.rangeStart.exprId == tsStart.exprId)
- assert(bi.rangeEnd.exprId == tsEnd.exprId)
- assert(bi.distributeColumns.map(_.exprId) == Seq(value.exprId))
- }
-
- test("multipart identifiers disambiguate same-name columns across a JOIN") {
- val t1Start = AttributeReference("ts_start", TimestampType, nullable =
true)()
- val t1End = AttributeReference("ts_end", TimestampType, nullable = true)()
- val t2Start = AttributeReference("ts_start", TimestampType, nullable =
true)()
- val t2End = AttributeReference("ts_end", TimestampType, nullable = true)()
- val t2Value = AttributeReference("value", LongType, nullable = true)()
- val t1 = SubqueryAlias("t1", LocalRelation(t1Start, t1End))
- val t2 = SubqueryAlias("t2", LocalRelation(t2Start, t2End, t2Value))
- val join = Join(t1, t2, Inner, None, JoinHint.NONE)
-
- // t1. picks t1's columns.
- val u1 = UnresolvedBinBy(
- binWidthExpr = fiveMinutes,
- rangeStartCol = UnresolvedAttribute(Seq("t1", "ts_start")),
- rangeEndCol = UnresolvedAttribute(Seq("t1", "ts_end")),
- originExpr = Some(ltzOrigin),
- distributeColumns = Seq(UnresolvedAttribute(Seq("t2", "value"))),
- outputAliases = BinByOutputAliases.empty,
- child = join)
- val resolved1 = ResolveBinBy.apply(u1).asInstanceOf[BinBy]
- assert(resolved1.rangeStart.exprId == t1Start.exprId)
- assert(resolved1.rangeEnd.exprId == t1End.exprId)
-
- // t2. picks t2's columns.
- val u2 = UnresolvedBinBy(
- binWidthExpr = fiveMinutes,
- rangeStartCol = UnresolvedAttribute(Seq("t2", "ts_start")),
- rangeEndCol = UnresolvedAttribute(Seq("t2", "ts_end")),
- originExpr = Some(ltzOrigin),
- distributeColumns = Seq(UnresolvedAttribute(Seq("t2", "value"))),
- outputAliases = BinByOutputAliases.empty,
- child = join)
- val resolved2 = ResolveBinBy.apply(u2).asInstanceOf[BinBy]
- assert(resolved2.rangeStart.exprId == t2Start.exprId)
- assert(resolved2.rangeEnd.exprId == t2End.exprId)
- }
-
- test("rejects unresolvable column references with BIN_BY_COLUMN_NOT_FOUND") {
- expectError(
- unresolved(rangeStart = UnresolvedAttribute("nonexistent")),
"BIN_BY_COLUMN_NOT_FOUND")
- expectError(
- unresolved(distribute = Seq(UnresolvedAttribute("nonexistent"))),
"BIN_BY_COLUMN_NOT_FOUND")
- }
-
- test("rejects nested column refs through the full analyzer
(RULE_ORDERING_DEPENDENCIES)") {
- val structCol = AttributeReference(
- "outer",
- StructType(Seq(
- StructField("ts_start", TimestampType, nullable = true),
- StructField("ts_end", TimestampType, nullable = true))),
- nullable = true)()
- val numeric = AttributeReference("value", LongType, nullable = true)()
- val plan = UnresolvedBinBy(
- binWidthExpr = fiveMinutes,
- rangeStartCol = UnresolvedAttribute(Seq("outer", "ts_start")),
- rangeEndCol = UnresolvedAttribute(Seq("outer", "ts_end")),
- originExpr = Some(ltzOrigin),
- distributeColumns = Seq(numeric),
- outputAliases = BinByOutputAliases.empty,
- child = LocalRelation(structCol, numeric))
- assertAnalysisErrorCondition(
- plan,
- "BIN_BY_REQUIRES_TOP_LEVEL_COLUMN",
- Map("columnName" -> "`ts_start`"))
- }
-
- test("rejects non-timestamp or mismatched RANGE columns") {
- // Range columns must be TIMESTAMP/TIMESTAMP_NTZ...
- expectError(unresolved(rangeStart = value), "BIN_BY_RANGE_TYPE_MISMATCH")
- // ...and both must share the same type.
- expectError(unresolved(rangeEnd = tsEndNtz), "BIN_BY_RANGE_TYPE_MISMATCH")
- }
-
- test("rejects invalid DISTRIBUTE UNIFORM columns") {
- // Empty (defensive: the parser also rejects this), non-numeric, and
duplicate columns.
- expectError(unresolved(distribute = Seq.empty),
"BIN_BY_MISSING_DISTRIBUTE")
- expectError(unresolved(distribute = Seq(label)),
"BIN_BY_DISTRIBUTE_TYPE_MISMATCH")
- expectError(unresolved(distribute = Seq(value, value)),
"BIN_BY_DUPLICATE_DISTRIBUTE_COLUMN")
- }
-
- test("BinBy survives full Analyzer + CheckAnalysis (regression for
producedAttributes)") {
- // Without BinBy.producedAttributes, CheckAnalysis would flag
MISSING_ATTRIBUTES.
- assertAnalysisSuccess(unresolved())
- assertAnalysisSuccess(
- unresolved(aliases = BinByOutputAliases(Some("ws"), Some("we"),
Some("frac"))))
- }
-
- test("self-join over a shared BinBy subtree is deduplicated") {
- // Reference the same resolved BinBy subtree on both sides of a join: the
appended
- // bin_start/bin_end/bin_distribute_ratio attributes start with identical
exprIds.
- val binBy = ResolveBinBy.apply(unresolved()).asInstanceOf[BinBy]
- val selfJoin = Join(binBy, binBy, Inner, None, JoinHint.NONE)
-
- // DeduplicateRelations must renew the right side's appended attributes;
without the
- // BinBy cases in both dedup phases the conflicting exprIds make analysis
fail.
- val analyzed = getAnalyzer.executeAndCheck(selfJoin, new
QueryPlanningTracker)
-
- val binBys = analyzed.collect { case b: BinBy => b }
- assert(binBys.size == 2, s"expected two BinBy nodes, got ${binBys.size}")
- val appendedExprIds = binBys.flatMap(_.appendedAttributes.map(_.exprId))
- assert(appendedExprIds.distinct.size == appendedExprIds.size,
- "appended BinBy attributes must have distinct exprIds across the two
join sides")
- }
-}
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
index e20071b3de62..fe1a7a81a2c5 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/parser/PlanParserSuite.scala
@@ -2753,174 +2753,6 @@ class PlanParserSuite extends AnalysisTest {
parameters = Map("error" -> "'ts'", "hint" -> ""))
}
- test("BIN BY: clause parses into the expected UnresolvedBinBy") {
- // Minimal clause: all optionals (ALIGN TO, output renames) omitted.
- val minimal = singleBinBy(parsePlan(
- """SELECT * FROM metrics BIN BY (
- | RANGE ts_start TO ts_end
- | BIN WIDTH INTERVAL '5' MINUTE
- | DISTRIBUTE UNIFORM (value)
- |)""".stripMargin))
- assert(minimal.rangeStartCol == UnresolvedAttribute(Seq("ts_start")))
- assert(minimal.rangeEndCol == UnresolvedAttribute(Seq("ts_end")))
- assert(minimal.originExpr.isEmpty)
- assert(minimal.distributeColumns == Seq(UnresolvedAttribute(Seq("value"))))
- assert(minimal.outputAliases == BinByOutputAliases.empty)
-
- // Qualified column references and a partial HOUR TO MINUTE interval as
BIN WIDTH.
- val qualified = singleBinBy(parsePlan(
- """SELECT * FROM t1 JOIN t2 ON t1.id = t2.id BIN BY (
- | RANGE t1.ts_start TO t1.ts_end
- | BIN WIDTH INTERVAL '5:30' HOUR TO MINUTE
- | DISTRIBUTE UNIFORM (t1.value)
- |)""".stripMargin))
- assert(qualified.rangeStartCol == UnresolvedAttribute(Seq("t1",
"ts_start")))
- assert(qualified.rangeEndCol == UnresolvedAttribute(Seq("t1", "ts_end")))
- assert(qualified.distributeColumns == Seq(UnresolvedAttribute(Seq("t1",
"value"))))
-
- // Maximal clause: explicit ALIGN TO, multiple DISTRIBUTE UNIFORM columns,
all three renames.
- val maximal = singleBinBy(parsePlan(
- """SELECT * FROM metrics BIN BY (
- | RANGE ts_start TO ts_end
- | BIN WIDTH INTERVAL '1' HOUR
- | ALIGN TO TIMESTAMP '2024-01-01 00:30:00'
- | DISTRIBUTE UNIFORM (bytes_sent, requests)
- | BIN_START AS ws
- | BIN_END AS we
- | BIN_DISTRIBUTE_RATIO AS frac
- |)""".stripMargin))
- assert(maximal.originExpr.contains(parseExpression("TIMESTAMP '2024-01-01
00:30:00'")))
- assert(maximal.distributeColumns == Seq(
- UnresolvedAttribute(Seq("bytes_sent")),
- UnresolvedAttribute(Seq("requests"))))
- assert(maximal.outputAliases.binStart.contains("ws"))
- assert(maximal.outputAliases.binEnd.contains("we"))
- assert(maximal.outputAliases.binRatio.contains("frac"))
- }
-
- test("BIN BY: malformed clauses raise PARSE_SYNTAX_ERROR") {
- def assertSyntaxError(query: String): Unit =
- assert(parseException(query).getCondition == "PARSE_SYNTAX_ERROR")
-
- // BIN WIDTH missing.
- assertSyntaxError(
- """SELECT * FROM metrics BIN BY (
- | RANGE ts_start TO ts_end
- | DISTRIBUTE UNIFORM (value)
- |)""".stripMargin)
-
- // DISTRIBUTE UNIFORM missing.
- assertSyntaxError(
- """SELECT * FROM metrics BIN BY (
- | RANGE ts_start TO ts_end
- | BIN WIDTH INTERVAL '5' MINUTE
- |)""".stripMargin)
-
- // DISTRIBUTE UNIFORM with an empty column list.
- assertSyntaxError(
- """SELECT * FROM metrics BIN BY (
- | RANGE ts_start TO ts_end
- | BIN WIDTH INTERVAL '5' MINUTE
- | DISTRIBUTE UNIFORM ()
- |)""".stripMargin)
-
- // RANGE missing.
- assertSyntaxError(
- """SELECT * FROM metrics BIN BY (
- | BIN WIDTH INTERVAL '5' MINUTE
- | DISTRIBUTE UNIFORM (value)
- |)""".stripMargin)
-
- // Clauses out of order: BIN WIDTH must come after RANGE.
- assertSyntaxError(
- """SELECT * FROM metrics BIN BY (
- | BIN WIDTH INTERVAL '5' MINUTE
- | RANGE ts_start TO ts_end
- | DISTRIBUTE UNIFORM (value)
- |)""".stripMargin)
- }
-
- test("BIN BY: new keywords stay non-reserved and `bin` still parses as a
table alias") {
- // The seven new keywords are in `ansiNonReserved`, so they stay usable as
plain identifiers
- // outside a BIN BY clause.
- assert(binByNodes(parsePlan(
- "SELECT bin, width, uniform, align, bin_start, bin_end,
bin_distribute_ratio " +
- "FROM t WHERE bin > 0")).isEmpty)
-
- // `bin` not followed by BY stays a table alias, both bare and when
referenced downstream.
- assert(binByNodes(parsePlan("SELECT * FROM t bin")).isEmpty)
- assert(binByNodes(parsePlan("SELECT bin.x FROM t bin WHERE bin.x >
0")).isEmpty)
- }
-
- test("BIN BY: composition, chaining, trailing alias, and pipe operator") {
- // Composes with a subquery in FROM and a downstream WHERE.
- assert(binByNodes(parsePlan(
- """SELECT * FROM (SELECT * FROM metrics) BIN BY (
- | RANGE ts_start TO ts_end
- | BIN WIDTH INTERVAL '5' MINUTE
- | DISTRIBUTE UNIFORM (value)
- |) WHERE value > 0""".stripMargin)).size == 1)
-
- // Composes after PIVOT.
- assert(binByNodes(parsePlan(
- """SELECT * FROM events
- |PIVOT (sum(amount) FOR category IN ('a', 'b'))
- |BIN BY (
- | RANGE ts_start TO ts_end
- | BIN WIDTH INTERVAL '5' MINUTE
- | DISTRIBUTE UNIFORM (a, b)
- |)""".stripMargin)).size == 1)
-
- // Two BIN BY clauses chain on the same relation.
- assert(binByNodes(parsePlan(
- """SELECT * FROM metrics
- |BIN BY (
- | RANGE ts_start TO ts_end
- | BIN WIDTH INTERVAL '5' MINUTE
- | DISTRIBUTE UNIFORM (value)
- |)
- |BIN BY (
- | RANGE bin_start TO bin_end
- | BIN WIDTH INTERVAL '1' HOUR
- | DISTRIBUTE UNIFORM (value)
- | BIN_START AS hr_start
- | BIN_END AS hr_end
- | BIN_DISTRIBUTE_RATIO AS hr_ratio
- |)""".stripMargin)).size == 2)
-
- // Trailing alias, with and without the AS keyword, wraps the BinBy in a
SubqueryAlias.
- Seq(") AS binned", ") binned").foreach { tail =>
- val parsed = parsePlan(
- s"""SELECT * FROM metrics BIN BY (
- | RANGE ts_start TO ts_end
- | BIN WIDTH INTERVAL '5' MINUTE
- | DISTRIBUTE UNIFORM (value)
- |$tail""".stripMargin)
- val aliases = parsed.collect { case s: SubqueryAlias if s.alias ==
"binned" => s }
- assert(aliases.size == 1, s"alias tail '$tail' did not produce a
SubqueryAlias")
- assert(aliases.head.child.isInstanceOf[UnresolvedBinBy])
- }
-
- // Works as a SQL pipe-operator stage with |>.
- assert(binByNodes(parsePlan(
- """FROM metrics
- ||> BIN BY (
- | RANGE ts_start TO ts_end
- | BIN WIDTH INTERVAL '5' MINUTE
- | DISTRIBUTE UNIFORM (value)
- |)""".stripMargin)).size == 1)
- }
-
private def intercept(sqlCommand: String, messages: String*): Unit =
interceptParseException(parsePlan)(sqlCommand, messages: _*)()
-
- // Helpers shared by the BIN BY parser tests.
- private def binByNodes(plan: LogicalPlan): Seq[UnresolvedBinBy] =
- plan.collect { case b: UnresolvedBinBy => b }
-
- private def singleBinBy(plan: LogicalPlan): UnresolvedBinBy = {
- val nodes = binByNodes(plan)
- assert(nodes.size == 1, s"expected exactly one BIN BY node, got
${nodes.size}")
- nodes.head
- }
}
diff --git
a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala
b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala
index 747d9dc5d0ff..1f525a541daa 100644
---
a/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala
+++
b/sql/connect/client/jdbc/src/test/scala/org/apache/spark/sql/connect/client/jdbc/SparkConnectDatabaseMetaDataSuite.scala
@@ -210,7 +210,7 @@ class SparkConnectDatabaseMetaDataSuite extends
ConnectFunSuite with RemoteSpark
val metadata = conn.getMetaData
// scalastyle:off line.size.limit
// CURRENT_PATH and SYSTEM are excluded: getSQLKeywords drops SQL:2003
reserved words (see companion).
- assert(metadata.getSQLKeywords ===
"ADD,AFTER,AGGREGATE,ALIGN,ALWAYS,ANALYZE,ANTI,ANY_VALUE,APPROX,ARCHIVE,ASC,BERNOULLI,BIN,BINDING,BIN_DISTRIBUTE_RATIO,BIN_END,BIN_START,BUCKET,BUCKETS,BYTE,CACHE,CASCADE,CATALOG,CATALOGS,CHANGE,CHANGES,CLEAR,CLUSTER,CLUSTERED,CODEGEN,COLLATION,COLLATIONS,COLLECTION,COLUMNS,COMMENT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONTAINS,CONTINUE,COST,CURRENT_DATABASE,CURRENT_SCHEMA,DATA,DATABASE,DATABASES,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,
[...]
+ assert(metadata.getSQLKeywords ===
"ADD,AFTER,AGGREGATE,ALWAYS,ANALYZE,ANTI,ANY_VALUE,APPROX,ARCHIVE,ASC,BERNOULLI,BINDING,BUCKET,BUCKETS,BYTE,CACHE,CASCADE,CATALOG,CATALOGS,CHANGE,CHANGES,CLEAR,CLUSTER,CLUSTERED,CODEGEN,COLLATION,COLLATIONS,COLLECTION,COLUMNS,COMMENT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE,CONCATENATE,CONTAINS,CONTINUE,COST,CURRENT_DATABASE,CURRENT_SCHEMA,DATA,DATABASE,DATABASES,DATEADD,DATEDIFF,DATE_ADD,DATE_DIFF,DAYOFYEAR,DAYS,DBPROPERTIES,DEFAULT_PATH,DEFINED,
[...]
// scalastyle:on line.size.limit
}
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
index 5bc729193535..1729e2dd0dfc 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/SparkStrategies.scala
@@ -937,8 +937,6 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
throw SparkException.internalError(
"Deduplicate operator for non streaming data source should have been
replaced " +
"by aggregate in the optimizer")
- case _: logical.BinBy =>
- throw new
SparkUnsupportedOperationException("UNSUPPORTED_FEATURE.BIN_BY")
case logical.DeserializeToObject(deserializer, objAttr, child) =>
execution.DeserializeToObjectExec(deserializer, objAttr,
planLater(child)) :: Nil
diff --git
a/sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out
b/sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out
index 8482a10b2292..6bcbdd2840f9 100644
--- a/sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/keywords-enforced.sql.out
@@ -7,7 +7,6 @@ struct<keyword:string,reserved:boolean>
ADD false
AFTER false
AGGREGATE false
-ALIGN false
ALL true
ALTER false
ALWAYS false
@@ -29,12 +28,8 @@ BEGIN false
BERNOULLI false
BETWEEN false
BIGINT false
-BIN false
BINARY false
BINDING false
-BIN_DISTRIBUTE_RATIO false
-BIN_END false
-BIN_START false
BOOLEAN false
BOTH true
BUCKET false
@@ -397,7 +392,6 @@ TYPE false
UNARCHIVE false
UNBOUNDED false
UNCACHE false
-UNIFORM false
UNION true
UNIQUE true
UNKNOWN true
@@ -425,7 +419,6 @@ WEEKS false
WHEN true
WHERE true
WHILE false
-WIDTH false
WINDOW false
WITH true
WITHIN true
diff --git a/sql/core/src/test/resources/sql-tests/results/keywords.sql.out
b/sql/core/src/test/resources/sql-tests/results/keywords.sql.out
index 4a48c1c01d7b..a01034326446 100644
--- a/sql/core/src/test/resources/sql-tests/results/keywords.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/keywords.sql.out
@@ -7,7 +7,6 @@ struct<keyword:string,reserved:boolean>
ADD false
AFTER false
AGGREGATE false
-ALIGN false
ALL false
ALTER false
ALWAYS false
@@ -29,12 +28,8 @@ BEGIN false
BERNOULLI false
BETWEEN false
BIGINT false
-BIN false
BINARY false
BINDING false
-BIN_DISTRIBUTE_RATIO false
-BIN_END false
-BIN_START false
BOOLEAN false
BOTH false
BUCKET false
@@ -397,7 +392,6 @@ TYPE false
UNARCHIVE false
UNBOUNDED false
UNCACHE false
-UNIFORM false
UNION false
UNIQUE false
UNKNOWN false
@@ -425,7 +419,6 @@ WEEKS false
WHEN false
WHERE false
WHILE false
-WIDTH false
WINDOW false
WITH false
WITHIN false
diff --git
a/sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out
b/sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out
index 4a48c1c01d7b..a01034326446 100644
--- a/sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out
+++ b/sql/core/src/test/resources/sql-tests/results/nonansi/keywords.sql.out
@@ -7,7 +7,6 @@ struct<keyword:string,reserved:boolean>
ADD false
AFTER false
AGGREGATE false
-ALIGN false
ALL false
ALTER false
ALWAYS false
@@ -29,12 +28,8 @@ BEGIN false
BERNOULLI false
BETWEEN false
BIGINT false
-BIN false
BINARY false
BINDING false
-BIN_DISTRIBUTE_RATIO false
-BIN_END false
-BIN_START false
BOOLEAN false
BOTH false
BUCKET false
@@ -397,7 +392,6 @@ TYPE false
UNARCHIVE false
UNBOUNDED false
UNCACHE false
-UNIFORM false
UNION false
UNIQUE false
UNKNOWN false
@@ -425,7 +419,6 @@ WEEKS false
WHEN false
WHERE false
WHILE false
-WIDTH false
WINDOW false
WITH false
WITHIN false
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
deleted file mode 100644
index eab8a3bafc1b..000000000000
--- a/sql/core/src/test/scala/org/apache/spark/sql/BinBySuite.scala
+++ /dev/null
@@ -1,51 +0,0 @@
-/*
- * 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 org.apache.spark.SparkUnsupportedOperationException
-import org.apache.spark.sql.test.SharedSparkSession
-
-class BinBySuite extends QueryTest with SharedSparkSession {
-
- test("BIN BY analyzes but physical execution is not yet implemented") {
- withTempView("metrics") {
- spark.sql(
- """SELECT TIMESTAMP '2024-01-01 00:00:00' AS ts_start,
- | TIMESTAMP '2024-01-01 01:00:00' AS ts_end,
- | 1L AS
value""".stripMargin).createOrReplaceTempView("metrics")
- val df = spark.sql(
- """SELECT * FROM metrics BIN BY (
- | RANGE ts_start TO ts_end
- | BIN WIDTH INTERVAL '5' MINUTE
- | DISTRIBUTE UNIFORM (value)
- |)""".stripMargin)
-
- // Analysis is fully functional in this PR.
- df.queryExecution.assertAnalyzed()
-
- // Physical execution is stubbed until the follow-up PR; planning
surfaces a clean
- // UNSUPPORTED_FEATURE error rather than an internal error.
- checkError(
- exception = intercept[SparkUnsupportedOperationException] {
- df.collect()
- },
- condition = "UNSUPPORTED_FEATURE.BIN_BY",
- parameters = Map.empty[String, String])
- }
- }
-}
diff --git
a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala
b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala
index e20beaa15181..1ecf5b3dae4a 100644
---
a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala
+++
b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/ThriftServerWithSparkContextSuite.scala
@@ -214,7 +214,7 @@ trait ThriftServerWithSparkContextSuite extends
SharedThriftServer {
val sessionHandle = client.openSession(user, "")
val infoValue = client.getInfo(sessionHandle,
GetInfoType.CLI_ODBC_KEYWORDS)
// scalastyle:off line.size.limit
- assert(infoValue.getStringValue ==
"ADD,AFTER,AGGREGATE,ALIGN,ALL,ALTER,ALWAYS,ANALYZE,AND,ANTI,ANY,ANY_VALUE,APPROX,ARCHIVE,ARRAY,AS,ASC,ASENSITIVE,AT,ATOMIC,AUTHORIZATION,BEGIN,BERNOULLI,BETWEEN,BIGINT,BIN,BINARY,BINDING,BIN_DISTRIBUTE_RATIO,BIN_END,BIN_START,BOOLEAN,BOTH,BUCKET,BUCKETS,BY,BYTE,CACHE,CALL,CALLED,CASCADE,CASE,CAST,CATALOG,CATALOGS,CHANGE,CHANGES,CHAR,CHARACTER,CHECK,CLEAR,CLOSE,CLUSTER,CLUSTERED,CODEGEN,COLLATE,COLLATION,COLLATIONS,COLLECTION,COLUMN,COLUMNS,COMMEN
[...]
+ assert(infoValue.getStringValue ==
"ADD,AFTER,AGGREGATE,ALL,ALTER,ALWAYS,ANALYZE,AND,ANTI,ANY,ANY_VALUE,APPROX,ARCHIVE,ARRAY,AS,ASC,ASENSITIVE,AT,ATOMIC,AUTHORIZATION,BEGIN,BERNOULLI,BETWEEN,BIGINT,BINARY,BINDING,BOOLEAN,BOTH,BUCKET,BUCKETS,BY,BYTE,CACHE,CALL,CALLED,CASCADE,CASE,CAST,CATALOG,CATALOGS,CHANGE,CHANGES,CHAR,CHARACTER,CHECK,CLEAR,CLOSE,CLUSTER,CLUSTERED,CODEGEN,COLLATE,COLLATION,COLLATIONS,COLLECTION,COLUMN,COLUMNS,COMMENT,COMMIT,COMPACT,COMPACTIONS,COMPENSATION,COMPUTE
[...]
// scalastyle:on line.size.limit
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
