common_meta/ddl/event/
database.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;
16use std::collections::HashMap;
17
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 as EVENT_TABLE_CATALOG_NAME_COLUMN,
23    SCHEMA_NAME_COLUMN as EVENT_TABLE_SCHEMA_NAME_COLUMN, column_schemas, nullable_string,
24};
25use serde::Serialize;
26use snafu::ResultExt;
27
28use crate::rpc::ddl::{AlterDatabaseKind, SetDatabaseOption, UnsetDatabaseOption};
29
30pub(crate) const CREATE_DATABASE_EVENT_TYPE: &str = "create_database";
31pub(crate) const ALTER_DATABASE_EVENT_TYPE: &str = "alter_database";
32pub(crate) const DROP_DATABASE_EVENT_TYPE: &str = "drop_database";
33const PAYLOAD_VERSION: u8 = 1;
34const TTL_OPTION_NAME: &str = "ttl";
35
36#[derive(Debug)]
37pub(crate) struct DatabaseDdlEvent {
38    event_type: &'static str,
39    catalog_name: Option<String>,
40    schema_name: Option<String>,
41    payload: Option<DatabaseDdlPayload>,
42}
43
44#[derive(Debug, Serialize)]
45#[serde(untagged)]
46enum DatabaseDdlPayload {
47    Create(CreateDatabasePayload),
48    Alter(AlterDatabasePayload),
49    Drop(DropDatabasePayload),
50}
51
52#[derive(Debug, Serialize)]
53struct CreateDatabasePayload {
54    version: u8,
55    create_if_not_exists: bool,
56    options: Vec<DatabaseOptionIntent>,
57}
58
59#[derive(Debug, Serialize)]
60struct AlterDatabasePayload {
61    version: u8,
62    #[serde(flatten)]
63    intent: AlterDatabaseIntent,
64}
65
66#[derive(Debug, Serialize)]
67#[serde(tag = "action", rename_all = "snake_case")]
68enum AlterDatabaseIntent {
69    Set { options: Vec<DatabaseOptionIntent> },
70    Unset { options: Vec<String> },
71}
72
73#[derive(Debug, Serialize)]
74struct DropDatabasePayload {
75    version: u8,
76    drop_if_exists: bool,
77}
78
79#[derive(Debug, Serialize)]
80struct DatabaseOptionIntent {
81    key: String,
82    value: String,
83}
84
85impl DatabaseDdlEvent {
86    pub(crate) fn create_submitted(
87        catalog_name: &str,
88        schema_name: &str,
89        create_if_not_exists: bool,
90        options: &HashMap<String, String>,
91    ) -> Self {
92        let mut options = options.iter().collect::<Vec<_>>();
93        options.sort_unstable_by_key(|(left, _)| *left);
94        let options = options
95            .into_iter()
96            .map(|(key, value)| DatabaseOptionIntent {
97                key: key.clone(),
98                value: value.clone(),
99            })
100            .collect();
101        Self::submitted(
102            CREATE_DATABASE_EVENT_TYPE,
103            catalog_name,
104            schema_name,
105            DatabaseDdlPayload::Create(CreateDatabasePayload {
106                version: PAYLOAD_VERSION,
107                create_if_not_exists,
108                options,
109            }),
110        )
111    }
112
113    pub(crate) fn alter_submitted(
114        catalog_name: &str,
115        schema_name: &str,
116        kind: &AlterDatabaseKind,
117    ) -> Self {
118        let intent = match kind {
119            AlterDatabaseKind::SetDatabaseOptions(options) => AlterDatabaseIntent::Set {
120                options: options
121                    .0
122                    .iter()
123                    .map(|option| match option {
124                        SetDatabaseOption::Ttl(ttl) => DatabaseOptionIntent {
125                            key: TTL_OPTION_NAME.to_string(),
126                            value: ttl.to_string(),
127                        },
128                        SetDatabaseOption::Other(key, value) => DatabaseOptionIntent {
129                            key: key.clone(),
130                            value: value.clone(),
131                        },
132                    })
133                    .collect(),
134            },
135            AlterDatabaseKind::UnsetDatabaseOptions(options) => {
136                let options = options
137                    .0
138                    .iter()
139                    .map(|option| match option {
140                        UnsetDatabaseOption::Ttl => TTL_OPTION_NAME.to_string(),
141                        UnsetDatabaseOption::Other(key) => key.clone(),
142                    })
143                    .collect();
144                AlterDatabaseIntent::Unset { options }
145            }
146        };
147        Self::submitted(
148            ALTER_DATABASE_EVENT_TYPE,
149            catalog_name,
150            schema_name,
151            DatabaseDdlPayload::Alter(AlterDatabasePayload {
152                version: PAYLOAD_VERSION,
153                intent,
154            }),
155        )
156    }
157
158    pub(crate) fn drop_submitted(
159        catalog_name: &str,
160        schema_name: &str,
161        drop_if_exists: bool,
162    ) -> Self {
163        Self::submitted(
164            DROP_DATABASE_EVENT_TYPE,
165            catalog_name,
166            schema_name,
167            DatabaseDdlPayload::Drop(DropDatabasePayload {
168                version: PAYLOAD_VERSION,
169                drop_if_exists,
170            }),
171        )
172    }
173
174    pub(crate) fn create_lifecycle(catalog_name: &str, schema_name: &str) -> Self {
175        Self::lifecycle(CREATE_DATABASE_EVENT_TYPE, catalog_name, schema_name)
176    }
177
178    pub(crate) fn alter_lifecycle(catalog_name: &str, schema_name: &str) -> Self {
179        Self::lifecycle(ALTER_DATABASE_EVENT_TYPE, catalog_name, schema_name)
180    }
181
182    pub(crate) fn drop_lifecycle(catalog_name: &str, schema_name: &str) -> Self {
183        Self::lifecycle(DROP_DATABASE_EVENT_TYPE, catalog_name, schema_name)
184    }
185
186    fn submitted(
187        event_type: &'static str,
188        catalog_name: &str,
189        schema_name: &str,
190        payload: DatabaseDdlPayload,
191    ) -> Self {
192        Self {
193            event_type,
194            catalog_name: Some(catalog_name.to_string()),
195            schema_name: Some(schema_name.to_string()),
196            payload: Some(payload),
197        }
198    }
199
200    fn lifecycle(event_type: &'static str, catalog_name: &str, schema_name: &str) -> Self {
201        Self {
202            event_type,
203            catalog_name: Some(catalog_name.to_string()),
204            schema_name: Some(schema_name.to_string()),
205            payload: None,
206        }
207    }
208}
209
210impl Event for DatabaseDdlEvent {
211    fn event_type(&self) -> &str {
212        self.event_type
213    }
214
215    fn json_payload(&self) -> Result<serde_json::Value> {
216        match &self.payload {
217            Some(payload) => serde_json::to_value(payload).context(SerializeEventSnafu),
218            None => Ok(serde_json::Value::Null),
219        }
220    }
221
222    fn extra_schema(&self) -> Vec<ColumnSchema> {
223        column_schemas([
224            &EVENT_TABLE_CATALOG_NAME_COLUMN,
225            &EVENT_TABLE_SCHEMA_NAME_COLUMN,
226        ])
227    }
228
229    fn extra_rows(&self) -> Result<Vec<Row>> {
230        Ok(vec![Row {
231            values: vec![
232                nullable_string(self.catalog_name.as_deref()),
233                nullable_string(self.schema_name.as_deref()),
234            ],
235        }])
236    }
237
238    fn as_any(&self) -> &dyn Any {
239        self
240    }
241}
common_meta/ddl/event/database.rs

common_meta/ddl/event/
database.rs
