You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
{{ message }}
Repository navigation
Native window over a sorted cache fails with "Expects PARTITION BY expression to be ordered" #7223
Spark drops the sort below a window when its input already reports the order the window needs.
A relation sorted before it was cached reports that order through InMemoryTableScanExec, so a
window partitioned and ordered by the cache's sort keys gets no sort of its own. When Comet reads
the cache from the JVM (spark.comet.convert.inMemoryCache.enabled=true with spark.comet.exec.inMemoryCache.enabled=false), the rows enter the native plan through a ScanExec, which declares no order to DataFusion. The native BoundedWindowAggExec runs in InputOrderMode::Sorted (native/core/src/execution/planner/window.rs), checks its input's
declared order and fails the task:
org.apache.comet.CometNativeException: Expects PARTITION BY expression to be ordered
The rows are in fact sorted. Only the declaration is missing.
With #6785, an inner hash join between the cache and the window no longer hits this: the join
sends the streamed side's order to the native planner, which declares it on the probe input. A
window directly over the cache has no such operator in between. A native window directly over a
sorted bucketed table read by the native scan (one file per bucket, with spark.sql.legacy.bucketedTableScan.outputOrdering=true) fails on main with the same error.
Steps to reproduce
On Spark 4.1.3:
spark.conf.set("spark.sql.adaptive.enabled", "false")
spark.conf.set("spark.sql.autoBroadcastJoinThreshold", "-1")
spark.conf.set("spark.sql.shuffle.partitions", "2")
spark.conf.set("spark.comet.exec.range.enabled", "true")
spark.conf.set("spark.comet.exec.inMemoryCache.enabled", "false")
spark.conf.set("spark.comet.convert.inMemoryCache.enabled", "true")
valcached= spark
.range(2000)
.selectExpr("id % 20 AS k", "id AS v")
.repartition(2, col("k"))
.sortWithinPartitions("k", "v")
.cache()
cached.count()
cached.selectExpr("k", "v", "dense_rank() OVER (PARTITION BY k ORDER BY v) AS r").collect()
Expected behavior
The 2,000 ranked rows Spark returns.
Additional context
One fix is to carry the order Spark reports for the window's input to the native side, for
example an input_sort_orders field on the Window proto that the planner declares on the
window's input with with_declared_order, guarded the same way CometHashJoin guards streamedSortOrders (mayReportFalseOrder). Another is to keep a sort below the native window
when its input reads from the JVM and reports an order.
Activity
fix: declare the input order of a native window over a sorted cache