feat(sink): produce Upsert chunk in compaction for SinkType::Upsert

[BugenZhao]

I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.

What's changed and what's your intention?

Extend StreamChunkCompactor to allow specifying OutputKind, which decides whether to produce chunks in

• Retract format, which convert Update record to a pair of U- and U+ rows
• or Upsert format, which only keep the new row of Update record and convert it to an Insert row

Dispatch different OutputKind based on whether the SinkType is Upsert or Retract (introduced in last PR: #23593).

---

Previously...

• For "inline" compaction, the actual effective Update rows might result in inconsecutive, forcing us to rewrite it into 2 operations (Delete and Insert) (Unexpected Op in output of StreamChunkCompactor into_compacted_chunks #23540)
• Also, some sink implementation may not handle Update operation in the optimal way

This led to unnecessary Delete operation to external system, along with I/O overhead and temporary inconsistent state.

Directly produce chunk in upsert format in compaction minimize Delete operation as much as possible. This approach also better aligns with the semantics of SinkType::Upsert. If the sink explicitly requires old value for updates, typically DEBEZIUM, it should be marked SinkType::Retract instead.

By using upsert format in sink-into-table, we also fixes #22579, since the update won't be considered as a pair of Delete and Insert when handling table conflict.

Checklist

• I have written necessary rustdoc comments.
• I have added necessary unit tests and integration tests.
• I have added test labels as necessary.
• I have added fuzzing tests or opened an issue to track them.
• My PR contains breaking changes.
• My PR changes performance-critical code, so I will run (micro) benchmarks and present the results.
• I have checked the Release Timeline and Currently Supported Versions to determine which release branches I need to cherry-pick this PR into.

Documentation

• My PR needs documentation updates.

Release note

[BugenZhao]

• feat(sink): produce Upsert chunk in compaction for SinkType::Upsert #23581 👈 (View in Graphite)
• feat(sink): introduce Retract sink type #23593
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

[Copilot]

Pull Request Overview

This PR renames StreamChunkCompactor to StreamChunkUpsertCompactor and modifies the compaction logic to produce chunks in upsert format by removing UpdateDelete operations and rewriting UpdateInsert operations as Insert operations. This change minimizes unnecessary Delete operations to external systems and aligns better with SinkType::Upsert semantics.

Key Changes:

• Renamed StreamChunkCompactor to StreamChunkUpsertCompactor and compact_chunk_inline to into_upsert_compacted_chunk
• Modified compaction methods to produce upsert format chunks using new Record::into_upsert() method
• Added Kind type parameter to ChangeBuffer::into_chunk and into_chunks methods to support both upsert and retract formats
• Updated StreamSink to correctly derive and set stream kind based on sink type

Reviewed Changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
src/stream/src/executor/sink.rs	Updates sink executor to use renamed StreamChunkUpsertCompactor and into_upsert_compacted_chunk
src/stream/src/executor/mview/materialize.rs	Updates materialize executor to explicitly use RETRACT kind for change buffer
src/stream/src/common/upsert_compact.rs	Renames compactor class, updates methods to produce upsert format chunks, and improves documentation
src/stream/src/common/mod.rs	Renames module from compact_chunk to upsert_compact
src/stream/src/common/change_buffer.rs	Adds Kind type parameter to support both upsert and retract formats in chunk conversion
src/frontend/src/optimizer/plan_node/stream_sink.rs	Updates sink type derivation logic and explicitly sets stream kind based on sink type
src/common/src/array/stream_record.rs	Adds into_upsert() method to convert Update records to Insert format
src/common/src/array/stream_chunk_builder.rs	Refactors append_record to use helper method for cleaner code

[BugenZhao]

I realized that there's an exception: FORMAT DEBEZIUM actually requires retractable output.

[BugenZhao]

I realized that there's an exception: FORMAT DEBEZIUM actually requires retractable output.

Addressed by introducing SinkType::Retract first in #23593, and dispatch different behavior based on it's retractable or upsert.