DataChunk::from_rows

Author: Liang Zhao

src/stream/src/executor/now.rs

@@ -16,11 +16,10 @@ use async_stack_trace::StackTrace;
 use chrono::NaiveDateTime;
 use futures::{pin_mut, StreamExt};
 use futures_async_stream::try_stream;
-use risingwave_common::array::{Op, StreamChunk};
+use risingwave_common::array::{DataChunk, Op, StreamChunk};
 use risingwave_common::catalog::{Field, Schema};
 use risingwave_common::row::Row;
 use risingwave_common::types::{DataType, NaiveDateTimeWrapper, ScalarImpl};
-use risingwave_common::util::chunk_coalesce::DataChunkBuilder;
 use risingwave_common::util::epoch::Epoch;
 use risingwave_storage::StateStore;
 use tokio::sync::mpsc::UnboundedReceiver;
@@ -106,18 +105,14 @@ impl<S: StateStore> NowExecutor<S> {
                     ),
                 )));
 
-                let mut data_chunk_builder = DataChunkBuilder::new(
-                    schema.data_types(),
-                    if last_timestamp.is_some() { 2 } else { 1 },
+                let data_chunk = DataChunk::from_rows(
+                    &[Row::new(if last_timestamp.is_some() {
+                        vec![last_timestamp.clone(), timestamp.clone()]
+                    } else {
+                        vec![timestamp.clone()]
+                    })],
+                    &schema.data_types(),
                 );
-                if last_timestamp.is_some() {
-                    let chunk_popped = data_chunk_builder
-                        .append_one_row_from_datums([&last_timestamp].into_iter());
-                    debug_assert!(chunk_popped.is_none());
-                }
-                let data_chunk = data_chunk_builder
-                    .append_one_row_from_datums([&timestamp].into_iter())
-                    .unwrap();
                 let mut ops = if last_timestamp.is_some() {
                     vec![Op::Delete]
                 } else {