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

Code before:
            }

            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
            };

            builder.append(row_id);
        }

        builder.finish().into_ref()
    }

    #[try_stream(ok = Message, error = StreamExecutorError)]
    async fn execute_inner(mut self) {
        let mut upstream = self.upstream.take().unwrap().execute();

        // The first barrier mush propagated.
        let barrier = expect_first_barrier(&mut upstream).await?;
        yield Message::Barrier(barrier);

        #[for_await]
        for msg in upstream {
            let msg = msg?;

            match msg {
                Message::Chunk(chunk) => {
                    // For chunk message, we fill the row id column and then yield it.
                    let (ops, mut columns, bitmap) = chunk.into_inner();
                    columns[self.row_id_index] =
                        self.gen_row_id_column_by_op(&columns[self.row_id_index], &ops, &bitmap);
                    yield Message::Chunk(StreamChunk::with_visibility(ops, columns, bitmap));
                }
                Message::Barrier(barrier) => {
                    // Update row id generator if vnode mapping changes.
                    // Note that: since `Update` barrier only exists between `Pause` and `Resume`
                    // barrier, duplicated row id won't be generated.
                    if let Some(vnodes) = barrier.as_update_vnode_bitmap(self.ctx.id) {
                        self.row_id_generator = Self::new_generator(&vnodes);
                    }

Review comment (suggestion):
Make them different arms to use different `LogSuppressor` instances.

Code after:
                            "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,
                            );
                        }
                        Some(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
            };

            builder.append(row_id);
        }

        builder.finish().into_ref()
    }

    #[try_stream(ok = Message, error = StreamExecutorError)]
    async fn execute_inner(mut self) {
        let mut upstream = self.upstream.take().unwrap().execute();

        // The first barrier mush propagated.
        let barrier = expect_first_barrier(&mut upstream).await?;
        yield Message::Barrier(barrier);

        #[for_await]
        for msg in upstream {
            let msg = msg?;

            match msg {