Parallel bounded RANGE-frame window functions without PARTITION BY (draft)

[avantgardnerio]

Which issue does this PR close?

None, yet.

Rationale for this change

This is a draft PR as a PoC up for architecture feedback, not approval.

In Coralogix, roughly 80% of the observed slow/failed/OOM queries are window queries with an order, bounds, and no patitionby. These move all the data to 1 core of 1 node and are never likely to be successful this way. This PR demonstrates what I believe is a relatively simple & straight forward solution to parallelizing these queries.

What changes are included in this PR?

1. Exploiting an observed fact within DataFusion. Although it is generally a streaming engine, pipeline-breaking methods like sorts load every row into memory/disk before they emit a single result. This is analogous to stages in Ballista. So this PR exploits that by adding a run time PartitionExtremes method to physical plan nodes to expose the min/max values of their partition, if known. SortExec can easily know this once it has seen all the data and before it emits it's first batch.

2. RangeRepartitionExec. I know this is duplicate work ATM, and before a PR like this one is ever merged, I would unify this operator with PartitionExec::Range to align directionally with upstream. But for reasons of expedience it is a new operator in this PR. Rather than trying to add dynamic numbers of output partitions, it uses PartitionExtremes at run time to calculate the ranges of values that should go into each partition, including the new concept of Halo Rows (also under investigation in this PR). The output partitions from RRE can have overlaps, and if a row belongs in multiple output partitions, it gets duplicated - this gives BoundedWindowAgg (BWAG) what it needs to correctly calculate totals, and the halo rows will be dropped later

3. HaloDropExec. A new operator that checks rows against what should be in PartitionExtremes, dropping any rows that are outside ownership of the partition. This is how we remove halo rows later

4. ParallelWindow optimizer rule. This detects candidate BWAGs and inserts the appropriate RRE & HaloDrop

Target physical plan shape (from the included SLT):

SortPreservingMergeExec  (N → 1)
  HaloDropExec  (N partitions; per-partition primary filter)
    BoundedWindowAggExec  (N partitions, parallel_aware)
      RangeRepartitionExec  (N, halo duplicates intended)
        SortExec  (preserve_partitioning, N)
          DataSource

flowchart TB
    subgraph Before["Before ParallelWindow"]
        direction TB
        DS1[DataSource<br/>4 partitions] --> S1[SortExec<br/>preserve_partitioning]
        S1 --> SPM1[SortPreservingMergeExec<br/><b>N → 1</b>]
        SPM1 --> BWAG1["BoundedWindowAggExec<br/><b>SinglePartition</b><br/>(serial)"]
        BWAG1 --> Out1[Projection + SortLimit]
    end
    subgraph After["After ParallelWindow"]
        direction TB
        DS2[DataSource<br/>4 partitions] --> S2[SortExec<br/>preserve_partitioning]
        S2 --> RRE["RangeRepartitionExec<br/>K-way coordinator<br/>+ halo duplication"]
        RRE --> BWAG2["BoundedWindowAggExec<br/><b>parallel_aware</b><br/>N partitions"]
        BWAG2 --> HDE["HaloDropExec<br/>per-partition filter<br/>reads runtime_partition_extremes"]
        HDE --> SPM2[SortPreservingMergeExec<br/>N → 1, fetch=5]
        SPM2 --> Out2[Projection]
    end
    style BWAG1 fill:#f9b,stroke:#a00
    style BWAG2 fill:#bf9,stroke:#0a0
    style RRE fill:#bf9,stroke:#0a0
    style HDE fill:#bf9,stroke:#0a0

Loading

Input (post-sort, single sequence 0..99):
  ┌─────────────────────────────────────────────────────────────────────────┐
  │ 0 1 2 ... 23 24 │ 25 26 ... 48 49 │ 50 51 ... 73 74 │ 75 76 ... 98 99   │
  └─────────────────┴─────────────────┴─────────────────┴───────────────────┘
   primary [0, 25)   primary [25, 50)  primary [50, 75)  primary [75, 100)

RangeRepartitionExec — boundaries computed from runtime extremes,
                       halo_preceding = 5 (from RANGE 5 PRECEDING):

  bucket 0:                  [   0 .. 24 ]              25 rows
  bucket 1:        [20..24 │ 25 .. 49 ]                 30 rows  ← 5 halo + 25 primary
  bucket 2:        [45..49 │ 50 .. 74 ]                 30 rows  ← 5 halo + 25 primary
  bucket 3:        [70..74 │ 75 .. 99 ]                 30 rows  ← 5 halo + 25 primary
                    ─────                              ─────
                    halo                          total routed = 115

BoundedWindowAggExec (per-partition, parallel_aware):
  bucket 1 at row 25 sees rows 20..25 in its frame → rolling_sum correct at the seam.
  Same for buckets 2, 3.

HaloDropExec (per-partition, reads bucket's intended primary range):
  bucket 0 keeps [0, 25)     → 25 rows
  bucket 1 drops [20, 25), keeps [25, 50) → 25 rows
  bucket 2 drops [45, 50), keeps [50, 75) → 25 rows
  bucket 3 drops [70, 75), keeps [75, 100) → 25 rows
                                              ─────
                                       total emitted = 100

The "useful lie": each bucket's runtime_partition_extremes returns the
INTENDED range [primary_lo, primary_hi], not the ACTUAL data range
[primary_lo - halo, primary_hi]. HaloDropExec consumes that lie to strip
exactly the halo rows.

Are these changes tested?

sqllogictest/test_files/parallel_window.slt is the fixture:

• 4 parquet files with overlapping seq ranges (so routing actually
  has to move rows around),
• EXPLAIN assertion on the new plan shape,
• value assertion on a 5-row truncation,
• row-count assertion that catches halo leakage
  (count(rolling_sum) = 100, not 115).

I have not yet added a feature flag + equality SLT comparing parallel
vs. serial outputs row-for-row. That's the next baby step and is what
will catch slide-accumulator boundary bugs that the current count
assertion can't see. Posting now to drive the design conversation
before investing in that loop.

Are there any user-facing changes?

• New nodes in plan trees: RangeRepartitionExec, HaloDropExec.
• New trait method ExecutionPlan::runtime_partition_extremes with a
  default Ok(None) — existing implementations are unaffected.
• New PartitionExtremes re-export from physical-plan.
• No SQL surface changes.

Known scope cuts (PoC)

• Int64-only leading sort key with Int64-bounded RANGE distances.
  The runtime_partition_extremes type is generic
  (Vec<ScalarValue>); the coordinator's bucketing arithmetic is
  what's Int64-only. DataType-dispatch is the permanent architecture
  here, not sample-based bucketing.
• Single sort key (LexOrdering len 1). Multi-key extremes are
  in the type from day one; coordinator + halo math are not.
• No feature flag yet. The rule fires unconditionally on eligible
  windows. A session config optimizer.enable_parallel_window (default
  off) is the next commit.
• No equality SLT comparing parallel-on vs. parallel-off outputs.
  Next commit after the flag.
• No Ballista distribution yet. The coordinator is the AQE
  reducer; lifting it scheduler-side is a follow-up (~300-500 LoC).
• No benchmark numbers yet. Coming after the equality SLT lands.

Coexistence with #22395

Partitioning::Range + RangePartitioning { ordering, split_points }
have already merged from @gabotechs. The execution slot in
RepartitionExec for Partitioning::Range is still
not_impl_err!. This PR keeps RangeRepartitionExec as a separate
operator for now because it does something RepartitionExec is
unlikely to absorb (halo duplication). Once the design is settled here
I expect to thread the existing RangePartitioning shape through
where the boundaries live, rather than carry a parallel
Vec<ScalarValue> representation forever.

Draft status

This is up for architecture review, not approval. CI will be red —
clippy / fmt / unrelated tests are not yet tended. I will not push
forward to "green" until reviewers agree the shape is roughly right.