feat: adjust max refresh rate

Signed-off-by: discord9 <discord9@163.com>

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. |
+| `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.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. |
+| `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.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,17 +91,14 @@ impl TaskState {
         time_window_size: &Option<Duration>,
         max_timeout: Option<Duration>,
     ) -> Instant {
-        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);
+        // = 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
+        };
 
         METRIC_FLOW_BATCHING_ENGINE_WAIT_TIME
             .with_label_values(&[
@@ -240,29 +237,30 @@ impl DirtyTimeWindows {
                 break;
             }
 
-            if let Some(x) = end.sub(start) {
-                if cur_time_range + x <= max_time_range {
-                    to_be_query.insert(*start, Some(end));
-                    new_windows.remove(start);
-                    cur_time_range += x;
-                } 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 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;
+            } 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));
+                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;
-                }
+                // remove the original window
+                new_windows.remove(start);
+                new_windows.insert(split_at, Some(end));
+                cur_time_range += split_offset;
+                break;
             }
         }