rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-07-07 06:20:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor: simplify soft-drop table tests

Browse Source 
Consolidate redundant soft-drop lifecycle assertions into existing end-to-end tests and share dropped-table metadata setup to keep the branch coverage focused.

Files:

- `src/common/meta/src/ddl/tests/drop_table.rs`

- `src/common/meta/src/key.rs`

- `src/mito2/src/engine/open_test.rs`

Signed-off-by: Lei, HUANG <mrsatangel@gmail.com>
This commit is contained in:
Lei, HUANG
2026-05-03 22:35:52 +08:00
committed by Lei, HUANG
parent 7f6fd9e61f
commit c3eb0bb447
3 changed files with 188 additions and 357 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
src/common/meta/src/ddl/tests/drop_table.rs
View File

@@ -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,

362
src/common/meta/src/key.rs
View File

@@ -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<u64>,
) -> 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<RegionRoute>,
region_wal_options: HashMap<RegionNumber, WalOptions>,
) -> (
Arc<MemoryKvBackend<Error>>,
TableMetadataManager,
TableName,
TableInfo,
Vec<RegionRoute>,
HashMap<RegionNumber, WalOptions>,
) {
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::<HashMap<_, _>>();
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,
&region_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<RegionNumber, WalOptions>,
) {
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,
&region_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,
&region_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,
&region_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,
&region_routes,
&options,
);
}
#[tokio::test]

54
src/mito2/src/engine/open_test.rs
View File

@@ -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::<usize>());
}
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;