From 8a2610938c375f1a77903f31173ada23d10346a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nikolina=20Vrane=C5=A1?= Date: Thu, 9 Jul 2026 13:16:14 +0000 Subject: [PATCH 1/4] [SPARK-58063][SQL] Add column pruning for BIN BY --- .../sql/catalyst/analysis/ResolveBinBy.scala | 1 + .../sql/catalyst/optimizer/Optimizer.scala | 11 +++++ .../plans/logical/basicLogicalOperators.scala | 12 ++++- .../optimizer/ColumnPruningSuite.scala | 48 ++++++++++++++++++- 4 files changed, 70 insertions(+), 2 deletions(-) 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 d90326cc3e0b1..d6203f2c5a7ac 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 @@ -76,6 +76,7 @@ object ResolveBinBy extends Rule[LogicalPlan] { distributeColumns = distributeAttributes, scaledDistributeColumns = scaledDistributeColumns, appendedAttributes = appendedAttributes, + unrequiredChildIndex = Nil, child = child, timeZoneId = parameters.timeZoneId) } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala index 597a70e01a6e9..424e2361e3b01 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala @@ -1135,6 +1135,17 @@ object ColumnPruning extends Rule[LogicalPlan] { .map(_._2) p.copy(child = g.copy(child = newChild, unrequiredChildIndex = unrequiredIndices)) + // prune unrequired pass-through columns from child of BinBy. + case p @ Project(_, b: BinBy) if p.references != b.outputSet => + val requiredAttrs = p.references -- b.producedAttributes ++ b.references + val newChild = prunedChild(b.child, requiredAttrs) + // Drop from output the kernel-read columns the project does not forward; they stay in + // newChild. DISTRIBUTE columns are excluded (swapped to scaled attrs in output). + val unrequired = b.references -- AttributeSet(b.distributeColumns) -- p.references + val unrequiredIndices = newChild.output.zipWithIndex + .filter(t => unrequired.contains(t._1)).map(_._2) + p.copy(child = b.copy(child = newChild, unrequiredChildIndex = unrequiredIndices)) + // prune unrequired nested fields from `Generate`. case GeneratorNestedColumnAliasing(rewrittenPlan) => rewrittenPlan 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 b7c441063874d..0d812ece53401 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 @@ -1804,6 +1804,9 @@ case class UnresolvedBinBy( * `ExprId`s, same names/types as `distributeColumns`); they replace * `distributeColumns` in `output`. * @param appendedAttributes The three output attributes appended after the child columns. + * @param unrequiredChildIndex Child-output positions not forwarded to `output`; starts as `Nil` and + * is filled by `ColumnPruning`. The kernel-read range and DISTRIBUTE + * columns stay in the child even when listed here. * @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 @@ -1817,6 +1820,7 @@ case class BinBy( distributeColumns: Seq[Attribute], scaledDistributeColumns: Seq[Attribute], appendedAttributes: Seq[Attribute], + unrequiredChildIndex: Seq[Int], child: LogicalPlan, timeZoneId: Option[String]) extends UnaryNode { @@ -1834,8 +1838,14 @@ case class BinBy( private lazy val distributeReplacements: AttributeMap[Attribute] = AttributeMap(distributeColumns.zip(scaledDistributeColumns)) + /** Forwarded child columns after column pruning. */ + lazy val requiredChildOutput: Seq[Attribute] = { + val unrequiredSet = unrequiredChildIndex.toSet + child.output.zipWithIndex.filterNot(t => unrequiredSet.contains(t._2)).map(_._1) + } + override def output: Seq[Attribute] = - child.output.map(a => distributeReplacements.getOrElse(a, a)) ++ appendedAttributes + requiredChildOutput.map(a => distributeReplacements.getOrElse(a, a)) ++ appendedAttributes override def producedAttributes: AttributeSet = AttributeSet(scaledDistributeColumns ++ appendedAttributes) diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala index 266c369894eca..73ddad1bd034c 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala @@ -28,7 +28,7 @@ import org.apache.spark.sql.catalyst.plans.{Inner, PlanTest} import org.apache.spark.sql.catalyst.plans.logical._ import org.apache.spark.sql.catalyst.rules.RuleExecutor import org.apache.spark.sql.internal.SQLConf -import org.apache.spark.sql.types.{StringType, StructType} +import org.apache.spark.sql.types.{DoubleType, StringType, StructType, TimestampType} class ColumnPruningSuite extends PlanTest { @@ -482,4 +482,50 @@ class ColumnPruningSuite extends PlanTest { comparePlans(CustomOptimize.execute(originalQuery.analyze), correctAnswer.analyze) } + + test("Column pruning for BinBy") { + // 4-column relation: two range inputs, one DISTRIBUTE column, one unused pass-through. + val tsStart = AttributeReference("ts_start", TimestampType, nullable = false)() + val tsEnd = AttributeReference("ts_end", TimestampType, nullable = false)() + val value = AttributeReference("value", DoubleType, nullable = false)() + val label = AttributeReference("label", StringType, nullable = true)() + val relation = LocalRelation(tsStart, tsEnd, value, label) + + // Minted produced attributes: scaled DISTRIBUTE column + three BIN BY output columns. + val scaledValue = AttributeReference("value", DoubleType, nullable = true)() + val binStart = AttributeReference("bin_start", TimestampType, nullable = true)() + val binEnd = AttributeReference("bin_end", TimestampType, nullable = true)() + val binRatio = AttributeReference("bin_distribute_ratio", DoubleType, nullable = true)() + val appended = Seq(binStart, binEnd, binRatio) + + val binBy = BinBy( + binWidthMicros = 300000000L, + rangeStart = tsStart, + rangeEnd = tsEnd, + originMicros = 0L, + distributeColumns = Seq(value), + scaledDistributeColumns = Seq(scaledValue), + appendedAttributes = appended, + unrequiredChildIndex = Nil, + child = relation, + timeZoneId = Some("UTC")) + + // Project keeps only bin_start (a produced attr). `label` is forwarded by neither the project + // nor the kernel, so it is pruned out of the child entirely. The range columns stay in the + // child (the kernel reads them) but are not forwarded to output, so they populate + // unrequiredChildIndex. + val query = Project(Seq(binStart), binBy) + + val optimized = Optimize.execute(query) + + val binByResult = optimized.collectFirst { case b: BinBy => b }.get + assert(binByResult.unrequiredChildIndex.nonEmpty, + "unforwarded range columns must populate unrequiredChildIndex after ColumnPruning") + // range inputs and DISTRIBUTE column remain in the child output (the kernel reads them). + val keptNames = binByResult.child.output.map(_.name).toSet + assert(keptNames.contains("ts_start"), "rangeStart must remain in child output") + assert(keptNames.contains("ts_end"), "rangeEnd must remain in child output") + assert(keptNames.contains("value"), "DISTRIBUTE column must remain in child output") + assert(!keptNames.contains("label"), "label must have been pruned from child output") + } } From 025ae4a11f93dabba77a55de59dbbaab153b57c8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nikolina=20Vrane=C5=A1?= Date: Thu, 9 Jul 2026 14:49:11 +0000 Subject: [PATCH 2/4] [SPARK-58063][SQL] Add binBy test DSL helper and use it in the column-pruning test --- .../spark/sql/catalyst/dsl/package.scala | 14 ++++++ .../plans/logical/basicLogicalOperators.scala | 4 +- .../optimizer/ColumnPruningSuite.scala | 45 +++++++------------ 3 files changed, 30 insertions(+), 33 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala index 3ba538c0e2160..c5616b5df3d4d 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala @@ -524,6 +524,20 @@ package object dsl extends SQLConfHelper { Generate(generator, unrequiredChildIndex, outer, alias, outputNames.map(UnresolvedAttribute(_)), logicalPlan) + def binBy( + rangeStart: Attribute, + rangeEnd: Attribute, + distributeColumns: Seq[Attribute], + scaledDistributeColumns: Seq[Attribute], + appendedAttributes: Seq[Attribute], + binWidthMicros: Long = 300000000L, + originMicros: Long = 0L, + unrequiredChildIndex: Seq[Int] = Nil, + timeZoneId: Option[String] = Some("UTC")): LogicalPlan = + BinBy(binWidthMicros, rangeStart, rangeEnd, originMicros, distributeColumns, + scaledDistributeColumns, appendedAttributes, unrequiredChildIndex, logicalPlan, + timeZoneId) + def insertInto(tableName: String): LogicalPlan = insertInto(table(tableName)) def insertInto( 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 0d812ece53401..6e70f3aa00b22 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 @@ -1804,9 +1804,7 @@ case class UnresolvedBinBy( * `ExprId`s, same names/types as `distributeColumns`); they replace * `distributeColumns` in `output`. * @param appendedAttributes The three output attributes appended after the child columns. - * @param unrequiredChildIndex Child-output positions not forwarded to `output`; starts as `Nil` and - * is filled by `ColumnPruning`. The kernel-read range and DISTRIBUTE - * columns stay in the child even when listed here. + * @param unrequiredChildIndex Child-output positions not forwarded to `output`. * @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 diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala index 73ddad1bd034c..fc472888d5842 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala @@ -484,48 +484,33 @@ class ColumnPruningSuite extends PlanTest { } test("Column pruning for BinBy") { - // 4-column relation: two range inputs, one DISTRIBUTE column, one unused pass-through. + // Two range inputs, one DISTRIBUTE column, one unused pass-through. val tsStart = AttributeReference("ts_start", TimestampType, nullable = false)() val tsEnd = AttributeReference("ts_end", TimestampType, nullable = false)() val value = AttributeReference("value", DoubleType, nullable = false)() val label = AttributeReference("label", StringType, nullable = true)() val relation = LocalRelation(tsStart, tsEnd, value, label) - // Minted produced attributes: scaled DISTRIBUTE column + three BIN BY output columns. + // Produced attributes: scaled DISTRIBUTE column + three appended columns. val scaledValue = AttributeReference("value", DoubleType, nullable = true)() val binStart = AttributeReference("bin_start", TimestampType, nullable = true)() val binEnd = AttributeReference("bin_end", TimestampType, nullable = true)() val binRatio = AttributeReference("bin_distribute_ratio", DoubleType, nullable = true)() - val appended = Seq(binStart, binEnd, binRatio) - - val binBy = BinBy( - binWidthMicros = 300000000L, - rangeStart = tsStart, - rangeEnd = tsEnd, - originMicros = 0L, - distributeColumns = Seq(value), - scaledDistributeColumns = Seq(scaledValue), - appendedAttributes = appended, - unrequiredChildIndex = Nil, - child = relation, - timeZoneId = Some("UTC")) - - // Project keeps only bin_start (a produced attr). `label` is forwarded by neither the project - // nor the kernel, so it is pruned out of the child entirely. The range columns stay in the - // child (the kernel reads them) but are not forwarded to output, so they populate - // unrequiredChildIndex. - val query = Project(Seq(binStart), binBy) + + val query = relation + .binBy(tsStart, tsEnd, Seq(value), Seq(scaledValue), Seq(binStart, binEnd, binRatio)) + .select(binStart) val optimized = Optimize.execute(query) - val binByResult = optimized.collectFirst { case b: BinBy => b }.get - assert(binByResult.unrequiredChildIndex.nonEmpty, - "unforwarded range columns must populate unrequiredChildIndex after ColumnPruning") - // range inputs and DISTRIBUTE column remain in the child output (the kernel reads them). - val keptNames = binByResult.child.output.map(_.name).toSet - assert(keptNames.contains("ts_start"), "rangeStart must remain in child output") - assert(keptNames.contains("ts_end"), "rangeEnd must remain in child output") - assert(keptNames.contains("value"), "DISTRIBUTE column must remain in child output") - assert(!keptNames.contains("label"), "label must have been pruned from child output") + // `label` is pruned from the child; the range columns stay in the child (the kernel reads + // them) but unforwarded, so they are unrequired; `value` stays (DISTRIBUTE, scaled in place). + val correctAnswer = relation + .select(tsStart, tsEnd, value) + .binBy(tsStart, tsEnd, Seq(value), Seq(scaledValue), Seq(binStart, binEnd, binRatio), + unrequiredChildIndex = Seq(0, 1)) + .select(binStart) + + comparePlans(optimized, correctAnswer, checkAnalysis = false) } } From b37456c7857b9cb7f543a1990e539151cba3d309 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nikolina=20Vrane=C5=A1?= Date: Thu, 9 Jul 2026 22:47:27 +0000 Subject: [PATCH 3/4] [SPARK-58063][SQL] Simplify BIN BY column pruning to child-pruning only --- .../spark/sql/catalyst/analysis/ResolveBinBy.scala | 1 - .../apache/spark/sql/catalyst/dsl/package.scala | 4 +--- .../spark/sql/catalyst/optimizer/Optimizer.scala | 14 +++++--------- .../plans/logical/basicLogicalOperators.scala | 10 +--------- .../catalyst/optimizer/ColumnPruningSuite.scala | 6 ++---- 5 files changed, 9 insertions(+), 26 deletions(-) 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 d6203f2c5a7ac..d90326cc3e0b1 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 @@ -76,7 +76,6 @@ object ResolveBinBy extends Rule[LogicalPlan] { distributeColumns = distributeAttributes, scaledDistributeColumns = scaledDistributeColumns, appendedAttributes = appendedAttributes, - unrequiredChildIndex = Nil, child = child, timeZoneId = parameters.timeZoneId) } diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala index c5616b5df3d4d..b4acde7026b6b 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/dsl/package.scala @@ -532,11 +532,9 @@ package object dsl extends SQLConfHelper { appendedAttributes: Seq[Attribute], binWidthMicros: Long = 300000000L, originMicros: Long = 0L, - unrequiredChildIndex: Seq[Int] = Nil, timeZoneId: Option[String] = Some("UTC")): LogicalPlan = BinBy(binWidthMicros, rangeStart, rangeEnd, originMicros, distributeColumns, - scaledDistributeColumns, appendedAttributes, unrequiredChildIndex, logicalPlan, - timeZoneId) + scaledDistributeColumns, appendedAttributes, logicalPlan, timeZoneId) def insertInto(tableName: String): LogicalPlan = insertInto(table(tableName)) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala index 424e2361e3b01..babfcc0f9ad66 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala @@ -1135,16 +1135,12 @@ object ColumnPruning extends Rule[LogicalPlan] { .map(_._2) p.copy(child = g.copy(child = newChild, unrequiredChildIndex = unrequiredIndices)) - // prune unrequired pass-through columns from child of BinBy. - case p @ Project(_, b: BinBy) if p.references != b.outputSet => + // prune unused pass-through columns from the child of BinBy. The range and DISTRIBUTE inputs + // the kernel reads are kept via `b.references`; only fully-unused pass-throughs are dropped. + case p @ Project(_, b: BinBy) + if !b.child.outputSet.subsetOf(p.references -- b.producedAttributes ++ b.references) => val requiredAttrs = p.references -- b.producedAttributes ++ b.references - val newChild = prunedChild(b.child, requiredAttrs) - // Drop from output the kernel-read columns the project does not forward; they stay in - // newChild. DISTRIBUTE columns are excluded (swapped to scaled attrs in output). - val unrequired = b.references -- AttributeSet(b.distributeColumns) -- p.references - val unrequiredIndices = newChild.output.zipWithIndex - .filter(t => unrequired.contains(t._1)).map(_._2) - p.copy(child = b.copy(child = newChild, unrequiredChildIndex = unrequiredIndices)) + p.copy(child = b.copy(child = prunedChild(b.child, requiredAttrs))) // prune unrequired nested fields from `Generate`. case GeneratorNestedColumnAliasing(rewrittenPlan) => rewrittenPlan 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 6e70f3aa00b22..b7c441063874d 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 @@ -1804,7 +1804,6 @@ case class UnresolvedBinBy( * `ExprId`s, same names/types as `distributeColumns`); they replace * `distributeColumns` in `output`. * @param appendedAttributes The three output attributes appended after the child columns. - * @param unrequiredChildIndex Child-output positions not forwarded to `output`. * @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 @@ -1818,7 +1817,6 @@ case class BinBy( distributeColumns: Seq[Attribute], scaledDistributeColumns: Seq[Attribute], appendedAttributes: Seq[Attribute], - unrequiredChildIndex: Seq[Int], child: LogicalPlan, timeZoneId: Option[String]) extends UnaryNode { @@ -1836,14 +1834,8 @@ case class BinBy( private lazy val distributeReplacements: AttributeMap[Attribute] = AttributeMap(distributeColumns.zip(scaledDistributeColumns)) - /** Forwarded child columns after column pruning. */ - lazy val requiredChildOutput: Seq[Attribute] = { - val unrequiredSet = unrequiredChildIndex.toSet - child.output.zipWithIndex.filterNot(t => unrequiredSet.contains(t._2)).map(_._1) - } - override def output: Seq[Attribute] = - requiredChildOutput.map(a => distributeReplacements.getOrElse(a, a)) ++ appendedAttributes + child.output.map(a => distributeReplacements.getOrElse(a, a)) ++ appendedAttributes override def producedAttributes: AttributeSet = AttributeSet(scaledDistributeColumns ++ appendedAttributes) diff --git a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala index fc472888d5842..e66cb800f773f 100644 --- a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala +++ b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/ColumnPruningSuite.scala @@ -503,12 +503,10 @@ class ColumnPruningSuite extends PlanTest { val optimized = Optimize.execute(query) - // `label` is pruned from the child; the range columns stay in the child (the kernel reads - // them) but unforwarded, so they are unrequired; `value` stays (DISTRIBUTE, scaled in place). + // `label` is pruned from the child; the range and DISTRIBUTE inputs the kernel reads stay. val correctAnswer = relation .select(tsStart, tsEnd, value) - .binBy(tsStart, tsEnd, Seq(value), Seq(scaledValue), Seq(binStart, binEnd, binRatio), - unrequiredChildIndex = Seq(0, 1)) + .binBy(tsStart, tsEnd, Seq(value), Seq(scaledValue), Seq(binStart, binEnd, binRatio)) .select(binStart) comparePlans(optimized, correctAnswer, checkAnalysis = false) From 7ebee25195fab24b31e32c47913760c63c522157 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Nikolina=20Vrane=C5=A1?= Date: Fri, 10 Jul 2026 16:05:31 +0000 Subject: [PATCH 4/4] [SPARK-58063][SQL] Rely on generic ColumnPruning fallback for BIN BY --- .../apache/spark/sql/catalyst/optimizer/Optimizer.scala | 7 ------- 1 file changed, 7 deletions(-) diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala index babfcc0f9ad66..597a70e01a6e9 100644 --- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala +++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala @@ -1135,13 +1135,6 @@ object ColumnPruning extends Rule[LogicalPlan] { .map(_._2) p.copy(child = g.copy(child = newChild, unrequiredChildIndex = unrequiredIndices)) - // prune unused pass-through columns from the child of BinBy. The range and DISTRIBUTE inputs - // the kernel reads are kept via `b.references`; only fully-unused pass-throughs are dropped. - case p @ Project(_, b: BinBy) - if !b.child.outputSet.subsetOf(p.references -- b.producedAttributes ++ b.references) => - val requiredAttrs = p.references -- b.producedAttributes ++ b.references - p.copy(child = b.copy(child = prunedChild(b.child, requiredAttrs))) - // prune unrequired nested fields from `Generate`. case GeneratorNestedColumnAliasing(rewrittenPlan) => rewrittenPlan