common_meta/ddl/drop_table/
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::collections::HashMap;
16
17use api::v1::region::{
18    region_request, CloseRequest as PbCloseRegionRequest, DropRequest as PbDropRegionRequest,
19    RegionRequest, RegionRequestHeader,
20};
21use common_error::ext::ErrorExt;
22use common_error::status_code::StatusCode;
23use common_telemetry::tracing_context::TracingContext;
24use common_telemetry::{debug, error};
25use common_wal::options::WalOptions;
26use futures::future::join_all;
27use snafu::ensure;
28use store_api::storage::{RegionId, RegionNumber};
29use table::metadata::TableId;
30use table::table_name::TableName;
31
32use crate::cache_invalidator::Context;
33use crate::ddl::utils::{add_peer_context_if_needed, convert_region_routes_to_detecting_regions};
34use crate::ddl::DdlContext;
35use crate::error::{self, Result};
36use crate::instruction::CacheIdent;
37use crate::key::table_name::TableNameKey;
38use crate::key::table_route::TableRouteValue;
39use crate::rpc::router::{
40    find_follower_regions, find_followers, find_leader_regions, find_leaders,
41    operating_leader_regions, RegionRoute,
42};
43
44/// [Control] indicated to the caller whether to go to the next step.
45#[derive(Debug)]
46pub enum Control<T> {
47    Continue(T),
48    Stop,
49}
50
51impl<T> Control<T> {
52    /// Returns true if it's [Control::Stop].
53    pub fn stop(&self) -> bool {
54        matches!(self, Control::Stop)
55    }
56}
57
58impl DropTableExecutor {
59    /// Returns the [DropTableExecutor].
60    pub fn new(table: TableName, table_id: TableId, drop_if_exists: bool) -> Self {
61        Self {
62            table,
63            table_id,
64            drop_if_exists,
65        }
66    }
67}
68
69/// [DropTableExecutor] performs:
70/// - Drops the metadata of the table.
71/// - Invalidates the cache on the Frontend nodes.
72/// - Drops the regions on the Datanode nodes.
73pub struct DropTableExecutor {
74    table: TableName,
75    table_id: TableId,
76    drop_if_exists: bool,
77}
78
79impl DropTableExecutor {
80    /// Checks whether table exists.
81    /// - Early returns if table not exists and `drop_if_exists` is `true`.
82    /// - Throws an error if table not exists and `drop_if_exists` is `false`.
83    pub async fn on_prepare(&self, ctx: &DdlContext) -> Result<Control<()>> {
84        let table_ref = self.table.table_ref();
85
86        let exist = ctx
87            .table_metadata_manager
88            .table_name_manager()
89            .exists(TableNameKey::new(
90                table_ref.catalog,
91                table_ref.schema,
92                table_ref.table,
93            ))
94            .await?;
95
96        if !exist && self.drop_if_exists {
97            return Ok(Control::Stop);
98        }
99
100        ensure!(
101            exist,
102            error::TableNotFoundSnafu {
103                table_name: table_ref.to_string()
104            }
105        );
106
107        Ok(Control::Continue(()))
108    }
109
110    /// Deletes the table metadata **logically**.
111    pub async fn on_delete_metadata(
112        &self,
113        ctx: &DdlContext,
114        table_route_value: &TableRouteValue,
115        region_wal_options: &HashMap<RegionNumber, WalOptions>,
116    ) -> Result<()> {
117        ctx.table_metadata_manager
118            .delete_table_metadata(
119                self.table_id,
120                &self.table,
121                table_route_value,
122                region_wal_options,
123            )
124            .await
125    }
126
127    /// Deletes the table metadata tombstone **permanently**.
128    pub async fn on_delete_metadata_tombstone(
129        &self,
130        ctx: &DdlContext,
131        table_route_value: &TableRouteValue,
132        region_wal_options: &HashMap<u32, WalOptions>,
133    ) -> Result<()> {
134        ctx.table_metadata_manager
135            .delete_table_metadata_tombstone(
136                self.table_id,
137                &self.table,
138                table_route_value,
139                region_wal_options,
140            )
141            .await
142    }
143
144    /// Deletes metadata for table **permanently**.
145    pub async fn on_destroy_metadata(
146        &self,
147        ctx: &DdlContext,
148        table_route_value: &TableRouteValue,
149        region_wal_options: &HashMap<u32, WalOptions>,
150    ) -> Result<()> {
151        ctx.table_metadata_manager
152            .destroy_table_metadata(
153                self.table_id,
154                &self.table,
155                table_route_value,
156                region_wal_options,
157            )
158            .await?;
159
160        let detecting_regions = if table_route_value.is_physical() {
161            // Safety: checked.
162            let regions = table_route_value.region_routes().unwrap();
163            convert_region_routes_to_detecting_regions(regions)
164        } else {
165            vec![]
166        };
167        ctx.deregister_failure_detectors(detecting_regions).await;
168        Ok(())
169    }
170
171    /// Restores the table metadata.
172    pub async fn on_restore_metadata(
173        &self,
174        ctx: &DdlContext,
175        table_route_value: &TableRouteValue,
176        region_wal_options: &HashMap<u32, WalOptions>,
177    ) -> Result<()> {
178        ctx.table_metadata_manager
179            .restore_table_metadata(
180                self.table_id,
181                &self.table,
182                table_route_value,
183                region_wal_options,
184            )
185            .await
186    }
187
188    /// Invalidates frontend caches
189    pub async fn invalidate_table_cache(&self, ctx: &DdlContext) -> Result<()> {
190        let cache_invalidator = &ctx.cache_invalidator;
191        let ctx = Context {
192            subject: Some("Invalidate table cache by dropping table".to_string()),
193        };
194
195        cache_invalidator
196            .invalidate(
197                &ctx,
198                &[
199                    CacheIdent::TableName(self.table.table_ref().into()),
200                    CacheIdent::TableId(self.table_id),
201                ],
202            )
203            .await?;
204
205        Ok(())
206    }
207
208    /// Drops region on datanode.
209    pub async fn on_drop_regions(
210        &self,
211        ctx: &DdlContext,
212        region_routes: &[RegionRoute],
213        fast_path: bool,
214    ) -> Result<()> {
215        // Drops leader regions on datanodes.
216        let leaders = find_leaders(region_routes);
217        let mut drop_region_tasks = Vec::with_capacity(leaders.len());
218        let table_id = self.table_id;
219        for datanode in leaders {
220            let requester = ctx.node_manager.datanode(&datanode).await;
221            let regions = find_leader_regions(region_routes, &datanode);
222            let region_ids = regions
223                .iter()
224                .map(|region_number| RegionId::new(table_id, *region_number))
225                .collect::<Vec<_>>();
226
227            for region_id in region_ids {
228                debug!("Dropping region {region_id} on Datanode {datanode:?}");
229                let request = RegionRequest {
230                    header: Some(RegionRequestHeader {
231                        tracing_context: TracingContext::from_current_span().to_w3c(),
232                        ..Default::default()
233                    }),
234                    body: Some(region_request::Body::Drop(PbDropRegionRequest {
235                        region_id: region_id.as_u64(),
236                        fast_path,
237                    })),
238                };
239                let datanode = datanode.clone();
240                let requester = requester.clone();
241                drop_region_tasks.push(async move {
242                    if let Err(err) = requester.handle(request).await {
243                        if err.status_code() != StatusCode::RegionNotFound {
244                            return Err(add_peer_context_if_needed(datanode)(err));
245                        }
246                    }
247                    Ok(())
248                });
249            }
250        }
251
252        join_all(drop_region_tasks)
253            .await
254            .into_iter()
255            .collect::<Result<Vec<_>>>()?;
256
257        // Drops follower regions on datanodes.
258        let followers = find_followers(region_routes);
259        let mut close_region_tasks = Vec::with_capacity(followers.len());
260        for datanode in followers {
261            let requester = ctx.node_manager.datanode(&datanode).await;
262            let regions = find_follower_regions(region_routes, &datanode);
263            let region_ids = regions
264                .iter()
265                .map(|region_number| RegionId::new(table_id, *region_number))
266                .collect::<Vec<_>>();
267
268            for region_id in region_ids {
269                debug!("Closing region {region_id} on Datanode {datanode:?}");
270                let request = RegionRequest {
271                    header: Some(RegionRequestHeader {
272                        tracing_context: TracingContext::from_current_span().to_w3c(),
273                        ..Default::default()
274                    }),
275                    body: Some(region_request::Body::Close(PbCloseRegionRequest {
276                        region_id: region_id.as_u64(),
277                    })),
278                };
279
280                let datanode = datanode.clone();
281                let requester = requester.clone();
282                close_region_tasks.push(async move {
283                    if let Err(err) = requester.handle(request).await {
284                        if err.status_code() != StatusCode::RegionNotFound {
285                            return Err(add_peer_context_if_needed(datanode)(err));
286                        }
287                    }
288                    Ok(())
289                });
290            }
291        }
292
293        // Failure to close follower regions is not critical.
294        // When a leader region is dropped, follower regions will be unable to renew their leases via metasrv.
295        // Eventually, these follower regions will be automatically closed by the region livekeeper.
296        if let Err(err) = join_all(close_region_tasks)
297            .await
298            .into_iter()
299            .collect::<Result<Vec<_>>>()
300        {
301            error!(err; "Failed to close follower regions on datanodes, table_id: {}", table_id);
302        }
303
304        // Deletes the leader region from registry.
305        let region_ids = operating_leader_regions(region_routes);
306        ctx.leader_region_registry
307            .batch_delete(region_ids.into_iter().map(|(region_id, _)| region_id));
308
309        Ok(())
310    }
311}
312
313#[cfg(test)]
314mod tests {
315    use std::assert_matches::assert_matches;
316    use std::collections::HashMap;
317    use std::sync::Arc;
318
319    use api::v1::{ColumnDataType, SemanticType};
320    use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
321    use table::metadata::RawTableInfo;
322    use table::table_name::TableName;
323
324    use super::*;
325    use crate::ddl::test_util::columns::TestColumnDefBuilder;
326    use crate::ddl::test_util::create_table::{
327        build_raw_table_info_from_expr, TestCreateTableExprBuilder,
328    };
329    use crate::key::table_route::TableRouteValue;
330    use crate::test_util::{new_ddl_context, MockDatanodeManager};
331
332    fn test_create_raw_table_info(name: &str) -> RawTableInfo {
333        let create_table = TestCreateTableExprBuilder::default()
334            .column_defs([
335                TestColumnDefBuilder::default()
336                    .name("ts")
337                    .data_type(ColumnDataType::TimestampMillisecond)
338                    .semantic_type(SemanticType::Timestamp)
339                    .build()
340                    .unwrap()
341                    .into(),
342                TestColumnDefBuilder::default()
343                    .name("host")
344                    .data_type(ColumnDataType::String)
345                    .semantic_type(SemanticType::Tag)
346                    .build()
347                    .unwrap()
348                    .into(),
349                TestColumnDefBuilder::default()
350                    .name("cpu")
351                    .data_type(ColumnDataType::Float64)
352                    .semantic_type(SemanticType::Field)
353                    .build()
354                    .unwrap()
355                    .into(),
356            ])
357            .time_index("ts")
358            .primary_keys(["host".into()])
359            .table_name(name)
360            .build()
361            .unwrap()
362            .into();
363        build_raw_table_info_from_expr(&create_table)
364    }
365
366    #[tokio::test]
367    async fn test_on_prepare() {
368        // Drops if exists
369        let node_manager = Arc::new(MockDatanodeManager::new(()));
370        let ctx = new_ddl_context(node_manager);
371        let executor = DropTableExecutor::new(
372            TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_table"),
373            1024,
374            true,
375        );
376        let ctrl = executor.on_prepare(&ctx).await.unwrap();
377        assert!(ctrl.stop());
378
379        // Drops a non-exists table
380        let executor = DropTableExecutor::new(
381            TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_table"),
382            1024,
383            false,
384        );
385        let err = executor.on_prepare(&ctx).await.unwrap_err();
386        assert_matches!(err, error::Error::TableNotFound { .. });
387
388        // Drops a exists table
389        let executor = DropTableExecutor::new(
390            TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, "my_table"),
391            1024,
392            false,
393        );
394        let raw_table_info = test_create_raw_table_info("my_table");
395        ctx.table_metadata_manager
396            .create_table_metadata(
397                raw_table_info,
398                TableRouteValue::physical(vec![]),
399                HashMap::new(),
400            )
401            .await
402            .unwrap();
403        let ctrl = executor.on_prepare(&ctx).await.unwrap();
404        assert!(!ctrl.stop());
405    }
406}