Repository: risingwavelabs/risingwave
File: src/common/src/array/stream_chunk.rs (Rust)

Code before:
                    // 2. Insert then Delete
                    //
                    // Note that after eliminating `U+` and `U-` here, we will get a new
                    // pair of `U-` and `U+` consisting of `prev.prev` and the `next` row.
                    // `prev.prev` and `prev`, `curr` and `next` share the same stream key
                    // because they are `U-` and `U+` pairs, while `prev` and `curr` share
                    // the same stream key because they are equal, so `prev-prev` and `next`
                    // must also share the same stream key. Therefore, it's okay to leave
                    // their `Update` ops unchanged.
                    //
                    // Also, they can't be the same, otherwise these 4 rows are the same,
                    // which should already be eliminated by the first branch.
                    (matches!(c.op(prev), Op::UpdateInsert | Op::Insert)
                    && matches!(c.op(curr), Op::UpdateDelete | Op::Delete))
                )
                && c.row_ref(prev) == c.row_ref(curr)
            {
                c.set_vis(prev, false);
                c.set_vis(curr, false);
                prev_r = None;
            } else {
                prev_r = Some(curr);
            }
        }
        c.into()
    }

    /// Reorder columns and set visibility.
    pub fn project_with_vis(&self, indices: &[usize], vis: Bitmap) -> Self {
        Self {
            ops: self.ops.clone(),
            data: self.data.project_with_vis(indices, vis),
        }
    }

    /// Clone the `StreamChunk` with a new visibility.
    pub fn clone_with_vis(&self, vis: Bitmap) -> Self {
        Self {
            ops: self.ops.clone(),
            data: self.data.with_visibility(vis),
        }
    }
}

impl Deref for StreamChunk {
    type Target = DataChunk;

    fn deref(&self) -> &Self::Target {
        &self.data
    }
}

Review comment (suggestion):
Seems that ideally some cases do not need such normalization, i.e., when `U- U+ U- U+` becomes `U- . . U+`. 

Code after:
                    // 2. Insert then Delete
                    //
                    // Note that after eliminating `U+` and `U-` here, we will get a new
                    // pair of `U-` and `U+` consisting of `prev.prev` and the `next` row.
                    // `prev.prev` and `prev`, `curr` and `next` share the same stream key
                    // because they are `U-` and `U+` pairs, while `prev` and `curr` share
                    // the same stream key because they are equal, so `prev-prev` and `next`
                    // must also share the same stream key. Therefore, it's okay to leave
                    // their `Update` ops unchanged.
                    //
                    // Also, they can't be the same, otherwise these 4 rows are the same,
                    // which should already be eliminated by the first branch.
                    (matches!(c.op(prev), Op::UpdateInsert | Op::Insert)
                    && matches!(c.op(curr), Op::UpdateDelete | Op::Delete))
                )
                && c.row_ref(prev) == c.row_ref(curr)
            {
                c.set_vis(prev, false);
                c.set_vis(curr, false);
                prev_r = None;
            } else {
                prev_r = Some(curr);
            }
        }

        // Normalize update pairs that became partially invisible.
        // If only U- is visible, turn it into Delete; if only U+ is visible, turn it into Insert.
        for idx in 0..len.saturating_sub(1) {
            if c.op(idx) == Op::UpdateDelete && c.op(idx + 1) == Op::UpdateInsert {
                let delete_vis = c.vis(idx);
                let insert_vis = c.vis(idx + 1);
                if delete_vis && !insert_vis {
                    c.set_op(idx, Op::Delete);
                } else if !delete_vis && insert_vis {
                    c.set_op(idx + 1, Op::Insert);
                }
            }
        }
        c.into()
    }

    /// Reorder columns and set visibility.
    pub fn project_with_vis(&self, indices: &[usize], vis: Bitmap) -> Self {
        Self {
            ops: self.ops.clone(),
            data: self.data.project_with_vis(indices, vis),
        }
    }

    /// Clone the `StreamChunk` with a new visibility.
    pub fn clone_with_vis(&self, vis: Bitmap) -> Self {