This is an automated email from the ASF dual-hosted git repository.
cloud-fan 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 99f574b0b4bd [SPARK-58003][SQL] Add physical execution for BIN BY
99f574b0b4bd is described below
commit 99f574b0b4bdca6c6a80424b956838dc97ff2127
Author: Nikolina Vraneš <[email protected]>
AuthorDate: Wed Jul 15 10:53:57 2026 +0800
[SPARK-58003][SQL] Add physical execution for BIN BY
### What changes were proposed in this pull request?
This PR makes the `BIN BY` relation operator execute end-to-end. It adds
the `BinByExec` physical operator and wires it into planning, replacing the
stub that raised `UNSUPPORTED_FEATURE.BIN_BY`.
For each input row, `BinByExec` emits one row per bin that overlaps the
row's `[range_start, range_end)`:
- `DISTRIBUTE UNIFORM` columns are rescaled by the bin's overlap fraction.
- All other columns, including the range columns, are copied unchanged.
- `bin_start`, `bin_end`, and `bin_distribute_ratio` are appended.
Bin boundaries reuse `DateTimeUtils.timeBucketDTInterval` /
`timestampAddDayTime`, matching `time_bucket`: sub-day widths use UTC
microsecond arithmetic; multi-day widths use civil-time arithmetic in the
session zone (UTC for `TIMESTAMP_NTZ`).
Per-row edge cases:
- Zero-length range (`range_start == range_end`): one row with ratio `1.0`.
- Inverted range (`range_start > range_end`): raises `BIN_BY_INVALID_RANGE`.
- NULL in either range column: one row with the scaled DISTRIBUTE values
and all three appended columns NULL (no valid bin to scale into); passthrough
columns are unaffected.
### Why are the changes needed?
`BIN BY` parsed, resolved, and type-checked, but planning threw
`UNSUPPORTED_FEATURE.BIN_BY` because it had no physical execution. This PR adds
that execution so `BIN BY` queries can run.
### Does this PR introduce _any_ user-facing change?
No. `BIN BY` is gated off by default
(`spark.sql.binByRelationOperator.enabled`).
### How was this patch tested?
- `BinBySuite`: end-to-end execution covering proportional splits,
single-bin passthrough, FLOAT + DOUBLE scaling, downstream `GROUP BY`,
zero-length and NULL ranges, the inverted-range error, nested-struct
passthrough, renamed outputs, and civil-time boundaries across a DST transition
(LTZ and NTZ).
- `bin-by.sql` golden file: executing scenarios plus analysis-error cases.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic)
Closes #57086 from vranes/bin-by-physical-exec.
Authored-by: Nikolina Vraneš <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../sql/catalyst/analysis/BinByResolution.scala | 7 +-
.../spark/sql/errors/QueryExecutionErrors.scala | 7 +
.../org/apache/spark/sql/execution/BinByExec.scala | 193 +++++++++++++++++
.../spark/sql/execution/SparkStrategies.scala | 14 +-
.../sql-tests/analyzer-results/bin-by.sql.out | 122 +++++++++++
.../src/test/resources/sql-tests/inputs/bin-by.sql | 58 +++++
.../resources/sql-tests/results/bin-by.sql.out | 121 +++++++++++
.../scala/org/apache/spark/sql/BinBySuite.scala | 236 ++++++++++++++++++---
8 files changed, 718 insertions(+), 40 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
index d7dad79b07f6..ec8afaefc52c 100644
---
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
@@ -168,9 +168,10 @@ object BinByResolution {
AttributeReference(aliases.effectiveBinRatio, DoubleType, nullable =
true)())
/**
- * Mints a produced output attribute for each DISTRIBUTE input column: same
name, type, and
- * nullability, but a fresh `ExprId` so the rescaled value is a distinct
attribute from the input.
+ * Mints a produced output attribute for each DISTRIBUTE input column: same
name, type, and a
+ * fresh `ExprId` so the rescaled value is a distinct attribute from the
input. Always nullable:
+ * a NULL range column produces a NULL scaled value regardless of the input
column's nullability.
*/
def scaledDistributeAttributes(distributeColumns: Seq[Attribute]):
Seq[Attribute] =
- distributeColumns.map(a => AttributeReference(a.name, a.dataType,
a.nullable)())
+ distributeColumns.map(a => AttributeReference(a.name, a.dataType, nullable
= true)())
}
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
index b53267e8583f..ef7d3a1317a8 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/errors/QueryExecutionErrors.scala
@@ -3481,4 +3481,11 @@ private[sql] object QueryExecutionErrors extends
QueryErrorsBase with ExecutionE
)
)
}
+
+ def binByInvalidRangeError(rangeStart: String, rangeEnd: String): Throwable
= {
+ new SparkRuntimeException(
+ errorClass = "BIN_BY_INVALID_RANGE",
+ messageParameters = Map("rangeStart" -> rangeStart, "rangeEnd" ->
rangeEnd)
+ )
+ }
}
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/BinByExec.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/BinByExec.scala
new file mode 100644
index 000000000000..d95addd93f40
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/BinByExec.scala
@@ -0,0 +1,193 @@
+/*
+ * 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.execution
+
+import java.time.ZoneOffset
+
+import org.apache.spark.SparkException
+import org.apache.spark.rdd.RDD
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeMap,
AttributeSet, BindReferences, BoundReference, Cast, Expression,
GenericInternalRow, JoinedRow, Literal, Multiply, UnsafeProjection}
+import org.apache.spark.sql.catalyst.util.{DateTimeUtils, TimestampFormatter}
+import org.apache.spark.sql.errors.QueryExecutionErrors
+import org.apache.spark.sql.execution.metric.{SQLMetric, SQLMetrics}
+import org.apache.spark.sql.types.DoubleType
+
+/**
+ * Physical node for the `BIN BY` relation operator. For each input row it
emits one output row per
+ * bin overlapping `[rangeStart, rangeEnd)`, scaling the DISTRIBUTE UNIFORM
columns by the overlap
+ * fraction and appending `bin_start`, `bin_end`, `bin_distribute_ratio`.
+ *
+ * Bin boundaries reuse [[DateTimeUtils.timeBucketDTInterval]] /
+ * [[DateTimeUtils.timestampAddDayTime]], matching `time_bucket`: sub-day
widths use UTC microsecond
+ * arithmetic, multi-day widths use civil-time arithmetic in the session zone
(UTC for
+ * TIMESTAMP_NTZ). DISTRIBUTE UNIFORM columns are FLOAT or DOUBLE only and
scale by multiplication.
+ */
+case class BinByExec(
+ binWidthMicros: Long,
+ originMicros: Long,
+ rangeStart: Attribute,
+ rangeEnd: Attribute,
+ distributeColumns: Seq[Attribute],
+ scaledDistributeColumns: Seq[Attribute],
+ appendedAttributes: Seq[Attribute],
+ timeZoneId: Option[String],
+ child: SparkPlan)
+ extends UnaryExecNode {
+
+ override lazy val metrics: Map[String, SQLMetric] = Map(
+ "numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output
rows"))
+
+ private val distributeReplacements: AttributeMap[Attribute] =
+ AttributeMap(distributeColumns.zip(scaledDistributeColumns))
+
+ private val numAppended: Int = appendedAttributes.length
+ assert(numAppended == 3,
+ s"BinBy appends exactly 3 columns (bin_start, bin_end,
bin_distribute_ratio), got $numAppended")
+
+ override def output: Seq[Attribute] =
+ child.output.map(a => distributeReplacements.getOrElse(a, a)) ++
appendedAttributes
+
+ override def producedAttributes: AttributeSet =
+ AttributeSet(scaledDistributeColumns ++ appendedAttributes)
+
+ override protected def withNewChildInternal(newChild: SparkPlan): BinByExec =
+ copy(child = newChild)
+
+ protected override def doExecute(): RDD[InternalRow] = {
+ val width = binWidthMicros
+ val origin = originMicros
+ val zone =
timeZoneId.map(DateTimeUtils.getZoneId).getOrElse(ZoneOffset.UTC)
+ val numOutputRows = longMetric("numOutputRows")
+
+ val rsIdx = bindOrdinal(rangeStart)
+ val reIdx = bindOrdinal(rangeEnd)
+ val childLen = child.output.length
+
+ // Bound over JoinedRow(childRow, appendedRow); reused for every sub-row.
+ val projExprs = buildOutputExpressions(childLen)
+ // Bound over the child row alone, for the null-range path.
+ val nullProjExprs = buildNullRangeExpressions()
+
+ child.execute().mapPartitionsInternal { rows =>
+ val proj = UnsafeProjection.create(projExprs)
+ val nullProj = UnsafeProjection.create(nullProjExprs)
+ val appended = new GenericInternalRow(numAppended)
+ val joined = new JoinedRow
+ val fmt = TimestampFormatter.getFractionFormatter(zone)
+
+ rows.flatMap { row =>
+ joined.withLeft(row)
+ val rowIter: Iterator[InternalRow] =
+ if (row.isNullAt(rsIdx) || row.isNullAt(reIdx)) {
+ Iterator.single(nullProj(row))
+ } else {
+ val rs = row.getLong(rsIdx)
+ val re = row.getLong(reIdx)
+ if (rs > re) {
+ throw
QueryExecutionErrors.binByInvalidRangeError(fmt.format(rs), fmt.format(re))
+ } else if (rs == re) {
+ val binStart = DateTimeUtils.timeBucketDTInterval(width, rs,
origin, zone)
+ val binEnd = DateTimeUtils.timestampAddDayTime(binStart, width,
zone)
+ appended.update(0, binStart)
+ appended.update(1, binEnd)
+ appended.update(2, 1.0d)
+ Iterator.single(proj(joined.withRight(appended)))
+ } else {
+ val total = Math.subtractExact(re, rs)
+ new Iterator[InternalRow] {
+ private var curStart =
+ DateTimeUtils.timeBucketDTInterval(width, rs, origin, zone)
+
+ override def hasNext: Boolean = curStart < re
+
+ override def next(): InternalRow = {
+ val curEnd = DateTimeUtils.timestampAddDayTime(curStart,
width, zone)
+ val overlap = math.min(re, curEnd) - math.max(rs, curStart)
+ appended.update(0, curStart)
+ appended.update(1, curEnd)
+ appended.update(2, overlap.toDouble / total.toDouble)
+ curStart = curEnd
+ proj(joined.withRight(appended))
+ }
+ }
+ }
+ }
+ rowIter.map { r =>
+ numOutputRows += 1
+ r
+ }
+ }
+ }
+ }
+
+ private def bindOrdinal(a: Attribute): Int =
+ (BindReferences.bindReference(a, child.output): Expression) match {
+ case b: BoundReference => b.ordinal
+ case _ =>
+ throw SparkException.internalError(
+ s"BinByExec attribute ${a.name}#${a.exprId.id} is not bound in child
output " +
+ s"${child.output.mkString("[", ", ", "]")}")
+ }
+
+ /**
+ * Output projection over `JoinedRow(childRow, appendedRow)`: DISTRIBUTE
columns scaled by the
+ * ratio, other columns passed through.
+ */
+ private def buildOutputExpressions(childLen: Int): Seq[Expression] = {
+ val distSet = distributeColumns.map(bindOrdinal).toSet
+ // bin_distribute_ratio is the last appended column.
+ val ratioRef = BoundReference(childLen + numAppended - 1, DoubleType,
nullable = false)
+
+ val forwarded = child.output.zipWithIndex.map { case (attr, ordinal) =>
+ val ref = BoundReference(ordinal, attr.dataType, attr.nullable)
+ if (distSet.contains(ordinal)) {
+ // Cast to double, multiply, narrow back. These expressions bind at
execution time, so the
+ // analyzer's operand coercion never runs; a bare Multiply(FloatType,
DoubleType) fails.
+ Cast(Multiply(Cast(ref, DoubleType), ratioRef), attr.dataType)
+ } else {
+ ref
+ }
+ }
+
+ val appendedRefs = appendedAttributes.zipWithIndex.map { case (attr, i) =>
+ BoundReference(childLen + i, attr.dataType, attr.nullable)
+ }
+
+ forwarded ++ appendedRefs
+ }
+
+ /**
+ * Null-range projection over the child row: DISTRIBUTE and appended columns
are NULL (no valid
+ * bin), other columns passed through.
+ */
+ private def buildNullRangeExpressions(): Seq[Expression] = {
+ val distSet = distributeColumns.map(bindOrdinal).toSet
+ val forwarded = child.output.zipWithIndex.map { case (attr, ordinal) =>
+ if (distSet.contains(ordinal)) {
+ Literal.create(null, attr.dataType)
+ } else {
+ BoundReference(ordinal, attr.dataType, attr.nullable)
+ }
+ }
+ val nullAppended = appendedAttributes.map { attr =>
+ Literal.create(null, attr.dataType)
+ }
+ forwarded ++ nullAppended
+ }
+}
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 e8d29dfa7c5e..752ba33bdad2 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
@@ -1043,9 +1043,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
case logical.SerializeFromObject(serializer, child) =>
@@ -1234,6 +1231,17 @@ abstract class SparkStrategies extends
QueryPlanner[SparkPlan] {
staticPartitions) :: Nil
case MultiResult(children) =>
MultiResultExec(children.map(planLater)) :: Nil
+ case b: logical.BinBy =>
+ execution.BinByExec(
+ binWidthMicros = b.binWidthMicros,
+ originMicros = b.originMicros,
+ rangeStart = b.rangeStart,
+ rangeEnd = b.rangeEnd,
+ distributeColumns = b.distributeColumns,
+ scaledDistributeColumns = b.scaledDistributeColumns,
+ appendedAttributes = b.appendedAttributes,
+ timeZoneId = b.timeZoneId,
+ child = planLater(b.child)) :: Nil
case _ => Nil
}
}
diff --git
a/sql/core/src/test/resources/sql-tests/analyzer-results/bin-by.sql.out
b/sql/core/src/test/resources/sql-tests/analyzer-results/bin-by.sql.out
new file mode 100644
index 000000000000..91f951d8a403
--- /dev/null
+++ b/sql/core/src/test/resources/sql-tests/analyzer-results/bin-by.sql.out
@@ -0,0 +1,122 @@
+-- Automatically generated by SQLQueryTestSuite
+-- !query
+SET spark.sql.binByRelationOperator.enabled = true
+-- !query analysis
+SetCommand (spark.sql.binByRelationOperator.enabled,Some(true))
+
+
+-- !query
+SET TIME ZONE 'UTC'
+-- !query analysis
+SetCommand (spark.sql.session.timeZone,Some(UTC))
+
+
+-- !query
+CREATE OR REPLACE TEMP VIEW metrics AS
+SELECT * FROM VALUES
+ (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01 00:05:00', 100.0D),
+ (TIMESTAMP '2024-01-01 00:02:00', TIMESTAMP '2024-01-01 00:12:00', 300.0D)
+AS metrics(ts_start, ts_end, value)
+-- !query analysis
+CreateViewCommand `metrics`, SELECT * FROM VALUES
+ (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01 00:05:00', 100.0D),
+ (TIMESTAMP '2024-01-01 00:02:00', TIMESTAMP '2024-01-01 00:12:00', 300.0D)
+AS metrics(ts_start, ts_end, value), false, true, LocalTempView, UNSUPPORTED,
true
+ +- Project [ts_start#x, ts_end#x, value#x]
+ +- SubqueryAlias metrics
+ +- LocalRelation [ts_start#x, ts_end#x, value#x]
+
+
+-- !query
+SELECT * FROM metrics BIN BY (
+ RANGE ts_start TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ ALIGN TO TIMESTAMP '2024-01-01 00:00:00'
+ DISTRIBUTE UNIFORM (value)
+)
+-- !query analysis
+Project [ts_start#x, ts_end#x, value#x, bin_start#x, bin_end#x,
bin_distribute_ratio#x]
++- BinBy 300000000, ts_start#x: timestamp, ts_end#x: timestamp,
1704067200000000, [value#x], [value#x], [bin_start#x, bin_end#x,
bin_distribute_ratio#x], UTC
+ +- SubqueryAlias metrics
+ +- View (`metrics`, [ts_start#x, ts_end#x, value#x])
+ +- Project [cast(ts_start#x as timestamp) AS ts_start#x,
cast(ts_end#x as timestamp) AS ts_end#x, cast(value#x as double) AS value#x]
+ +- Project [ts_start#x, ts_end#x, value#x]
+ +- SubqueryAlias metrics
+ +- LocalRelation [ts_start#x, ts_end#x, value#x]
+
+
+-- !query
+SELECT bin_start, SUM(value) AS total
+FROM metrics BIN BY (
+ RANGE ts_start TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ ALIGN TO TIMESTAMP '2024-01-01 00:00:00'
+ DISTRIBUTE UNIFORM (value)
+)
+GROUP BY bin_start
+ORDER BY bin_start
+-- !query analysis
+Sort [bin_start#x ASC NULLS FIRST], true
++- Aggregate [bin_start#x], [bin_start#x, sum(value#x) AS total#x]
+ +- BinBy 300000000, ts_start#x: timestamp, ts_end#x: timestamp,
1704067200000000, [value#x], [value#x], [bin_start#x, bin_end#x,
bin_distribute_ratio#x], UTC
+ +- SubqueryAlias metrics
+ +- View (`metrics`, [ts_start#x, ts_end#x, value#x])
+ +- Project [cast(ts_start#x as timestamp) AS ts_start#x,
cast(ts_end#x as timestamp) AS ts_end#x, cast(value#x as double) AS value#x]
+ +- Project [ts_start#x, ts_end#x, value#x]
+ +- SubqueryAlias metrics
+ +- LocalRelation [ts_start#x, ts_end#x, value#x]
+
+
+-- !query
+SELECT * FROM VALUES
+ (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01 00:08:00', 100.0D)
+ AS t(ts_start, ts_end, value)
+BIN BY (
+ RANGE ts_start TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ ALIGN TO TIMESTAMP '2024-01-01 00:01:00'
+ DISTRIBUTE UNIFORM (value)
+ BIN_START AS window_start
+ BIN_END AS window_end
+ BIN_DISTRIBUTE_RATIO AS frac
+)
+-- !query analysis
+Project [ts_start#x, ts_end#x, value#x, window_start#x, window_end#x, frac#x]
++- BinBy 300000000, ts_start#x: timestamp, ts_end#x: timestamp,
1704067260000000, [value#x], [value#x], [window_start#x, window_end#x, frac#x],
UTC
+ +- SubqueryAlias t
+ +- LocalRelation [ts_start#x, ts_end#x, value#x]
+
+
+-- !query
+SELECT * FROM metrics BIN BY (
+ RANGE ts_start TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ DISTRIBUTE UNIFORM (value, value)
+)
+-- !query analysis
+org.apache.spark.sql.AnalysisException
+{
+ "errorClass" : "BIN_BY_DUPLICATE_DISTRIBUTE_COLUMN",
+ "sqlState" : "42713",
+ "messageParameters" : {
+ "columnName" : "`value`"
+ }
+}
+
+
+-- !query
+SELECT * FROM metrics BIN BY (
+ RANGE value TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ DISTRIBUTE UNIFORM (value)
+)
+-- !query analysis
+org.apache.spark.sql.AnalysisException
+{
+ "errorClass" : "BIN_BY_RANGE_TYPE_MISMATCH",
+ "sqlState" : "42K09",
+ "messageParameters" : {
+ "columnName" : "`value`",
+ "columnType" : "\"DOUBLE\""
+ }
+}
diff --git a/sql/core/src/test/resources/sql-tests/inputs/bin-by.sql
b/sql/core/src/test/resources/sql-tests/inputs/bin-by.sql
new file mode 100644
index 000000000000..5413a148ef77
--- /dev/null
+++ b/sql/core/src/test/resources/sql-tests/inputs/bin-by.sql
@@ -0,0 +1,58 @@
+SET spark.sql.binByRelationOperator.enabled = true;
+SET TIME ZONE 'UTC';
+
+CREATE OR REPLACE TEMP VIEW metrics AS
+SELECT * FROM VALUES
+ (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01 00:05:00', 100.0D),
+ (TIMESTAMP '2024-01-01 00:02:00', TIMESTAMP '2024-01-01 00:12:00', 300.0D)
+AS metrics(ts_start, ts_end, value);
+
+
+-- Basic split: one row fits a single bin, the other spans multiple bins.
+SELECT * FROM metrics BIN BY (
+ RANGE ts_start TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ ALIGN TO TIMESTAMP '2024-01-01 00:00:00'
+ DISTRIBUTE UNIFORM (value)
+);
+
+
+-- Composability: feed BIN BY into a downstream GROUP BY.
+SELECT bin_start, SUM(value) AS total
+FROM metrics BIN BY (
+ RANGE ts_start TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ ALIGN TO TIMESTAMP '2024-01-01 00:00:00'
+ DISTRIBUTE UNIFORM (value)
+)
+GROUP BY bin_start
+ORDER BY bin_start;
+
+
+-- Custom ALIGN TO origin and renamed output columns.
+SELECT * FROM VALUES
+ (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01 00:08:00', 100.0D)
+ AS t(ts_start, ts_end, value)
+BIN BY (
+ RANGE ts_start TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ ALIGN TO TIMESTAMP '2024-01-01 00:01:00'
+ DISTRIBUTE UNIFORM (value)
+ BIN_START AS window_start
+ BIN_END AS window_end
+ BIN_DISTRIBUTE_RATIO AS frac
+);
+
+
+-- Analysis errors: duplicate DISTRIBUTE column and a non-timestamp RANGE
column.
+SELECT * FROM metrics BIN BY (
+ RANGE ts_start TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ DISTRIBUTE UNIFORM (value, value)
+);
+
+SELECT * FROM metrics BIN BY (
+ RANGE value TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ DISTRIBUTE UNIFORM (value)
+);
diff --git a/sql/core/src/test/resources/sql-tests/results/bin-by.sql.out
b/sql/core/src/test/resources/sql-tests/results/bin-by.sql.out
new file mode 100644
index 000000000000..7630f54276a2
--- /dev/null
+++ b/sql/core/src/test/resources/sql-tests/results/bin-by.sql.out
@@ -0,0 +1,121 @@
+-- Automatically generated by SQLQueryTestSuite
+-- !query
+SET spark.sql.binByRelationOperator.enabled = true
+-- !query schema
+struct<key:string,value:string>
+-- !query output
+spark.sql.binByRelationOperator.enabled true
+
+
+-- !query
+SET TIME ZONE 'UTC'
+-- !query schema
+struct<key:string,value:string>
+-- !query output
+spark.sql.session.timeZone UTC
+
+
+-- !query
+CREATE OR REPLACE TEMP VIEW metrics AS
+SELECT * FROM VALUES
+ (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01 00:05:00', 100.0D),
+ (TIMESTAMP '2024-01-01 00:02:00', TIMESTAMP '2024-01-01 00:12:00', 300.0D)
+AS metrics(ts_start, ts_end, value)
+-- !query schema
+struct<>
+-- !query output
+
+
+
+-- !query
+SELECT * FROM metrics BIN BY (
+ RANGE ts_start TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ ALIGN TO TIMESTAMP '2024-01-01 00:00:00'
+ DISTRIBUTE UNIFORM (value)
+)
+-- !query schema
+struct<ts_start:timestamp,ts_end:timestamp,value:double,bin_start:timestamp,bin_end:timestamp,bin_distribute_ratio:double>
+-- !query output
+2024-01-01 00:00:00 2024-01-01 00:05:00 100.0 2024-01-01 00:00:00
2024-01-01 00:05:00 1.0
+2024-01-01 00:02:00 2024-01-01 00:12:00 150.0 2024-01-01 00:05:00
2024-01-01 00:10:00 0.5
+2024-01-01 00:02:00 2024-01-01 00:12:00 60.0 2024-01-01 00:10:00
2024-01-01 00:15:00 0.2
+2024-01-01 00:02:00 2024-01-01 00:12:00 90.0 2024-01-01 00:00:00
2024-01-01 00:05:00 0.3
+
+
+-- !query
+SELECT bin_start, SUM(value) AS total
+FROM metrics BIN BY (
+ RANGE ts_start TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ ALIGN TO TIMESTAMP '2024-01-01 00:00:00'
+ DISTRIBUTE UNIFORM (value)
+)
+GROUP BY bin_start
+ORDER BY bin_start
+-- !query schema
+struct<bin_start:timestamp,total:double>
+-- !query output
+2024-01-01 00:00:00 190.0
+2024-01-01 00:05:00 150.0
+2024-01-01 00:10:00 60.0
+
+
+-- !query
+SELECT * FROM VALUES
+ (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01 00:08:00', 100.0D)
+ AS t(ts_start, ts_end, value)
+BIN BY (
+ RANGE ts_start TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ ALIGN TO TIMESTAMP '2024-01-01 00:01:00'
+ DISTRIBUTE UNIFORM (value)
+ BIN_START AS window_start
+ BIN_END AS window_end
+ BIN_DISTRIBUTE_RATIO AS frac
+)
+-- !query schema
+struct<ts_start:timestamp,ts_end:timestamp,value:double,window_start:timestamp,window_end:timestamp,frac:double>
+-- !query output
+2024-01-01 00:00:00 2024-01-01 00:08:00 12.5 2023-12-31 23:56:00
2024-01-01 00:01:00 0.125
+2024-01-01 00:00:00 2024-01-01 00:08:00 25.0 2024-01-01 00:06:00
2024-01-01 00:11:00 0.25
+2024-01-01 00:00:00 2024-01-01 00:08:00 62.5 2024-01-01 00:01:00
2024-01-01 00:06:00 0.625
+
+
+-- !query
+SELECT * FROM metrics BIN BY (
+ RANGE ts_start TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ DISTRIBUTE UNIFORM (value, value)
+)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.sql.AnalysisException
+{
+ "errorClass" : "BIN_BY_DUPLICATE_DISTRIBUTE_COLUMN",
+ "sqlState" : "42713",
+ "messageParameters" : {
+ "columnName" : "`value`"
+ }
+}
+
+
+-- !query
+SELECT * FROM metrics BIN BY (
+ RANGE value TO ts_end
+ BIN WIDTH INTERVAL '5' MINUTE
+ DISTRIBUTE UNIFORM (value)
+)
+-- !query schema
+struct<>
+-- !query output
+org.apache.spark.sql.AnalysisException
+{
+ "errorClass" : "BIN_BY_RANGE_TYPE_MISMATCH",
+ "sqlState" : "42K09",
+ "messageParameters" : {
+ "columnName" : "`value`",
+ "columnType" : "\"DOUBLE\""
+ }
+}
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 2004aa89aa06..3e00a0645e0d 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
@@ -17,12 +17,23 @@
package org.apache.spark.sql
-import org.apache.spark.{SparkThrowable, SparkUnsupportedOperationException}
+import java.sql.Timestamp
+import java.time.{LocalDateTime, ZoneId, ZoneOffset}
+
+import org.apache.spark.SparkThrowable
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
class BinBySuite extends QueryTest with SharedSparkSession {
+ private def tsAt(s: String, zone: ZoneId = ZoneOffset.UTC): Timestamp =
+ Timestamp.from(LocalDateTime.parse(s.replace(' ',
'T')).atZone(zone).toInstant)
+
+ private def ntz(s: String): LocalDateTime = LocalDateTime.parse(s.replace('
', 'T'))
+
+ private def ratio(overlapMicros: Long, totalMicros: Long): Double =
+ overlapMicros.toDouble / totalMicros.toDouble
+
private def createMetricsView(): Unit = {
spark.sql(
"""SELECT TIMESTAMP '2024-01-01 00:00:00' AS ts_start,
@@ -37,24 +48,169 @@ class BinBySuite extends QueryTest with SharedSparkSession
{
| DISTRIBUTE UNIFORM (value)
|)""".stripMargin
- test("BIN BY analyzes but physical execution is not yet implemented") {
- withSQLConf(SQLConf.BIN_BY_ENABLED.key -> "true") {
- withTempView("metrics") {
- createMetricsView()
- val df = spark.sql(binByQuery)
+ test("BIN BY splits a range into proportional sub-rows") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ val df = spark.sql(
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01
00:10:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY bin_start""".stripMargin)
+ checkAnswer(df, Seq(
+ Row(tsAt("2024-01-01 00:00:00"), tsAt("2024-01-01 00:05:00"), 0.5,
50.0),
+ Row(tsAt("2024-01-01 00:05:00"), tsAt("2024-01-01 00:10:00"), 0.5,
50.0)))
+ }
+ }
- // Analysis is fully functional when the operator is enabled.
- df.queryExecution.assertAnalyzed()
+ test("BIN BY scales FLOAT and DOUBLE DISTRIBUTE columns in one clause") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ val df = spark.sql(
+ """SELECT bin_start, f, d
+ |FROM VALUES
+ | (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01 00:10:00',
+ | CAST(100.0 AS FLOAT), 200.0D)
+ | AS metrics(ts_start, ts_end, f, d)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM (f,
d))
+ |ORDER BY bin_start""".stripMargin)
+ checkAnswer(df, Seq(
+ Row(tsAt("2024-01-01 00:00:00"), 50.0f, 100.0),
+ Row(tsAt("2024-01-01 00:05:00"), 50.0f, 100.0)))
+ }
+ }
- // 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])
- }
+ test("BIN BY passes a single-bin range through with ratio 1.0") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ val df = spark.sql(
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01
00:05:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))""".stripMargin)
+ checkAnswer(df, Seq(
+ Row(tsAt("2024-01-01 00:00:00"), tsAt("2024-01-01 00:05:00"), 1.0,
100.0)))
+ }
+ }
+
+ test("BIN BY emits a NULL-range row with all computed columns NULL") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ val df = spark.sql(
+ """SELECT id, bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (1, CAST(NULL AS TIMESTAMP), TIMESTAMP '2024-01-01 00:10:00',
100.0D),
+ | (2, TIMESTAMP '2024-01-01 00:00:00', CAST(NULL AS TIMESTAMP),
200.0D)
+ | AS metrics(id, ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY id""".stripMargin)
+ // A NULL range nulls every computed column; only the `id` passthrough
survives.
+ checkAnswer(df, Seq(
+ Row(1, null, null, null, null),
+ Row(2, null, null, null, null)))
+ }
+ }
+
+ test("BIN BY raises BIN_BY_INVALID_RANGE for an inverted range") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ val df = spark.sql(
+ """SELECT * FROM VALUES
+ | (TIMESTAMP '2024-01-01 00:10:00', TIMESTAMP '2024-01-01
00:00:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))""".stripMargin)
+ checkError(
+ exception = intercept[SparkThrowable] {
+ df.collect()
+ },
+ condition = "BIN_BY_INVALID_RANGE",
+ parameters = Map(
+ "rangeStart" -> "2024-01-01 00:10:00",
+ "rangeEnd" -> "2024-01-01 00:00:00"))
+ }
+ }
+
+ test("BIN BY emits a single ratio-1.0 row for a zero-length range") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ val df = spark.sql(
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (TIMESTAMP '2024-01-01 00:02:00', TIMESTAMP '2024-01-01
00:02:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))""".stripMargin)
+ // rangeStart == rangeEnd: one row, ratio 1.0, value kept.
+ checkAnswer(df, Seq(
+ Row(tsAt("2024-01-01 00:00:00"), tsAt("2024-01-01 00:05:00"), 1.0,
100.0)))
+ }
+ }
+
+ test("BIN BY uses civil-time bin boundaries across a DST spring-forward") {
+ val la = ZoneId.of("America/Los_Angeles")
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
+ // A 1-DAY bin spanning the 2024-03-10 spring-forward. Multi-day widths
use civil-time
+ // arithmetic in the session zone, so boundaries land on civil midnight
and the 2024-03-10
+ // bin is 23h wide (02:00 PST -> 03:00 PDT) while 2024-03-11 is 24h. The
ratios split the
+ // range by real elapsed microseconds, so the 23h bin gets a smaller
share than the 24h bin.
+ val df = spark.sql(
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (TIMESTAMP '2024-03-10 00:00:00', TIMESTAMP '2024-03-12
00:00:00', 100.0D)
+ | AS metrics(ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '1' DAY
+ | ALIGN TO TIMESTAMP '2024-03-10 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY bin_start""".stripMargin)
+ val h = 3600L * 1000000L // one hour in micros
+ val total = 47 * h // 23h (DST day) + 24h
+ checkAnswer(df, Seq(
+ Row(tsAt("2024-03-10 00:00:00", la), tsAt("2024-03-11 00:00:00", la),
+ ratio(23 * h, total), 100.0 * ratio(23 * h, total)),
+ Row(tsAt("2024-03-11 00:00:00", la), tsAt("2024-03-12 00:00:00", la),
+ ratio(24 * h, total), 100.0 * ratio(24 * h, total))))
+ }
+ }
+
+ test("BIN BY replicates a nested struct passthrough column across a
multi-bin split") {
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "UTC") {
+ // A nested struct passthrough must appear identically on every split
sub-row.
+ val df = spark.sql(
+ """SELECT s, bin_start, value
+ |FROM VALUES
+ | (named_struct('a', 1, 'b', 'x'),
+ | TIMESTAMP '2024-01-01 00:00:00', TIMESTAMP '2024-01-01
00:10:00', 100.0D)
+ | AS metrics(s, ts_start, ts_end, value)
+ |BIN BY (
+ | RANGE ts_start TO ts_end BIN WIDTH INTERVAL '5' MINUTE
+ | ALIGN TO TIMESTAMP '2024-01-01 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY bin_start""".stripMargin)
+ checkAnswer(df, Seq(
+ Row(Row(1, "x"), tsAt("2024-01-01 00:00:00"), 50.0),
+ Row(Row(1, "x"), tsAt("2024-01-01 00:05:00"), 50.0)))
}
}
@@ -74,28 +230,40 @@ class BinBySuite extends QueryTest with SharedSparkSession
{
}
}
- test("BIN BY analyzes NTZ inputs and a custom ALIGN TO with renamed
outputs") {
+ test("BIN BY executes on NTZ inputs with the epoch default origin") {
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)
+ // NTZ inputs default the origin to the wall-clock epoch.
+ val df = spark.sql(
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (TIMESTAMP_NTZ'2024-01-01 00:00:00', TIMESTAMP_NTZ'2024-01-01
00:10:00', 100.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()
+ |ORDER BY bin_start""".stripMargin)
+ checkAnswer(df, Seq(
+ Row(ntz("2024-01-01 00:00:00"), ntz("2024-01-01 00:05:00"), 0.5, 50.0),
+ Row(ntz("2024-01-01 00:05:00"), ntz("2024-01-01 00:10:00"), 0.5,
50.0)))
+ }
+ }
- // 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()
+ test("BIN BY on NTZ inputs uses UTC arithmetic and ignores the session zone
across DST") {
+ // A multi-day NTZ bin over the LA spring-forward: NTZ ignores the session
zone, so every day is
+ // a full 24h (no DST shortening), unlike the LTZ civil-time path. Two
1-day bins split evenly.
+ withSQLConf(
+ SQLConf.BIN_BY_ENABLED.key -> "true",
+ SQLConf.SESSION_LOCAL_TIMEZONE.key -> "America/Los_Angeles") {
+ val df = spark.sql(
+ """SELECT bin_start, bin_end, bin_distribute_ratio, value
+ |FROM VALUES
+ | (TIMESTAMP_NTZ'2024-03-10 00:00:00', TIMESTAMP_NTZ'2024-03-12
00:00:00', 100.0D)
+ | AS t(ts_start, ts_end, value)
+ |BIN BY (RANGE ts_start TO ts_end BIN WIDTH INTERVAL '1' DAY
+ | ALIGN TO TIMESTAMP_NTZ'2024-03-10 00:00:00' DISTRIBUTE UNIFORM
(value))
+ |ORDER BY bin_start""".stripMargin)
+ checkAnswer(df, Seq(
+ Row(ntz("2024-03-10 00:00:00"), ntz("2024-03-11 00:00:00"), 0.5, 50.0),
+ Row(ntz("2024-03-11 00:00:00"), ntz("2024-03-12 00:00:00"), 0.5,
50.0)))
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
