feat: change EXPIRE WHEN to EXPIRE AFTER

Cargo.toml

@@ -120,7 +120,7 @@ etcd-client = { git = "https://github.com/MichaelScofield/etcd-client.git", rev
 fst = "0.4.7"
 futures = "0.3"
 futures-util = "0.3"
-greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "a11db14b8502f55ca5348917fd18e6fcf140f55e" }
+greptime-proto = { git = "https://github.com/GreptimeTeam/greptime-proto.git", rev = "48ddbd96d0d9beadaa50e2dfe2066b01e80a268c" }
 humantime = "2.1"
 humantime-serde = "1.1"
 itertools = "0.10"

src/common/meta/src/cache/flow/table_flownode.rs

@@ -180,7 +180,7 @@ mod tests {
                     catalog_name: DEFAULT_CATALOG_NAME.to_string(),
                     flow_name: "my_flow".to_string(),
                     raw_sql: "sql".to_string(),
-                    expire_when: "expire".to_string(),
+                    expire_after: "expire".to_string(),
                     comment: "comment".to_string(),
                     options: Default::default(),
                 },

src/common/meta/src/ddl/create_flow.rs

@@ -283,7 +283,7 @@ impl From<&CreateFlowData> for CreateRequest {
             sink_table_name: Some(value.task.sink_table_name.clone().into()),
             // Always be true
             create_if_not_exists: true,
-            expire_when: value.task.expire_when.clone(),
+            expire_after: value.task.expire_after.clone(),
             comment: value.task.comment.clone(),
             sql: value.task.sql.clone(),
             flow_options: value.task.flow_options.clone(),
@@ -297,7 +297,7 @@ impl From<&CreateFlowData> for FlowInfoValue {
             catalog_name,
             flow_name,
             sink_table_name,
-            expire_when,
+            expire_after,
             comment,
             sql,
             flow_options: options,
@@ -318,7 +318,7 @@ impl From<&CreateFlowData> for FlowInfoValue {
             catalog_name,
             flow_name,
             raw_sql: sql,
-            expire_when,
+            expire_after,
             comment,
             options,
         }

src/common/meta/src/ddl/tests/create_flow.rs

@@ -44,7 +44,7 @@ pub(crate) fn test_create_flow_task(
         sink_table_name,
         or_replace: false,
         create_if_not_exists,
-        expire_when: "".to_string(),
+        expire_after: "".to_string(),
         comment: "".to_string(),
         sql: "raw_sql".to_string(),
         flow_options: Default::default(),

src/common/meta/src/key/flow.rs

@@ -328,7 +328,7 @@ mod tests {
             sink_table_name,
             flownode_ids,
             raw_sql: "raw".to_string(),
-            expire_when: "expr".to_string(),
+            expire_after: "expr".to_string(),
             comment: "hi".to_string(),
             options: Default::default(),
         }
@@ -420,7 +420,7 @@ mod tests {
             sink_table_name: another_sink_table_name,
             flownode_ids: [(0, 1u64)].into(),
             raw_sql: "raw".to_string(),
-            expire_when: "expr".to_string(),
+            expire_after: "expr".to_string(),
             comment: "hi".to_string(),
             options: Default::default(),
         };

src/common/meta/src/key/flow/flow_info.rs

@@ -123,7 +123,7 @@ pub struct FlowInfoValue {
     /// The raw sql.
     pub(crate) raw_sql: String,
     /// The expr of expire.
-    pub(crate) expire_when: String,
+    pub(crate) expire_after: String,
     /// The comment.
     pub(crate) comment: String,
     /// The options.

src/common/meta/src/rpc/ddl.rs

@@ -898,7 +898,7 @@ pub struct CreateFlowTask {
     pub sink_table_name: TableName,
     pub or_replace: bool,
     pub create_if_not_exists: bool,
-    pub expire_when: String,
+    pub expire_after: String,
     pub comment: String,
     pub sql: String,
     pub flow_options: HashMap<String, String>,
@@ -915,7 +915,7 @@ impl TryFrom<PbCreateFlowTask> for CreateFlowTask {
             sink_table_name,
             or_replace,
             create_if_not_exists,
-            expire_when,
+            expire_after,
             comment,
             sql,
             flow_options,
@@ -934,7 +934,7 @@ impl TryFrom<PbCreateFlowTask> for CreateFlowTask {
                 .into(),
             or_replace,
             create_if_not_exists,
-            expire_when,
+            expire_after,
             comment,
             sql,
             flow_options,
@@ -951,7 +951,7 @@ impl From<CreateFlowTask> for PbCreateFlowTask {
             sink_table_name,
             or_replace,
             create_if_not_exists,
-            expire_when,
+            expire_after,
             comment,
             sql,
             flow_options,
@@ -965,7 +965,7 @@ impl From<CreateFlowTask> for PbCreateFlowTask {
                 sink_table_name: Some(sink_table_name.into()),
                 or_replace,
                 create_if_not_exists,
-                expire_when,
+                expire_after,
                 comment,
                 sql,
                 flow_options,

src/flow/src/adapter.rs

@@ -565,7 +565,7 @@ impl FlownodeManager {
     /// Return task id if a new task is created, otherwise return None
     ///
     /// steps to create task:
-    /// 1. parse query into typed plan(and optional parse expire_when expr)
+    /// 1. parse query into typed plan(and optional parse expire_after expr)
     /// 2. render source/sink with output table id and used input table id
     #[allow(clippy::too_many_arguments)]
     pub async fn create_flow(
@@ -574,7 +574,7 @@ impl FlownodeManager {
         sink_table_name: TableName,
         source_table_ids: &[TableId],
         create_if_not_exist: bool,
-        expire_when: Option<String>,
+        expire_after: Option<String>,
         comment: Option<String>,
         sql: String,
         flow_options: HashMap<String, String>,
@@ -608,7 +608,7 @@ impl FlownodeManager {
         debug!("Flow {:?}'s Plan is {:?}", flow_id, flow_plan);
         node_ctx.assign_table_schema(&sink_table_name, flow_plan.typ.clone())?;
 
-        let expire_when = expire_when
+        let expire_after = expire_after
             .and_then(|s| {
                 if s.is_empty() || s.split_whitespace().join("").is_empty() {
                     None
@@ -656,7 +656,7 @@ impl FlownodeManager {
             sink_sender,
             source_ids,
             src_recvs: source_receivers,
-            expire_when,
+            expire_after,
             create_if_not_exist,
             err_collector,
         };

src/flow/src/adapter/flownode_impl.rs

@@ -45,7 +45,7 @@ impl Flownode for FlownodeManager {
                 source_table_ids,
                 sink_table_name: Some(sink_table_name),
                 create_if_not_exists,
-                expire_when,
+                expire_after,
                 comment,
                 sql,
                 flow_options,
@@ -62,7 +62,7 @@ impl Flownode for FlownodeManager {
                         sink_table_name,
                         &source_table_ids,
                         create_if_not_exists,
-                        Some(expire_when),
+                        Some(expire_after),
                         Some(comment),
                         sql,
                         flow_options,

src/flow/src/adapter/worker.rs

@@ -232,7 +232,7 @@ impl<'s> Worker<'s> {
         source_ids: &[GlobalId],
         src_recvs: Vec<broadcast::Receiver<DiffRow>>,
         // TODO(discord9): set expire duration for all arrangement and compare to sys timestamp instead
-        expire_when: Option<repr::Duration>,
+        expire_after: Option<repr::Duration>,
         create_if_not_exist: bool,
         err_collector: ErrCollector,
     ) -> Result<Option<FlowId>, Error> {
@@ -247,7 +247,7 @@ impl<'s> Worker<'s> {
             err_collector,
             ..Default::default()
         };
-        cur_task_state.state.set_expire_after(expire_when);
+        cur_task_state.state.set_expire_after(expire_after);
 
         {
             let mut ctx = cur_task_state.new_ctx(sink_id);
@@ -319,7 +319,7 @@ impl<'s> Worker<'s> {
                 sink_sender,
                 source_ids,
                 src_recvs,
-                expire_when,
+                expire_after,
                 create_if_not_exist,
                 err_collector,
             } => {
@@ -330,7 +330,7 @@ impl<'s> Worker<'s> {
                     sink_sender,
                     &source_ids,
                     src_recvs,
-                    expire_when,
+                    expire_after,
                     create_if_not_exist,
                     err_collector,
                 );
@@ -368,7 +368,7 @@ pub enum Request {
         sink_sender: mpsc::UnboundedSender<DiffRow>,
         source_ids: Vec<GlobalId>,
         src_recvs: Vec<broadcast::Receiver<DiffRow>>,
-        expire_when: Option<repr::Duration>,
+        expire_after: Option<repr::Duration>,
         create_if_not_exist: bool,
         err_collector: ErrCollector,
     },
@@ -524,7 +524,7 @@ mod test {
             sink_sender: sink_tx,
             source_ids: src_ids,
             src_recvs: vec![rx],
-            expire_when: None,
+            expire_after: None,
             create_if_not_exist: true,
             err_collector: ErrCollector::default(),
         };

src/operator/src/expr_factory.rs

@@ -591,8 +591,8 @@ pub fn to_create_flow_task_expr(
         sink_table_name: Some(sink_table_name),
         or_replace: create_flow.or_replace,
         create_if_not_exists: create_flow.if_not_exists,
-        expire_when: create_flow
-            .expire_when
+        expire_after: create_flow
+            .expire_after
             .map(|e| e.to_string())
             .unwrap_or_default(),
         comment: create_flow.comment.unwrap_or_default(),

src/sql/src/parsers/create_parser.rs

@@ -44,7 +44,7 @@ pub const ENGINE: &str = "ENGINE";
 pub const MAXVALUE: &str = "MAXVALUE";
 pub const SINK: &str = "SINK";
 pub const EXPIRE: &str = "EXPIRE";
-pub const WHEN: &str = "WHEN";
+pub const AFTER: &str = "AFTER";
 
 const DB_OPT_KEY_TTL: &str = "ttl";
 
@@ -235,9 +235,9 @@ impl<'a> ParserContext<'a> {
 
         let output_table_name = self.intern_parse_table_name()?;
 
-        let expire_when = if self
+        let expire_after = if self
             .parser
-            .consume_tokens(&[Token::make_keyword(EXPIRE), Token::make_keyword(WHEN)])
+            .consume_tokens(&[Token::make_keyword(EXPIRE), Token::make_keyword(AFTER)])
         {
             Some(self.parser.parse_expr().context(error::SyntaxSnafu)?)
         } else {
@@ -272,7 +272,7 @@ impl<'a> ParserContext<'a> {
             sink_table_name: output_table_name,
             or_replace,
             if_not_exists,
-            expire_when,
+            expire_after,
             comment,
             query,
         }))
@@ -877,7 +877,7 @@ mod tests {
     use common_catalog::consts::FILE_ENGINE;
     use common_error::ext::ErrorExt;
     use sqlparser::ast::ColumnOption::NotNull;
-    use sqlparser::ast::{BinaryOperator, Expr, Function, Interval, ObjectName, Value};
+    use sqlparser::ast::{BinaryOperator, Expr, Interval, ObjectName, Value};
 
     use super::*;
     use crate::dialect::GreptimeDbDialect;
@@ -1103,7 +1103,7 @@ mod tests {
         let sql = r"
 CREATE OR REPLACE FLOW IF NOT EXISTS task_1
 SINK TO schema_1.table_1
-EXPIRE WHEN timestamp < now() - INTERVAL '5m'
+EXPIRE AFTER INTERVAL '5m'
 COMMENT 'test comment'
 AS
 SELECT max(c1), min(c2) FROM schema_2.table_2;";
@@ -1133,36 +1133,13 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;";
             ]),
             or_replace: true,
             if_not_exists: true,
-            expire_when: Some(Expr::BinaryOp {
-                left: Box::new(Expr::Identifier(Ident {
-                    value: "timestamp".to_string(),
-                    quote_style: None,
-                })),
-                op: BinaryOperator::Lt,
-                right: Box::new(Expr::BinaryOp {
-                    left: Box::new(Expr::Function(Function {
-                        name: ObjectName(vec![Ident {
-                            value: "now".to_string(),
-                            quote_style: None,
-                        }]),
-                        args: vec![],
-                        filter: None,
-                        null_treatment: None,
-                        over: None,
-                        distinct: false,
-                        special: false,
-                        order_by: vec![],
-                    })),
-                    op: BinaryOperator::Minus,
-                    right: Box::new(Expr::Interval(Interval {
-                        value: Box::new(Expr::Value(Value::SingleQuotedString("5m".to_string()))),
-                        leading_field: None,
-                        leading_precision: None,
-                        last_field: None,
-                        fractional_seconds_precision: None,
-                    })),
-                }),
-            }),
+            expire_after: Some(Expr::Interval(Interval {
+                value: Box::new(Expr::Value(Value::SingleQuotedString("5m".to_string()))),
+                leading_field: None,
+                leading_precision: None,
+                last_field: None,
+                fractional_seconds_precision: None,
+            })),
             comment: Some("test comment".to_string()),
             // ignore query parse result
             query: create_task.query.clone(),
@@ -1185,7 +1162,7 @@ SELECT max(c1), min(c2) FROM schema_2.table_2;";
         };
         assert!(!create_task.or_replace);
         assert!(!create_task.if_not_exists);
-        assert!(create_task.expire_when.is_none());
+        assert!(create_task.expire_after.is_none());
         assert!(create_task.comment.is_none());
     }
 

src/sql/src/statements/create.rs

@@ -255,8 +255,8 @@ pub struct CreateFlow {
     pub or_replace: bool,
     /// Create if not exist
     pub if_not_exists: bool,
-    /// `EXPIRE_WHEN`
-    pub expire_when: Option<Expr>,
+    /// `EXPIRE AFTER`
+    pub expire_after: Option<Expr>,
     /// Comment string
     pub comment: Option<String>,
     /// SQL statement
@@ -275,8 +275,8 @@ impl Display for CreateFlow {
         }
         write!(f, "{} ", &self.flow_name)?;
         write!(f, "OUTPUT AS {} ", &self.sink_table_name)?;
-        if let Some(expire_when) = &self.expire_when {
-            write!(f, "EXPIRE WHEN {} ", expire_when)?;
+        if let Some(expire_after) = &self.expire_after {
+            write!(f, "EXPIRE AFTER {} ", expire_after)?;
         }
         if let Some(comment) = &self.comment {
             write!(f, "COMMENT '{}' ", comment)?;