Grouped streaming AsOf join uses a lot of memory when the right side has many group matches wrt the left side

Checks

• I have checked that this issue has not already been reported.
• I have confirmed this bug exists on the latest version of Polars.

Reproducible example

import polars as pl

n_groups = 50_000
left_rows_per_group = 4
right_rows_per_group = 2000

left = pl.DataFrame({
    "group": [g for g in range(n_groups) for _ in range(left_rows_per_group)],
    "time": [t for _ in range(n_groups) for t in range(left_rows_per_group)],
}).sort("group", "time")

right = pl.DataFrame({
    "group": [g for g in range(n_groups) for _ in range(right_rows_per_group)],
    "time": [t for _ in range(n_groups) for t in range(right_rows_per_group)],
    "val": list(range(n_groups * right_rows_per_group)),
}).sort("group", "time")

out = left.lazy().join_asof(right.lazy(), on="time", by="group").collect(engine="streaming")

Log output

Issue description

In crates/polars-stream/src/nodes/joins/asof_join.rs, distrubute_work_task processes a left morsel as an single unit:

while need_more_right_side(&left_df, right_buffer, params)? && !right_done { // pull more right morsels into right buffer … }

// … dispatch left_df + right_buffer, then prune_right_side(left_df, …)

need_more_right_side / prune_right_side key off the first/last group of the morsels needs right-side rows far down the right stream, so the loop keeps pulling right morsels until the buffer spans every group in the morsel. Nothing is pruned until dispatch, so the buffer never shrinks between groups.

Expected behavior

Dispatch one by-group at a time. Slice the left morsel on group boundaries and run the existing strategy-aware need_more_right_side / prune_right_side / check_left_continuity helpers per group. This bound the right side buffer to a single group’s live range and works for all strategies.

In a local test on proprietary data, this took a process from ~500G -> ~20G peak and ran slight faster.

Happy to create a PR with the patch or can leave to someone more familiar with the code base.

Installed versions

Replace this line with the output of pl.show_versions(). Leave the backticks in place.