common_meta/ddl/event/
flow.rs

1// Copyright 2023 Greptime Team
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use std::any::Any;
16
17use api::v1::value::ValueData;
18use api::v1::{ColumnSchema, Row};
19use common_event_recorder::Event;
20use common_event_recorder::error::{Result, SerializeEventSnafu};
21use common_event_recorder::event_table::{
22    CATALOG_NAME_COLUMN, FLOW_ID_COLUMN, FLOW_NAME_COLUMN, column_schemas, nullable_string,
23    nullable_value,
24};
25use serde::Serialize;
26use snafu::ResultExt;
27
28pub(crate) const CREATE_FLOW_EVENT_TYPE: &str = "create_flow";
29pub(crate) const DROP_FLOW_EVENT_TYPE: &str = "drop_flow";
30
31const PAYLOAD_VERSION: u8 = 1;
32
33/// The bounded Create Flow intent allowed in a submitted event payload.
34#[derive(Debug)]
35pub(crate) struct CreateFlowEventIntent {
36    pub(crate) or_replace: bool,
37    pub(crate) create_if_not_exists: bool,
38    pub(crate) expire_after: Option<i64>,
39    pub(crate) eval_interval_secs: Option<i64>,
40}
41
42#[derive(Debug, Serialize)]
43struct CreateFlowPayload {
44    version: u8,
45    or_replace: bool,
46    create_if_not_exists: bool,
47    expire_after: Option<i64>,
48    eval_interval_secs: Option<i64>,
49}
50
51#[derive(Debug, Serialize)]
52struct DropFlowPayload {
53    version: u8,
54    drop_if_exists: bool,
55}
56
57#[derive(Debug, Serialize)]
58#[serde(untagged)]
59enum FlowDdlPayload {
60    Create(CreateFlowPayload),
61    Drop(DropFlowPayload),
62}
63
64/// A Flow DDL event with a schema shared by every lifecycle trigger.
65#[derive(Debug)]
66pub(crate) struct FlowDdlEvent {
67    event_type: &'static str,
68    catalog_name: Option<String>,
69    flow_name: Option<String>,
70    flow_id: Option<u32>,
71    payload: Option<FlowDdlPayload>,
72}
73
74impl FlowDdlEvent {
75    /// Builds the bounded event emitted when creating a Flow is submitted.
76    pub(crate) fn create_submitted(
77        catalog_name: &str,
78        flow_name: &str,
79        intent: CreateFlowEventIntent,
80    ) -> Self {
81        Self {
82            event_type: CREATE_FLOW_EVENT_TYPE,
83            catalog_name: Some(catalog_name.to_string()),
84            flow_name: Some(flow_name.to_string()),
85            flow_id: None,
86            payload: Some(FlowDdlPayload::Create(CreateFlowPayload {
87                version: PAYLOAD_VERSION,
88                or_replace: intent.or_replace,
89                create_if_not_exists: intent.create_if_not_exists,
90                expire_after: intent.expire_after,
91                eval_interval_secs: intent.eval_interval_secs,
92            })),
93        }
94    }
95
96    /// Builds the bounded event emitted when dropping a Flow is submitted.
97    pub(crate) fn drop_submitted(
98        catalog_name: &str,
99        flow_name: &str,
100        flow_id: u32,
101        drop_if_exists: bool,
102    ) -> Self {
103        Self {
104            event_type: DROP_FLOW_EVENT_TYPE,
105            catalog_name: Some(catalog_name.to_string()),
106            flow_name: Some(flow_name.to_string()),
107            flow_id: Some(flow_id),
108            payload: Some(FlowDdlPayload::Drop(DropFlowPayload {
109                version: PAYLOAD_VERSION,
110                drop_if_exists,
111            })),
112        }
113    }
114
115    /// Builds a Create Flow lifecycle event with its submitted locator.
116    pub(crate) fn create_lifecycle(catalog_name: &str, flow_name: &str) -> Self {
117        Self::lifecycle(CREATE_FLOW_EVENT_TYPE, catalog_name, flow_name)
118    }
119
120    /// Builds a successful Create Flow event with its submitted locator and resolved ID.
121    pub(crate) fn create_succeeded(
122        catalog_name: &str,
123        flow_name: &str,
124        flow_id: Option<u32>,
125    ) -> Self {
126        Self {
127            flow_id,
128            ..Self::lifecycle(CREATE_FLOW_EVENT_TYPE, catalog_name, flow_name)
129        }
130    }
131
132    /// Builds a Drop Flow lifecycle event with its submitted locator.
133    pub(crate) fn drop_lifecycle(catalog_name: &str, flow_name: &str, flow_id: u32) -> Self {
134        Self {
135            flow_id: Some(flow_id),
136            ..Self::lifecycle(DROP_FLOW_EVENT_TYPE, catalog_name, flow_name)
137        }
138    }
139
140    fn lifecycle(event_type: &'static str, catalog_name: &str, flow_name: &str) -> Self {
141        Self {
142            event_type,
143            catalog_name: Some(catalog_name.to_string()),
144            flow_name: Some(flow_name.to_string()),
145            flow_id: None,
146            payload: None,
147        }
148    }
149}
150
151impl Event for FlowDdlEvent {
152    fn event_type(&self) -> &str {
153        self.event_type
154    }
155
156    fn json_payload(&self) -> Result<serde_json::Value> {
157        match &self.payload {
158            Some(payload) => serde_json::to_value(payload).context(SerializeEventSnafu),
159            None => Ok(serde_json::Value::Null),
160        }
161    }
162
163    fn extra_schema(&self) -> Vec<ColumnSchema> {
164        column_schemas([&CATALOG_NAME_COLUMN, &FLOW_NAME_COLUMN, &FLOW_ID_COLUMN])
165    }
166
167    fn extra_rows(&self) -> Result<Vec<Row>> {
168        Ok(vec![Row {
169            values: vec![
170                nullable_string(self.catalog_name.as_deref()),
171                nullable_string(self.flow_name.as_deref()),
172                nullable_value(self.flow_id.map(ValueData::U32Value)),
173            ],
174        }])
175    }
176
177    fn as_any(&self) -> &dyn Any {
178        self
179    }
180}
common_meta/ddl/event/flow.rs

common_meta/ddl/event/
flow.rs
