You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/10/19 13:08:04 UTC
(datafusion-comet) 01/01: refactor: share window frame bound offset serialization between lower and upper bounds (#7107)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-7107-f31b0def60cab6603ca3464f678d0c813a5f643f
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit a2cbe266e28c18696a2bbfa162510d387314c382
Author: ChenChen Lai <72...@users.noreply.github.com>
AuthorDate: Mon Oct 19 13:07:46 2026 +0000
refactor: share window frame bound offset serialization between lower and upper bounds (#7107)
---
.../apache/spark/sql/comet/CometWindowExec.scala | 119 ++++++++++-----------
.../apache/comet/exec/CometWindowExecSuite.scala | 18 +++-
2 files changed, 73 insertions(+), 64 deletions(-)
diff --git a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala
index 33c2da2117..aefa252615 100644
--- a/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala
+++ b/spark/src/main/scala/org/apache/spark/sql/comet/CometWindowExec.scala
@@ -21,7 +21,7 @@ package org.apache.spark.sql.comet
import scala.jdk.CollectionConverters._
-import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, Cast, CumeDist, CurrentRow, DenseRank, Divide, Expression, Lag, Lead, Literal, MakeDecimal, NamedExpression, NthValue, NTile, PercentRank, RangeFrame, Rank, RowFrame, RowNumber, SortOrder, SpecialFrameBoundary, SpecifiedWindowFrame, UnboundedFollowing, UnboundedPreceding, UnscaledValue, WindowExpression}
+import org.apache.spark.sql.catalyst.expressions.{Alias, Attribute, AttributeSet, Cast, CumeDist, CurrentRow, DenseRank, Divide, Expression, FrameType, Lag, Lead, Literal, MakeDecimal, NamedExpression, NthValue, NTile, PercentRank, RangeFrame, Rank, RowFrame, RowNumber, SortOrder, SpecialFrameBoundary, SpecifiedWindowFrame, UnboundedFollowing, UnboundedPreceding, UnscaledValue, WindowExpression}
import org.apache.spark.sql.catalyst.expressions.aggregate.{AggregateExpression, Average, Complete, Count, First, Last, Max, Min, Sum}
import org.apache.spark.sql.catalyst.plans.physical.Partitioning
import org.apache.spark.sql.execution.SparkPlan
@@ -511,44 +511,26 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] {
.newBuilder()
.setCurrentRow(OperatorOuterClass.CurrentRow.newBuilder().build())
.build()
- case e if frameType == RowFrame =>
- val offset = e.eval() match {
- case i: Integer => i.toLong
- case l: Long => l
- case _ =>
- withFallbackReason(
- windowExpr,
- s"Unsupported ROWS frame lower offset: $e (${e.dataType})")
- return None
- }
- OperatorOuterClass.LowerWindowFrameBound
- .newBuilder()
- .setPreceding(
+ case e =>
+ val preceding = boundOffset(e, frameType, isLower = true, output) match {
+ case Right(RowOffset(offset)) =>
OperatorOuterClass.Preceding
.newBuilder()
.setOffset(offset)
- .build())
- .build()
- case e if frameType == RangeFrame && e.dataType.isInstanceOf[NumericType] =>
- rangeBoundLiteral(e, isLower = true, output) match {
- case Some(lit) =>
- OperatorOuterClass.LowerWindowFrameBound
+ .build()
+ case Right(RangeOffset(lit)) =>
+ OperatorOuterClass.Preceding
.newBuilder()
- .setPreceding(
- OperatorOuterClass.Preceding
- .newBuilder()
- .setRangeOffset(lit)
- .build())
+ .setRangeOffset(lit)
.build()
- case None =>
- withFallbackReason(windowExpr, s"Unsupported RANGE frame lower offset: $e")
+ case Left(reason) =>
+ withFallbackReason(windowExpr, reason)
return None
}
- case e =>
- withFallbackReason(
- windowExpr,
- s"RANGE frame with non-numeric offset is not supported: ${e.dataType}")
- return None
+ OperatorOuterClass.LowerWindowFrameBound
+ .newBuilder()
+ .setPreceding(preceding)
+ .build()
}
val uBoundProto = uBound match {
@@ -562,44 +544,26 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] {
.newBuilder()
.setCurrentRow(OperatorOuterClass.CurrentRow.newBuilder().build())
.build()
- case e if frameType == RowFrame =>
- val offset = e.eval() match {
- case i: Integer => i.toLong
- case l: Long => l
- case _ =>
- withFallbackReason(
- windowExpr,
- s"Unsupported ROWS frame upper offset: $e (${e.dataType})")
- return None
- }
- OperatorOuterClass.UpperWindowFrameBound
- .newBuilder()
- .setFollowing(
+ case e =>
+ val following = boundOffset(e, frameType, isLower = false, output) match {
+ case Right(RowOffset(offset)) =>
OperatorOuterClass.Following
.newBuilder()
.setOffset(offset)
- .build())
- .build()
- case e if frameType == RangeFrame && e.dataType.isInstanceOf[NumericType] =>
- rangeBoundLiteral(e, isLower = false, output) match {
- case Some(lit) =>
- OperatorOuterClass.UpperWindowFrameBound
+ .build()
+ case Right(RangeOffset(lit)) =>
+ OperatorOuterClass.Following
.newBuilder()
- .setFollowing(
- OperatorOuterClass.Following
- .newBuilder()
- .setRangeOffset(lit)
- .build())
+ .setRangeOffset(lit)
.build()
- case None =>
- withFallbackReason(windowExpr, s"Unsupported RANGE frame upper offset: $e")
+ case Left(reason) =>
+ withFallbackReason(windowExpr, reason)
return None
}
- case e =>
- withFallbackReason(
- windowExpr,
- s"RANGE frame with non-numeric offset is not supported: ${e.dataType}")
- return None
+ OperatorOuterClass.UpperWindowFrameBound
+ .newBuilder()
+ .setFollowing(following)
+ .build()
}
(frameProto, lBoundProto, uBoundProto)
@@ -700,6 +664,35 @@ object CometWindowExec extends CometOperatorSerde[WindowExec] {
}
}
+ private sealed trait BoundOffset
+ private case class RowOffset(offset: Long) extends BoundOffset
+ private case class RangeOffset(literal: LiteralOuterClass.Literal) extends BoundOffset
+
+ // Resolves the offset of a PRECEDING/FOLLOWING frame bound, shared by the
+ // lower and upper bound so both apply the same checks. Returns the fallback
+ // reason on Left when the offset cannot be serialized.
+ private def boundOffset(
+ bound: Expression,
+ frameType: FrameType,
+ isLower: Boolean,
+ output: Seq[Attribute]): Either[String, BoundOffset] = {
+ val side = if (isLower) "lower" else "upper"
+ bound match {
+ case e if frameType == RowFrame =>
+ e.eval() match {
+ case i: Integer => Right(RowOffset(i.toLong))
+ case l: Long => Right(RowOffset(l))
+ case _ => Left(s"Unsupported ROWS frame $side offset: $e (${e.dataType})")
+ }
+ case e if frameType == RangeFrame && e.dataType.isInstanceOf[NumericType] =>
+ rangeBoundLiteral(e, isLower, output)
+ .map(RangeOffset)
+ .toRight(s"Unsupported RANGE frame $side offset: $e")
+ case e =>
+ Left(s"RANGE frame with non-numeric offset is not supported: ${e.dataType}")
+ }
+ }
+
// Folds a RANGE frame bound expression to a constant and serializes its
// magnitude as a typed Literal proto. Spark encodes PRECEDING/FOLLOWING via
// the sign of the literal (negative => PRECEDING, positive => FOLLOWING),
diff --git a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala
index b392706d12..7f4bf22491 100644
--- a/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/exec/CometWindowExecSuite.scala
@@ -414,7 +414,23 @@ class CometWindowExecSuite extends CometTestBase {
val df2 =
df.select($"value", sum($"value").over(window.rangeBetween(1L, Window.unboundedFollowing)))
- checkSparkAnswer(df2)
+ checkSparkAnswerAndFallbackReason(df2, "Unsupported RANGE frame lower offset")
+ val cometWindowExecs = collect(df2.queryExecution.executedPlan) { case w: CometWindowExec =>
+ w
+ }
+ assert(cometWindowExecs.isEmpty)
+ }
+
+ test("window query with rangeBetween PRECEDING upper bound falls back to Spark") {
+ // `rangeBetween(unboundedPreceding, -1L)` puts a negative offset (PRECEDING semantic)
+ // in the upper bound position, which the proto only encodes as Following. We fall
+ // back to Spark rather than serialize the magnitude as a FOLLOWING bound.
+ val df = Seq(1, 2, 4, 3, 2, 1).toDF("value")
+ val window = Window.orderBy($"value".desc)
+ val df2 =
+ df.select($"value", sum($"value").over(window.rangeBetween(Window.unboundedPreceding, -1L)))
+
+ checkSparkAnswerAndFallbackReason(df2, "Unsupported RANGE frame upper offset")
val cometWindowExecs = collect(df2.queryExecution.executedPlan) { case w: CometWindowExec =>
w
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org