common_meta/ddl/drop_database/
executor.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 common_procedure::Status;
18use common_telemetry::info;
19use serde::{Deserialize, Serialize};
20use snafu::OptionExt;
21use table::metadata::TableId;
22use table::table_name::TableName;
23
24use crate::ddl::DdlContext;
25use crate::ddl::drop_database::cursor::DropDatabaseCursor;
26use crate::ddl::drop_database::{DropDatabaseContext, DropTableTarget, State};
27use crate::ddl::drop_table::executor::DropTableExecutor;
28use crate::ddl::utils::get_region_wal_options;
29use crate::error::{self, Result};
30use crate::key::table_route::TableRouteValue;
31use crate::region_keeper::OperatingRegionGuard;
32use crate::rpc::router::{RegionRoute, operating_leader_regions};
33
34#[derive(Debug, Serialize, Deserialize)]
35pub(crate) struct DropDatabaseExecutor {
36    table_id: TableId,
37    physical_table_id: TableId,
38    table_name: TableName,
39    /// The physical table region routes.
40    pub(crate) physical_region_routes: Vec<RegionRoute>,
41    pub(crate) target: DropTableTarget,
42    #[serde(skip)]
43    dropping_regions: Vec<OperatingRegionGuard>,
44}
45
46impl DropDatabaseExecutor {
47    /// Returns a new [DropDatabaseExecutor].
48    pub fn new(
49        table_id: TableId,
50        physical_table_id: TableId,
51        table_name: TableName,
52        physical_region_routes: Vec<RegionRoute>,
53        target: DropTableTarget,
54    ) -> Self {
55        Self {
56            table_id,
57            physical_table_id,
58            table_name,
59            physical_region_routes,
60            target,
61            dropping_regions: vec![],
62        }
63    }
64}
65
66impl DropDatabaseExecutor {
67    /// Registers the operating regions.
68    pub(crate) fn register_dropping_regions(&mut self, ddl_ctx: &DdlContext) -> Result<()> {
69        if !self.dropping_regions.is_empty() {
70            return Ok(());
71        }
72        let dropping_regions = operating_leader_regions(&self.physical_region_routes);
73        let mut dropping_region_guards = Vec::with_capacity(dropping_regions.len());
74        for (region_id, datanode_id) in dropping_regions {
75            let guard = ddl_ctx
76                .memory_region_keeper
77                .register(datanode_id, region_id)
78                .context(error::RegionOperatingRaceSnafu {
79                    region_id,
80                    peer_id: datanode_id,
81                })?;
82            dropping_region_guards.push(guard);
83        }
84        self.dropping_regions = dropping_region_guards;
85        Ok(())
86    }
87}
88
89#[async_trait::async_trait]
90#[typetag::serde]
91impl State for DropDatabaseExecutor {
92    fn recover(&mut self, ddl_ctx: &DdlContext) -> Result<()> {
93        self.register_dropping_regions(ddl_ctx)
94    }
95
96    async fn next(
97        &mut self,
98        ddl_ctx: &DdlContext,
99        ctx: &mut DropDatabaseContext,
100    ) -> Result<(Box<dyn State>, Status)> {
101        self.register_dropping_regions(ddl_ctx)?;
102        let executor = DropTableExecutor::new(self.table_name.clone(), self.table_id, true);
103        if ctx.retrying {
104            info!(
105                "Remapping region routes addresses for retrying drop regions for table_id: {}",
106                self.table_id
107            );
108            let storage = ddl_ctx
109                .table_metadata_manager
110                .table_route_manager()
111                .table_route_storage();
112            // The peer addresses may change during retries,
113            // so we always remap the region routes.
114            storage
115                .remap_region_routes(&mut self.physical_region_routes)
116                .await?;
117        }
118        // Deletes metadata for table permanently.
119        let table_route_value = TableRouteValue::new(
120            self.table_id,
121            self.physical_table_id,
122            self.physical_region_routes.clone(),
123        );
124
125        // Deletes topic-region mapping if dropping physical table
126        let region_wal_options = get_region_wal_options(
127            &ddl_ctx.table_metadata_manager,
128            &table_route_value,
129            self.physical_table_id,
130        )
131        .await?;
132
133        executor
134            .on_destroy_metadata(ddl_ctx, &table_route_value, &region_wal_options)
135            .await?;
136        executor.invalidate_table_cache(ddl_ctx).await?;
137        executor
138            .on_drop_regions(
139                &ddl_ctx.node_manager,
140                &ddl_ctx.leader_region_registry,
141                &self.physical_region_routes,
142                true,
143                false,
144                false,
145            )
146            .await?;
147        info!("Table: {}({}) is dropped", self.table_name, self.table_id);
148
149        Ok((
150            Box::new(DropDatabaseCursor::new(self.target)),
151            Status::executing(false),
152        ))
153    }
154
155    fn as_any(&self) -> &dyn Any {
156        self
157    }
158}
159
160#[cfg(test)]
161mod tests {
162    use std::sync::Arc;
163
164    use api::region::RegionResponse;
165    use api::v1::region::RegionRequest;
166    use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
167    use common_error::ext::BoxedError;
168    use common_query::request::QueryRequest;
169    use common_recordbatch::SendableRecordBatchStream;
170    use table::table_name::TableName;
171
172    use crate::ddl::drop_database::cursor::DropDatabaseCursor;
173    use crate::ddl::drop_database::executor::DropDatabaseExecutor;
174    use crate::ddl::drop_database::{DropDatabaseContext, DropTableTarget, State};
175    use crate::ddl::test_util::datanode_handler::DatanodeWatcher;
176    use crate::ddl::test_util::{
177        create_logical_table, create_physical_table, put_datanode_address,
178    };
179    use crate::error::{self, Error, Result};
180    use crate::key::datanode_table::DatanodeTableKey;
181    use crate::peer::Peer;
182    use crate::rpc::router::region_distribution;
183    use crate::test_util::{MockDatanodeHandler, MockDatanodeManager, new_ddl_context};
184
185    #[derive(Clone)]
186    pub struct NaiveDatanodeHandler;
187
188    #[async_trait::async_trait]
189    impl MockDatanodeHandler for NaiveDatanodeHandler {
190        async fn handle(&self, _peer: &Peer, _request: RegionRequest) -> Result<RegionResponse> {
191            Ok(RegionResponse::new(0))
192        }
193
194        async fn handle_query(
195            &self,
196            _peer: &Peer,
197            _request: QueryRequest,
198        ) -> Result<SendableRecordBatchStream> {
199            unreachable!()
200        }
201    }
202
203    #[tokio::test]
204    async fn test_next_with_physical_table() {
205        let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
206        let ddl_context = new_ddl_context(node_manager);
207        let physical_table_id = create_physical_table(&ddl_context, "phy").await;
208        let (_, table_route) = ddl_context
209            .table_metadata_manager
210            .table_route_manager()
211            .get_physical_table_route(physical_table_id)
212            .await
213            .unwrap();
214        {
215            let mut state = DropDatabaseExecutor::new(
216                physical_table_id,
217                physical_table_id,
218                TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "phy"),
219                table_route.region_routes.clone(),
220                DropTableTarget::Physical,
221            );
222            let mut ctx = DropDatabaseContext {
223                catalog: DEFAULT_CATALOG_NAME.to_string(),
224                schema: DEFAULT_SCHEMA_NAME.to_string(),
225                drop_if_exists: false,
226                tables: None,
227                retrying: false,
228            };
229            let (state, status) = state.next(&ddl_context, &mut ctx).await.unwrap();
230            assert!(!status.need_persist());
231            let cursor = state.as_any().downcast_ref::<DropDatabaseCursor>().unwrap();
232            assert_eq!(cursor.target, DropTableTarget::Physical);
233        }
234        // Execute again
235        let mut ctx = DropDatabaseContext {
236            catalog: DEFAULT_CATALOG_NAME.to_string(),
237            schema: DEFAULT_SCHEMA_NAME.to_string(),
238            drop_if_exists: false,
239            tables: None,
240            retrying: false,
241        };
242        let mut state = DropDatabaseExecutor::new(
243            physical_table_id,
244            physical_table_id,
245            TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "phy"),
246            table_route.region_routes.clone(),
247            DropTableTarget::Physical,
248        );
249        let (state, status) = state.next(&ddl_context, &mut ctx).await.unwrap();
250        assert!(!status.need_persist());
251        let cursor = state.as_any().downcast_ref::<DropDatabaseCursor>().unwrap();
252        assert_eq!(cursor.target, DropTableTarget::Physical);
253    }
254
255    #[tokio::test]
256    async fn test_next_logical_table() {
257        let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
258        let ddl_context = new_ddl_context(node_manager);
259        let physical_table_id = create_physical_table(&ddl_context, "phy").await;
260        create_logical_table(ddl_context.clone(), physical_table_id, "metric").await;
261        let logical_table_id = physical_table_id + 1;
262        let (_, table_route) = ddl_context
263            .table_metadata_manager
264            .table_route_manager()
265            .get_physical_table_route(logical_table_id)
266            .await
267            .unwrap();
268        {
269            let mut state = DropDatabaseExecutor::new(
270                logical_table_id,
271                physical_table_id,
272                TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "metric"),
273                table_route.region_routes.clone(),
274                DropTableTarget::Logical,
275            );
276            let mut ctx = DropDatabaseContext {
277                catalog: DEFAULT_CATALOG_NAME.to_string(),
278                schema: DEFAULT_SCHEMA_NAME.to_string(),
279                drop_if_exists: false,
280                tables: None,
281                retrying: false,
282            };
283            let (state, status) = state.next(&ddl_context, &mut ctx).await.unwrap();
284            assert!(!status.need_persist());
285            let cursor = state.as_any().downcast_ref::<DropDatabaseCursor>().unwrap();
286            assert_eq!(cursor.target, DropTableTarget::Logical);
287        }
288        // Execute again
289        let mut ctx = DropDatabaseContext {
290            catalog: DEFAULT_CATALOG_NAME.to_string(),
291            schema: DEFAULT_SCHEMA_NAME.to_string(),
292            drop_if_exists: false,
293            tables: None,
294            retrying: false,
295        };
296        let mut state = DropDatabaseExecutor::new(
297            logical_table_id,
298            physical_table_id,
299            TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "metric"),
300            table_route.region_routes,
301            DropTableTarget::Logical,
302        );
303        let (state, status) = state.next(&ddl_context, &mut ctx).await.unwrap();
304        assert!(!status.need_persist());
305        let cursor = state.as_any().downcast_ref::<DropDatabaseCursor>().unwrap();
306        assert_eq!(cursor.target, DropTableTarget::Logical);
307        // Checks table info
308        ddl_context
309            .table_metadata_manager
310            .table_info_manager()
311            .get(physical_table_id)
312            .await
313            .unwrap()
314            .unwrap();
315        // Checks table route
316        let table_route = ddl_context
317            .table_metadata_manager
318            .table_route_manager()
319            .table_route_storage()
320            .get(physical_table_id)
321            .await
322            .unwrap()
323            .unwrap();
324        let region_routes = table_route.region_routes().unwrap();
325        for datanode_id in region_distribution(region_routes).into_keys() {
326            ddl_context
327                .table_metadata_manager
328                .datanode_table_manager()
329                .get(&DatanodeTableKey::new(datanode_id, physical_table_id))
330                .await
331                .unwrap()
332                .unwrap();
333        }
334    }
335
336    #[derive(Clone)]
337    pub struct RetryErrorDatanodeHandler;
338
339    #[async_trait::async_trait]
340    impl MockDatanodeHandler for RetryErrorDatanodeHandler {
341        async fn handle(&self, _peer: &Peer, _request: RegionRequest) -> Result<RegionResponse> {
342            Err(Error::RetryLater {
343                source: BoxedError::new(
344                    error::UnexpectedSnafu {
345                        err_msg: "retry later",
346                    }
347                    .build(),
348                ),
349                clean_poisons: false,
350            })
351        }
352
353        async fn handle_query(
354            &self,
355            _peer: &Peer,
356            _request: QueryRequest,
357        ) -> Result<SendableRecordBatchStream> {
358            unreachable!()
359        }
360    }
361
362    #[tokio::test]
363    async fn test_next_retryable_err() {
364        let node_manager = Arc::new(MockDatanodeManager::new(RetryErrorDatanodeHandler));
365        let ddl_context = new_ddl_context(node_manager);
366        let physical_table_id = create_physical_table(&ddl_context, "phy").await;
367        let (_, table_route) = ddl_context
368            .table_metadata_manager
369            .table_route_manager()
370            .get_physical_table_route(physical_table_id)
371            .await
372            .unwrap();
373        let mut state = DropDatabaseExecutor::new(
374            physical_table_id,
375            physical_table_id,
376            TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "phy"),
377            table_route.region_routes,
378            DropTableTarget::Physical,
379        );
380        let mut ctx = DropDatabaseContext {
381            catalog: DEFAULT_CATALOG_NAME.to_string(),
382            schema: DEFAULT_SCHEMA_NAME.to_string(),
383            drop_if_exists: false,
384            tables: None,
385            retrying: false,
386        };
387        let err = state.next(&ddl_context, &mut ctx).await.unwrap_err();
388        assert!(err.is_retry_later());
389    }
390
391    #[tokio::test]
392    async fn test_on_recovery() {
393        let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler));
394        let ddl_context = new_ddl_context(node_manager);
395        let physical_table_id = create_physical_table(&ddl_context, "phy").await;
396        let (_, table_route) = ddl_context
397            .table_metadata_manager
398            .table_route_manager()
399            .get_physical_table_route(physical_table_id)
400            .await
401            .unwrap();
402        {
403            let mut state = DropDatabaseExecutor::new(
404                physical_table_id,
405                physical_table_id,
406                TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "phy"),
407                table_route.region_routes.clone(),
408                DropTableTarget::Physical,
409            );
410            let mut ctx = DropDatabaseContext {
411                catalog: DEFAULT_CATALOG_NAME.to_string(),
412                schema: DEFAULT_SCHEMA_NAME.to_string(),
413                drop_if_exists: false,
414                tables: None,
415                retrying: false,
416            };
417            state.recover(&ddl_context).unwrap();
418            assert_eq!(state.dropping_regions.len(), 1);
419            let (state, status) = state.next(&ddl_context, &mut ctx).await.unwrap();
420            assert!(!status.need_persist());
421            let cursor = state.as_any().downcast_ref::<DropDatabaseCursor>().unwrap();
422            assert_eq!(cursor.target, DropTableTarget::Physical);
423        }
424    }
425
426    #[tokio::test]
427    async fn test_next_remaps_addresses_when_retrying() {
428        let (tx, mut rx) = tokio::sync::mpsc::channel(8);
429        let node_manager = Arc::new(MockDatanodeManager::new(DatanodeWatcher::new(tx)));
430        let ddl_context = new_ddl_context(node_manager);
431        let physical_table_id = create_physical_table(&ddl_context, "phy").await;
432        let (_, table_route) = ddl_context
433            .table_metadata_manager
434            .table_route_manager()
435            .get_physical_table_route(physical_table_id)
436            .await
437            .unwrap();
438
439        let mut state = DropDatabaseExecutor::new(
440            physical_table_id,
441            physical_table_id,
442            TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "phy"),
443            table_route.region_routes,
444            DropTableTarget::Physical,
445        );
446        state.physical_region_routes[0]
447            .leader_peer
448            .as_mut()
449            .unwrap()
450            .addr = "old-addr".to_string();
451        let mut ctx = DropDatabaseContext {
452            catalog: DEFAULT_CATALOG_NAME.to_string(),
453            schema: DEFAULT_SCHEMA_NAME.to_string(),
454            drop_if_exists: false,
455            tables: None,
456            retrying: true,
457        };
458
459        put_datanode_address(&ddl_context, 0, "new-addr").await;
460
461        state.next(&ddl_context, &mut ctx).await.unwrap();
462
463        let (peer, _) = rx.try_recv().unwrap();
464        assert_eq!(peer.addr, "new-addr");
465    }
466}
common_meta/ddl/drop_database/executor.rs

common_meta/ddl/drop_database/
executor.rs
