avantgardnerio opened a new pull request, #2392:
URL: https://github.com/apache/datafusion-ballista/pull/2392
## What this is for
`BoundedWindowAggExec` requires a single input partition, so a window with no
PARTITION BY runs serially no matter how wide the cluster is.
`ParallelWindowRule`
already fixes that for bounded RANGE frames and `PrefixWindowRule` for
UNBOUNDED
PRECEDING. This adds the third sibling, `HaloRowRule`, for bounded ROWS
frames:
```sql
avg(v2) OVER (ORDER BY id3 ROWS BETWEEN 100 PRECEDING AND CURRENT ROW)
```
On h2o window q6 at 1e7 that window is 10.03s of a 19.33s wall clock, in a
stage
of exactly one task: stage 0's writer declares `UnknownPartitioning`, so the
SPM
above the reader collapses 8 partitions to 1 and BWAG does the whole window
alone.
## Why the RANGE halo does not extend to ROWS
`ParallelWindowRule`'s halo is a value delta end to end: files route by
sketched
`[min, max]` overlap against `cuts[k] +/- halo`, and `RangeFilterExec` keeps
`cuts[k-1] - halo_lo <= v < cuts[k] + halo_hi`. A ROWS halo is a rank delta,
and
nothing on that path expresses rank. Widening a cut by any value can land in
a gap
holding no rows at all, so "100 preceding rows" is not reachable by a choice
of
value width.
## Where the rank bound comes from
The range shuffle sidecar index from #2364 carries one row per IPC message,
with
that message's first key and its `num_rows`. Merging those index rows
descending by
first key, across every candidate file, and accumulating `num_rows` until
the total
reaches `n` gives a key: reading from it up to the cut yields a superset of
the true
`n` predecessors. Comparison only, so strings, timestamps, multi-column and
DESC
keys all work. A message straddling the cut counts as zero, since
over-counting
would stop the walk short of the rows asked for.
The walk runs at two granularities:
- **coarse, on the scheduler**, at file granularity, over `key_min` /
`key_max` /
`num_rows` the runtime stats reports already carry. This fetches nothing,
and it
is what keeps the file list per consumer from growing with K.
- **fine, on the consumer**, at message granularity, over the index files
`RangeShuffleReaderExec` already downloads. This is byte reduction, not
correctness.
Cuts stay approximate KLL value cuts. Halo exactness rests on the index
walk's exact
`num_rows`, not on cut exactness.
## What does not change
`OrderedRangeRepartitionExec`, `RangeFilterExec` and
`PartitionedBoundedWindowAggExec`
learn nothing about ROWS. The filter takes a resolved half-open band per
partition,
which is what it already takes; the rank lives entirely in the
scheduler-side walk
that produces those bands. Only the rule's frame-units gate mentions ROWS at
all.
## State of this PR
Draft, and deliberately incomplete: the rewrite fires and the halo is still
zero.
`ballista/client/tests/halo_row.rs` runs the standalone cluster against
single-node
DataFusion as the oracle, at K=4 over 48 unique keys:
- `the_rewrite_actually_fires`: green. ORRE, RuntimeStatsExec,
RangeFilterExec and
PBWAG all present.
- `CURRENT ROW AND CURRENT ROW`: green. No row from outside a task's own
range is
needed, so this isolates the plan shape.
- `5 PRECEDING AND CURRENT ROW`: red on 15 rows, the first 5 after each of
the 3
cuts. That is `(K-1) * n`, and nothing else in the answer moves. The halo
is the
only defect.
Next commit is the coarse walk, which turns that test green without an index
fetch
or a reader change.
Behaviour changes only for queries in the matched shape, and only when
`ballista.planner.parallel_window.enabled` is set, which defaults to false.
The rule
shares that flag with `ParallelWindowRule` rather than adding a second knob.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
