Repository: risingwavelabs/risingwave
File: src/stream/src/executor/dispatch.rs (Rust)

Code before:
                }
            } else {
                new_ops.push(op);
            }
        }
        assert!(last_update_delete_row_idx.is_none(), "missing U+ after U-");

        let ops = new_ops;
        // Apply output mapping after calculating the vnode and new visibility maps.
        // The output mapping may project columns and eliminate noop updates.
        let chunk = self.output_mapping.apply(chunk);
        // Get the visibility after noop update elimination to incorporate into the final visibility.
        let chunk_vis = chunk.visibility();

        // individually output StreamChunk integrated with vis_map
        futures::future::try_join_all(
            vis_maps
                .into_iter()
                .zip_eq_fast(self.outputs.iter_mut())
                .map(|(vis_map, output)| async {
                    let vis_map = vis_map.finish();
                    // Combine hash routing visibility with noop update elimination visibility.
                    // A row is visible only if it passes BOTH:
                    // 1. Hash routing (the row should go to this output)
                    // 2. Noop update elimination (the row was not eliminated as a noop)
                    let combined_vis = &vis_map & chunk_vis;
                    let new_stream_chunk =
                        StreamChunk::with_visibility(ops.clone(), chunk.columns().into(), combined_vis);
                    if new_stream_chunk.cardinality() > 0 {
                        event!(
                            tracing::Level::TRACE,
                            msg = "chunk",
                            downstream = %output.actor_id(),
                            "send = \n{:#?}",
                            new_stream_chunk
                        );
                        output
                            .send(DispatcherMessageBatch::Chunk(new_stream_chunk))
                            .await?;
                    }
                    StreamResult::Ok(())
                }),
        )
        .await?;

        Ok(())
    }

    fn remove_outputs(&mut self, actor_ids: &HashSet<ActorId>) {
        self.outputs
            .extract_if(.., |output| actor_ids.contains(&output.actor_id()))

Review comment (bug):
`combined_vis` is built via `&vis_map & chunk_vis`, which allocates/clones a new `Bitmap` per output even when `chunk_vis` is fully visible (the common case when no noop updates are eliminated). Since this dispatcher is on a hot path, consider short-circuiting: if `chunk_vis.all()` then reuse `vis_map` directly; otherwise compute the bitand. This keeps the bug fix while avoiding extra allocations when noop elimination doesn’t actually hide rows.
```suggestion
                    // Fast path: when noop elimination keeps all rows visible, reuse `vis_map`
                    // directly to avoid allocating a new bitmap on this hot path.
                    let combined_vis = if chunk_vis.all() {
                        vis_map
                    } else {
                        &vis_map & chunk_vis
                    };
```

Code after:
                }
            } else {
                new_ops.push(op);
            }
        }
        assert!(last_update_delete_row_idx.is_none(), "missing U+ after U-");

        let ops = new_ops;
        // Apply output mapping after calculating the vnode and new visibility maps.
        // The output mapping may project columns and eliminate noop updates.
        let chunk = self.output_mapping.apply(chunk);
        // Get the visibility after noop update elimination to incorporate into the final visibility.
        let chunk_vis = chunk.visibility();

        // individually output StreamChunk integrated with vis_map
        futures::future::try_join_all(
            vis_maps
                .into_iter()
                .zip_eq_fast(self.outputs.iter_mut())
                .map(|(vis_map, output)| async {
                    let vis_map = vis_map.finish();
                    // Combine hash routing visibility with noop update elimination visibility.
                    // A row is visible only if it passes BOTH:
                    // 1. Hash routing (the row should go to this output)
                    // 2. Noop update elimination (the row was not eliminated as a noop)
                    let combined_vis = &vis_map & chunk_vis;
                    let new_stream_chunk = StreamChunk::with_visibility(
                        ops.clone(),
                        chunk.columns().into(),
                        combined_vis,
                    );
                    if new_stream_chunk.cardinality() > 0 {
                        event!(
                            tracing::Level::TRACE,
                            msg = "chunk",
                            downstream = %output.actor_id(),
                            "send = \n{:#?}",
                            new_stream_chunk
                        );
                        output
                            .send(DispatcherMessageBatch::Chunk(new_stream_chunk))
                            .await?;
                    }
                    StreamResult::Ok(())
                }),
        )
        .await?;

        Ok(())
    }