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:03 UTC
(datafusion-comet) branch gh-readonly-queue/main/pr-7107-f31b0def60cab6603ca3464f678d0c813a5f643f created (now a2cbe266e2)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-7107-f31b0def60cab6603ca3464f678d0c813a5f643f
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
at a2cbe266e2 refactor: share window frame bound offset serialization between lower and upper bounds (#7107)
This branch includes the following new commits:
new a2cbe266e2 refactor: share window frame bound offset serialization between lower and upper bounds (#7107)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion-comet) 01/01: refactor: share window frame bound offset serialization between lower and upper bounds (#7107)
Posted by gi...@apache.org.
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