feat: compact stream chunk with sink key when sink key mismatch

e2e_test/batch/aggregate/two_phase_agg.slt.part

@@ -3,6 +3,7 @@ SET RW_IMPLICIT_FLUSH TO true;
 
 statement ok
 SET RW_ENABLE_TWO_PHASE_AGG=false;
+
 # This should override `RW_ENABLE_TWO_PHASE_AGG`, enabling it.
 statement ok
 SET RW_FORCE_TWO_PHASE_AGG=true;

src/common/src/array/compact_chunk.rs

@@ -114,8 +114,13 @@ impl<'a, 'b> RowOpMap<'a, 'b> {
                 RowOp::Delete(ref old_v) => {
                     e.insert(RowOp::Update((*old_v, v)));
                 }
-                RowOp::Insert(_) | RowOp::Update(_) => {
-                    tracing::warn!("double insert");
+                RowOp::Insert(_) => {
+                    tracing::warn!("double insert for the same pk");
+                    e.insert(RowOp::Insert(v));
+                }
+                RowOp::Update((ref old_v, _)) => {
+                    tracing::warn!("double insert for the same pk");
+                    e.insert(RowOp::Update((*old_v, v)));
                 }
             },
         }
@@ -135,7 +140,8 @@ impl<'a, 'b> RowOpMap<'a, 'b> {
                     e.insert(RowOp::Delete(*prev));
                 }
                 RowOp::Delete(_) => {
-                    tracing::warn!("double delete");
+                    tracing::warn!("double delete for the same pk");
+                    e.insert(RowOp::Delete(v));
                 }
             },
         }

src/stream/src/executor/sink.rs

@@ -187,16 +187,19 @@ impl<F: LogStoreFactory> SinkExecutor<F> {
         // So the delete event is not to delete the inserted record in our internal streaming SQL semantic.
         let need_advance_delete =
             stream_key_sink_pk_mismatch && self.sink_param.sink_type != SinkType::AppendOnly;
-
+        // NOTE(st1page): reconstruct with sink pk need extra cost to buffer a barrier's data, so currently we bind it with mismatch case.
+        let re_construct_with_sink_pk = need_advance_delete
+            && self.sink_param.sink_type == SinkType::Upsert
+            && !self.sink_param.downstream_pk.is_empty();
         let processed_input = Self::process_msg(
             input,
             self.sink_param.sink_type,
             stream_key,
             need_advance_delete,
-            // NOTE(st1page): reconstruct with sink pk need extra cost to buffer a barrier's data, so currently we bind it with mismatch case.
-            need_advance_delete,
+            re_construct_with_sink_pk,
             self.chunk_size,
             self.input_data_types,
+            self.sink_param.downstream_pk.clone(),
         );
 
         if self.sink.is_sink_into_table() {
@@ -287,6 +290,7 @@ impl<F: LogStoreFactory> SinkExecutor<F> {
         }
     }
 
+    #[allow(clippy::too_many_arguments)]
     #[try_stream(ok = Message, error = StreamExecutorError)]
     async fn process_msg(
         input: impl MessageStream,
@@ -296,6 +300,7 @@ impl<F: LogStoreFactory> SinkExecutor<F> {
         re_construct_with_sink_pk: bool,
         chunk_size: usize,
         input_data_types: Vec<DataType>,
+        down_stream_pk: Vec<usize>,
     ) {
         // need to buffer chunks during one barrier
         if need_advance_delete || re_construct_with_sink_pk {
@@ -333,7 +338,7 @@ impl<F: LogStoreFactory> SinkExecutor<F> {
                             chunks
                         };
                         let chunks = if re_construct_with_sink_pk {
-                            StreamChunkCompactor::new(stream_key.clone(), chunks)
+                            StreamChunkCompactor::new(down_stream_pk.clone(), chunks)
                                 .reconstructed_compacted_chunks(
                                     chunk_size,
                                     input_data_types.clone(),
@@ -712,8 +717,7 @@ mod test {
             StreamChunk::from_pretty(
                 " I I I
                 U- 1 1 10
-                U+ 1 1 40
-                +  1 3 30",
+                U+ 1 1 40",
             )
         );