diff --git a/src/common/meta/src/ddl/tests/drop_table.rs b/src/common/meta/src/ddl/tests/drop_table.rs index ddc8d7edc0..6fc4723c0e 100644 --- a/src/common/meta/src/ddl/tests/drop_table.rs +++ b/src/common/meta/src/ddl/tests/drop_table.rs @@ -243,8 +243,10 @@ async fn test_soft_drop_closes_regions_and_keeps_tombstone() { let (tx, mut rx) = mpsc::channel(8); let datanode_handler = DatanodeWatcher::new(tx); let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler)); + let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default()); let mut ddl_context = new_ddl_context(node_manager); ddl_context.soft_drop_enabled = true; + ddl_context.region_failure_detector_controller = detector_controller.clone(); let table_id = 1024; let table_name = "foo"; let task = test_create_table_task(table_name, table_id); @@ -318,44 +320,13 @@ async fn test_soft_drop_closes_regions_and_keeps_tombstone() { .await .unwrap(); assert!(dropped_table.is_some()); -} - -#[tokio::test] -async fn test_soft_drop_deregisters_failure_detectors() { - let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); - let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default()); - let mut ddl_context = new_ddl_context(node_manager); - ddl_context.soft_drop_enabled = true; - ddl_context.region_failure_detector_controller = detector_controller.clone(); - let table_id = 1024; - let table_name = "foo"; - let task = test_create_table_task(table_name, table_id); - ddl_context - .table_metadata_manager - .create_table_metadata( - task.table_info.clone(), - TableRouteValue::physical(vec![RegionRoute { - region: Region::new_test(RegionId::new(table_id, 1)), - leader_peer: Some(Peer::empty(1)), - follower_peers: vec![Peer::empty(2)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }]), - HashMap::new(), - ) - .await - .unwrap(); - - let mut procedure = DropTableProcedure::new( - new_drop_table_task(table_name, table_id, false), - ddl_context, - ); - execute_procedure_until_done(&mut procedure).await; assert_eq!( detector_controller.deregistered().await, - vec![(1, RegionId::new(table_id, 1))] + vec![ + (1, RegionId::new(table_id, 1)), + (2, RegionId::new(table_id, 2)) + ] ); } @@ -542,8 +513,10 @@ async fn test_undrop_table_restores_metadata_and_reopens_regions() { let (tx, mut rx) = mpsc::channel(8); let datanode_handler = DatanodeWatcher::new(tx); let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler)); + let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default()); let mut ddl_context = new_ddl_context(node_manager); ddl_context.soft_drop_enabled = true; + ddl_context.region_failure_detector_controller = detector_controller.clone(); let table_id = 1024; let table_name = "foo"; let task = test_create_table_task(table_name, table_id); @@ -569,6 +542,7 @@ async fn test_undrop_table_restores_metadata_and_reopens_regions() { ddl_context.clone(), ); execute_procedure_until_done(&mut drop_procedure).await; + detector_controller.clear().await; while rx.try_recv().is_ok() {} @@ -606,44 +580,6 @@ async fn test_undrop_table_restores_metadata_and_reopens_regions() { }; assert_eq!(req.region_id, RegionId::new(table_id, 1).as_u64()); assert!(rx.try_recv().is_err()); -} - -#[tokio::test] -async fn test_undrop_table_registers_failure_detectors() { - let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); - let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default()); - let mut ddl_context = new_ddl_context(node_manager); - ddl_context.soft_drop_enabled = true; - ddl_context.region_failure_detector_controller = detector_controller.clone(); - let table_id = 1024; - let table_name = "foo"; - let task = test_create_table_task(table_name, table_id); - ddl_context - .table_metadata_manager - .create_table_metadata( - task.table_info.clone(), - TableRouteValue::physical(vec![RegionRoute { - region: Region::new_test(RegionId::new(table_id, 1)), - leader_peer: Some(Peer::empty(1)), - follower_peers: vec![], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }]), - HashMap::new(), - ) - .await - .unwrap(); - let mut drop_procedure = DropTableProcedure::new( - new_drop_table_task(table_name, table_id, false), - ddl_context.clone(), - ); - execute_procedure_until_done(&mut drop_procedure).await; - detector_controller.clear().await; - - let mut procedure = - UndropTableProcedure::new(new_undrop_table_task(table_name, table_id), ddl_context); - execute_procedure_until_done(&mut procedure).await; assert_eq!( detector_controller.registered().await, @@ -817,8 +753,10 @@ async fn test_purge_dropped_table_drops_regions_and_deletes_tombstone() { let (tx, mut rx) = mpsc::channel(8); let datanode_handler = DatanodeWatcher::new(tx); let node_manager = Arc::new(MockDatanodeManager::new(datanode_handler)); + let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default()); let mut ddl_context = new_ddl_context(node_manager); ddl_context.soft_drop_enabled = true; + ddl_context.region_failure_detector_controller = detector_controller.clone(); let table_id = 1024; let table_name = "foo"; let task = test_create_table_task(table_name, table_id); @@ -844,6 +782,7 @@ async fn test_purge_dropped_table_drops_regions_and_deletes_tombstone() { ); execute_procedure_until_done(&mut drop_procedure).await; while rx.try_recv().is_ok() {} + detector_controller.clear().await; let mut procedure = PurgeDroppedTableProcedure::new( new_purge_dropped_table_task(table_name, Some(table_id)), @@ -868,50 +807,6 @@ async fn test_purge_dropped_table_drops_regions_and_deletes_tombstone() { .unwrap() .is_none() ); -} - -#[tokio::test] -async fn test_purge_dropped_table_deregisters_failure_detectors() { - let node_manager = Arc::new(MockDatanodeManager::new(NaiveDatanodeHandler)); - let detector_controller = Arc::new(RecordingRegionFailureDetectorController::default()); - let mut ddl_context = new_ddl_context(node_manager); - ddl_context.region_failure_detector_controller = detector_controller.clone(); - let table_id = 1024; - let table_name = "foo"; - let task = test_create_table_task(table_name, table_id); - let table_route_value = TableRouteValue::physical(vec![RegionRoute { - region: Region::new_test(RegionId::new(table_id, 1)), - leader_peer: Some(Peer::empty(1)), - follower_peers: vec![], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }]); - ddl_context - .table_metadata_manager - .create_table_metadata( - task.table_info.clone(), - table_route_value.clone(), - HashMap::new(), - ) - .await - .unwrap(); - ddl_context - .table_metadata_manager - .delete_table_metadata( - table_id, - &task.table_name(), - &table_route_value, - &HashMap::new(), - ) - .await - .unwrap(); - - let mut procedure = PurgeDroppedTableProcedure::new( - new_purge_dropped_table_task(table_name, Some(table_id)), - ddl_context, - ); - execute_procedure_until_done(&mut procedure).await; assert_eq!( detector_controller.deregistered().await, diff --git a/src/common/meta/src/key.rs b/src/common/meta/src/key.rs index 4f52822d8c..6fa160e432 100644 --- a/src/common/meta/src/key.rs +++ b/src/common/meta/src/key.rs @@ -1640,7 +1640,7 @@ mod tests { use common_wal::options::{KafkaWalOptions, WalOptions}; use futures::TryStreamExt; use store_api::storage::{RegionId, RegionNumber}; - use table::metadata::TableInfo; + use table::metadata::{TableId, TableInfo}; use table::table_name::TableName; use super::datanode_table::DatanodeTableKey; @@ -1648,7 +1648,7 @@ mod tests { use crate::ddl::allocator::wal_options::WalOptionsAllocator; use crate::ddl::test_util::create_table::test_create_table_task; use crate::ddl::utils::region_storage_path; - use crate::error::Result; + use crate::error::{Error, Result}; use crate::key::datanode_table::RegionInfo; use crate::key::node_address::{NodeAddressKey, NodeAddressValue}; use crate::key::table_info::TableInfoValue; @@ -1656,8 +1656,8 @@ mod tests { use crate::key::table_route::TableRouteValue; use crate::key::topic_region::TopicRegionKey; use crate::key::{ - DeserializedValueWithBytes, MetadataValue, RegionDistribution, RegionRoleSet, - TOPIC_REGION_PREFIX, TableMetadataManager, ViewInfoValue, + DeserializedValueWithBytes, DroppedTableMetadata, MetadataValue, RegionDistribution, + RegionRoleSet, TOPIC_REGION_PREFIX, TableMetadataManager, ViewInfoValue, }; use crate::kv_backend::KvBackend; use crate::kv_backend::memory::MemoryKvBackend; @@ -1779,6 +1779,100 @@ mod tests { ]) } + fn test_physical_region_route( + table_id: TableId, + region_number: RegionNumber, + leader_peer: u64, + follower_peers: Vec, + ) -> RegionRoute { + RegionRoute { + region: Region::new_test(RegionId::new(table_id, region_number)), + leader_peer: Some(Peer::empty(leader_peer)), + follower_peers: follower_peers.into_iter().map(Peer::empty).collect(), + leader_state: None, + leader_down_since: None, + write_route_policy: None, + } + } + + async fn create_dropped_physical_table_metadata( + table_id: TableId, + table_name: &str, + region_routes: Vec, + region_wal_options: HashMap, + ) -> ( + Arc>, + TableMetadataManager, + TableName, + TableInfo, + Vec, + HashMap, + ) { + let mem_kv = Arc::new(MemoryKvBackend::default()); + let table_metadata_manager = TableMetadataManager::new(mem_kv.clone()); + let task = test_create_table_task(table_name, table_id); + let table_info = task.table_info.clone(); + let serialized_options = region_wal_options + .iter() + .map(|(k, v)| (*k, serde_json::to_string(v).unwrap())) + .collect::>(); + table_metadata_manager + .create_table_metadata( + table_info.clone(), + TableRouteValue::physical(region_routes), + serialized_options, + ) + .await + .unwrap(); + + let table_route_value = table_metadata_manager + .table_route_manager + .table_route_storage() + .get_with_raw_bytes(table_id) + .await + .unwrap() + .unwrap(); + let region_routes = table_route_value.region_routes().unwrap().clone(); + let table_name = TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, table_name); + let table_route_value = TableRouteValue::physical(region_routes.clone()); + table_metadata_manager + .delete_table_metadata( + table_id, + &table_name, + &table_route_value, + ®ion_wal_options, + ) + .await + .unwrap(); + + ( + mem_kv, + table_metadata_manager, + table_name, + table_info, + region_routes, + region_wal_options, + ) + } + + fn assert_dropped_table_metadata( + dropped_table: &DroppedTableMetadata, + table_id: TableId, + table_name: &TableName, + table_info: &TableInfo, + region_routes: &[RegionRoute], + region_wal_options: &HashMap, + ) { + assert_eq!(dropped_table.table_id, table_id); + assert_eq!(&dropped_table.table_name, table_name); + assert_eq!(&dropped_table.table_info_value.table_info, table_info); + assert_eq!( + dropped_table.table_route_value.region_routes().unwrap(), + region_routes + ); + assert_eq!(&dropped_table.region_wal_options, region_wal_options); + } + #[tokio::test] async fn test_raft_engine_topic_region_map() { let mem_kv = Arc::new(MemoryKvBackend::default()); @@ -2857,62 +2951,20 @@ mod tests { #[tokio::test] async fn test_dropped_table_metadata_enumeration_and_lookup() { - let mem_kv = Arc::new(MemoryKvBackend::default()); - let table_metadata_manager = TableMetadataManager::new(mem_kv.clone()); let table_id = 1025; let table_name = "foo"; - let task = test_create_table_task(table_name, table_id); - let table_info = task.table_info.clone(); - let options = create_mixed_region_wal_options(); - let serialized_options = options.clone(); - table_metadata_manager - .create_table_metadata( - table_info.clone(), - TableRouteValue::physical(vec![ - RegionRoute { - region: Region::new_test(RegionId::new(table_id, 1)), - leader_peer: Some(Peer::empty(1)), - follower_peers: vec![Peer::empty(5)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - RegionRoute { - region: Region::new_test(RegionId::new(table_id, 2)), - leader_peer: Some(Peer::empty(2)), - follower_peers: vec![Peer::empty(4)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - RegionRoute { - region: Region::new_test(RegionId::new(table_id, 3)), - leader_peer: Some(Peer::empty(3)), - follower_peers: vec![], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - ]), - serialized_options, + let (_, table_metadata_manager, table_name, table_info, region_routes, options) = + create_dropped_physical_table_metadata( + table_id, + table_name, + vec![ + test_physical_region_route(table_id, 1, 1, vec![5]), + test_physical_region_route(table_id, 2, 2, vec![4]), + test_physical_region_route(table_id, 3, 3, vec![]), + ], + create_mixed_region_wal_options(), ) - .await - .unwrap(); - let table_route_value = table_metadata_manager - .table_route_manager - .table_route_storage() - .get_with_raw_bytes(table_id) - .await - .unwrap() - .unwrap(); - let region_routes = table_route_value.region_routes().unwrap(); - let table_name = TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, table_name); - let table_route_value = TableRouteValue::physical(region_routes.clone()); - - table_metadata_manager - .delete_table_metadata(table_id, &table_name, &table_route_value, &options) - .await - .unwrap(); + .await; let dropped_tables = table_metadata_manager.list_dropped_tables().await.unwrap(); assert_eq!(dropped_tables.len(), 1); @@ -2924,104 +2976,64 @@ mod tests { .await .unwrap() .unwrap(); - assert_eq!(dropped_table.table_id, table_id); - assert_eq!(dropped_table.table_name, table_name); - assert_eq!(dropped_table.table_info_value.table_info, table_info); - assert_eq!( - dropped_table.table_route_value.region_routes().unwrap(), - region_routes + assert_dropped_table_metadata( + &dropped_table, + table_id, + &table_name, + &table_info, + ®ion_routes, + &options, ); - assert_eq!(dropped_table.region_wal_options, options); let dropped_table_by_id = table_metadata_manager .get_dropped_table_by_id(table_id) .await .unwrap() .unwrap(); - assert_eq!(dropped_table_by_id.table_id, table_id); - assert_eq!(dropped_table_by_id.table_name, table_name); - assert_eq!(dropped_table_by_id.table_info_value.table_info, table_info); - assert_eq!( - dropped_table_by_id - .table_route_value - .region_routes() - .unwrap(), - region_routes + assert_dropped_table_metadata( + &dropped_table_by_id, + table_id, + &table_name, + &table_info, + ®ion_routes, + &options, ); - assert_eq!(dropped_table_by_id.region_wal_options, options); } #[tokio::test] async fn test_dropped_table_lookup_survives_live_name_recreation() { - let mem_kv = Arc::new(MemoryKvBackend::default()); - let table_metadata_manager = TableMetadataManager::new(mem_kv.clone()); let dropped_table_id = 1025; let recreated_table_id = 1026; let table_name = "foo"; - let dropped_task = test_create_table_task(table_name, dropped_table_id); - let dropped_table_info = dropped_task.table_info.clone(); - let options = create_mock_region_wal_options(); - let serialized_options = options.clone(); - table_metadata_manager - .create_table_metadata( - dropped_table_info.clone(), - TableRouteValue::physical(vec![ - RegionRoute { - region: Region::new_test(RegionId::new(dropped_table_id, 1)), - leader_peer: Some(Peer::empty(1)), - follower_peers: vec![Peer::empty(5)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - RegionRoute { - region: Region::new_test(RegionId::new(dropped_table_id, 2)), - leader_peer: Some(Peer::empty(2)), - follower_peers: vec![Peer::empty(4)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - ]), - serialized_options.clone(), - ) - .await - .unwrap(); - - let dropped_table_name = - TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, table_name); - let dropped_table_route = table_metadata_manager - .table_route_manager - .table_route_storage() - .get_with_raw_bytes(dropped_table_id) - .await - .unwrap() - .unwrap(); - let dropped_table_route = - TableRouteValue::physical(dropped_table_route.region_routes().unwrap().clone()); - table_metadata_manager - .delete_table_metadata( - dropped_table_id, - &dropped_table_name, - &dropped_table_route, - &options, - ) - .await - .unwrap(); + let ( + _, + table_metadata_manager, + dropped_table_name, + dropped_table_info, + region_routes, + options, + ) = create_dropped_physical_table_metadata( + dropped_table_id, + table_name, + vec![ + test_physical_region_route(dropped_table_id, 1, 1, vec![5]), + test_physical_region_route(dropped_table_id, 2, 2, vec![4]), + ], + create_mock_region_wal_options(), + ) + .await; let recreated_task = test_create_table_task(table_name, recreated_table_id); table_metadata_manager .create_table_metadata( recreated_task.table_info, - TableRouteValue::physical(vec![RegionRoute { - region: Region::new_test(RegionId::new(recreated_table_id, 1)), - leader_peer: Some(Peer::empty(4)), - follower_peers: vec![], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }]), - serialized_options, + TableRouteValue::physical(vec![test_physical_region_route( + recreated_table_id, + 1, + 4, + vec![], + )]), + HashMap::new(), ) .await .unwrap(); @@ -3042,11 +3054,13 @@ mod tests { .await .unwrap() .unwrap(); - assert_eq!(dropped_table.table_id, dropped_table_id); - assert_eq!(dropped_table.table_name, dropped_table_name); - assert_eq!( - dropped_table.table_info_value.table_info, - dropped_table_info + assert_dropped_table_metadata( + &dropped_table, + dropped_table_id, + &dropped_table_name, + &dropped_table_info, + ®ion_routes, + &options, ); let dropped_tables = table_metadata_manager.list_dropped_tables().await.unwrap(); @@ -3057,55 +3071,19 @@ mod tests { #[tokio::test] async fn test_dropped_table_lookup_ignores_unrelated_malformed_datanode_tombstones() { - let mem_kv = Arc::new(MemoryKvBackend::default()); - let table_metadata_manager = TableMetadataManager::new(mem_kv.clone()); let table_id = 1025; let table_name = "foo"; - let task = test_create_table_task(table_name, table_id); - let table_info = task.table_info.clone(); - let options = create_mixed_region_wal_options(); - let serialized_options = options.clone(); - table_metadata_manager - .create_table_metadata( - table_info.clone(), - TableRouteValue::physical(vec![ - RegionRoute { - region: Region::new_test(RegionId::new(table_id, 1)), - leader_peer: Some(Peer::empty(1)), - follower_peers: vec![Peer::empty(5)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - RegionRoute { - region: Region::new_test(RegionId::new(table_id, 2)), - leader_peer: Some(Peer::empty(2)), - follower_peers: vec![Peer::empty(4)], - leader_state: None, - leader_down_since: None, - write_route_policy: None, - }, - ]), - serialized_options, + let (mem_kv, table_metadata_manager, table_name, table_info, region_routes, options) = + create_dropped_physical_table_metadata( + table_id, + table_name, + vec![ + test_physical_region_route(table_id, 1, 1, vec![5]), + test_physical_region_route(table_id, 2, 2, vec![4]), + ], + create_mixed_region_wal_options(), ) - .await - .unwrap(); - - let table_route_value = table_metadata_manager - .table_route_manager - .table_route_storage() - .get_with_raw_bytes(table_id) - .await - .unwrap() - .unwrap(); - let region_routes = table_route_value.region_routes().unwrap(); - let table_name = TableName::new(DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME, table_name); - let table_route_value = TableRouteValue::physical(region_routes.clone()); - - table_metadata_manager - .delete_table_metadata(table_id, &table_name, &table_route_value, &options) - .await - .unwrap(); + .await; mem_kv .put( @@ -3121,10 +3099,14 @@ mod tests { .await .unwrap() .unwrap(); - assert_eq!(dropped_table.table_id, table_id); - assert_eq!(dropped_table.table_name, table_name); - assert_eq!(dropped_table.table_info_value.table_info, table_info); - assert_eq!(dropped_table.region_wal_options, options); + assert_dropped_table_metadata( + &dropped_table, + table_id, + &table_name, + &table_info, + ®ion_routes, + &options, + ); } #[tokio::test] diff --git a/src/mito2/src/engine/open_test.rs b/src/mito2/src/engine/open_test.rs index 1f0053aab5..cf4bfb1ff1 100644 --- a/src/mito2/src/engine/open_test.rs +++ b/src/mito2/src/engine/open_test.rs @@ -125,14 +125,8 @@ async fn test_engine_reopen_region() { test_engine_reopen_region_with_format(true).await; } -#[tokio::test] -async fn test_engine_reopens_closed_soft_dropped_region() { - test_engine_reopens_closed_soft_dropped_region_with_format(false).await; - test_engine_reopens_closed_soft_dropped_region_with_format(true).await; -} - -async fn test_engine_reopens_closed_soft_dropped_region_with_format(flat_format: bool) { - let mut env = TestEnv::with_prefix("reopen-soft-dropped-region").await; +async fn test_engine_reopen_region_with_format(flat_format: bool) { + let mut env = TestEnv::with_prefix("reopen-region").await; let engine = env .create_engine(MitoConfig { default_flat_format: flat_format, @@ -152,32 +146,13 @@ async fn test_engine_reopens_closed_soft_dropped_region_with_format(flat_format: &engine, region_id, Rows { - schema: column_schemas.clone(), + schema: column_schemas, rows: build_rows(0, 2), }, ) .await; - engine - .handle_request(region_id, RegionRequest::Close(RegionCloseRequest {})) - .await - .unwrap(); - assert!(!engine.is_region_exists(region_id)); - - engine - .handle_request( - region_id, - RegionRequest::Open(RegionOpenRequest { - engine: String::new(), - table_dir, - path_type: PathType::Bare, - options: HashMap::default(), - skip_wal_replay: false, - checkpoint: None, - }), - ) - .await - .unwrap(); + reopen_region(&engine, region_id, table_dir, false, Default::default()).await; assert!(engine.is_region_exists(region_id)); let scanner = engine @@ -189,27 +164,6 @@ async fn test_engine_reopens_closed_soft_dropped_region_with_format(flat_format: assert_eq!(2, batches.iter().map(|b| b.num_rows()).sum::()); } -async fn test_engine_reopen_region_with_format(flat_format: bool) { - let mut env = TestEnv::with_prefix("reopen-region").await; - let engine = env - .create_engine(MitoConfig { - default_flat_format: flat_format, - ..Default::default() - }) - .await; - - let region_id = RegionId::new(1, 1); - let request = CreateRequestBuilder::new().build(); - let table_dir = request.table_dir.clone(); - engine - .handle_request(region_id, RegionRequest::Create(request)) - .await - .unwrap(); - - reopen_region(&engine, region_id, table_dir, false, Default::default()).await; - assert!(engine.is_region_exists(region_id)); -} - #[tokio::test] async fn test_engine_open_readonly() { test_engine_open_readonly_with_format(false).await;