From 5d9dddffa77b0584d23dd6890fe911b8a1b4e1ab Mon Sep 17 00:00:00 2001
From: QuakeWang <45645138+QuakeWang@users.noreply.github.com>
Date: Tue, 14 Jul 2026 12:42:23 +0800
Subject: [PATCH] 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        | 22 +++++++++++++++++++++-
 src/flow/src/adapter/refill.rs |  5 ++++-
 src/flow/src/adapter/tests.rs  | 22 ++++++++++++++++++++++
 src/flow/src/engine.rs         |  1 +
 4 files changed, 48 insertions(+), 2 deletions(-)

diff --git a/src/flow/src/adapter.rs b/src/flow/src/adapter.rs
index 1e663582a2..e8dd10ad7d 100644
--- a/src/flow/src/adapter.rs
+++ b/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
diff --git a/src/flow/src/adapter/refill.rs b/src/flow/src/adapter/refill.rs
index 6d66505e89..3672c0427e 100644
--- a/src/flow/src/adapter/refill.rs
+++ b/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
diff --git a/src/flow/src/adapter/tests.rs b/src/flow/src/adapter/tests.rs
index 9ed7e676c2..df6a58c46c 100644
--- a/src/flow/src/adapter/tests.rs
+++ b/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 { .. }));
+    }
+}
diff --git a/src/flow/src/engine.rs b/src/flow/src/engine.rs
index 386610ad71..3a139a898a 100644
--- a/src/flow/src/engine.rs
+++ b/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>,