feat: wait for longer anyway&wait metrics

config/config.md

@@ -232,7 +232,7 @@
 | `grpc.bind_addr` | String | `127.0.0.1:4001` | The address to bind the gRPC server. |
 | `grpc.server_addr` | String | `127.0.0.1:4001` | The address advertised to the metasrv, and used for connections from outside the host.<br/>If left empty or unset, the server will automatically use the IP address of the first network interface<br/>on the host, with the same port number as the one specified in `grpc.bind_addr`. |
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
-| `grpc.flight_compression` | String | `arrow_ipc` | Compression mode for frontend side Arrow IPC service. Available options:<br/>- `none`: disable all compression<br/>- `transport`: only enable gRPC transport compression (zstd)<br/>- `arrow_ipc`: only enable Arrow IPC compression (lz4)<br/>- `all`: enable all compression.<br/>Default to `none` |
+| `grpc.flight_compression` | String | `arrow_ipc` | Compression mode for frontend side Arrow IPC service. Available options:<br/>- `none`: disable all compression<br/>- `transport`: only enable gRPC transport compression (zstd)<br/>- `arrow_ipc`: only enable Arrow IPC compression (lz4)<br/>- `all`: enable all compression. |
 | `grpc.tls` | -- | -- | gRPC server TLS options, see `mysql.tls` section. |
 | `grpc.tls.mode` | String | `disable` | TLS mode. |
 | `grpc.tls.cert_path` | String | Unset | Certificate file path. |
@@ -405,7 +405,7 @@
 | `grpc.runtime_size` | Integer | `8` | The number of server worker threads. |
 | `grpc.max_recv_message_size` | String | `512MB` | The maximum receive message size for gRPC server. |
 | `grpc.max_send_message_size` | String | `512MB` | The maximum send message size for gRPC server. |
-| `grpc.flight_compression` | String | `arrow_ipc` | Compression mode for datanode side Arrow IPC service. Available options:<br/>- `none`: disable all compression<br/>- `transport`: only enable gRPC transport compression (zstd)<br/>- `arrow_ipc`: only enable Arrow IPC compression (lz4)<br/>- `all`: enable all compression.<br/>Default to `none` |
+| `grpc.flight_compression` | String | `arrow_ipc` | Compression mode for datanode side Arrow IPC service. Available options:<br/>- `none`: disable all compression<br/>- `transport`: only enable gRPC transport compression (zstd)<br/>- `arrow_ipc`: only enable Arrow IPC compression (lz4)<br/>- `all`: enable all compression. |
 | `grpc.tls` | -- | -- | gRPC server TLS options, see `mysql.tls` section. |
 | `grpc.tls.mode` | String | `disable` | TLS mode. |
 | `grpc.tls.cert_path` | String | Unset | Certificate file path. |

src/flow/src/batching_mode/state.rs

@@ -91,14 +91,17 @@ impl TaskState {
         time_window_size: &Option<Duration>,
         max_timeout: Option<Duration>,
     ) -> Instant {
-        // = last query duration, capped by [max(min_run_interval, time_window_size), max_timeout], note at most `max_timeout`
-        let lower = MIN_REFRESH_DURATION.max(time_window_size.unwrap_or_default());
-        let next_duration = self.last_query_duration.max(lower);
-        let next_duration = if let Some(max_timeout) = max_timeout {
-            next_duration.min(max_timeout)
-        } else {
-            next_duration
-        };
+        let last_duration = max_timeout
+            .unwrap_or(self.last_query_duration)
+            .min(self.last_query_duration)
+            .max(MIN_REFRESH_DURATION);
+
+        let next_duration = time_window_size
+            .map(|t| {
+                let half = t / 2;
+                half.max(last_duration)
+            })
+            .unwrap_or(last_duration);
 
         METRIC_FLOW_BATCHING_ENGINE_WAIT_TIME
             .with_label_values(&[
@@ -217,60 +220,47 @@ impl DirtyTimeWindows {
 
         // get the first `window_cnt` time windows
         let max_time_range = window_size * window_cnt as i32;
+        let nth = {
+            let mut cur_time_range = chrono::Duration::zero();
+            let mut nth_key = None;
+            for (idx, (start, end)) in self.windows.iter().enumerate() {
+                // if time range is too long, stop
+                if cur_time_range > max_time_range {
+                    nth_key = Some(*start);
+                    break;
+                }
 
-        let mut to_be_query = BTreeMap::new();
-        let mut new_windows = self.windows.clone();
-        let mut cur_time_range = chrono::Duration::zero();
-        for (idx, (start, end)) in self.windows.iter().enumerate() {
-            let first_end = start
-                .add_duration(window_size.to_std().unwrap())
-                .context(TimeSnafu)?;
-            let end = end.unwrap_or(first_end);
+                // if we have enough time windows, stop
+                if idx >= window_cnt {
+                    nth_key = Some(*start);
+                    break;
+                }
 
-            // if time range is too long, stop
-            if cur_time_range >= max_time_range {
-                break;
+                if let Some(end) = end {
+                    if let Some(x) = end.sub(start) {
+                        cur_time_range += x;
+                    }
+                }
             }
 
-            // if we have enough time windows, stop
-            if idx >= window_cnt {
-                break;
-            }
+            nth_key
+        };
+        let first_nth = {
+            if let Some(nth) = nth {
+                let mut after = self.windows.split_off(&nth);
+                std::mem::swap(&mut self.windows, &mut after);
 
-            let Some(x) = end.sub(start) else {
-                continue;
-            };
-            if cur_time_range + x <= max_time_range {
-                to_be_query.insert(*start, Some(end));
-                new_windows.remove(start);
-                cur_time_range += x;
+                after
             } else {
-                // too large a window, split it
-                // split at window_size * times
-                let surplus = max_time_range - cur_time_range;
-                let times = surplus.num_seconds() / window_size.num_seconds();
-
-                let split_offset = window_size * times as i32;
-                let split_at = start
-                    .add_duration(split_offset.to_std().unwrap())
-                    .context(TimeSnafu)?;
-                to_be_query.insert(*start, Some(split_at));
-
-                // remove the original window
-                new_windows.remove(start);
-                new_windows.insert(split_at, Some(end));
-                cur_time_range += split_offset;
-                break;
+                std::mem::take(&mut self.windows)
             }
-        }
-
-        self.windows = new_windows;
+        };
 
         METRIC_FLOW_BATCHING_ENGINE_QUERY_WINDOW_CNT
             .with_label_values(&[flow_id.to_string().as_str()])
-            .observe(to_be_query.len() as f64);
+            .observe(first_nth.len() as f64);
 
-        let full_time_range = to_be_query
+        let full_time_range = first_nth
             .iter()
             .fold(chrono::Duration::zero(), |acc, (start, end)| {
                 if let Some(end) = end {
@@ -285,7 +275,7 @@ impl DirtyTimeWindows {
             .observe(full_time_range);
 
         let mut expr_lst = vec![];
-        for (start, end) in to_be_query.into_iter() {
+        for (start, end) in first_nth.into_iter() {
             // align using time window exprs
             let (start, end) = if let Some(ctx) = task_ctx {
                 let Some(time_window_expr) = &ctx.config.time_window_expr else {
@@ -519,64 +509,6 @@ mod test {
                     "((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:00:21' AS TIMESTAMP)))",
                 )
             ),
-            // split range
-            (
-                Vec::from_iter((0..20).map(|i|Timestamp::new_second(i*3)).chain(std::iter::once(
-                    Timestamp::new_second(60 + 3 * (DirtyTimeWindows::MERGE_DIST as i64 + 1)),
-                ))),
-                (chrono::Duration::seconds(3), None),
-                BTreeMap::from([
-                (
-                    Timestamp::new_second(0),
-                    Some(Timestamp::new_second(
-                        60
-                    )),
-                ),
-                (
-                    Timestamp::new_second(60 + 3 * (DirtyTimeWindows::MERGE_DIST as i64 + 1)),
-                    Some(Timestamp::new_second(
-                        60 + 3 * (DirtyTimeWindows::MERGE_DIST as i64 + 1) + 3
-                    )),
-                )]),
-                Some(
-                    "((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:01:00' AS TIMESTAMP)))",
-                )
-            ),
-            // split 2 min into 1 min
-            (
-                Vec::from_iter((0..40).map(|i|Timestamp::new_second(i*3))),
-                (chrono::Duration::seconds(3), None),
-                BTreeMap::from([
-                (
-                    Timestamp::new_second(0),
-                    Some(Timestamp::new_second(
-                        40 * 3
-                    )),
-                )]),
-                Some(
-                    "((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:01:00' AS TIMESTAMP)))",
-                )
-            ),
-            // split 3s + 1min into 3s + 57s
-            (
-                Vec::from_iter(std::iter::once(Timestamp::new_second(0)).chain((0..40).map(|i|Timestamp::new_second(20+i*3)))),
-                (chrono::Duration::seconds(3), None),
-                BTreeMap::from([
-                (
-                    Timestamp::new_second(0),
-                    Some(Timestamp::new_second(
-                        3
-                    )),
-                ),(
-                    Timestamp::new_second(20),
-                    Some(Timestamp::new_second(
-                        140
-                    )),
-                )]),
-                Some(
-                    "(((ts >= CAST('1970-01-01 00:00:00' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:00:03' AS TIMESTAMP))) OR ((ts >= CAST('1970-01-01 00:00:20' AS TIMESTAMP)) AND (ts < CAST('1970-01-01 00:01:17' AS TIMESTAMP))))",
-                )
-            ),
             // expired
             (
                 vec![
@@ -593,8 +525,6 @@ mod test {
                 None
             ),
         ];
-        // let len = testcases.len();
-        // let testcases = testcases[(len - 2)..(len - 1)].to_vec();
         for (lower_bounds, (window_size, expire_lower_bound), expected, expected_filter_expr) in
             testcases
         {