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

Code before:
impl NonAppendOnlyBehavior {
    /// NOTE(kwannoel):
    ///
    /// After the optimization in <https://github.com/risingwavelabs/risingwave/pull/12250>,
    /// `DELETE`s will be sequenced before `INSERT`s in JDBC sinks and PG rust sink.
    /// There's a risk that adjacent chunks with `DELETE`s on the same PK will get
    /// merged into a single chunk, since the logstore format doesn't preserve chunk
    /// boundaries. Then we will have double `DELETE`s followed by unspecified sequence
    /// of `INSERT`s, and lead to inconsistent data downstream.
    ///
    /// We only need to do the compaction for non-append-only sinks, when the upstream and
    /// downstream PKs are matched. When the upstream and downstream PKs are not matched,
    /// we will buffer the chunks between two barriers, so the compaction is not needed,
    /// since the barriers will preserve chunk boundaries.
    ///
    /// When `force_buffering` is true, we also skip compaction here, since the buffering
    /// will also make compaction.
    ///
    /// When the sink is an append-only sink, it is either `force_append_only` or
    /// `append_only`, we should only append to downstream, so there should not be any
    /// overlapping keys.
    fn should_compact_in_log_reader(self) -> bool {
        self.pk_specified_and_matched && !self.force_buffering
    }

    /// When stream key is different from the user defined primary key columns for sinks.
    /// The operations could be out of order.
    ///
    /// For example, we have a stream with derived stream key `a, b` and user-specified sink
    /// primary key `a`. Assume that we have `(1, 1)` in the table. Then, we perform two `UPDATE`
    /// operations:
    ///
    /// ```text
    /// UPDATE SET b = 2 WHERE a = 1 ... which issues:
    ///   - (1, 1)
    ///   + (1, 2)
    ///
    /// UPDATE SET b = 3 WHERE a = 1 ... which issues:
    ///   - (1, 2)
    ///   + (1, 3)
    /// ```
    ///
    /// When these changes go into streaming pipeline, they could be shuffled to different parallelism
    /// (actor), given that they are under different stream keys.
    ///
    /// ```text
    /// Actor 1:
    /// - (1, 1)
    ///
    /// Actor 2:
    /// + (1, 2)

Review comment (suggestion):
The documentation for `should_buffer_and_reorder_records` is now inaccurate. The comment states this function is called 'When stream key is different from the user defined primary key columns', but the implementation at line 195 shows it also returns true when `force_buffering` is enabled, even if the stream key matches the downstream pk. The comment should be updated to reflect this additional condition.

Code after:
impl NonAppendOnlyBehavior {
    /// NOTE(kwannoel):
    ///
    /// After the optimization in <https://github.com/risingwavelabs/risingwave/pull/12250>,
    /// `DELETE`s will be sequenced before `INSERT`s in JDBC sinks and PG rust sink.
    /// There's a risk that adjacent chunks with `DELETE`s on the same PK will get
    /// merged into a single chunk, since the logstore format doesn't preserve chunk
    /// boundaries. Then we will have double `DELETE`s followed by unspecified sequence
    /// of `INSERT`s, and lead to inconsistent data downstream.
    ///
    /// We only need to do the compaction for non-append-only sinks, when the upstream and
    /// downstream PKs are matched. When the upstream and downstream PKs are not matched,
    /// we will buffer the chunks between two barriers, so the compaction is not needed,
    /// since the barriers will preserve chunk boundaries.
    ///
    /// When `force_compaction` is true, we also skip compaction here, since the buffering
    /// will also make compaction.
    ///
    /// When the sink is an append-only sink, it is either `force_append_only` or
    /// `append_only`, we should only append to downstream, so there should not be any
    /// overlapping keys.
    fn should_compact_in_log_reader(self) -> bool {
        self.pk_specified_and_matched && !self.force_compaction
    }

    /// When stream key is different from the user defined primary key columns for sinks.
    /// The operations could be out of order.
    ///
    /// For example, we have a stream with derived stream key `a, b` and user-specified sink
    /// primary key `a`. Assume that we have `(1, 1)` in the table. Then, we perform two `UPDATE`
    /// operations:
    ///
    /// ```text
    /// UPDATE SET b = 2 WHERE a = 1 ... which issues:
    ///   - (1, 1)
    ///   + (1, 2)
    ///
    /// UPDATE SET b = 3 WHERE a = 1 ... which issues:
    ///   - (1, 2)
    ///   + (1, 3)
    /// ```
    ///
    /// When these changes go into streaming pipeline, they could be shuffled to different parallelism
    /// (actor), given that they are under different stream keys.
    ///
    /// ```text
    /// Actor 1:
    /// - (1, 1)
    ///
    /// Actor 2:
    /// + (1, 2)