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>,