Repository navigation
Description
Describe the bug
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")
val cached = 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.
Found in #6785 (#6785 (comment)).
Activity
fix: declare the input order of a native window over a sorted cache
fix: declare the input order of a native window over a sorted cache
fix: declare the input order of a native window over a sorted cache (a…