meta_srv/event/
wal_prune.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    LATEST_OFFSET_COLUMN, PRUNABLE_ENTRY_ID_COLUMN, TOPIC_NAME_COLUMN, column_schemas,
23    nullable_value,
24};
25use serde::Serialize;
26use snafu::ResultExt;
27use store_api::logstore::EntryId;
28
29/// Procedure event type for WAL pruning.
30pub(crate) const WAL_PRUNE_EVENT_TYPE: &str = "wal_prune";
31
32const PAYLOAD_VERSION: u8 = 1;
33
34#[derive(Debug, Serialize)]
35struct WalPrunePayload {
36    version: u8,
37    logical_delete: bool,
38}
39
40/// A lifecycle event for a WAL prune operation.
41#[derive(Debug)]
42pub(crate) struct WalPruneEvent {
43    topic_name: String,
44    prunable_entry_id: EntryId,
45    latest_offset: Option<u64>,
46    payload: WalPrunePayload,
47}
48
49impl WalPruneEvent {
50    /// Creates an event for a WAL prune operation.
51    pub(crate) fn new(
52        topic_name: &str,
53        prunable_entry_id: EntryId,
54        latest_offset: Option<u64>,
55        logical_delete: bool,
56    ) -> Self {
57        Self {
58            topic_name: topic_name.to_string(),
59            prunable_entry_id,
60            latest_offset,
61            payload: WalPrunePayload {
62                version: PAYLOAD_VERSION,
63                logical_delete,
64            },
65        }
66    }
67}
68
69impl Event for WalPruneEvent {
70    fn event_type(&self) -> &str {
71        WAL_PRUNE_EVENT_TYPE
72    }
73
74    fn json_payload(&self) -> Result<serde_json::Value> {
75        serde_json::to_value(&self.payload).context(SerializeEventSnafu)
76    }
77
78    fn extra_schema(&self) -> Vec<ColumnSchema> {
79        column_schemas([
80            &TOPIC_NAME_COLUMN,
81            &PRUNABLE_ENTRY_ID_COLUMN,
82            &LATEST_OFFSET_COLUMN,
83        ])
84    }
85
86    fn extra_rows(&self) -> Result<Vec<Row>> {
87        Ok(vec![Row {
88            values: vec![
89                ValueData::StringValue(self.topic_name.clone()).into(),
90                ValueData::U64Value(self.prunable_entry_id).into(),
91                nullable_value(self.latest_offset.map(ValueData::U64Value)),
92            ],
93        }])
94    }
95
96    fn as_any(&self) -> &dyn Any {
97        self
98    }
99}
100
101#[cfg(test)]
102mod tests {
103    use api::v1::Row;
104    use common_event_recorder::Event;
105    use common_event_recorder::event_table::{
106        LATEST_OFFSET_COLUMN, PRUNABLE_ENTRY_ID_COLUMN, TOPIC_NAME_COLUMN, column_schemas,
107    };
108    use common_event_recorder::testing::assert_event_contract;
109
110    use super::*;
111
112    #[test]
113    fn test_wal_prune_event_contract() {
114        let event = WalPruneEvent::new("greptimedb_wal_topic_0", 100, Some(200), false);
115
116        assert_event_contract(
117            &event,
118            WAL_PRUNE_EVENT_TYPE,
119            &column_schemas([
120                &TOPIC_NAME_COLUMN,
121                &PRUNABLE_ENTRY_ID_COLUMN,
122                &LATEST_OFFSET_COLUMN,
123            ]),
124            &[Row {
125                values: vec![
126                    ValueData::StringValue("greptimedb_wal_topic_0".to_string()).into(),
127                    ValueData::U64Value(100).into(),
128                    ValueData::U64Value(200).into(),
129                ],
130            }],
131        );
132        assert_eq!(
133            event.json_payload().unwrap(),
134            serde_json::json!({
135                "version": 1,
136                "logical_delete": false,
137            })
138        );
139
140        let event = WalPruneEvent::new("greptimedb_wal_topic_1", 42, None, true);
141        assert!(
142            event.extra_rows().unwrap()[0].values[2]
143                .value_data
144                .is_none()
145        );
146        assert_eq!(
147            event.json_payload().unwrap(),
148            serde_json::json!({
149                "version": 1,
150                "logical_delete": true,
151            })
152        );
153    }
154}
meta_srv/event/wal_prune.rs

meta_srv/event/
wal_prune.rs
