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

refactor(event): separate procedure submission context (#8856)

Browse Source 
* refactor(event): separate procedure submission context

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(event): map extensions and forward GC context

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(gc): initialize integration test context

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(test): pass procedure context to DDL helpers

Signed-off-by: WenyXu <wenymedia@gmail.com>

* refactor: simplify procedure submission contexts

Signed-off-by: WenyXu <wenymedia@gmail.com>

* refactor(event): separate procedure and query contexts

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(event): clarify procedure context propagation

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(event): preserve procedure submission context

Signed-off-by: WenyXu <wenymedia@gmail.com>

* refactor(event): move DDL context by value

Signed-off-by: WenyXu <wenymedia@gmail.com>

* fix(test): retain manual GC event context

Signed-off-by: WenyXu <wenymedia@gmail.com>

* refactor(event): tighten procedure context API

Signed-off-by: WenyXu <wenymedia@gmail.com>

* chore: update greptime-proto

Signed-off-by: WenyXu <wenymedia@gmail.com>

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>
(cherry picked from commit 1af4c33524)
(cherry picked from commit 1b6ee164f2b4a32afeb7c36fd926bcc651046294)
Signed-off-by: discord9 <discord9@163.com>
This commit is contained in:
Weny Xu
2026-08-20 14:46:35 +08:00
committed by discord9
parent 0d5585ba33
commit a451a16cd0
41 changed files with 1487 additions and 775 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests-integration/tests/gc_event.rs
+5 -5
View File
@@ -15,8 +15,8 @@
use std::sync::Arc;
use std::time::Duration;
use common_meta::rpc::ddl::{PersistentEventContext, TriggerReason};
use common_procedure::{ProcedureId, ProcedureWithId, watcher};
use common_event_recorder::{PersistentEventContext, TriggerReason};
use common_procedure::{ProcedureContext, ProcedureId, ProcedureWithId, watcher};
use common_test_util::temp_dir::create_temp_dir;
use meta_srv::gc::{BatchGcProcedure, GcSchedulerOptions};
use mito2::gc::GcConfig;
@@ -137,9 +137,9 @@ async fn test_batch_gc_event() {
.submit(ProcedureWithId {
id: procedure_id,
procedure: Box::new(procedure),
context: common_procedure::ProcedureContext::from_event_context(
PersistentEventContext::new(TriggerReason::Manual),
),
context: ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
})
.await
.unwrap();
tests-integration/tests/region_migration.rs
+110 -64
View File
@@ -23,10 +23,11 @@ use common_event_recorder::event_table::{
};
use common_event_recorder::{
DEFAULT_EVENTS_TABLE_NAME, DEFAULT_FLUSH_INTERVAL_SECONDS, EVENTS_TABLE_TIMESTAMP_COLUMN_NAME,
EVENTS_TABLE_TYPE_COLUMN_NAME,
EVENTS_TABLE_TYPE_COLUMN_NAME, PersistentEventContext, TriggerReason,
};
use common_meta::key::{RegionDistribution, RegionRoleSet, TableMetadataManagerRef};
use common_meta::peer::Peer;
use common_procedure::ProcedureContext;
use common_procedure::event::{
EVENTS_TABLE_PROCEDURE_ID_COLUMN_NAME, EVENTS_TABLE_PROCEDURE_STATE_COLUMN_NAME,
};
@@ -183,13 +184,18 @@ pub async fn test_region_migration(store_type: StorageType, endpoints: Vec<Strin
let region_id = RegionId::new(table_id, from_regions.leader_regions[0]);
// Trigger region migration.
let procedure = region_migration_manager
.submit_procedure(RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
))
.submit_procedure(
ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
),
)
.await
.unwrap();
info!("Started region procedure: {}!", procedure.unwrap());
@@ -235,13 +241,18 @@ pub async fn test_region_migration(store_type: StorageType, endpoints: Vec<Strin
// Triggers again.
let err = region_migration_manager
.submit_procedure(RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
))
.submit_procedure(
ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
),
)
.await
.unwrap_err();
assert!(matches!(err, error::Error::RegionMigrated { .. }));
@@ -523,13 +534,18 @@ pub async fn test_region_migration_by_sql(store_type: StorageType, endpoints: Ve
// Triggers again.
let err = region_migration_manager
.submit_procedure(RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
))
.submit_procedure(
ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
),
)
.await
.unwrap_err();
assert!(matches!(err, error::Error::RegionMigrated { .. }));
@@ -628,13 +644,18 @@ pub async fn test_region_migration_multiple_regions(
let region_id = RegionId::new(table_id, from_regions.leader_regions[0]);
// Trigger region migration.
let procedure = region_migration_manager
.submit_procedure(RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
))
.submit_procedure(
ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
),
)
.await
.unwrap();
info!("Started region procedure: {}!", procedure.unwrap());
@@ -689,13 +710,18 @@ pub async fn test_region_migration_multiple_regions(
// Triggers again.
let err = region_migration_manager
.submit_procedure(RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
))
.submit_procedure(
ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
),
)
.await
.unwrap_err();
assert!(matches!(err, error::Error::RegionMigrated { .. }));
@@ -779,13 +805,18 @@ pub async fn test_region_migration_all_regions(store_type: StorageType, endpoint
let region_id = RegionId::new(table_id, from_regions.leader_regions[0]);
// Trigger region migration.
let procedure = region_migration_manager
.submit_procedure(RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
))
.submit_procedure(
ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
),
)
.await
.unwrap();
info!("Started region procedure: {}!", procedure.unwrap());
@@ -838,13 +869,18 @@ pub async fn test_region_migration_all_regions(store_type: StorageType, endpoint
// Triggers again.
let err = region_migration_manager
.submit_procedure(RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
))
.submit_procedure(
ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
RegionMigrationProcedureTask::new(
region_id,
peer_factory(from_peer_id),
peer_factory(to_peer_id),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
),
)
.await
.unwrap_err();
assert!(matches!(err, error::Error::RegionMigrated { .. }));
@@ -917,13 +953,18 @@ pub async fn test_region_migration_incorrect_from_peer(
// Trigger region migration.
let err = region_migration_manager
.submit_procedure(RegionMigrationProcedureTask::new(
region_id,
peer_factory(5),
peer_factory(1),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
))
.submit_procedure(
ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
RegionMigrationProcedureTask::new(
region_id,
peer_factory(5),
peer_factory(1),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
),
)
.await
.unwrap_err();
@@ -1000,13 +1041,18 @@ pub async fn test_region_migration_incorrect_region_id(
// Trigger region migration.
let err = region_migration_manager
.submit_procedure(RegionMigrationProcedureTask::new(
region_id,
peer_factory(2),
peer_factory(1),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
))
.submit_procedure(
ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
RegionMigrationProcedureTask::new(
region_id,
peer_factory(2),
peer_factory(1),
Duration::from_millis(1000),
RegionMigrationTriggerReason::Manual,
),
)
.await
.unwrap_err();
tests-integration/tests/repartition.rs
+5 -1
View File
@@ -18,8 +18,9 @@ use std::time::Duration;
use client::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
use common_error::root_source;
use common_event_recorder::{PersistentEventContext, TriggerReason};
use common_meta::key::table_name::TableNameKey;
use common_procedure::{ProcedureWithId, watcher};
use common_procedure::{ProcedureContext, ProcedureWithId, watcher};
use common_query::Output;
use common_telemetry::info;
use common_test_util::recordbatch::check_output_stream;
@@ -854,6 +855,9 @@ async fn trigger_full_gc(ticker: &GcTickerRef) {
region_ids: None,
full_file_listing: None,
timeout: None,
procedure_context: ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
})
.await
.unwrap();
tests-integration/tests/table_ddl_event.rs
+10 -6
View File
@@ -29,6 +29,10 @@ use common_event_recorder::event_table::{
CATALOG_NAME_COLUMN, PAYLOAD_COLUMN, PHYSICAL_TABLE_ID_COLUMN, PROCEDURE_ID_COLUMN,
PROCEDURE_TRIGGER_COLUMN, SCHEMA_NAME_COLUMN, TABLE_ID_COLUMN, TABLE_NAME_COLUMN, TYPE_COLUMN,
};
#[cfg(feature = "enterprise")]
use common_event_recorder::{PersistentEventContext, TriggerReason};
#[cfg(feature = "enterprise")]
use common_procedure::ProcedureContext;
use common_test_util::temp_dir::create_temp_dir;
use frontend::instance::Instance;
use meta_srv::gc::GcSchedulerOptions;
@@ -377,9 +381,9 @@ async fn test_table_ddl_procedure_events() {
.ddl_manager()
.submit_undrop_table_task(
common_meta::rpc::ddl::UndropTableTask { table_id },
common_meta::rpc::ddl::PersistentEventContext::new(
common_meta::rpc::ddl::TriggerReason::Manual,
),
ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
)
.await
.unwrap();
@@ -419,9 +423,9 @@ async fn test_table_ddl_procedure_events() {
.ddl_manager()
.submit_purge_dropped_table_task(
common_meta::rpc::ddl::PurgeDroppedTableTask { table_id },
common_meta::rpc::ddl::PersistentEventContext::new(
common_meta::rpc::ddl::TriggerReason::Manual,
),
ProcedureContext::from_event_context(PersistentEventContext::new(
TriggerReason::Manual,
)),
)
.await
.unwrap();