Struct UpsertBehavior 

struct UpsertBehavior {
    pk_specified_and_matched: bool,
}

When the sink is an upsert sink, we need to do some extra work for correctness and performance.

Fields

pk_specified_and_matched: bool

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.

Implementations

impl UpsertBehavior

fn should_compact_in_log_reader(self) -> bool

NOTE(kwannoel):

After the optimization in https://github.com/risingwavelabs/risingwave/pull/12250, DELETEs will be sequenced before INSERTs in JDBC sinks and PG rust sink. There’s a risk that adjacent chunks with DELETEs 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 DELETEs followed by unspecified sequence of INSERTs, and lead to inconsistent data downstream.

We only need to do the compaction for upsert 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 the sink is not an upsert 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_reorder_records(self) -> bool

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:

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.

Actor 1:
- (1, 1)

Actor 2:
+ (1, 2)
- (1, 2)

Actor 3:
+ (1, 3)

When these records are merged back into sink actor, we may get the records from different parallelism in arbitrary order, like:

+ (1, 2) -- Actor 2, first row
+ (1, 3) -- Actor 3
- (1, 1) -- Actor 1
- (1, 2) -- Actor 2, second row

Note that in terms of stream key (a, b), the operations in the order above are completely correct, because we are operating on 3 different rows. However, in terms of user defined sink primary key a, we’re violating the unique constraint all the time.

Therefore, in this case, we have to do additional reordering in the sink executor per barrier. Specifically, we need to:

First, compact all the changes with the stream key, so we have:

+ (1, 3)
- (1, 1)

Then, sink all the delete events before sinking all insert events, so we have:

- (1, 1)
+ (1, 3)

Since we’ve compacted the chunk with the stream key, the DELETE records survived must be to delete an existing row, so we can safely move them to the front. After the deletion is done, we can then safely sink the insert events with uniqueness guarantee.

Trait Implementations

impl Clone for UpsertBehavior

fn clone(&self) -> UpsertBehavior

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for UpsertBehavior

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Copy for UpsertBehavior