common_meta/ddl/drop_table/
executor.rs1use 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#[derive(Debug)]
46pub enum Control<T> {
47 Continue(T),
48 Stop,
49}
50
51impl<T> Control<T> {
52 pub fn stop(&self) -> bool {
54 matches!(self, Control::Stop)
55 }
56}
57
58impl DropTableExecutor {
59 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
69pub struct DropTableExecutor {
74 table: TableName,
75 table_id: TableId,
76 drop_if_exists: bool,
77}
78
79impl DropTableExecutor {
80 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 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 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 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 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 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 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 pub async fn on_drop_regions(
210 &self,
211 ctx: &DdlContext,
212 region_routes: &[RegionRoute],
213 fast_path: bool,
214 ) -> Result<()> {
215 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 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 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 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 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 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 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}