fix(stream): sync ops with visibility after noop elimination in HashDataDispatcher

[jw-itq]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Fix a bug where HashDataDispatcher uses pre-elimination ops with post-elimination visibility, causing downstream panic: expect a U- before U+.

Root Cause:

In HashDataDispatcher::dispatch_data, new_ops is computed before output_mapping.apply(), which internally calls eliminate_adjacent_noop_update(). This function may:

1. Hide rows by setting visibility to false (noop pairs)
2. Normalize partially-visible U-/U+ pairs (U- → Delete, U+ → Insert) via fix(stream): fix eliminate no op #24659

However, combined_vis is computed after the elimination, incorporating the new visibility. The resulting chunk uses stale new_ops (without normalize) but updated combined_vis (with normalize), creating a mismatch — orphan UpdateInsert ops become visible without a preceding UpdateDelete, causing downstream iterators to panic.

Fix:

After output_mapping.apply(), merge new_ops (which contains dist-key-changed rewrites) with chunk.ops() (which contains post-elimination normalized ops). This ensures ops and visibility are always consistent.

How to Reproduce:

When the same stream key has multiple consecutive updates within a single chunk and the dispatcher's output columns are fewer than input columns (triggering noop elimination), the middle U-/U+ pair gets eliminated, leaving the outer pair's ops un-normalized.

Example:
U- row_A (visible) U+ row_B (eliminated by noop) U- row_B (eliminated by noop) U+ row_C (visible, but ops still says UpdateInsert → panic)

Related:

• fix: correctly handle vis map in hash dispatcher after noop elimination #24580 — introduced combined_vis but did not sync ops
• fix: remove adjacent update check #24645 — removed adjacent U-/U+ assertion
• fix(stream): fix eliminate no op #24659 — added normalize in eliminate_adjacent_noop_update, but HashDataDispatcher ignores it

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• I have checked the Release Timeline and Currently Supported Versions to determine which release branches I need to cherry-pick this PR into.

Documentation

• My PR needs documentation updates.

Release note

Fix a critical bug introduced in v2.8.0 where compute nodes crash with expect a U- before U+ panic when processing streams with consecutive updates on the same key through HashDataDispatcher with column pruning. This is caused by ops/visibility mismatch after noop update elimination in the dispatcher.