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

Code before:
            row_id_generator: Self::new_generator(&vnodes),
        }
    }

    /// Create a new row id generator based on the assigned vnodes.
    fn new_generator(vnodes: &Bitmap) -> RowIdGenerator {
        RowIdGenerator::new(vnodes.iter_vnodes(), vnodes.len())
    }

    /// Generate a row ID column according to ops.
    fn gen_row_id_column_by_op(
        &mut self,
        column: &ArrayRef,
        ops: &'_ [Op],
        vis: &Bitmap,
    ) -> ArrayRef {
        let len = column.len();
        let mut builder = SerialArrayBuilder::new(len);

        for ((datum, op), vis) in column.iter().zip_eq_fast(ops).zip_eq_fast(vis.iter()) {
            macro_rules! warn_absent_row_id {
                ($op:expr) => {{
                    static LOG_SUPPRESSOR: LazyLock<LogSuppressor> =
                        LazyLock::new(LogSuppressor::default);
                    if let Ok(suppressed_count) = LOG_SUPPRESSOR.check() {
                        tracing::warn!(
                            suppressed_count,
                            ?op,
                            "absent row id for non-insert operation"
                        );
                    }
                    None
                }};
            }

            let row_id = if vis {
                match datum {
                    // If the row id is already present, e.g., from `UPDATE` or `DELETE`, keep it.
                    Some(scalar) => Some(scalar.into_serial()),
                    // Otherwise, generate a new row id only if it's an `Insert` operation.
                    None => match op {
                        Op::Insert => Some(self.row_id_generator.next().into()),
                        Op::Delete => warn_absent_row_id!(op),
                        Op::UpdateDelete => warn_absent_row_id!(op),
                        Op::UpdateInsert => warn_absent_row_id!(op),
                    },
                }
            } else {
                None
            };


Review comment (suggestion):
We only want to generate row id for `INSERT`s instead of `UPDATE`s and `DELETE`s. To achieve this, we determine it by `Op`. However, with upsert DML `UPDATE`s will also yield `Op::Insert`.

Thus, this PR changed the logic to skipping generating once there's a non-NULL value for row id column, which also seems more reasonable.

Code after:
            ctx,
            upstream: Some(upstream),
            row_id_index,
            row_id_generator: Self::new_generator(&vnodes),
            vnodes: Arc::new(vnodes),
        }
    }

    /// Create a new row id generator based on the assigned vnodes.
    fn new_generator(vnodes: &Bitmap) -> RowIdGenerator {
        RowIdGenerator::new(vnodes.iter_vnodes(), vnodes.len())
    }

    /// Generate a row ID column according to ops.
    fn gen_row_id_column_by_op(
        &mut self,
        column: &ArrayRef,
        ops: &'_ [Op],
        vis: &Bitmap,
    ) -> ArrayRef {
        let column = column.as_serial();
        let len = column.len();
        let mut builder = SerialArrayBuilder::new(len);

        for ((serial, op), vis) in column.iter().zip_eq_fast(ops).zip_eq_fast(vis.iter()) {
            macro_rules! warn_absent_row_id {
                ($op:expr) => {{
                    static LOG_SUPPRESSOR: LazyLock<LogSuppressor> =
                        LazyLock::new(LogSuppressor::default);
                    if let Ok(suppressed_count) = LOG_SUPPRESSOR.check() {
                        tracing::warn!(
                            suppressed_count,
                            ?op,
                            "absent row id for non-insert operation"
                        );
                    }
                    None
                }};
            }

            let row_id = if vis {
                match serial {
                    // If the row id is already present, e.g., from `UPDATE` or `DELETE`, keep it.
                    Some(serial) => {
                        // Check if the data is shuffled correctly.
                        if cfg!(debug_assertions) {
                            check_vnode_is_set(
                                compute_vnode_from_row_id(serial.as_row_id(), self.vnodes.len()),
                                &self.vnodes,
                            );
                        }