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

Code before:
    /// Whether the user specifies a primary key for the sink, and it matches the derived stream
    /// key of the stream.
    ///
    /// By matching, we mean that stream key is a subset of the downstream pk.
    pk_specified_and_matched: bool,
    /// Whether the user forces buffering all chunks between two barriers.
    force_buffering: bool,
}

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)
    /// ```
    ///

Review comment (suggestion):
The phrase 'the buffering will also make compaction' is unclear. It should be rephrased to 'the buffering process will perform compaction' or 'compaction will be done during the buffering step' for better clarity.
```suggestion
    /// When `force_buffering` is true, we also skip compaction here, since compaction
    /// will be performed during the buffering step.
```

Code after:
    /// Whether the user specifies a primary key for the sink, and it matches the derived stream
    /// key of the stream.
    ///
    /// By matching, we mean that stream key is a subset of the downstream pk.
    pk_specified_and_matched: bool,
    /// Whether the user forces buffering all chunks between two barriers.
    force_compaction: bool,
}

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)
    /// ```
    ///