debug inserted rows

src/stream/src/executor/backfill.rs

@@ -152,10 +152,6 @@ where
         let key: &[Datum] = &[Some((arbitrary_vnode as i16).into())];
         let row = self.state_table.get_row(key).await?;
         println!("row: {:?}", row);
-        #[for_await]
-        for row in self.state_table.iter(Default::default()).await? {
-            println!("row {:?}", row);
-        }
         let is_finished = if let Some(row) = row
             && let Some(is_finished) = row.datum_at(state_len + 1)
         {
@@ -573,6 +569,8 @@ where
                     // fill the state
                     current_partial_state[0] = datum.clone();
                     old_state[0] = datum;
+                    println!("Updating old_row: {:?}", old_state);
+                    println!("Updating new_row: {:?}", current_partial_state);
                     table.write_record(Record::Update {
                         old_row: &old_state[..],
                         new_row: &(*current_partial_state),
@@ -589,6 +587,7 @@ where
                 let datum = Some((vnode as i16).into());
                 // fill the state
                 current_partial_state[0] = datum;
+                println!("Inserting new_row: {:?}", current_partial_state);
                 table.write_record(Record::Insert {
                     new_row: &(*current_partial_state),
                 })