fix(flow): convert streaming expiration to milliseconds (#8481)

* fix(flow): convert streaming expiration to milliseconds

Flow EXPIRE AFTER values are stored in seconds, while streaming flow timestamps and durations use milliseconds. Passing the value through unchanged expires state and limits refill scans 1000 times too early.

Convert the value at streaming create and refill boundaries, reject overflow, and document the CreateFlowArgs unit.

Signed-off-by: QuakeWang <wangfuzheng0814@foxmail.com>

* fix(flow): reject negative streaming expiration

Reject negative EXPIRE AFTER values at the streaming adapter boundary so refill ranges cannot move into the future. Keep zero valid and retain checked seconds-to-milliseconds conversion.

Signed-off-by: QuakeWang <wangfuzheng0814@foxmail.com>

---------

Signed-off-by: QuakeWang <wangfuzheng0814@foxmail.com>

src/flow/src/adapter.rs

@@ -76,6 +76,23 @@ pub(crate) mod table_source;
 use crate::FrontendInvoker;
 use crate::error::Error;
 
+fn expire_after_secs_to_millis(expire_after_secs: i64) -> Result<repr::Duration, Error> {
+    ensure!(
+        expire_after_secs >= 0,
+        InvalidQuerySnafu {
+            reason: format!("EXPIRE AFTER must be non-negative, got {expire_after_secs} seconds"),
+        }
+    );
+
+    expire_after_secs
+        .checked_mul(1_000)
+        .with_context(|| InvalidQuerySnafu {
+            reason: format!(
+                "EXPIRE AFTER value {expire_after_secs} seconds cannot be represented in milliseconds"
+            ),
+        })
+}
+
 // `GREPTIME_TIMESTAMP` is not used to distinguish when table is created automatically by flow
 pub const AUTO_CREATED_PLACEHOLDER_TS_COL: &str = "__ts_placeholder";
 
@@ -740,7 +757,7 @@ impl StreamingEngine {
             source_table_ids,
             create_if_not_exists,
             or_replace,
-            expire_after,
+            expire_after: expire_after_secs,
             eval_interval: _,
             comment,
             sql,
@@ -748,6 +765,9 @@ impl StreamingEngine {
             query_ctx,
             ..
         } = args;
+        let expire_after = expire_after_secs
+            .map(expire_after_secs_to_millis)
+            .transpose()?;
 
         let mut node_ctx = self.node_context.write().await;
         // assign global id to source and sink table

src/flow/src/adapter/refill.rs

@@ -85,7 +85,10 @@ impl StreamingEngine {
                 }
             }
 
-            let expire_after = info.expire_after();
+            let expire_after = info
+                .expire_after()
+                .map(super::expire_after_secs_to_millis)
+                .transpose()?;
             // TODO(discord9): better way to get last point
             let now = self.tick_manager.tick();
             let plan = self

src/flow/src/adapter/tests.rs

@@ -61,3 +61,25 @@ pub fn new_test_table_info_with_name<I: IntoIterator<Item = u32>>(
 ///
 /// containing several default table info and schema
 fn mock_harness_flow_node_manager() {}
+
+#[test]
+fn test_expire_after_secs_to_millis() {
+    assert_eq!(expire_after_secs_to_millis(300).unwrap(), 300_000);
+    assert_eq!(expire_after_secs_to_millis(0).unwrap(), 0);
+
+    let max_secs = i64::MAX / 1_000;
+    assert_eq!(
+        expire_after_secs_to_millis(max_secs).unwrap(),
+        max_secs * 1_000
+    );
+}
+
+#[test]
+fn test_expire_after_secs_to_millis_invalid() {
+    let invalid_values = [i64::MAX / 1_000 + 1, -1];
+
+    for invalid_secs in invalid_values {
+        let error = expire_after_secs_to_millis(invalid_secs).unwrap_err();
+        assert!(matches!(error, Error::InvalidQuery { .. }));
+    }
+}

src/flow/src/engine.rs

@@ -34,6 +34,7 @@ pub struct CreateFlowArgs {
     pub source_table_ids: Vec<TableId>,
     pub create_if_not_exists: bool,
     pub or_replace: bool,
+    /// Duration in seconds.
     pub expire_after: Option<i64>,
     pub eval_interval: Option<i64>,
     pub comment: Option<String>,