[PR] feat(scheduler): parallel bounded ROWS-frame windows via a rank-derived halo [datafusion-ballista]

["avantgardnerio (via GitHub)" <gi...@apache.org>]

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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org

Re: [PR] feat(scheduler): parallel bounded ROWS-frame windows via a rank-derived halo [datafusion-ballista]

["github-actions[bot] (via GitHub)" <gi...@apache.org>]

github-actions[bot] commented on PR #2392:
URL: https://github.com/apache/datafusion-ballista/pull/2392#issuecomment-6431810198

   Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.


-- 
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: github-unsubscribe@datafusion.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org