chore: per review

src/flow/src/adapter.rs

@@ -228,7 +228,7 @@ pub fn diff_row_to_request(rows: Vec<DiffRow>) -> Vec<DiffRequest> {
 pub fn batches_to_rows_req(batches: Vec<Batch>) -> Result<Vec<DiffRequest>, Error> {
     let mut reqs = Vec::new();
     for batch in batches {
-        let mut rows = Vec::new();
+        let mut rows = Vec::with_capacity(batch.row_count());
         for i in 0..batch.row_count() {
             let row = batch.get_row(i).context(EvalSnafu)?;
             rows.push((Row::new(row), 0));

src/flow/src/repr.rs

@@ -55,10 +55,9 @@ pub const BROADCAST_CAP: usize = 1024;
 /// The maximum capacity of the send buffer, to prevent the buffer from growing too large
 pub const SEND_BUF_CAP: usize = BROADCAST_CAP * 2;
 
+/// Flow worker will try to at least accumulate this many rows before processing them(if one second havn't passed)
 pub const BATCH_SIZE: usize = 32 * 16384;
 
-pub const SLEEP_DURATION: std::time::Duration = std::time::Duration::from_millis(100);
-
 /// Convert a value that is or can be converted to Datetime to internal timestamp
 ///
 /// support types are: `Date`, `DateTime`, `TimeStamp`, `i64`