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 529483475e73 [SPARK-58138][SQL] Preserve child output partitioning
through BIN BY
529483475e73 is described below
commit 529483475e73852e45e9d84152705af6594ac303
Author: Nikolina Vraneš <[email protected]>
AuthorDate: Thu Jul 16 01:14:14 2026 +0800
[SPARK-58138][SQL] Preserve child output partitioning through BIN BY
### What changes were proposed in this pull request?
This PR makes `BinByExec` extend `PartitioningPreservingUnaryExecNode` so
it reports an `outputPartitioning` derived from its child.
A child partitioning is kept only when its keys are in the operator's
output: a partitioning on a pass-through column survives, while one on a scaled
`DISTRIBUTE` column is dropped (that column has a fresh `ExprId` absent from
the output). Keeping the pass-through case is safe because `BIN BY` transforms
each partition independently and pass-through columns hold the same value
across an input row's sub-rows.
### Why are the changes needed?
A downstream operator clustered on a pass-through column that the child
already partitions on would otherwise get an unnecessary shuffle.
### 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?
`ProjectedOrderingAndPartitioningSuite`: a pass-through key is preserved as
`HashPartitioning`, a scaled `DISTRIBUTE` key is dropped to
`UnknownPartitioning`, and a mixed key is dropped whole.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Anthropic)
Closes #57275 from vranes/bin-by-output-partitioning.
Authored-by: Nikolina Vraneš <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
---
.../org/apache/spark/sql/execution/BinByExec.scala | 9 ++-
.../ProjectedOrderingAndPartitioningSuite.scala | 75 +++++++++++++++++++++-
2 files changed, 81 insertions(+), 3 deletions(-)
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
index d95addd93f40..22898f6dfdbd 100644
--- 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
@@ -22,7 +22,7 @@ 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.expressions.{Attribute, AttributeMap,
AttributeSet, BindReferences, BoundReference, Cast, Expression,
GenericInternalRow, JoinedRow, Literal, Multiply, NamedExpression,
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}
@@ -48,7 +48,7 @@ case class BinByExec(
appendedAttributes: Seq[Attribute],
timeZoneId: Option[String],
child: SparkPlan)
- extends UnaryExecNode {
+ extends UnaryExecNode with PartitioningPreservingUnaryExecNode {
override lazy val metrics: Map[String, SQLMetric] = Map(
"numOutputRows" -> SQLMetrics.createMetric(sparkContext, "number of output
rows"))
@@ -66,6 +66,11 @@ case class BinByExec(
override def producedAttributes: AttributeSet =
AttributeSet(scaledDistributeColumns ++ appendedAttributes)
+ // The trait keeps a child partitioning only when its keys are in the
output: a partitioning keyed
+ // on a pass-through column survives; one keyed on a scaled DISTRIBUTE
column (fresh ExprId) does
+ // not, because that column is absent from the output.
+ override protected def outputExpressions: Seq[NamedExpression] = output
+
override protected def withNewChildInternal(newChild: SparkPlan): BinByExec =
copy(child = newChild)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
index a70baece7784..cb5570bfdf2c 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/execution/ProjectedOrderingAndPartitioningSuite.scala
@@ -25,7 +25,7 @@ import
org.apache.spark.sql.connector.catalog.functions.{BucketFunction, YearsFu
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSparkSession
-import org.apache.spark.sql.types.{IntegerType, StringType}
+import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType,
TimestampType}
class ProjectedOrderingAndPartitioningSuite
extends SharedSparkSession with AdaptiveSparkPlanHelper {
@@ -602,6 +602,79 @@ class ProjectedOrderingAndPartitioningSuite
}
assert(e.getMessage.contains("partitionKeys"))
}
+
+ test("SPARK-58138: BIN BY preserves a child partitioning on a pass-through
column") {
+ val tsStart = AttributeReference("ts_start", TimestampType)()
+ val tsEnd = AttributeReference("ts_end", TimestampType)()
+ val value = AttributeReference("value", DoubleType)()
+ val host = AttributeReference("host", IntegerType)()
+ val binBy = BinByExec(
+ binWidthMicros = 300000000L, originMicros = 0L, rangeStart = tsStart,
rangeEnd = tsEnd,
+ distributeColumns = Seq(value),
+ scaledDistributeColumns = Seq(AttributeReference("value", DoubleType)()),
+ appendedAttributes = Seq(
+ AttributeReference("bin_start", TimestampType)(),
+ AttributeReference("bin_end", TimestampType)(),
+ AttributeReference("bin_distribute_ratio", DoubleType)()),
+ timeZoneId = None,
+ child = DummyLeafExecWithPartitioning(
+ output = Seq(tsStart, tsEnd, value, host),
+ partitioning = HashPartitioning(Seq(host), 4)))
+
+ binBy.outputPartitioning match {
+ case p: HashPartitioning =>
+ assert(p.expressions === Seq(host))
+ assert(p.numPartitions === 4)
+ case other => fail(s"Expected HashPartitioning, got $other")
+ }
+ }
+
+ test("SPARK-58138: BIN BY drops a child partitioning on a scaled DISTRIBUTE
column") {
+ val tsStart = AttributeReference("ts_start", TimestampType)()
+ val tsEnd = AttributeReference("ts_end", TimestampType)()
+ val value = AttributeReference("value", DoubleType)()
+ val binBy = BinByExec(
+ binWidthMicros = 300000000L, originMicros = 0L, rangeStart = tsStart,
rangeEnd = tsEnd,
+ distributeColumns = Seq(value),
+ scaledDistributeColumns = Seq(AttributeReference("value", DoubleType)()),
+ appendedAttributes = Seq(
+ AttributeReference("bin_start", TimestampType)(),
+ AttributeReference("bin_end", TimestampType)(),
+ AttributeReference("bin_distribute_ratio", DoubleType)()),
+ timeZoneId = None,
+ child = DummyLeafExecWithPartitioning(
+ output = Seq(tsStart, tsEnd, value),
+ partitioning = HashPartitioning(Seq(value), 4)))
+
+ binBy.outputPartitioning match {
+ case p: UnknownPartitioning => assert(p.numPartitions === 4)
+ case other => fail(s"Expected UnknownPartitioning, got $other")
+ }
+ }
+
+ test("SPARK-58138: BIN BY drops a mixed pass-through and DISTRIBUTE
partitioning whole") {
+ val tsStart = AttributeReference("ts_start", TimestampType)()
+ val tsEnd = AttributeReference("ts_end", TimestampType)()
+ val value = AttributeReference("value", DoubleType)()
+ val host = AttributeReference("host", IntegerType)()
+ val binBy = BinByExec(
+ binWidthMicros = 300000000L, originMicros = 0L, rangeStart = tsStart,
rangeEnd = tsEnd,
+ distributeColumns = Seq(value),
+ scaledDistributeColumns = Seq(AttributeReference("value", DoubleType)()),
+ appendedAttributes = Seq(
+ AttributeReference("bin_start", TimestampType)(),
+ AttributeReference("bin_end", TimestampType)(),
+ AttributeReference("bin_distribute_ratio", DoubleType)()),
+ timeZoneId = None,
+ child = DummyLeafExecWithPartitioning(
+ output = Seq(tsStart, tsEnd, value, host),
+ partitioning = HashPartitioning(Seq(value, host), 4)))
+
+ binBy.outputPartitioning match {
+ case p: UnknownPartitioning => assert(p.numPartitions === 4)
+ case other => fail(s"Expected UnknownPartitioning, got $other")
+ }
+ }
}
private case class DummyLeafExecWithPartitioning(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
