fix(stream): materialize should not compact input when handling conflict

src/stream/src/executor/mview/materialize.rs

 pub async fn check_with_script<F, Fut>( 
 /// Similar to [`check_until_pending`], but use a DSL test script as input. 
 /// 
 /// For each input event, it drives the executor until it is pending. 
 pub async fn check_with_script<F, Fut>( 
@@ -720,6 +720,94 @@ mod tests {
         }
     }
 
+    // https://github.com/risingwavelabs/risingwave/issues/13346
+    #[tokio::test]
+    async fn test_upsert_stream() {
+        // Prepare storage and memtable.
+        let memory_state_store = MemoryStateStore::new();
+        let table_id = TableId::new(1);
+        // Two columns of int32 type, the first column is PK.
+        let schema = Schema::new(vec![
+            Field::unnamed(DataType::Int32),
+            Field::unnamed(DataType::Int32),
+        ]);
+        let column_ids = vec![0.into(), 1.into()];
+
+        // test double insert one pk, the latter needs to override the former.
+        let chunk1 = StreamChunk::from_pretty(
+            " i i
+            + 1 1",
+        );
+
+        let chunk2 = StreamChunk::from_pretty(
+            " i i
+            + 1 2
+            - 1 2",
+        );
+
+        // Prepare stream executors.
+        let source = MockSource::with_messages(
+            schema.clone(),
+            PkIndices::new(),
+            vec![
+                Message::Barrier(Barrier::new_test_barrier(1)),
+                Message::Chunk(chunk1),
+                Message::Barrier(Barrier::new_test_barrier(2)),
+                Message::Chunk(chunk2),
+                Message::Barrier(Barrier::new_test_barrier(3)),
+            ],
+        );
+
+        let order_types = vec![OrderType::ascending()];
+        let column_descs = vec![
+            ColumnDesc::unnamed(column_ids[0], DataType::Int32),
+            ColumnDesc::unnamed(column_ids[1], DataType::Int32),
+        ];
+
+        let table = StorageTable::for_test(
+            memory_state_store.clone(),
+            table_id,
+            column_descs,
+            order_types,
+            vec![0],
+            vec![0, 1],
+        );
+
+        let mut materialize_executor = Box::new(
+            MaterializeExecutor::for_test(
+                Box::new(source),
+                memory_state_store,
+                table_id,
+                vec![ColumnOrder::new(0, OrderType::ascending())],
+                column_ids,
+                1,
+                Arc::new(AtomicU64::new(0)),
+                ConflictBehavior::Overwrite,
+            )
+            .await,
+        )
+        .execute();
+        materialize_executor.next().await.transpose().unwrap();
+
+        materialize_executor.next().await.transpose().unwrap();
+        materialize_executor.next().await.transpose().unwrap();
+        materialize_executor.next().await.transpose().unwrap();
+
+        match materialize_executor.next().await.transpose().unwrap() {
+            Some(Message::Barrier(_)) => {
+                let row = table
+                    .get_row(
+                        &OwnedRow::new(vec![Some(1_i32.into())]),
+                        HummockReadEpoch::NoWait(u64::MAX),
+                    )
+                    .await
+                    .unwrap();
+                assert!(row.is_none());
+            }
+            _ => unreachable!(),
+        }
+    }
+
     #[tokio::test]
     async fn test_check_insert_conflict() {
         // Prepare storage and memtable.