This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new e2785d4baa46 [SPARK-57858][SQL] Emit BIN BY scaled DISTRIBUTE columns
as produced attributes
e2785d4baa46 is described below
commit e2785d4baa463580c0c2b72997a1fead6d35af87
Author: Nikolina Vraneš <[email protected]>
AuthorDate: Thu Jul 2 20:55:22 2026 +0800
[SPARK-57858][SQL] Emit BIN BY scaled DISTRIBUTE columns as produced
attributes
### What changes were proposed in this pull request?
The `BIN BY` relation operator (SPARK-57133) proportionally rescales its
`DISTRIBUTE UNIFORM` columns. The logical `BinBy` node carried those columns
through `child.output` with the child's own `ExprId`, even though execution
rewrites their values.
This PR makes the rescaled `DISTRIBUTE` columns *produced* attributes with
fresh `ExprId`s (same names, types, nullability, and positions), shadowing the
inputs, mirroring `Generate.generatorOutput`:
- `BinBy` gains a `scaledDistributeColumns` field; `output` swaps each
`DISTRIBUTE` input slot for its scaled counterpart in place, and
`producedAttributes` includes them. The input `distributeColumns` stay on the
node as the executor's read inputs but leave `output`.
- `BinBy.scaledDistributeAttributes` mints the fresh attributes (qualifier
and metadata dropped, matching `expr AS value` computed-value semantics).
- `ResolveBinBy` mints them; `DeduplicateRelations` renews them in both
phases so self-joins over a shared `BinBy` subtree resolve.
The two `BinBy` constructor invariants (`timeZoneId` set iff LTZ range, and
one scaled attribute per `DISTRIBUTE` column) are both internal `assert`s:
neither is reachable from `ResolveBinBy`, which derives `timeZoneId` from the
same `rangeStart` type and mints `scaledDistributeColumns` by mapping over
`distributeColumns`.
### Why are the changes needed?
Catalyst relies on the invariant that the same `ExprId` everywhere implies
the same value. No other operator edits a value under a retained child
attribute (`Generate` / `Window` / `Expand` / `Aggregate` all mint fresh ids
for changed columns). Carrying the rescaled `DISTRIBUTE` column under the
child's `ExprId` violated that: any rule reasoning on `ExprId` (predicate
pushdown, constraint propagation, common-subexpression elimination) could read
the pre-scale value. It is harmless tod [...]
### Does this PR introduce _any_ user-facing change?
No. `BIN BY` is gated off by default (SPARK-57440) and its physical
execution is still stubbed, so the operator is not usable end-to-end yet; this
is an internal analyzer / plan-shape change. The output schema (column names,
types, positions) is unchanged.
### How was this patch tested?
`ResolveBinBySuite`, including new cases that the rescaled `DISTRIBUTE`
columns are produced attributes shadowing the input, that multiple `DISTRIBUTE`
columns are each replaced in place with distinct fresh ids, and that
qualifier/metadata are dropped on the produced column; plus the existing
self-join deduplication regression.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic)
Closes #56930 from vranes/bin-by-distribute-produced-attrs.
Authored-by: Nikolina Vraneš <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 3d8bbe8f3e37a2544a2ac28109b6dacdcc6aebee)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../sql/catalyst/analysis/BinByResolution.scala | 21 ++++++-
.../catalyst/analysis/DeduplicateRelations.scala | 6 +-
.../spark/sql/catalyst/analysis/ResolveBinBy.scala | 4 +-
.../plans/logical/basicLogicalOperators.scala | 43 ++++++++------
.../sql/catalyst/analysis/ResolveBinBySuite.scala | 69 +++++++++++++++++++++-
5 files changed, 117 insertions(+), 26 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 890df8eb83fc..d7dad79b07f6 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
@@ -20,7 +20,8 @@ package org.apache.spark.sql.catalyst.analysis
import scala.collection.mutable
import scala.util.control.NonFatal
-import org.apache.spark.sql.catalyst.expressions.{Attribute, EmptyRow,
Expression, ExprId}
+import org.apache.spark.sql.catalyst.expressions.{Attribute,
AttributeReference, EmptyRow, Expression, ExprId}
+import org.apache.spark.sql.catalyst.plans.logical.BinByOutputAliases
import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.errors.QueryCompilationErrors
import org.apache.spark.sql.internal.SQLConf
@@ -154,4 +155,22 @@ object BinByResolution {
timeZoneId = if (isLTZ) Some(sessionZone) else None
)
}
+
+ /**
+ * Builds the three appended output attributes (`bin_start`, `bin_end`,
`bin_distribute_ratio`),
+ * applying `aliases`; `rangeType` is the type of `bin_start` / `bin_end`.
+ */
+ 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)())
+
+ /**
+ * 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.
+ */
+ def scaledDistributeAttributes(distributeColumns: Seq[Attribute]):
Seq[Attribute] =
+ distributeColumns.map(a => AttributeReference(a.name, a.dataType,
a.nullable)())
}
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..1fb703814fb9 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
@@ -202,8 +202,9 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
existingRelations,
b,
_.producedAttributes.map(_.exprId.id).toSeq,
- newBinBy => newBinBy.copy(appendedAttributes =
- newBinBy.appendedAttributes.map(_.newInstance())))
+ newBinBy => newBinBy.copy(
+ scaledDistributeColumns =
newBinBy.scaledDistributeColumns.map(_.newInstance()),
+ appendedAttributes =
newBinBy.appendedAttributes.map(_.newInstance())))
case e: Expand =>
deduplicateAndRenew[Expand](
@@ -470,6 +471,7 @@ object DeduplicateRelations extends Rule[LogicalPlan] {
case oldVersion: BinBy
if
oldVersion.producedAttributes.intersect(conflictingAttributes).nonEmpty =>
val newVersion = oldVersion.copy(
+ scaledDistributeColumns =
oldVersion.scaledDistributeColumns.map(_.newInstance()),
appendedAttributes =
oldVersion.appendedAttributes.map(_.newInstance()))
newVersion.copyTagsFrom(oldVersion)
Seq((oldVersion, newVersion))
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
index 9cd225628d64..d90326cc3e0b 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBy.scala
@@ -65,7 +65,8 @@ object ResolveBinBy extends Rule[LogicalPlan] {
originExpr = b.originExpr)
val appendedAttributes =
- BinBy.appendedAttributesWithAliases(parameters.rangeType,
b.outputAliases)
+ BinByResolution.appendedAttributesWithAliases(parameters.rangeType,
b.outputAliases)
+ val scaledDistributeColumns =
BinByResolution.scaledDistributeAttributes(distributeAttributes)
BinBy(
binWidthMicros = parameters.binWidthMicros,
@@ -73,6 +74,7 @@ object ResolveBinBy extends Rule[LogicalPlan] {
rangeEnd = rangeEnd,
originMicros = parameters.originMicros,
distributeColumns = distributeAttributes,
+ scaledDistributeColumns = scaledDistributeColumns,
appendedAttributes = appendedAttributes,
child = child,
timeZoneId = parameters.timeZoneId)
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 673edaab1368..9ca9897c123b 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}
@@ -1799,8 +1798,13 @@ case class UnresolvedBinBy(
* @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`.
+ * @param distributeColumns Resolved input columns to proportionally
redistribute. Read by the
+ * operator to compute the rescaled values; not
part of `output`.
+ * @param scaledDistributeColumns
+ * Produced output attributes holding the rescaled
values (fresh
+ * `ExprId`s, same names/types as
`distributeColumns`); they replace
+ * `distributeColumns` in `output`.
+ * @param appendedAttributes The three output attributes appended after the
child columns.
* @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
@@ -1812,20 +1816,30 @@ case class BinBy(
rangeEnd: Attribute,
originMicros: Long,
distributeColumns: Seq[Attribute],
+ scaledDistributeColumns: 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")
- }
+ assert(timeZoneId.isDefined ==
rangeStart.dataType.isInstanceOf[TimestampType],
+ s"timeZoneId must be set iff rangeStart is TIMESTAMP (LTZ); got
rangeStart.dataType=" +
+ s"${rangeStart.dataType}, timeZoneId=$timeZoneId")
+
+ assert(distributeColumns.length == scaledDistributeColumns.length,
+ "BinBy requires one scaled attribute per DISTRIBUTE column, got " +
+ s"${distributeColumns.length} distribute columns and " +
+ s"${scaledDistributeColumns.length} scaled attributes")
+
+ // In `output`, each DISTRIBUTE input is replaced by its scaled produced
counterpart.
+ private lazy val distributeReplacements: AttributeMap[Attribute] =
+ AttributeMap(distributeColumns.zip(scaledDistributeColumns))
- override def output: Seq[Attribute] = child.output ++ appendedAttributes
+ override def output: Seq[Attribute] =
+ child.output.map(a => distributeReplacements.getOrElse(a, a)) ++
appendedAttributes
- override def producedAttributes: AttributeSet =
AttributeSet(appendedAttributes)
+ override def producedAttributes: AttributeSet =
+ AttributeSet(scaledDistributeColumns ++ appendedAttributes)
final override val nodePatterns: Seq[TreePattern] = Seq(BIN_BY)
@@ -1833,15 +1847,6 @@ case class 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/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
index 43134eed760d..ff44ae657411 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/analysis/ResolveBinBySuite.scala
@@ -47,6 +47,7 @@ class ResolveBinBySuite extends AnalysisTest {
private val tsEndNtz = $"ts_end".timestampNTZ
private val value = $"value".double
private val label = $"label".string
+ private val label2 = $"label2".string
private val ltzChild: LogicalPlan = LocalRelation(tsStart, tsEnd, value,
label)
private val ntzChild: LogicalPlan = LocalRelation(tsStartNtz, tsEndNtz,
value)
@@ -202,6 +203,67 @@ class ResolveBinBySuite extends AnalysisTest {
assert(bi.distributeColumns.map(_.exprId) == Seq(value.exprId))
}
+ test("resolved BinBy emits the DISTRIBUTE column as a produced attribute
replacing the input") {
+ // `value` sits mid-schema (not last) and carries a qualifier + metadata,
so this covers
+ // in-place replacement, produced identity, and the qualifier/metadata
drop in one go.
+ val md = new MetadataBuilder().putString("comment", "a measure").build()
+ val valueMd = AttributeReference("value", DoubleType, nullable = true,
md)()
+ val child = SubqueryAlias("m", LocalRelation(tsStart, tsEnd, valueMd,
label))
+ val bi = ResolveBinBy.apply(
+ unresolved(child = child, distribute = Seq(UnresolvedAttribute(Seq("m",
"value")))))
+ .asInstanceOf[BinBy]
+
+ // The input is read (held in distributeColumns) but not forwarded by
identity.
+ assert(bi.distributeColumns.head.qualifier == Seq("m"))
+ assert(bi.distributeColumns.head.metadata == md)
+ assert(!bi.output.exists(_.exprId == valueMd.exprId))
+
+ // It is replaced at its own position by a fresh-id, same-name produced
attribute.
+ val outValue = bi.output(child.output.indexWhere(_.exprId ==
valueMd.exprId))
+ assert(outValue.name == "value" && outValue.exprId != valueMd.exprId)
+ assert(bi.scaledDistributeColumns.map(_.exprId) == Seq(outValue.exprId))
+ assert(bi.producedAttributes.contains(outValue))
+
+ // The produced (computed) value drops the input's qualifier and metadata.
+ assert(outValue.qualifier.isEmpty && outValue.metadata == Metadata.empty)
+
+ // Forwarded (non-distribute) columns keep their identity.
+ assert(bi.output.exists(_.exprId == label.exprId))
+ assert(bi.output.exists(_.exprId == tsStart.exprId))
+ }
+
+ test("resolved BinBy emits each of multiple DISTRIBUTE columns as a produced
attribute " +
+ "replacing the input") {
+ // `v1`, `v2`, `v3` sit at non-adjacent schema positions with forwarded
columns between
+ // them, so this covers per-slot in-place replacement with distinct fresh
ids.
+ val v1 = AttributeReference("v1", DoubleType, nullable = true)()
+ val v2 = AttributeReference("v2", DoubleType, nullable = true)()
+ val v3 = AttributeReference("v3", DoubleType, nullable = true)()
+ val child = LocalRelation(tsStart, tsEnd, v1, label, v2, label2, v3)
+ val distribute = Seq(v1, v2, v3)
+ val bi = ResolveBinBy.apply(
+ unresolved(child = child, distribute = distribute)).asInstanceOf[BinBy]
+
+ // The inputs are read (held in distributeColumns) but none is forwarded
by identity.
+ assert(bi.distributeColumns.map(_.exprId) == distribute.map(_.exprId))
+ assert(distribute.forall(v => !bi.output.exists(_.exprId == v.exprId)))
+
+ // Each is replaced at its own position by a fresh-id, same-name produced
attribute.
+ val outputs = distribute.map(v =>
bi.output(child.output.indexWhere(_.exprId == v.exprId)))
+ outputs.zip(distribute).foreach { case (out, in) =>
+ assert(out.name == in.name && out.exprId != in.exprId)
+ }
+ assert(outputs.map(_.exprId).distinct.length == distribute.length)
+ assert(bi.scaledDistributeColumns.map(_.exprId) == outputs.map(_.exprId))
+ assert(outputs.forall(bi.producedAttributes.contains))
+
+ // The child portion of `output` keeps the child's column order and names.
+ assert(bi.output.take(child.output.length).map(_.name) ==
child.output.map(_.name))
+ assert(bi.output.exists(_.exprId == label.exprId))
+ assert(bi.output.exists(_.exprId == label2.exprId))
+ assert(bi.output.exists(_.exprId == tsStart.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)()
@@ -330,9 +392,10 @@ class ResolveBinBySuite extends AnalysisTest {
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")
+ val producedExprIds = binBys.flatMap(b =>
+ (b.scaledDistributeColumns ++ b.appendedAttributes).map(_.exprId))
+ assert(producedExprIds.distinct.size == producedExprIds.size,
+ "produced BinBy attributes must have distinct exprIds across the two
join sides")
}
// `super.test` escapes the suite-wide flag-on wrapper; pin the flag off
explicitly.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
