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

test(mito2): trim redundant compaction tests

Browse Source 
Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>
This commit is contained in:
Lei, HUANG
2026-07-24 11:57:23 +08:00
parent d46a35c4d4
commit 85ec9af053
9 changed files with 11 additions and 1728 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/mito2/src/compaction.rs
-967
View File
File diff suppressed because it is too large Load Diff
src/mito2/src/compaction/window.rs
+11 -29
View File
@@ -207,22 +207,17 @@ mod tests {
use std::sync::Arc;
use std::time::Duration;
use common_base::Plugins;
use common_time::Timestamp;
use common_time::range::TimestampRange;
use store_api::storage::{FileId, RegionId};
use crate::cache::CacheManager;
use crate::compaction::compactor::{CompactionRegion, CompactionVersion};
use crate::compaction::picker::Picker;
use crate::compaction::compactor::CompactionVersion;
use crate::compaction::window::{WindowedCompactionPicker, file_time_bucket_span};
use crate::config::MitoConfig;
use crate::region::options::RegionOptions;
use crate::sst::file::{FileMeta, Level};
use crate::sst::file_purger::NoopFilePurger;
use crate::sst::version::SstVersion;
use crate::test_util::memtable_util::metadata_for_test;
use crate::test_util::scheduler_util::SchedulerEnv;
fn build_version(
files: &[(FileId, i64, i64, Level)],
@@ -269,33 +264,20 @@ mod tests {
}
}
#[tokio::test]
async fn test_pick_expired_ssts_without_marking_compacting() {
#[test]
fn test_pick_expired_ssts_without_marking_compacting() {
let picker = WindowedCompactionPicker::new(None);
let files = vec![(FileId::random(), 0, 10, 0)];
let version = build_version(&files, Some(Duration::from_millis(1)));
let env = SchedulerEnv::new().await;
let manifest_ctx = env.mock_manifest_context(version.metadata.clone()).await;
let compaction_region = CompactionRegion {
region_id: version.metadata.region_id,
region_options: RegionOptions::default(),
engine_config: Arc::new(MitoConfig::default()),
region_metadata: version.metadata.clone(),
cache_manager: Arc::new(CacheManager::default()),
access_layer: env.access_layer,
manifest_ctx,
current_version: version,
file_purger: None,
ttl: None,
max_parallelism: 1,
plugins: Plugins::new(),
};
let (outputs, expired_ssts, _) = picker.pick_inner(
RegionId::new(0, 0),
&version,
Timestamp::new_millisecond(12),
);
let output = picker.pick(&compaction_region).unwrap();
assert!(output.outputs.is_empty());
assert!(!output.expired_ssts.is_empty());
assert!(output.expired_ssts.iter().all(|file| !file.compacting()));
assert!(outputs.is_empty());
assert_eq!(1, expired_ssts.len());
assert!(expired_ssts.iter().all(|file| !file.compacting()));
}
const HOUR: i64 = 60 * 60 * 1000;
src/mito2/src/engine/compaction_test.rs
-552
View File
@@ -18,7 +18,6 @@ use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use api::v1::region::{StrictWindow, compact_request};
use api::v1::{ColumnSchema, Rows};
use async_trait::async_trait;
use common_error::ext::ErrorExt;
@@ -306,18 +305,6 @@ async fn test_region_b_progresses_while_same_worker_region_a_is_picking() {
.await
.expect("region A planning did not reach the gate");
tokio::time::timeout(
Duration::from_secs(5),
put_and_flush(&engine, region_a, &column_schemas, 10..20),
)
.await
.expect("region A automatic compaction trigger did not finish");
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_schedule_attempts(2))
.await
.expect("region A automatic trigger did not reach the scheduler");
assert_eq!(2, gate.schedule_attempt_count());
assert_eq!(1, gate.invocation_count());
let engine_for_region_b = engine.clone();
let mut region_b_work = tokio::spawn(async move {
put_and_flush(&engine_for_region_b, region_b, &column_schemas, 0..10).await;
@@ -326,308 +313,16 @@ async fn test_region_b_progresses_while_same_worker_region_a_is_picking() {
.await
.expect("region B was blocked by region A compaction planning")
.expect("region B work task panicked");
let followup_guard = gate.arm();
gate_guard.release();
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_entered())
.await
.expect("coalesced region A trigger did not start its follow-up plan");
tokio::time::timeout(Duration::from_secs(5), region_a_compaction)
.await
.expect("region A compaction task did not finish after gate release")
.expect("region A compaction task panicked")
.expect("region A compaction failed");
assert_eq!(2, gate.invocation_count());
followup_guard.release();
}
#[tokio::test]
async fn test_regular_trigger_while_picking_replans_after_no_plan() {
common_telemetry::init_default_ut_logging();
let mut env = TestEnv::new().await;
let region_id = RegionId::new(7, 1);
let gate = Arc::new(CompactionPlanningGate::new(region_id));
let engine = env
.create_engine_with(
MitoConfig {
min_compaction_interval: Duration::ZERO,
..Default::default()
},
None,
Some(gate.clone()),
None,
)
.await;
env.get_schema_metadata_manager()
.register_region_table_info(
region_id.table_id(),
"replan_after_no_plan",
"test_catalog",
"test_schema",
None,
env.get_kv_backend(),
)
.await;
let create = CreateRequestBuilder::new()
.insert_option("compaction.type", "twcs")
.build();
let column_schemas = create
.column_metadatas
.iter()
.map(column_metadata_to_column_schema)
.collect::<Vec<_>>();
engine
.handle_request(region_id, RegionRequest::Create(create))
.await
.unwrap();
let first_plan_guard = gate.arm();
put_and_flush(&engine, region_id, &column_schemas, 0..10).await;
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_entered())
.await
.expect("first automatic compaction did not reach the planning gate");
put_and_flush(&engine, region_id, &column_schemas, 5..20).await;
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_schedule_attempts(2))
.await
.expect("second flush did not trigger automatic compaction");
assert_eq!(1, gate.invocation_count());
let second_plan_guard = gate.arm();
first_plan_guard.release();
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_entered())
.await
.expect("coalesced regular trigger was lost after the first plan returned no plan");
assert_eq!(2, gate.invocation_count());
second_plan_guard.release();
}
#[tokio::test]
async fn test_regular_trigger_while_picking_replans_after_prepared_execution() {
common_telemetry::init_default_ut_logging();
let mut env = TestEnv::new().await;
let region_id = RegionId::new(8, 1);
let gate = Arc::new(CompactionPlanningGate::new(region_id));
let engine = env
.create_engine_with(
MitoConfig {
min_compaction_interval: Duration::from_secs(60 * 60),
..Default::default()
},
None,
Some(gate.clone()),
None,
)
.await;
env.get_schema_metadata_manager()
.register_region_table_info(
region_id.table_id(),
"replan_after_prepared",
"test_catalog",
"test_schema",
None,
env.get_kv_backend(),
)
.await;
let create = CreateRequestBuilder::new()
.insert_option("compaction.type", "twcs")
.build();
let column_schemas = create
.column_metadatas
.iter()
.map(column_metadata_to_column_schema)
.collect::<Vec<_>>();
engine
.handle_request(region_id, RegionRequest::Create(create))
.await
.unwrap();
put_and_flush(&engine, region_id, &column_schemas, 0..10).await;
put_and_flush(&engine, region_id, &column_schemas, 5..20).await;
let first_plan_guard = gate.arm();
let first_engine = engine.clone();
let first = tokio::spawn(async move {
first_engine
.handle_request(
region_id,
RegionRequest::Compact(RegionCompactRequest::default()),
)
.await
});
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_entered())
.await
.expect("first regular compaction did not reach the planning gate");
let expected_attempts = gate.schedule_attempt_count() + 1;
let second_engine = engine.clone();
let second = tokio::spawn(async move {
second_engine
.handle_request(
region_id,
RegionRequest::Compact(RegionCompactRequest::default()),
)
.await
});
tokio::time::timeout(
Duration::from_secs(5),
gate.wait_until_schedule_attempts(expected_attempts),
)
.await
.expect("second regular compaction did not reach the scheduler");
let commit_guard = gate.arm_commit();
first_plan_guard.release();
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_commit_entered())
.await
.expect("first regular compaction did not produce a prepared execution");
let followup_plan_guard = gate.arm();
commit_guard.release();
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_entered())
.await
.expect("prepared execution did not immediately admit the retained regular follow-up");
tokio::time::timeout(Duration::from_secs(5), first)
.await
.expect("first regular compaction waiter was not notified")
.expect("first regular compaction task panicked")
.expect("first regular compaction failed");
assert!(!second.is_finished());
followup_plan_guard.release();
tokio::time::timeout(Duration::from_secs(5), second)
.await
.expect("retained regular compaction waiter was not notified")
.expect("retained regular compaction task panicked")
.expect("retained regular compaction failed");
}
#[tokio::test]
async fn test_pending_manual_compaction_finishes_before_queued_ddl() {
common_telemetry::init_default_ut_logging();
let mut env = TestEnv::new().await;
let region_id = RegionId::new(6, 1);
let gate = Arc::new(CompactionPlanningGate::new(region_id));
let engine = env
.create_engine_with(
MitoConfig {
num_workers: 1,
min_compaction_interval: Duration::from_secs(60 * 60),
..Default::default()
},
None,
Some(gate.clone()),
None,
)
.await;
env.get_schema_metadata_manager()
.register_region_table_info(
region_id.table_id(),
"pending_manual_before_ddl",
"test_catalog",
"test_schema",
None,
env.get_kv_backend(),
)
.await;
let create = CreateRequestBuilder::new()
.insert_option("compaction.type", "twcs")
.build();
let column_schemas = create
.column_metadatas
.iter()
.map(column_metadata_to_column_schema)
.collect::<Vec<_>>();
engine
.handle_request(region_id, RegionRequest::Create(create))
.await
.unwrap();
put_and_flush(&engine, region_id, &column_schemas, 0..10).await;
put_and_flush(&engine, region_id, &column_schemas, 5..20).await;
let commit_guard = gate.arm_commit();
let regular_engine = engine.clone();
let regular_task = tokio::spawn(async move {
regular_engine
.handle_request(
region_id,
RegionRequest::Compact(RegionCompactRequest::default()),
)
.await
});
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_commit_entered())
.await
.expect("regular compaction did not reach its non-cancellable commit gate");
let manual_plan_guard = gate.arm();
let manual_engine = engine.clone();
let manual_task = tokio::spawn(async move {
manual_engine
.handle_request(
region_id,
RegionRequest::Compact(RegionCompactRequest {
options: compact_request::Options::StrictWindow(StrictWindow {
window_seconds: 60,
}),
..Default::default()
}),
)
.await
});
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_schedule_attempts(2))
.await
.expect("manual compaction did not reach the scheduler");
assert!(!manual_task.is_finished());
let ddl_engine = engine.clone();
let ddl_task = tokio::spawn(async move {
ddl_engine
.handle_request(
region_id,
RegionRequest::EnterStaging(EnterStagingRequest {
partition_directive: StagingPartitionDirective::RejectAllWrites,
}),
)
.await
});
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_cancel_requested())
.await
.expect("enter-staging DDL was not queued behind regular compaction");
assert!(!ddl_task.is_finished());
commit_guard.release();
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_entered())
.await
.expect("pending manual compaction was not planned after regular completion");
assert_eq!(2, gate.invocation_count());
assert!(!manual_task.is_finished());
assert!(!ddl_task.is_finished());
let pending_ddl_guard = gate.arm_pending_ddl_dispatch();
manual_plan_guard.release();
tokio::time::timeout(
Duration::from_secs(5),
gate.wait_until_pending_ddl_dispatch(),
)
.await
.expect("manual result was not notified before pending DDL dispatch");
tokio::time::timeout(Duration::from_secs(5), manual_task)
.await
.expect("manual compaction did not notify its waiter")
.expect("manual compaction task panicked")
.expect("manual compaction failed");
assert!(!ddl_task.is_finished());
pending_ddl_guard.release();
tokio::time::timeout(Duration::from_secs(5), regular_task)
.await
.expect("regular compaction did not finish after commit release")
.expect("regular compaction task panicked")
.expect("regular compaction failed");
tokio::time::timeout(Duration::from_secs(5), ddl_task)
.await
.expect("queued DDL did not finish after manual compaction")
.expect("queued DDL task panicked")
.expect("queued DDL failed");
assert!(engine.get_region(region_id).unwrap().is_staging());
}
#[tokio::test]
async fn test_picking_close_reopen_ignores_old_plan() {
@@ -717,7 +412,6 @@ async fn test_picking_close_reopen_ignores_old_plan() {
.await
.expect("replacement compaction was blocked by the stale plan");
assert!(engine.is_region_exists(region_id));
assert_eq!(2, gate.invocation_count());
}
#[tokio::test]
@@ -805,108 +499,6 @@ async fn test_enter_staging_waits_for_picking_logical_cancellation_ack() {
assert!(engine.get_region(region_id).unwrap().is_staging());
}
#[tokio::test]
async fn test_truncate_waits_for_cancellable_compaction() {
common_telemetry::init_default_ut_logging();
let mut env = TestEnv::new().await;
let region_id = RegionId::new(9, 1);
let gate = Arc::new(CompactionPlanningGate::new(region_id));
let engine = env
.create_engine_with(
MitoConfig {
num_workers: 1,
min_compaction_interval: Duration::from_secs(60 * 60),
..Default::default()
},
None,
Some(gate.clone()),
None,
)
.await;
env.get_schema_metadata_manager()
.register_region_table_info(
region_id.table_id(),
"truncate_during_cancellable_compaction",
"test_catalog",
"test_schema",
None,
env.get_kv_backend(),
)
.await;
let create = CreateRequestBuilder::new()
.insert_option("compaction.type", "twcs")
.build();
let column_schemas = create
.column_metadatas
.iter()
.map(column_metadata_to_column_schema)
.collect::<Vec<_>>();
engine
.handle_request(region_id, RegionRequest::Create(create))
.await
.unwrap();
put_and_flush(&engine, region_id, &column_schemas, 0..10).await;
put_and_flush(&engine, region_id, &column_schemas, 5..20).await;
let planning_guard = gate.arm();
let compact_engine = engine.clone();
let compact_task = tokio::spawn(async move {
compact_engine
.handle_request(
region_id,
RegionRequest::Compact(RegionCompactRequest::default()),
)
.await
});
tokio::time::timeout(Duration::from_secs(5), gate.wait_until_entered())
.await
.expect("compaction did not reach the cancellable planning gate");
let truncate_engine = engine.clone();
let mut truncate_task = tokio::spawn(async move {
truncate_engine
.handle_request(
region_id,
RegionRequest::Truncate(RegionTruncateRequest::All),
)
.await
});
tokio::time::timeout(Duration::from_secs(5), async {
tokio::select! {
biased;
result = &mut truncate_task => {
panic!("truncate completed before cancellable compaction terminated: {result:?}");
}
() = gate.wait_until_cancel_requested() => {}
}
})
.await
.expect("truncate did not request compaction cancellation");
assert!(!truncate_task.is_finished());
let pending_ddl_guard = gate.arm_pending_ddl_dispatch();
planning_guard.release();
tokio::time::timeout(
Duration::from_secs(5),
gate.wait_until_pending_ddl_dispatch(),
)
.await
.expect("cancelled compaction did not reach pending truncate dispatch");
let compact_err = tokio::time::timeout(Duration::from_secs(5), compact_task)
.await
.expect("cancelled compaction waiter was not released")
.expect("compaction task panicked")
.unwrap_err();
assert_eq!(compact_err.status_code(), StatusCode::Cancelled);
assert!(!truncate_task.is_finished());
pending_ddl_guard.release();
tokio::time::timeout(Duration::from_secs(5), truncate_task)
.await
.expect("queued truncate did not finish after compaction cancellation")
.expect("truncate task panicked")
.expect("queued truncate failed");
}
#[tokio::test]
async fn test_truncate_waits_for_non_cancellable_compaction_commit() {
@@ -1751,151 +1343,7 @@ async fn test_local_compaction_cancellation_notifies_before_pending_ddl_dispatch
assert!(engine.get_region(region_id).unwrap().is_staging());
}
#[tokio::test]
async fn test_enter_staging_cancels_inflight_local_compaction_before_commit() {
common_telemetry::init_default_ut_logging();
let mut env = TestEnv::new().await;
let listener = Arc::new(CompactionListener::default());
let engine = env
.create_engine_with(
MitoConfig {
max_background_purges: 1,
..Default::default()
},
None,
Some(listener.clone()),
None,
)
.await;
let region_id = RegionId::new(2048, 1);
env.get_schema_metadata_manager()
.register_region_table_info(
region_id.table_id(),
"test_table",
"test_catalog",
"test_schema",
None,
env.get_kv_backend(),
)
.await;
let request = CreateRequestBuilder::new()
.insert_option("compaction.type", "twcs")
.build();
let column_schemas = request
.column_metadatas
.iter()
.map(column_metadata_to_column_schema)
.collect::<Vec<_>>();
engine
.handle_request(region_id, RegionRequest::Create(request))
.await
.unwrap();
let _listener_guard = CompactionListenerGuard::new(listener.clone());
put_and_flush(&engine, region_id, &column_schemas, 0..10).await;
put_and_flush(&engine, region_id, &column_schemas, 5..20).await;
tokio::time::timeout(Duration::from_secs(5), listener.wait_handle_finished())
.await
.expect("local compaction did not reach its pre-commit gate");
tokio::time::timeout(
Duration::from_secs(5),
engine.handle_request(
region_id,
RegionRequest::EnterStaging(EnterStagingRequest {
partition_directive: StagingPartitionDirective::RejectAllWrites,
}),
),
)
.await
.expect("enter-staging waited for the blocked local compaction")
.expect("enter-staging request failed");
assert!(engine.get_region(region_id).unwrap().is_staging());
}
#[tokio::test]
async fn test_manual_compaction_returns_cancelled_when_enter_staging_cancels_it() {
common_telemetry::init_default_ut_logging();
let mut env = TestEnv::new().await;
let listener = Arc::new(CompactionListener::default());
let engine = env
.create_engine_with(
MitoConfig {
max_background_purges: 1,
..Default::default()
},
None,
Some(listener.clone()),
None,
)
.await;
let region_id = RegionId::new(2050, 1);
env.get_schema_metadata_manager()
.register_region_table_info(
region_id.table_id(),
"test_table",
"test_catalog",
"test_schema",
None,
env.get_kv_backend(),
)
.await;
let request = CreateRequestBuilder::new()
.insert_option("compaction.type", "twcs")
.build();
let column_schemas = request
.column_metadatas
.iter()
.map(column_metadata_to_column_schema)
.collect::<Vec<_>>();
engine
.handle_request(region_id, RegionRequest::Create(request))
.await
.unwrap();
let _listener_guard = CompactionListenerGuard::new(listener.clone());
put_and_flush(&engine, region_id, &column_schemas, 0..10).await;
put_and_flush(&engine, region_id, &column_schemas, 5..20).await;
let engine_cloned = engine.clone();
let compact = tokio::spawn(async move {
engine_cloned
.handle_request(
region_id,
RegionRequest::Compact(RegionCompactRequest::default()),
)
.await
});
tokio::time::timeout(Duration::from_secs(5), listener.wait_handle_finished())
.await
.expect("manual compaction did not reach its pre-commit gate");
tokio::time::timeout(
Duration::from_secs(5),
engine.handle_request(
region_id,
RegionRequest::EnterStaging(EnterStagingRequest {
partition_directive: StagingPartitionDirective::RejectAllWrites,
}),
),
)
.await
.expect("enter-staging waited for the blocked manual compaction")
.expect("enter-staging request failed");
let err = tokio::time::timeout(Duration::from_secs(5), compact)
.await
.expect("cancelled manual compaction waiter was not released")
.expect("manual compaction task panicked")
.unwrap_err();
assert_eq!(err.status_code(), StatusCode::Cancelled);
}
#[tokio::test]
async fn test_compaction_update_time_window() {
src/mito2/src/engine/listener.rs
-48
View File
@@ -71,9 +71,6 @@ pub trait EventListener: Send + Sync {
/// Notifies the listener that the compaction is scheduled.
fn on_compaction_scheduled(&self, _region_id: RegionId) {}
/// Notifies the listener immediately before a worker asks the scheduler for compaction.
fn on_compaction_schedule_attempt(&self, _region_id: RegionId) {}
/// Notifies the listener immediately before compaction planning invokes the picker.
async fn on_compaction_pick_begin(&self, _region_id: RegionId) {}
@@ -111,9 +108,6 @@ pub struct CompactionPlanningGate {
entered: Notify,
cancel_requested: Notify,
permits: Semaphore,
invocation_count: AtomicUsize,
schedule_attempted: Notify,
schedule_attempt_count: AtomicUsize,
commit_armed: AtomicBool,
commit_entered: Notify,
commit_permits: Semaphore,
@@ -187,9 +181,6 @@ impl CompactionPlanningGate {
entered: Notify::new(),
cancel_requested: Notify::new(),
permits: Semaphore::new(0),
invocation_count: AtomicUsize::new(0),
schedule_attempted: Notify::new(),
schedule_attempt_count: AtomicUsize::new(0),
commit_armed: AtomicBool::new(false),
commit_entered: Notify::new(),
commit_permits: Semaphore::new(0),
@@ -214,16 +205,6 @@ impl CompactionPlanningGate {
self.cancel_requested.notified().await;
}
pub async fn wait_until_schedule_attempts(&self, expected: usize) {
while self.schedule_attempt_count() < expected {
self.schedule_attempted.notified().await;
}
}
pub fn schedule_attempt_count(&self) -> usize {
self.schedule_attempt_count.load(Ordering::Relaxed)
}
pub fn arm_commit(self: &Arc<Self>) -> CompactionCommitGateGuard {
self.commit_armed.store(true, Ordering::Relaxed);
CompactionCommitGateGuard {
@@ -250,10 +231,6 @@ impl CompactionPlanningGate {
self.permits.add_permits(1);
}
pub fn invocation_count(&self) -> usize {
self.invocation_count.load(Ordering::Relaxed)
}
fn release_commit(&self) {
self.commit_permits.add_permits(1);
}
@@ -265,19 +242,11 @@ impl CompactionPlanningGate {
#[async_trait]
impl EventListener for CompactionPlanningGate {
fn on_compaction_schedule_attempt(&self, region_id: RegionId) {
if region_id == self.region_id {
self.schedule_attempt_count.fetch_add(1, Ordering::Relaxed);
self.schedule_attempted.notify_one();
}
}
async fn on_compaction_pick_begin(&self, region_id: RegionId) {
if region_id != self.region_id {
return;
}
self.invocation_count.fetch_add(1, Ordering::Relaxed);
if !self.armed.swap(false, Ordering::Relaxed) {
return;
}
@@ -788,20 +757,3 @@ impl EventListener for GateIndexBuildListener {
self.stop_notify.notify_one();
}
}
#[cfg(test)]
mod tests {
use super::*;
#[tokio::test]
async fn test_compaction_planning_gate_counts_every_matching_callback() {
let region_id = RegionId::new(1, 1);
let gate = CompactionPlanningGate::new(region_id);
gate.on_compaction_pick_begin(region_id).await;
gate.on_compaction_pick_begin(RegionId::new(2, 1)).await;
gate.on_compaction_pick_begin(region_id).await;
assert_eq!(2, gate.invocation_count());
}
}
src/mito2/src/schedule/remote_job_scheduler.rs
-79
View File
@@ -218,9 +218,6 @@ impl Notifier for DefaultNotifier {
#[cfg(test)]
mod tests {
use super::*;
use crate::compaction::{CompactionExecution, CompactionExecutionKind};
use crate::error::InvalidSchedulerStateSnafu;
use crate::test_util::version_util::VersionControlBuilder;
#[test]
fn test_job_id() {
@@ -228,80 +225,4 @@ mod tests {
let job_id = JobId::parse_str(&id).unwrap();
assert_eq!(job_id.to_string(), id);
}
#[tokio::test]
async fn test_default_notifier_carries_remote_execution_on_success() {
let (tx, mut rx) = tokio::sync::mpsc::channel(1);
let execution = CompactionExecution::for_test(
Arc::new(VersionControlBuilder::new().build()),
CompactionExecutionKind::Remote,
);
let expected = execution.clone();
let notifier = DefaultNotifier::new(tx, execution);
let region_id = RegionId::new(1, 1);
notifier
.notify(
RemoteJobResult::CompactionJobResult(CompactionJobResult {
job_id: JobId::parse_str("00000000-0000-0000-0000-000000000003").unwrap(),
region_id,
start_time: Instant::now(),
region_edit: Ok(RegionEdit {
files_to_add: Vec::new(),
files_to_remove: Vec::new(),
timestamp_ms: None,
compaction_time_window: None,
flushed_entry_id: None,
flushed_sequence: None,
committed_sequence: None,
}),
}),
Vec::new(),
)
.await;
let request = rx.recv().await.unwrap();
let WorkerRequest::Background {
notify: BackgroundNotify::CompactionFinished(finished),
..
} = request.request
else {
panic!("expected remote compaction success notification");
};
assert!(finished.execution.matches(&expected));
}
#[tokio::test]
async fn test_default_notifier_carries_remote_execution_on_failure() {
let (tx, mut rx) = tokio::sync::mpsc::channel(1);
let execution = CompactionExecution::for_test(
Arc::new(VersionControlBuilder::new().build()),
CompactionExecutionKind::Remote,
);
let expected = execution.clone();
let notifier = DefaultNotifier::new(tx, execution);
let region_id = RegionId::new(1, 1);
notifier
.notify(
RemoteJobResult::CompactionJobResult(CompactionJobResult {
job_id: JobId::parse_str("00000000-0000-0000-0000-000000000004").unwrap(),
region_id,
start_time: Instant::now(),
region_edit: Err(InvalidSchedulerStateSnafu.build()),
}),
Vec::new(),
)
.await;
let request = rx.recv().await.unwrap();
let WorkerRequest::Background {
notify: BackgroundNotify::CompactionFailed(failed),
..
} = request.request
else {
panic!("expected remote compaction failure notification");
};
assert!(failed.execution.matches(&expected));
}
}
src/mito2/src/sst/file.rs
-12
View File
@@ -838,18 +838,6 @@ mod tests {
}
}
#[test]
fn test_try_set_compacting() {
let file = FileHandle::new(
create_file_meta(FileId::random(), 0),
crate::test_util::new_noop_file_purger(),
);
assert!(file.try_set_compacting());
assert!(!file.try_set_compacting());
file.set_compacting(false);
assert!(file.try_set_compacting());
}
#[test]
fn test_deserialize_file_meta() {
src/mito2/src/sst/version.rs
-31
View File
@@ -286,37 +286,6 @@ mod tests {
});
}
#[test]
fn test_file_for_compaction_returns_unambiguous_matching_level() {
let purger = new_noop_file_purger();
let file_id = FileId::random();
let file = FileMeta {
file_id,
level: 1,
..Default::default()
};
let selected = FileHandle::new(file.clone(), purger.clone());
let mut version = SstVersion::new();
version.add_files(purger, std::iter::once(file));
assert_eq!(
version
.file_for_compaction(&selected)
.unwrap()
.file_id()
.file_id(),
file_id
);
let missing = FileHandle::new(
FileMeta {
file_id: FileId::random(),
level: 1,
..Default::default()
},
new_noop_file_purger(),
);
assert!(version.file_for_compaction(&missing).is_none());
}
#[test]
fn test_usage_only_counts_owned_files() {
src/mito2/src/worker.rs
-7
View File
@@ -1418,13 +1418,6 @@ impl WorkerListener {
}
}
pub(crate) fn on_compaction_schedule_attempt(&self, _region_id: RegionId) {
#[cfg(any(test, feature = "test"))]
if let Some(listener) = &self.listener {
listener.on_compaction_schedule_attempt(_region_id);
}
}
pub(crate) async fn on_compaction_pick_begin(&self, _region_id: RegionId) {
#[cfg(any(test, feature = "test"))]
if let Some(listener) = &self.listener {
src/mito2/src/worker/handle_compaction.rs
-3
View File
@@ -80,7 +80,6 @@ impl<S> RegionWorkerLoop<S> {
};
COMPACTION_REQUEST_COUNT.inc();
let parallelism = req.parallelism.unwrap_or(1) as usize;
self.listener.on_compaction_schedule_attempt(region_id);
if let Err(e) = self.compaction_scheduler.schedule_compaction(
region.region_id,
req.options,
@@ -250,8 +249,6 @@ impl<S> RegionWorkerLoop<S> {
"minimal compaction interval time {:?} has passed, scheduling next compaction",
self.config.min_compaction_interval
);
self.listener
.on_compaction_schedule_attempt(region.region_id);
match self.compaction_scheduler.schedule_compaction(
region.region_id,
compact_request::Options::Regular(Default::default()),