Skip to content

Native window over a sorted cache fails with "Expects PARTITION BY expression to be ordered" #7223

Open

Description

@dwsmith1983

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

  1. added
    bugSomething isn't working
    priority:mediumFunctional bugs, performance regressions, broken features
    and removed on Oct 26, 2026
  2. added this to the 1.3.0 milestone on Oct 26, 2026
  3. added
    priority:highCrashes, panics, segfaults, major functional breakage
    and removed
    priority:mediumFunctional bugs, performance regressions, broken features
    on Oct 26, 2026
  4. added a commit that references this issue on Oct 27, 2026
    68bff2e
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

Labels

bugSomething isn't workingpriority:highCrashes, panics, segfaults, major functional breakage

Type

No type

Projects

No projects

    Milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions