From 32e215cad0aa0803991d751bc7b4f44ac1865e62 Mon Sep 17 00:00:00 2001 From: Weny Xu Date: Tue, 11 Aug 2026 15:12:45 +0800 Subject: [PATCH] fix(event): preserve procedure lifecycle locators (#8787) * fix(event): preserve procedure lifecycle locators Signed-off-by: WenyXu * fix(event): preserve dropped table lifecycle locators Signed-off-by: WenyXu * test(event): cover lifecycle locators Signed-off-by: WenyXu * test(event): fix lifecycle context expectations Signed-off-by: WenyXu --------- Signed-off-by: WenyXu --- src/common/meta/src/ddl/alter_database.rs | 2 +- .../meta/src/ddl/alter_logical_tables.rs | 13 +- src/common/meta/src/ddl/alter_table.rs | 9 +- src/common/meta/src/ddl/create_database.rs | 2 +- src/common/meta/src/ddl/create_flow.rs | 11 +- .../meta/src/ddl/create_logical_tables.rs | 66 +++++----- src/common/meta/src/ddl/create_table.rs | 17 +-- src/common/meta/src/ddl/create_view.rs | 29 ++++- src/common/meta/src/ddl/drop_database.rs | 2 +- src/common/meta/src/ddl/drop_flow.rs | 6 +- src/common/meta/src/ddl/drop_table.rs | 20 ++- src/common/meta/src/ddl/drop_view.rs | 10 +- src/common/meta/src/ddl/event/database.rs | 18 +-- src/common/meta/src/ddl/event/flow.rs | 31 +++-- src/common/meta/src/ddl/event/table.rs | 31 +++-- src/common/meta/src/ddl/event/view.rs | 64 +++++++--- .../meta/src/ddl/purge_dropped_table.rs | 29 +++-- .../meta/src/ddl/tests/event/database.rs | 29 +++-- src/common/meta/src/ddl/tests/event/flow.rs | 43 +++++-- src/common/meta/src/ddl/tests/event/table.rs | 44 +++++-- src/common/meta/src/ddl/tests/event/view.rs | 114 +++++++++++++----- src/common/meta/src/ddl/truncate_table.rs | 20 ++- src/common/meta/src/ddl/undrop_table.rs | 33 ++--- src/meta-srv/src/event/repartition.rs | 54 +++++---- src/meta-srv/src/procedure/repartition.rs | 2 +- .../src/procedure/repartition/group.rs | 2 +- tests-integration/tests/database_ddl_event.rs | 4 +- tests-integration/tests/flow_ddl_event.rs | 10 +- tests-integration/tests/repartition_event.rs | 49 ++++++++ tests-integration/tests/table_ddl_event.rs | 99 +++++++-------- tests-integration/tests/view_ddl_event.rs | 14 +-- 31 files changed, 562 insertions(+), 315 deletions(-) diff --git a/src/common/meta/src/ddl/alter_database.rs b/src/common/meta/src/ddl/alter_database.rs index 7e16275da3..f934421302 100644 --- a/src/common/meta/src/ddl/alter_database.rs +++ b/src/common/meta/src/ddl/alter_database.rs @@ -196,7 +196,7 @@ impl Procedure for AlterDatabaseProcedure { self.data.event_context.clone(), ) } else { - DatabaseDdlEvent::alter_lifecycle() + DatabaseDdlEvent::alter_lifecycle(self.data.catalog(), self.data.schema()) }; Some(Box::new(event)) } diff --git a/src/common/meta/src/ddl/alter_logical_tables.rs b/src/common/meta/src/ddl/alter_logical_tables.rs index 816db0a6af..f2f1543093 100644 --- a/src/common/meta/src/ddl/alter_logical_tables.rs +++ b/src/common/meta/src/ddl/alter_logical_tables.rs @@ -335,6 +335,11 @@ impl Procedure for AlterLogicalTablesProcedure { if ctx.trigger != EventTrigger::Submitted { return Some(Box::new(TableDdlEvent::lifecycle( TableDdlEventType::AlterLogicalTables, + self.data.tasks.iter().map(|task| { + let table_ref = task.table_ref(); + TableDdlLocator::new(table_ref.catalog, table_ref.schema, table_ref.table) + .with_physical_table_id(self.data.physical_table_id) + }), ))); } @@ -375,12 +380,12 @@ pub struct AlterTablesData { } impl AlterTablesData { - /// Clears all data fields except `state` and `table_cache_keys_to_invalidate` after metadata update. - /// This is done to avoid persisting unnecessary data after the update metadata step. + /// Clears metadata snapshots after the update metadata step. + /// + /// Keep the tasks and physical table ID until the procedure finishes: lifecycle + /// events use them to retain the logical table locator. fn clear_metadata_fields(&mut self) { - self.tasks.clear(); self.table_info_values.clear(); - self.physical_table_id = 0; self.physical_table_info = None; self.physical_columns.clear(); } diff --git a/src/common/meta/src/ddl/alter_table.rs b/src/common/meta/src/ddl/alter_table.rs index acc02c4d1a..21975fab71 100644 --- a/src/common/meta/src/ddl/alter_table.rs +++ b/src/common/meta/src/ddl/alter_table.rs @@ -579,12 +579,11 @@ impl Procedure for AlterTableProcedure { { return None; } + let table_ref = self.data.table_ref(); + let locator = TableDdlLocator::new(table_ref.catalog, table_ref.schema, table_ref.table) + .with_table_id(self.data.table_id()); let event = match &ctx.trigger { EventTrigger::Submitted => { - let table_ref = self.data.table_ref(); - let locator = - TableDdlLocator::new(table_ref.catalog, table_ref.schema, table_ref.table) - .with_table_id(self.data.table_id()); let kind = self .data .task @@ -594,7 +593,7 @@ impl Procedure for AlterTableProcedure { .and_then(alter_table_kind_name); TableDdlEvent::alter_table_submitted(locator, kind, self.data.event_context.clone()) } - _ => TableDdlEvent::lifecycle(TableDdlEventType::AlterTable), + _ => TableDdlEvent::lifecycle(TableDdlEventType::AlterTable, [locator]), }; Some(Box::new(event)) diff --git a/src/common/meta/src/ddl/create_database.rs b/src/common/meta/src/ddl/create_database.rs index 8f9c5bee1a..b48e700a51 100644 --- a/src/common/meta/src/ddl/create_database.rs +++ b/src/common/meta/src/ddl/create_database.rs @@ -297,7 +297,7 @@ impl Procedure for CreateDatabaseProcedure { self.data.event_context.clone(), ) } else { - DatabaseDdlEvent::create_lifecycle() + DatabaseDdlEvent::create_lifecycle(&self.data.catalog, &self.data.schema) }; Some(Box::new(event)) } diff --git a/src/common/meta/src/ddl/create_flow.rs b/src/common/meta/src/ddl/create_flow.rs index 66083f5949..3ef4dadbc6 100644 --- a/src/common/meta/src/ddl/create_flow.rs +++ b/src/common/meta/src/ddl/create_flow.rs @@ -440,9 +440,16 @@ impl Procedure for CreateFlowProcedure { .or(self.data.flow_id), _ => self.data.flow_id, }; - FlowDdlEvent::create_succeeded(flow_id) + FlowDdlEvent::create_succeeded( + &self.data.task.catalog_name, + &self.data.task.flow_name, + flow_id, + ) } - _ => FlowDdlEvent::create_lifecycle(), + _ => FlowDdlEvent::create_lifecycle( + &self.data.task.catalog_name, + &self.data.task.flow_name, + ), }; Some(Box::new(event)) diff --git a/src/common/meta/src/ddl/create_logical_tables.rs b/src/common/meta/src/ddl/create_logical_tables.rs index 8b93ecfcc0..1c9efb865e 100644 --- a/src/common/meta/src/ddl/create_logical_tables.rs +++ b/src/common/meta/src/ddl/create_logical_tables.rs @@ -212,6 +212,19 @@ impl CreateLogicalTablesProcedure { } } +impl CreateLogicalTablesProcedure { + fn event_locators(&self) -> impl Iterator + '_ { + self.data.tasks.iter().map(|task| { + TableDdlLocator::new( + &task.create_table.catalog_name, + &task.create_table.schema_name, + &task.create_table.table_name, + ) + .with_physical_table_id(self.data.physical_table_id) + }) + } +} + #[async_trait] impl Procedure for CreateLogicalTablesProcedure { fn type_name(&self) -> &str { @@ -268,21 +281,11 @@ impl Procedure for CreateLogicalTablesProcedure { return None; } let event = match &ctx.trigger { - EventTrigger::Submitted => { - let locators = self.data.tasks.iter().map(|task| { - TableDdlLocator::new( - &task.create_table.catalog_name, - &task.create_table.schema_name, - &task.create_table.table_name, - ) - .with_physical_table_id(self.data.physical_table_id) - }); - TableDdlEvent::create_logical_tables_submitted( - locators, - self.data.tasks.len(), - self.data.event_context.clone(), - ) - } + EventTrigger::Submitted => TableDdlEvent::create_logical_tables_submitted( + self.event_locators(), + self.data.tasks.len(), + self.data.event_context.clone(), + ), EventTrigger::Succeeded => match ctx.lifecycle_state { ProcedureState::Done { output: Some(output), @@ -290,28 +293,27 @@ impl Procedure for CreateLogicalTablesProcedure { .downcast_ref::>() .map(|table_ids| { debug_assert_eq!(self.data.tasks.len(), table_ids.len()); - let locators = - self.data - .tasks - .iter() - .zip(table_ids) - .map(|(task, table_id)| { - TableDdlLocator::new( - &task.create_table.catalog_name, - &task.create_table.schema_name, - &task.create_table.table_name, - ) - .with_table_id(*table_id) - .with_physical_table_id(self.data.physical_table_id) - }); + let locators = self + .event_locators() + .zip(table_ids) + .map(|(locator, table_id)| locator.with_table_id(*table_id)); TableDdlEvent::create_logical_tables_succeeded(locators) }) .unwrap_or_else(|| { - TableDdlEvent::lifecycle(TableDdlEventType::CreateLogicalTables) + TableDdlEvent::lifecycle( + TableDdlEventType::CreateLogicalTables, + self.event_locators(), + ) }), - _ => TableDdlEvent::lifecycle(TableDdlEventType::CreateLogicalTables), + _ => TableDdlEvent::lifecycle( + TableDdlEventType::CreateLogicalTables, + self.event_locators(), + ), }, - _ => TableDdlEvent::lifecycle(TableDdlEventType::CreateLogicalTables), + _ => TableDdlEvent::lifecycle( + TableDdlEventType::CreateLogicalTables, + self.event_locators(), + ), }; Some(Box::new(event)) diff --git a/src/common/meta/src/ddl/create_table.rs b/src/common/meta/src/ddl/create_table.rs index c1690416a2..c63d7f7a34 100644 --- a/src/common/meta/src/ddl/create_table.rs +++ b/src/common/meta/src/ddl/create_table.rs @@ -406,11 +406,10 @@ impl Procedure for CreateTableProcedure { { return None; } + let table_ref = self.data.table_ref(); + let locator = TableDdlLocator::new(table_ref.catalog, table_ref.schema, table_ref.table); let event = match &ctx.trigger { EventTrigger::Submitted => { - let table_ref = self.data.table_ref(); - let locator = - TableDdlLocator::new(table_ref.catalog, table_ref.schema, table_ref.table); let create_table = &self.data.task.create_table; TableDdlEvent::create_table_submitted( locator, @@ -425,11 +424,15 @@ impl Procedure for CreateTableProcedure { } => output .downcast_ref::() .copied() - .map(TableDdlEvent::create_table_succeeded) - .unwrap_or_else(|| TableDdlEvent::lifecycle(TableDdlEventType::CreateTable)), - _ => TableDdlEvent::lifecycle(TableDdlEventType::CreateTable), + .map(|table_id| { + TableDdlEvent::create_table_succeeded(locator.clone(), table_id) + }) + .unwrap_or_else(|| { + TableDdlEvent::lifecycle(TableDdlEventType::CreateTable, [locator.clone()]) + }), + _ => TableDdlEvent::lifecycle(TableDdlEventType::CreateTable, [locator.clone()]), }, - _ => TableDdlEvent::lifecycle(TableDdlEventType::CreateTable), + _ => TableDdlEvent::lifecycle(TableDdlEventType::CreateTable, [locator]), }; Some(Box::new(event)) diff --git a/src/common/meta/src/ddl/create_view.rs b/src/common/meta/src/ddl/create_view.rs index 0cb622774f..a7175854b7 100644 --- a/src/common/meta/src/ddl/create_view.rs +++ b/src/common/meta/src/ddl/create_view.rs @@ -297,12 +297,33 @@ impl Procedure for CreateViewProcedure { ProcedureState::Done { output: Some(output), } => output.downcast_ref::().copied().map_or_else( - ViewDdlEvent::create_lifecycle, - ViewDdlEvent::create_succeeded, + || { + ViewDdlEvent::create_lifecycle( + &self.data.task.create_view.catalog_name, + &self.data.task.create_view.schema_name, + &self.data.task.create_view.view_name, + ) + }, + |view_id| { + ViewDdlEvent::create_succeeded( + &self.data.task.create_view.catalog_name, + &self.data.task.create_view.schema_name, + &self.data.task.create_view.view_name, + view_id, + ) + }, + ), + _ => ViewDdlEvent::create_lifecycle( + &self.data.task.create_view.catalog_name, + &self.data.task.create_view.schema_name, + &self.data.task.create_view.view_name, ), - _ => ViewDdlEvent::create_lifecycle(), }, - _ => ViewDdlEvent::create_lifecycle(), + _ => ViewDdlEvent::create_lifecycle( + &self.data.task.create_view.catalog_name, + &self.data.task.create_view.schema_name, + &self.data.task.create_view.view_name, + ), }; Some(Box::new(event)) diff --git a/src/common/meta/src/ddl/drop_database.rs b/src/common/meta/src/ddl/drop_database.rs index d84a5e6dc8..7772e966cb 100644 --- a/src/common/meta/src/ddl/drop_database.rs +++ b/src/common/meta/src/ddl/drop_database.rs @@ -202,7 +202,7 @@ impl Procedure for DropDatabaseProcedure { self.event_context.clone(), ) } else { - DatabaseDdlEvent::drop_lifecycle() + DatabaseDdlEvent::drop_lifecycle(&self.context.catalog, &self.context.schema) }; Some(Box::new(event)) } diff --git a/src/common/meta/src/ddl/drop_flow.rs b/src/common/meta/src/ddl/drop_flow.rs index badded386a..99c18ebc13 100644 --- a/src/common/meta/src/ddl/drop_flow.rs +++ b/src/common/meta/src/ddl/drop_flow.rs @@ -239,7 +239,11 @@ impl Procedure for DropFlowProcedure { self.data.task.drop_if_exists, self.data.event_context.clone(), ), - _ => FlowDdlEvent::drop_lifecycle(), + _ => FlowDdlEvent::drop_lifecycle( + &self.data.task.catalog_name, + &self.data.task.flow_name, + self.data.task.flow_id, + ), }; Some(Box::new(event)) diff --git a/src/common/meta/src/ddl/drop_table.rs b/src/common/meta/src/ddl/drop_table.rs index 2dcfa2f7d8..7c8dec55e4 100644 --- a/src/common/meta/src/ddl/drop_table.rs +++ b/src/common/meta/src/ddl/drop_table.rs @@ -421,18 +421,16 @@ impl Procedure for DropTableProcedure { { return None; } + let task = &self.data.task; + let locator = TableDdlLocator::new(&task.catalog, &task.schema, &task.table) + .with_table_id(task.table_id); let event = match &ctx.trigger { - EventTrigger::Submitted => { - let task = &self.data.task; - let locator = TableDdlLocator::new(&task.catalog, &task.schema, &task.table) - .with_table_id(task.table_id); - TableDdlEvent::drop_table_submitted( - locator, - task.drop_if_exists, - self.data.event_context.clone(), - ) - } - _ => TableDdlEvent::lifecycle(TableDdlEventType::DropTable), + EventTrigger::Submitted => TableDdlEvent::drop_table_submitted( + locator, + task.drop_if_exists, + self.data.event_context.clone(), + ), + _ => TableDdlEvent::lifecycle(TableDdlEventType::DropTable, [locator]), }; Some(Box::new(event)) diff --git a/src/common/meta/src/ddl/drop_view.rs b/src/common/meta/src/ddl/drop_view.rs index 53a62e7615..ece1d2b06b 100644 --- a/src/common/meta/src/ddl/drop_view.rs +++ b/src/common/meta/src/ddl/drop_view.rs @@ -231,7 +231,15 @@ impl Procedure for DropViewProcedure { self.data.event_context.clone(), ) } - _ => ViewDdlEvent::drop_lifecycle(), + _ => { + let table_ref = self.data.table_ref(); + ViewDdlEvent::drop_lifecycle( + table_ref.catalog, + table_ref.schema, + table_ref.table, + self.data.view_id(), + ) + } }; Some(Box::new(event)) diff --git a/src/common/meta/src/ddl/event/database.rs b/src/common/meta/src/ddl/event/database.rs index 9d70ea35a5..b24611799c 100644 --- a/src/common/meta/src/ddl/event/database.rs +++ b/src/common/meta/src/ddl/event/database.rs @@ -179,16 +179,16 @@ impl DatabaseDdlEvent { ) } - pub(crate) fn create_lifecycle() -> Self { - Self::lifecycle(CREATE_DATABASE_EVENT_TYPE) + pub(crate) fn create_lifecycle(catalog_name: &str, schema_name: &str) -> Self { + Self::lifecycle(CREATE_DATABASE_EVENT_TYPE, catalog_name, schema_name) } - pub(crate) fn alter_lifecycle() -> Self { - Self::lifecycle(ALTER_DATABASE_EVENT_TYPE) + pub(crate) fn alter_lifecycle(catalog_name: &str, schema_name: &str) -> Self { + Self::lifecycle(ALTER_DATABASE_EVENT_TYPE, catalog_name, schema_name) } - pub(crate) fn drop_lifecycle() -> Self { - Self::lifecycle(DROP_DATABASE_EVENT_TYPE) + pub(crate) fn drop_lifecycle(catalog_name: &str, schema_name: &str) -> Self { + Self::lifecycle(DROP_DATABASE_EVENT_TYPE, catalog_name, schema_name) } fn submitted( @@ -207,11 +207,11 @@ impl DatabaseDdlEvent { } } - fn lifecycle(event_type: &'static str) -> Self { + fn lifecycle(event_type: &'static str, catalog_name: &str, schema_name: &str) -> Self { Self { event_type, - catalog_name: None, - schema_name: None, + catalog_name: Some(catalog_name.to_string()), + schema_name: Some(schema_name.to_string()), payload: None, event_context: None, } diff --git a/src/common/meta/src/ddl/event/flow.rs b/src/common/meta/src/ddl/event/flow.rs index 8a52486ddf..7d0d6cb539 100644 --- a/src/common/meta/src/ddl/event/flow.rs +++ b/src/common/meta/src/ddl/event/flow.rs @@ -117,29 +117,36 @@ impl FlowDdlEvent { } } - /// Builds a lightweight Create Flow lifecycle event. - pub(crate) fn create_lifecycle() -> Self { - Self::lifecycle(CREATE_FLOW_EVENT_TYPE) + /// Builds a Create Flow lifecycle event with its submitted locator. + pub(crate) fn create_lifecycle(catalog_name: &str, flow_name: &str) -> Self { + Self::lifecycle(CREATE_FLOW_EVENT_TYPE, catalog_name, flow_name) } - /// Builds a successful Create Flow event containing only a resolved ID. - pub(crate) fn create_succeeded(flow_id: Option) -> Self { + /// Builds a successful Create Flow event with its submitted locator and resolved ID. + pub(crate) fn create_succeeded( + catalog_name: &str, + flow_name: &str, + flow_id: Option, + ) -> Self { Self { flow_id, - ..Self::lifecycle(CREATE_FLOW_EVENT_TYPE) + ..Self::lifecycle(CREATE_FLOW_EVENT_TYPE, catalog_name, flow_name) } } - /// Builds a lightweight Drop Flow lifecycle event. - pub(crate) fn drop_lifecycle() -> Self { - Self::lifecycle(DROP_FLOW_EVENT_TYPE) + /// Builds a Drop Flow lifecycle event with its submitted locator. + pub(crate) fn drop_lifecycle(catalog_name: &str, flow_name: &str, flow_id: u32) -> Self { + Self { + flow_id: Some(flow_id), + ..Self::lifecycle(DROP_FLOW_EVENT_TYPE, catalog_name, flow_name) + } } - fn lifecycle(event_type: &'static str) -> Self { + fn lifecycle(event_type: &'static str, catalog_name: &str, flow_name: &str) -> Self { Self { event_type, - catalog_name: None, - flow_name: None, + catalog_name: Some(catalog_name.to_string()), + flow_name: Some(flow_name.to_string()), flow_id: None, payload: None, event_context: None, diff --git a/src/common/meta/src/ddl/event/table.rs b/src/common/meta/src/ddl/event/table.rs index 8608b3151a..9ae6f53076 100644 --- a/src/common/meta/src/ddl/event/table.rs +++ b/src/common/meta/src/ddl/event/table.rs @@ -103,6 +103,7 @@ impl TableDdlLocator { } /// Creates a locator containing only a table ID. + #[cfg(feature = "enterprise")] pub(crate) fn from_table_id(table_id: TableId) -> Self { Self { table_id: Some(table_id), @@ -361,36 +362,32 @@ impl TableDdlEvent { ) } - /// Builds a lightweight lifecycle event with null domain columns and payload. - pub(crate) fn lifecycle(event_type: TableDdlEventType) -> Self { + /// Builds a lifecycle event with stable object locators and no intent payload. + pub(crate) fn lifecycle( + event_type: TableDdlEventType, + locators: impl IntoIterator, + ) -> Self { Self { event_type, - locators: vec![TableDdlLocator::default()], + locators: locators.into_iter().collect(), payload: None, event_context: None, } } - /// Builds a Create Table success event containing only the allocated table ID. - pub(crate) fn create_table_succeeded(table_id: TableId) -> Self { - Self { - event_type: TableDdlEventType::CreateTable, - locators: vec![TableDdlLocator::from_table_id(table_id)], - payload: None, - event_context: None, - } + /// Builds a Create Table success event containing the submitted locator and allocated ID. + pub(crate) fn create_table_succeeded(locator: TableDdlLocator, table_id: TableId) -> Self { + Self::lifecycle( + TableDdlEventType::CreateTable, + [locator.with_table_id(table_id)], + ) } /// Builds Create Logical Tables success rows from their allocated locators. pub(crate) fn create_logical_tables_succeeded( locators: impl IntoIterator, ) -> Self { - Self { - event_type: TableDdlEventType::CreateLogicalTables, - locators: locators.into_iter().collect(), - payload: None, - event_context: None, - } + Self::lifecycle(TableDdlEventType::CreateLogicalTables, locators) } fn submitted( diff --git a/src/common/meta/src/ddl/event/view.rs b/src/common/meta/src/ddl/event/view.rs index d6dcf2c5f1..c1d5846fcf 100644 --- a/src/common/meta/src/ddl/event/view.rs +++ b/src/common/meta/src/ddl/event/view.rs @@ -121,19 +121,38 @@ impl ViewDdlEvent { ) } - /// Builds a lightweight create-view lifecycle event with no locator data. - pub(crate) fn create_lifecycle() -> Self { - Self::lifecycle(CREATE_VIEW_EVENT_TYPE) + /// Builds a create-view lifecycle event with its submitted locator. + pub(crate) fn create_lifecycle(catalog_name: &str, schema_name: &str, view_name: &str) -> Self { + Self::lifecycle(CREATE_VIEW_EVENT_TYPE, catalog_name, schema_name, view_name) } - /// Builds the successful create-view row that carries only the allocated id. - pub(crate) fn create_succeeded(view_id: u32) -> Self { - Self::succeeded(CREATE_VIEW_EVENT_TYPE, view_id) + /// Builds the successful create-view row with its submitted locator and allocated ID. + pub(crate) fn create_succeeded( + catalog_name: &str, + schema_name: &str, + view_name: &str, + view_id: u32, + ) -> Self { + Self::succeeded( + CREATE_VIEW_EVENT_TYPE, + catalog_name, + schema_name, + view_name, + view_id, + ) } - /// Builds a lightweight drop-view lifecycle event with no locator data. - pub(crate) fn drop_lifecycle() -> Self { - Self::lifecycle(DROP_VIEW_EVENT_TYPE) + /// Builds a drop-view lifecycle event with its submitted locator and ID. + pub(crate) fn drop_lifecycle( + catalog_name: &str, + schema_name: &str, + view_name: &str, + view_id: u32, + ) -> Self { + Self { + view_id: Some(view_id), + ..Self::lifecycle(DROP_VIEW_EVENT_TYPE, catalog_name, schema_name, view_name) + } } fn submitted( @@ -156,24 +175,35 @@ impl ViewDdlEvent { } } - fn lifecycle(event_type: &'static str) -> Self { + fn lifecycle( + event_type: &'static str, + catalog_name: &str, + schema_name: &str, + view_name: &str, + ) -> Self { Self { event_type, - catalog_name: None, - schema_name: None, - view_name: None, + catalog_name: Some(catalog_name.to_string()), + schema_name: Some(schema_name.to_string()), + view_name: Some(view_name.to_string()), view_id: None, payload: None, event_context: None, } } - fn succeeded(event_type: &'static str, view_id: u32) -> Self { + fn succeeded( + event_type: &'static str, + catalog_name: &str, + schema_name: &str, + view_name: &str, + view_id: u32, + ) -> Self { Self { event_type, - catalog_name: None, - schema_name: None, - view_name: None, + catalog_name: Some(catalog_name.to_string()), + schema_name: Some(schema_name.to_string()), + view_name: Some(view_name.to_string()), view_id: Some(view_id), payload: None, event_context: None, diff --git a/src/common/meta/src/ddl/purge_dropped_table.rs b/src/common/meta/src/ddl/purge_dropped_table.rs index 111b79cabe..c81aa043dc 100644 --- a/src/common/meta/src/ddl/purge_dropped_table.rs +++ b/src/common/meta/src/ddl/purge_dropped_table.rs @@ -226,6 +226,20 @@ impl PurgeDroppedTableProcedure { fn executor(&self) -> DropTableExecutor { DropTableExecutor::new(self.data.table_name().clone(), self.data.table_id(), false) } + + fn event_locator(&self) -> TableDdlLocator { + let table_id = self.data.task.table_id; + let Some(table_name) = self.data.table_name.as_ref() else { + return TableDdlLocator::from_table_id(table_id); + }; + + TableDdlLocator::new( + &table_name.catalog_name, + &table_name.schema_name, + &table_name.table_name, + ) + .with_table_id(table_id) + } } #[async_trait] @@ -274,17 +288,12 @@ impl Procedure for PurgeDroppedTableProcedure { { return None; } - let event = match &ctx.trigger { - EventTrigger::Submitted => { - let locator = TableDdlLocator::from_table_id(self.data.task.table_id); - TableDdlEvent::purge_dropped_table_submitted( - locator, - self.data.event_context.clone(), - ) - } - _ => TableDdlEvent::lifecycle(TableDdlEventType::PurgeDroppedTable), + let locator = self.event_locator(); + let event = if ctx.trigger == EventTrigger::Submitted { + TableDdlEvent::purge_dropped_table_submitted(locator, self.data.event_context.clone()) + } else { + TableDdlEvent::lifecycle(TableDdlEventType::PurgeDroppedTable, [locator]) }; - Some(Box::new(event)) } } diff --git a/src/common/meta/src/ddl/tests/event/database.rs b/src/common/meta/src/ddl/tests/event/database.rs index 8aa85c2fbb..29cb48f7b8 100644 --- a/src/common/meta/src/ddl/tests/event/database.rs +++ b/src/common/meta/src/ddl/tests/event/database.rs @@ -156,30 +156,33 @@ fn test_drop_database_submitted_event_contract() { } #[test] -fn test_database_lifecycle_events_have_fixed_schema_and_null_intent() { +fn test_database_lifecycle_events_preserve_locator_and_null_intent() { for (event, event_type) in [ ( - DatabaseDdlEvent::create_lifecycle(), + DatabaseDdlEvent::create_lifecycle("greptime", "metrics"), CREATE_DATABASE_EVENT_TYPE, ), ( - DatabaseDdlEvent::alter_lifecycle(), + DatabaseDdlEvent::alter_lifecycle("greptime", "metrics"), ALTER_DATABASE_EVENT_TYPE, ), - (DatabaseDdlEvent::drop_lifecycle(), DROP_DATABASE_EVENT_TYPE), + ( + DatabaseDdlEvent::drop_lifecycle("greptime", "metrics"), + DROP_DATABASE_EVENT_TYPE, + ), ] { - assert_event_locator(&event, event_type, None, None); + assert_event_locator(&event, event_type, Some("greptime"), Some("metrics")); assert_eq!(event.json_payload().unwrap(), serde_json::Value::Null); } assert_eq!( - DatabaseDdlEvent::create_lifecycle().extra_schema(), + DatabaseDdlEvent::create_lifecycle("greptime", "metrics").extra_schema(), DatabaseDdlEvent::create_submitted( "c", "s", false, &HashMap::new(), - EventContext::default() + EventContext::default(), ) .extra_schema() ); @@ -202,7 +205,7 @@ fn test_database_events_preserve_procedure_envelope_contract() { ); let lifecycle = ProcedureEvent::new( procedure_id, - Box::new(DatabaseDdlEvent::create_lifecycle()), + Box::new(DatabaseDdlEvent::create_lifecycle("greptime", "metrics")), ProcedureState::Done { output: None }, EventTrigger::Succeeded, ); @@ -220,8 +223,8 @@ fn test_database_events_preserve_procedure_envelope_contract() { CREATE_DATABASE_EVENT_TYPE, "Done", "Succeeded", - None, - None, + Some("greptime"), + Some("metrics"), ); assert_eq!(lifecycle.json_payload().unwrap(), serde_json::Value::Null); } @@ -282,6 +285,7 @@ fn assert_event_locator( catalog_name: Option<&str>, schema_name: Option<&str>, ) { + let has_event_context = !event.json_payload().unwrap().is_null(); assert_event_contract( event, event_type, @@ -298,7 +302,7 @@ fn assert_event_locator( Value { value_data: schema_name.map(|value| ValueData::StringValue(value.to_string())), }, - if catalog_name.is_some() { + if has_event_context { default_event_context_value() } else { Value { value_data: None } @@ -316,6 +320,7 @@ fn assert_procedure_event_contract( catalog_name: Option<&str>, schema_name: Option<&str>, ) { + let has_event_context = procedure_trigger == "Submitted"; assert_event_contract( event, event_type, @@ -348,7 +353,7 @@ fn assert_procedure_event_contract( Value { value_data: schema_name.map(|value| ValueData::StringValue(value.to_string())), }, - if catalog_name.is_some() { + if has_event_context { default_event_context_value() } else { Value { value_data: None } diff --git a/src/common/meta/src/ddl/tests/event/flow.rs b/src/common/meta/src/ddl/tests/event/flow.rs index b275819791..39f24b74e5 100644 --- a/src/common/meta/src/ddl/tests/event/flow.rs +++ b/src/common/meta/src/ddl/tests/event/flow.rs @@ -99,9 +99,18 @@ fn test_flow_submitted_event_contracts() { #[test] fn test_flow_lifecycle_events_have_fixed_schema_and_null_intent() { for (event, event_type) in [ - (FlowDdlEvent::create_lifecycle(), CREATE_FLOW_EVENT_TYPE), - (FlowDdlEvent::create_succeeded(None), CREATE_FLOW_EVENT_TYPE), - (FlowDdlEvent::drop_lifecycle(), DROP_FLOW_EVENT_TYPE), + ( + FlowDdlEvent::create_lifecycle("greptime", "metrics"), + CREATE_FLOW_EVENT_TYPE, + ), + ( + FlowDdlEvent::create_succeeded("greptime", "metrics", None), + CREATE_FLOW_EVENT_TYPE, + ), + ( + FlowDdlEvent::drop_lifecycle("greptime", "metrics", 42), + DROP_FLOW_EVENT_TYPE, + ), ] { assert_event_contract( &event, @@ -109,9 +118,13 @@ fn test_flow_lifecycle_events_have_fixed_schema_and_null_intent() { &flow_schema(), &[Row { values: vec![ - Value { value_data: None }, - Value { value_data: None }, - Value { value_data: None }, + ValueData::StringValue("greptime".to_string()).into(), + ValueData::StringValue("metrics".to_string()).into(), + if event_type == DROP_FLOW_EVENT_TYPE { + ValueData::U32Value(42).into() + } else { + Value { value_data: None } + }, Value { value_data: None }, ], }], @@ -119,15 +132,15 @@ fn test_flow_lifecycle_events_have_fixed_schema_and_null_intent() { assert_eq!(event.json_payload().unwrap(), serde_json::Value::Null); } - let event = FlowDdlEvent::create_succeeded(Some(42)); + let event = FlowDdlEvent::create_succeeded("greptime", "metrics", Some(42)); assert_event_contract( &event, CREATE_FLOW_EVENT_TYPE, &flow_schema(), &[Row { values: vec![ - Value { value_data: None }, - Value { value_data: None }, + ValueData::StringValue("greptime".to_string()).into(), + ValueData::StringValue("metrics".to_string()).into(), ValueData::U32Value(42).into(), Value { value_data: None }, ], @@ -156,7 +169,11 @@ fn test_flow_events_preserve_procedure_envelope_contract() { ); let succeeded = ProcedureEvent::new( procedure_id, - Box::new(FlowDdlEvent::create_succeeded(Some(42))), + Box::new(FlowDdlEvent::create_succeeded( + "greptime", + "metrics", + Some(42), + )), ProcedureState::Done { output: None }, EventTrigger::Succeeded, ); @@ -178,8 +195,8 @@ fn test_flow_events_preserve_procedure_envelope_contract() { "Done", "Succeeded", FlowEventLocator { - catalog_name: None, - flow_name: None, + catalog_name: Some("greptime"), + flow_name: Some("metrics"), flow_id: Some(42), }, ); @@ -257,7 +274,7 @@ fn assert_procedure_event_contract( .map(ValueData::U32Value) .map(Into::into) .unwrap_or(Value { value_data: None }), - if locator.catalog_name.is_some() { + if trigger == "Submitted" { default_event_context_value() } else { Value { value_data: None } diff --git a/src/common/meta/src/ddl/tests/event/table.rs b/src/common/meta/src/ddl/tests/event/table.rs index 90e7d83b3f..3326cb0680 100644 --- a/src/common/meta/src/ddl/tests/event/table.rs +++ b/src/common/meta/src/ddl/tests/event/table.rs @@ -108,13 +108,26 @@ fn submitted_event_contracts_are_bounded_and_fixed() { .all(|column| column.semantic_type == SemanticType::Field as i32) ); - let lifecycle = TableDdlEvent::lifecycle(case.event_type); + let lifecycle = TableDdlEvent::lifecycle( + case.event_type, + [TableDdlLocator::new( + DEFAULT_CATALOG_NAME, + DEFAULT_SCHEMA_NAME, + "lifecycle", + )], + ); assert_eq!(lifecycle.extra_schema(), schema); assert_eq!(lifecycle.json_payload().unwrap(), JsonValue::Null); - assert_eq!( - lifecycle.extra_rows().unwrap()[0].values, - vec![Value::default(); schema.len()] - ); + assert_eq!(lifecycle.extra_rows().unwrap()[0].values, { + let mut values = table_locator_values(Some("lifecycle"), None); + if matches!( + case.event_type, + TableDdlEventType::CreateLogicalTables | TableDdlEventType::AlterLogicalTables + ) { + values.push(Value::default()); + } + values + }); } } @@ -158,6 +171,16 @@ fn later_lifecycle_events_are_uniform() { for case in procedure_cases() { let submitted = event_for(case.procedure.as_ref(), EventTrigger::Submitted); let schema = submitted.extra_schema(); + let expected_rows = submitted + .extra_rows() + .unwrap() + .into_iter() + .map(|mut row| { + // Lifecycle events retain locators but not submitted event context. + *row.values.last_mut().unwrap() = Value::default(); + row.values + }) + .collect::>(); for trigger in &triggers { let event = event_for(case.procedure.as_ref(), trigger.clone()); @@ -166,8 +189,13 @@ fn later_lifecycle_events_are_uniform() { assert_eq!(event.extra_schema(), schema); assert_eq!(event.json_payload().unwrap(), JsonValue::Null); assert_eq!( - event.extra_rows().unwrap()[0].values, - vec![Value::default(); schema.len()] + event + .extra_rows() + .unwrap() + .into_iter() + .map(|row| row.values) + .collect::>(), + expected_rows ); } } @@ -193,7 +221,7 @@ fn create_success_events_keep_allocated_ids() { assert_eq!(event.json_payload().unwrap(), JsonValue::Null); assert_eq!( event.extra_rows().unwrap()[0].values, - table_locator_values(None, Some(42)) + table_locator_values(Some("create_success"), Some(42)) ); let logical_tables = CreateLogicalTablesProcedure::new( diff --git a/src/common/meta/src/ddl/tests/event/view.rs b/src/common/meta/src/ddl/tests/event/view.rs index 10947e45ee..cabe7418ff 100644 --- a/src/common/meta/src/ddl/tests/event/view.rs +++ b/src/common/meta/src/ddl/tests/event/view.rs @@ -102,26 +102,44 @@ fn test_view_submitted_event_contracts() { #[test] fn test_view_lifecycle_event_contracts() { for (event, event_type) in [ - (ViewDdlEvent::create_lifecycle(), CREATE_VIEW_EVENT_TYPE), - (ViewDdlEvent::drop_lifecycle(), DROP_VIEW_EVENT_TYPE), + ( + ViewDdlEvent::create_lifecycle("greptime", "public", "v_metrics"), + CREATE_VIEW_EVENT_TYPE, + ), + ( + ViewDdlEvent::drop_lifecycle("greptime", "public", "v_metrics", 42), + DROP_VIEW_EVENT_TYPE, + ), ] { - assert_lightweight_event(&event, event_type); + assert_eq!(event.json_payload().unwrap(), serde_json::Value::Null); + assert_view_event_contract( + &event, + event_type, + ViewEventLocator { + catalog_name: Some("greptime"), + schema_name: Some("public"), + view_name: Some("v_metrics"), + view_id: (event_type == DROP_VIEW_EVENT_TYPE).then_some(42), + }, + ); } - let event = ViewDdlEvent::create_succeeded(84); + let event = ViewDdlEvent::create_succeeded("greptime", "public", "v_metrics", 84); assert_view_event_contract( &event, CREATE_VIEW_EVENT_TYPE, ViewEventLocator { + catalog_name: Some("greptime"), + schema_name: Some("public"), + view_name: Some("v_metrics"), view_id: Some(84), - ..Default::default() }, ); assert_eq!(event.json_payload().unwrap(), serde_json::Value::Null); } #[test] -fn test_view_procedures_emit_lightweight_lifecycle_events() { +fn test_view_procedures_preserve_lifecycle_locators() { let create = CreateViewProcedure::new( test_create_view_task("view_name"), EventContext::default(), @@ -147,18 +165,36 @@ fn test_view_procedures_emit_lightweight_lifecycle_events() { EventTrigger::Poisoned, ]; - for (procedure, event_type) in [ - (&create as &dyn Procedure, CREATE_VIEW_EVENT_TYPE), - (&drop as &dyn Procedure, DROP_VIEW_EVENT_TYPE), + for (procedure, event_type, view_id) in [ + (&create as &dyn Procedure, CREATE_VIEW_EVENT_TYPE, None), + (&drop as &dyn Procedure, DROP_VIEW_EVENT_TYPE, Some(42)), ] { for trigger in &triggers { let event = event_for(procedure, trigger.clone()); - assert_lightweight_event(event.as_ref(), event_type); + assert_view_event_contract( + event.as_ref(), + event_type, + ViewEventLocator { + catalog_name: Some("greptime"), + schema_name: Some("public"), + view_name: Some("view_name"), + view_id, + }, + ); } } let event = event_for(&drop, EventTrigger::Succeeded); - assert_lightweight_event(event.as_ref(), DROP_VIEW_EVENT_TYPE); + assert_view_event_contract( + event.as_ref(), + DROP_VIEW_EVENT_TYPE, + ViewEventLocator { + catalog_name: Some("greptime"), + schema_name: Some("public"), + view_name: Some("view_name"), + view_id: Some(42), + }, + ); } #[test] @@ -177,8 +213,10 @@ fn test_create_view_succeeded_output_mapping() { event.as_ref(), CREATE_VIEW_EVENT_TYPE, ViewEventLocator { + catalog_name: Some("greptime"), + schema_name: Some("public"), + view_name: Some("view_name"), view_id: Some(84), - ..Default::default() }, ); assert_eq!(event.json_payload().unwrap(), serde_json::Value::Null); @@ -187,7 +225,16 @@ fn test_create_view_succeeded_output_mapping() { for output in invalid_outputs { let state = ProcedureState::Done { output }; let event = event_for_state(&procedure, EventTrigger::Succeeded, &state); - assert_lightweight_event(event.as_ref(), CREATE_VIEW_EVENT_TYPE); + assert_view_event_contract( + event.as_ref(), + CREATE_VIEW_EVENT_TYPE, + ViewEventLocator { + catalog_name: Some("greptime"), + schema_name: Some("public"), + view_name: Some("view_name"), + view_id: None, + }, + ); } } @@ -233,7 +280,12 @@ fn test_view_event_procedure_envelope_contract() { ); let succeeded = ProcedureEvent::new( procedure_id, - Box::new(ViewDdlEvent::create_succeeded(42)), + Box::new(ViewDdlEvent::create_succeeded( + "greptime", + "public", + "view_name", + 42, + )), ProcedureState::Done { output: None }, EventTrigger::Succeeded, ); @@ -256,8 +308,10 @@ fn test_view_event_procedure_envelope_contract() { "Done", "Succeeded", ViewEventLocator { + catalog_name: Some("greptime"), + schema_name: Some("public"), + view_name: Some("view_name"), view_id: Some(42), - ..Default::default() }, ); } @@ -280,11 +334,6 @@ impl ViewEventLocator<'_> { .map(ValueData::U32Value) .map(Into::into) .unwrap_or_default(), - if self.catalog_name.is_some() { - default_event_context_value() - } else { - Value { value_data: None } - }, ] } } @@ -300,19 +349,13 @@ fn view_schema() -> Vec { } fn assert_view_event_contract(event: &dyn Event, event_type: &str, locator: ViewEventLocator<'_>) { - assert_event_contract( - event, - event_type, - &view_schema(), - &[Row { - values: locator.values(), - }], - ); -} - -fn assert_lightweight_event(event: &dyn Event, event_type: &str) { - assert_view_event_contract(event, event_type, ViewEventLocator::default()); - assert_eq!(event.json_payload().unwrap(), serde_json::Value::Null); + let mut values = locator.values(); + values.push(if event.json_payload().unwrap().is_null() { + Value { value_data: None } + } else { + default_event_context_value() + }); + assert_event_contract(event, event_type, &view_schema(), &[Row { values }]); } fn assert_procedure_event_contract( @@ -337,6 +380,11 @@ fn assert_procedure_event_contract( procedure_trigger_value(trigger), ]; values.extend(locator.values()); + values.push(if trigger == "Submitted" { + default_event_context_value() + } else { + Value { value_data: None } + }); assert_event_contract(event, event_type, &schema, &[Row { values }]); } diff --git a/src/common/meta/src/ddl/truncate_table.rs b/src/common/meta/src/ddl/truncate_table.rs index 9d0ed09e8c..6a561b62af 100644 --- a/src/common/meta/src/ddl/truncate_table.rs +++ b/src/common/meta/src/ddl/truncate_table.rs @@ -99,18 +99,16 @@ impl Procedure for TruncateTableProcedure { { return None; } + let task = &self.data.task; + let locator = TableDdlLocator::new(&task.catalog, &task.schema, &task.table) + .with_table_id(task.table_id); let event = match &ctx.trigger { - EventTrigger::Submitted => { - let task = &self.data.task; - let locator = TableDdlLocator::new(&task.catalog, &task.schema, &task.table) - .with_table_id(task.table_id); - TableDdlEvent::truncate_table_submitted( - locator, - task.time_ranges.len(), - self.data.event_context.clone(), - ) - } - _ => TableDdlEvent::lifecycle(TableDdlEventType::TruncateTable), + EventTrigger::Submitted => TableDdlEvent::truncate_table_submitted( + locator, + task.time_ranges.len(), + self.data.event_context.clone(), + ), + _ => TableDdlEvent::lifecycle(TableDdlEventType::TruncateTable, [locator]), }; Some(Box::new(event)) diff --git a/src/common/meta/src/ddl/undrop_table.rs b/src/common/meta/src/ddl/undrop_table.rs index 635d15ee6a..3295b1163e 100644 --- a/src/common/meta/src/ddl/undrop_table.rs +++ b/src/common/meta/src/ddl/undrop_table.rs @@ -380,6 +380,23 @@ impl UndropTableProcedure { } } +impl UndropTableProcedure { + fn event_locator(&self) -> TableDdlLocator { + self.data + .table_name + .as_ref() + .map(|table_name| { + TableDdlLocator::new( + &table_name.catalog_name, + &table_name.schema_name, + &table_name.table_name, + ) + }) + .unwrap_or_default() + .with_table_id(self.data.task.table_id) + } +} + #[async_trait] impl Procedure for UndropTableProcedure { fn type_name(&self) -> &str { @@ -437,24 +454,12 @@ impl Procedure for UndropTableProcedure { { return None; } + let locator = self.event_locator(); let event = match &ctx.trigger { EventTrigger::Submitted => { - let locator = self - .data - .table_name - .as_ref() - .map(|table_name| { - TableDdlLocator::new( - &table_name.catalog_name, - &table_name.schema_name, - &table_name.table_name, - ) - }) - .unwrap_or_default() - .with_table_id(self.data.task.table_id); TableDdlEvent::undrop_table_submitted(locator, self.data.event_context.clone()) } - _ => TableDdlEvent::lifecycle(TableDdlEventType::UndropTable), + _ => TableDdlEvent::lifecycle(TableDdlEventType::UndropTable, [locator]), }; Some(Box::new(event)) diff --git a/src/meta-srv/src/event/repartition.rs b/src/meta-srv/src/event/repartition.rs index ad47fbf9d4..958bc4e11d 100644 --- a/src/meta-srv/src/event/repartition.rs +++ b/src/meta-srv/src/event/repartition.rs @@ -95,12 +95,12 @@ impl RepartitionEvent { } } - pub(crate) fn lifecycle() -> Self { + pub(crate) fn lifecycle(persistent_ctx: &RepartitionPersistentContext) -> Self { Self { - catalog_name: None, - schema_name: None, - table_name: None, - table_id: None, + catalog_name: Some(persistent_ctx.catalog_name.clone()), + schema_name: Some(persistent_ctx.schema_name.clone()), + table_name: Some(persistent_ctx.table_name.clone()), + table_id: Some(persistent_ctx.table_id), payload: None, event_context: None, } @@ -279,14 +279,14 @@ impl RepartitionGroupEvent { } } - pub(crate) fn lifecycle() -> Self { + pub(crate) fn lifecycle(persistent_ctx: &GroupPersistentContext) -> Self { Self { - catalog_name: None, - schema_name: None, - table_name: None, - table_id: None, - parent_procedure_id: None, - group_id: None, + catalog_name: Some(persistent_ctx.catalog_name.clone()), + schema_name: Some(persistent_ctx.schema_name.clone()), + table_name: persistent_ctx.table_name.clone(), + table_id: Some(persistent_ctx.table_id), + parent_procedure_id: persistent_ctx.parent_procedure_id.map(|id| id.to_string()), + group_id: Some(persistent_ctx.group_id.to_string()), topology: None, payload: None, } @@ -377,7 +377,7 @@ mod tests { use std::time::Duration; use api::v1::value::ValueData; - use api::v1::{ColumnSchema, Row, Value}; + use api::v1::{ColumnSchema, Row}; use common_event_recorder::Event; use common_event_recorder::event_table::{ PROCEDURE_ERROR_COLUMN, PROCEDURE_ID_COLUMN, PROCEDURE_STATE_COLUMN, @@ -641,22 +641,32 @@ mod tests { } #[test] - fn test_lifecycle_events_are_lightweight() { - let parent = RepartitionEvent::lifecycle(); + fn test_lifecycle_events_preserve_locators_and_null_payloads() { + let parent_ctx = parent_persistent_ctx(); + let parent = RepartitionEvent::lifecycle(&parent_ctx); assert_event_contract( &parent, REPARTITION_EVENT_TYPE, &parent_schema(), - &[null_row(parent_schema().len())], + &[Row { + values: vec![ + ValueData::StringValue("greptime".to_string()).into(), + ValueData::StringValue("public".to_string()).into(), + ValueData::StringValue("repartition_events".to_string()).into(), + ValueData::U32Value(1024).into(), + Default::default(), + ], + }], ); assert_eq!(parent.json_payload().unwrap(), serde_json::Value::Null); - let group = RepartitionGroupEvent::lifecycle(); + let group_ctx = new_persistent_context(1024, vec![], vec![]); + let group = RepartitionGroupEvent::lifecycle(&group_ctx); assert_event_contract( &group, REPARTITION_GROUP_EVENT_TYPE, &group_schema(), - &[null_row(group_schema().len())], + &[group.extra_row(None)], ); assert_eq!(group.json_payload().unwrap(), serde_json::Value::Null); } @@ -762,12 +772,4 @@ mod tests { ], } } - - fn null_row(column_count: usize) -> Row { - Row { - values: (0..column_count) - .map(|_| Value { value_data: None }) - .collect(), - } - } } diff --git a/src/meta-srv/src/procedure/repartition.rs b/src/meta-srv/src/procedure/repartition.rs index 62f9f26aab..d8b585b28e 100644 --- a/src/meta-srv/src/procedure/repartition.rs +++ b/src/meta-srv/src/procedure/repartition.rs @@ -809,7 +809,7 @@ impl Procedure for RepartitionProcedure { let start = self.state.as_any().downcast_ref::()?; RepartitionEvent::submitted(&self.context.persistent_ctx, start) } else { - RepartitionEvent::lifecycle() + RepartitionEvent::lifecycle(&self.context.persistent_ctx) }; Some(Box::new(event)) } diff --git a/src/meta-srv/src/procedure/repartition/group.rs b/src/meta-srv/src/procedure/repartition/group.rs index 700c2fdcd1..a3c38b5955 100644 --- a/src/meta-srv/src/procedure/repartition/group.rs +++ b/src/meta-srv/src/procedure/repartition/group.rs @@ -273,7 +273,7 @@ impl Procedure for RepartitionGroupProcedure { let event = if matches!(ctx.trigger, EventTrigger::Submitted) { RepartitionGroupEvent::submitted(&self.context.persistent_ctx) } else { - RepartitionGroupEvent::lifecycle() + RepartitionGroupEvent::lifecycle(&self.context.persistent_ctx) }; Some(Box::new(event)) } diff --git a/tests-integration/tests/database_ddl_event.rs b/tests-integration/tests/database_ddl_event.rs index 79b741b977..6cbc65a267 100644 --- a/tests-integration/tests/database_ddl_event.rs +++ b/tests-integration/tests/database_ddl_event.rs @@ -138,8 +138,8 @@ FROM greptime_private.events WHERE type = '{event_type}' AND procedure_state = 'Done' AND json_path_match(procedure_trigger, '$.type == "Succeeded"') - AND catalog_name IS NULL - AND schema_name IS NULL + AND catalog_name = 'greptime' + AND schema_name = '{DATABASE_NAME}' AND json_is_null(payload)"# ); assert_single_event(instance, &lifecycle).await; diff --git a/tests-integration/tests/flow_ddl_event.rs b/tests-integration/tests/flow_ddl_event.rs index 004d8382ed..c5fc9be94c 100644 --- a/tests-integration/tests/flow_ddl_event.rs +++ b/tests-integration/tests/flow_ddl_event.rs @@ -98,8 +98,8 @@ WHERE type = '{CREATE_FLOW_EVENT_TYPE}' AND procedure_id = '{procedure_id}' AND procedure_state = 'Done' AND json_path_match(procedure_trigger, '$.type == "Succeeded"') - AND catalog_name IS NULL - AND flow_name IS NULL + AND catalog_name = 'greptime' + AND flow_name = '{flow}' AND flow_id IS NOT NULL AND json_is_null(payload)"#, ), @@ -152,9 +152,9 @@ WHERE type = '{DROP_FLOW_EVENT_TYPE}' AND procedure_id = '{procedure_id}' AND procedure_state = 'Done' AND json_path_match(procedure_trigger, '$.type == "Succeeded"') - AND catalog_name IS NULL - AND flow_name IS NULL - AND flow_id IS NULL + AND catalog_name = 'greptime' + AND flow_name = '{flow}' + AND flow_id IS NOT NULL AND json_is_null(payload)"#, ), ) diff --git a/tests-integration/tests/repartition_event.rs b/tests-integration/tests/repartition_event.rs index 3871406cfa..65503a4a8e 100644 --- a/tests-integration/tests/repartition_event.rs +++ b/tests-integration/tests/repartition_event.rs @@ -129,9 +129,32 @@ WHERE type = 'repartition' +-------------+--------------+-------------+--------------------+----------+-----------------+----------------------+---------------+"; assert_eventually_eq(instance, &query, expected).await; + assert_repartition_lifecycle_event(instance, &procedure_id).await; procedure_id } +async fn assert_repartition_lifecycle_event( + instance: &Arc, + procedure_id: &str, +) { + let query = format!( + r#"SELECT type, catalog_name, schema_name, table_name, table_id +FROM greptime_private.events +WHERE type = 'repartition' + AND procedure_id = '{procedure_id}' + AND json_path_match(procedure_trigger, '$.type == "Succeeded"') + AND json_is_null(payload)"# + ); + let expected = "\ ++-------------+--------------+-------------+--------------------+----------+ +| type | catalog_name | schema_name | table_name | table_id | ++-------------+--------------+-------------+--------------------+----------+ +| repartition | greptime | public | repartition_events | 1024 | ++-------------+--------------+-------------+--------------------+----------+"; + + assert_eventually_eq(instance, &query, expected).await; +} + async fn assert_repartition_group_event( instance: &Arc, parent_procedure_id: &str, @@ -170,6 +193,32 @@ ORDER BY target_region_id"# | repartition_group | greptime | public | repartition_events | 1024 | true | true | 4398046511104 | 0 | true | 4398046511105 | 1 | host >= 10 | +-------------------+--------------+-------------+--------------------+----------+-----------------------------+--------------+------------------+----------------------+----------------+------------------+----------------------+-----------------------+"; + assert_eventually_eq(instance, &query, expected).await; + assert_repartition_group_lifecycle_event(instance, &procedure_id, parent_procedure_id).await; +} + +async fn assert_repartition_group_lifecycle_event( + instance: &Arc, + procedure_id: &str, + parent_procedure_id: &str, +) { + let query = format!( + r#"SELECT type, catalog_name, schema_name, table_name, table_id, + parent_procedure_id = '{parent_procedure_id}' AS matches_parent_procedure_id, + repartition_group_id IS NOT NULL AS has_group_id +FROM greptime_private.events +WHERE type = 'repartition_group' + AND procedure_id = '{procedure_id}' + AND json_path_match(procedure_trigger, '$.type == "Succeeded"') + AND json_is_null(payload)"# + ); + let expected = "\ ++-------------------+--------------+-------------+--------------------+----------+-----------------------------+--------------+ +| type | catalog_name | schema_name | table_name | table_id | matches_parent_procedure_id | has_group_id | ++-------------------+--------------+-------------+--------------------+----------+-----------------------------+--------------+ +| repartition_group | greptime | public | repartition_events | 1024 | true | true | ++-------------------+--------------+-------------+--------------------+----------+-----------------------------+--------------+"; + assert_eventually_eq(instance, &query, expected).await; } diff --git a/tests-integration/tests/table_ddl_event.rs b/tests-integration/tests/table_ddl_event.rs index c53842d66d..5c23e101de 100644 --- a/tests-integration/tests/table_ddl_event.rs +++ b/tests-integration/tests/table_ddl_event.rs @@ -230,23 +230,17 @@ async fn test_table_ddl_procedure_events() { +--------------+------------------+", ) .await; - assert_id_terminal_event( + assert_terminal_event( &frontend, "create_table", &create_table_procedure_id, - table_id, "Succeeded", - "\ -+--------------+ -| type | -+--------------+ -| create_table | -+--------------+", + Some(TABLE), + Some(table_id), ) .await; - // Act / Assert: Alter and Truncate have a rich submitted row and lightweight - // terminal lifecycle row. + // Act / Assert: Alter and Truncate retain their submitted locators on terminal rows. run_sql_with_context( &frontend, &format!("ALTER TABLE {TABLE} ADD COLUMN extra STRING"), @@ -278,11 +272,13 @@ async fn test_table_ddl_procedure_events() { +-------------+------------------+", ) .await; - assert_lightweight_terminal_event( + assert_terminal_event( &frontend, "alter_table", &alter_table_procedure_id, "Succeeded", + Some(TABLE), + Some(table_id), ) .await; run_sql( @@ -322,11 +318,13 @@ async fn test_table_ddl_procedure_events() { +----------------+------------------+", ) .await; - assert_lightweight_terminal_event( + assert_terminal_event( &frontend, "truncate_table", &truncate_table_procedure_id, "Succeeded", + Some(TABLE), + Some(table_id), ) .await; @@ -363,11 +361,13 @@ async fn test_table_ddl_procedure_events() { +------------+------------------+", ) .await; - assert_lightweight_terminal_event( + assert_terminal_event( &frontend, "drop_table", &drop_table_procedure_id, "Succeeded", + Some(TABLE), + Some(table_id), ) .await; #[cfg(feature = "enterprise")] @@ -404,11 +404,13 @@ async fn test_table_ddl_procedure_events() { +--------------+", ) .await; - assert_lightweight_terminal_event( + assert_terminal_event( &frontend, "undrop_table", &undrop_table_procedure_id, "Succeeded", + Some(TABLE), + Some(table_id), ) .await; run_sql(&frontend, &format!("DROP TABLE {TABLE}")).await; @@ -444,11 +446,13 @@ async fn test_table_ddl_procedure_events() { +---------------------+", ) .await; - assert_lightweight_terminal_event( + assert_terminal_event( &frontend, "purge_dropped_table", &purge_table_procedure_id, "Succeeded", + Some(TABLE), + Some(table_id), ) .await; } @@ -518,7 +522,7 @@ async fn test_table_ddl_procedure_events() { .await; // Act / Assert: logical Alter preserves the one-row-per-logical-table submitted - // contract and emits a lightweight terminal row. + // contract and retains the logical-table locator on its terminal row. run_sql_with_context( &frontend, &format!("ALTER TABLE {LOGICAL_TABLE} ADD COLUMN rack STRING PRIMARY KEY"), @@ -551,11 +555,13 @@ async fn test_table_ddl_procedure_events() { +----------------------+--------------------------+", ) .await; - assert_lightweight_terminal_event( + assert_terminal_event( &frontend, "alter_logical_tables", &alter_logical_tables_procedure_id, "Succeeded", + Some(LOGICAL_TABLE), + None, ) .await; } @@ -719,27 +725,46 @@ async fn assert_id_submitted_event( assert_eventually_eq(instance, &query, expected).await; } -async fn assert_id_terminal_event( +async fn assert_terminal_event( instance: &Arc, event_type: &str, procedure_id: &str, - table_id: u32, terminal_trigger: &str, - expected: &str, + table_name: Option<&str>, + table_id: Option, ) { + let table_name_predicate = table_name.map_or_else( + || format!("{} IS NULL", TABLE_NAME_COLUMN.name()), + |table_name| format!("{} = '{table_name}'", TABLE_NAME_COLUMN.name()), + ); + let catalog_and_schema_predicate = table_name.map_or_else( + || { + format!( + "{} IS NULL AND {} IS NULL", + CATALOG_NAME_COLUMN.name(), + SCHEMA_NAME_COLUMN.name(), + ) + }, + |_| { + format!( + "{} = 'greptime' AND {} = 'public'", + CATALOG_NAME_COLUMN.name(), + SCHEMA_NAME_COLUMN.name(), + ) + }, + ); + let table_id_predicate = table_id.map_or_else( + || format!("{} IS NULL", TABLE_ID_COLUMN.name()), + |table_id| format!("{} = {table_id}", TABLE_ID_COLUMN.name()), + ); let query = format!( - "SELECT {} FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{procedure_id}' AND json_path_match({}, '$.type == \"{terminal_trigger}\"') AND {} = {table_id} AND json_is_null({}) AND {} IS NULL AND {} IS NULL AND {} IS NULL", - TYPE_COLUMN.name(), + "SELECT count(*) AS event_count FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{procedure_id}' AND json_path_match({}, '$.type == \"{terminal_trigger}\"') AND json_is_null({}) AND {catalog_and_schema_predicate} AND {table_name_predicate} AND {table_id_predicate}", TYPE_COLUMN.name(), PROCEDURE_ID_COLUMN.name(), PROCEDURE_TRIGGER_COLUMN.name(), - TABLE_ID_COLUMN.name(), PAYLOAD_COLUMN.name(), - CATALOG_NAME_COLUMN.name(), - SCHEMA_NAME_COLUMN.name(), - TABLE_NAME_COLUMN.name(), ); - assert_eventually_eq(instance, &query, expected).await; + assert_single_event(instance, &query).await; } async fn assert_logical_terminal_event( @@ -764,23 +789,3 @@ async fn assert_logical_terminal_event( ); assert_eventually_eq(instance, &query, expected).await; } - -async fn assert_lightweight_terminal_event( - instance: &Arc, - event_type: &str, - procedure_id: &str, - terminal_trigger: &str, -) { - let query = format!( - "SELECT count(*) AS event_count FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{procedure_id}' AND json_path_match({}, '$.type == \"{terminal_trigger}\"') AND json_is_null({}) AND {} IS NULL AND {} IS NULL AND {} IS NULL AND {} IS NULL", - TYPE_COLUMN.name(), - PROCEDURE_ID_COLUMN.name(), - PROCEDURE_TRIGGER_COLUMN.name(), - PAYLOAD_COLUMN.name(), - CATALOG_NAME_COLUMN.name(), - SCHEMA_NAME_COLUMN.name(), - TABLE_NAME_COLUMN.name(), - TABLE_ID_COLUMN.name(), - ); - assert_single_event(instance, &query).await; -} diff --git a/tests-integration/tests/view_ddl_event.rs b/tests-integration/tests/view_ddl_event.rs index a25b58b684..fce33d09bc 100644 --- a/tests-integration/tests/view_ddl_event.rs +++ b/tests-integration/tests/view_ddl_event.rs @@ -134,9 +134,9 @@ WHERE type = '{CREATE_VIEW_EVENT_TYPE}' AND procedure_id = '{procedure_id}' AND procedure_state = 'Done' AND json_path_match(procedure_trigger, '$.type == "Succeeded"') - AND catalog_name IS NULL - AND schema_name IS NULL - AND view_name IS NULL + AND catalog_name = 'greptime' + AND schema_name = 'public' + AND view_name = '{view}' AND view_id IS NOT NULL AND json_is_null(payload)"#, ), @@ -190,10 +190,10 @@ WHERE type = '{DROP_VIEW_EVENT_TYPE}' AND procedure_id = '{procedure_id}' AND procedure_state = 'Done' AND json_path_match(procedure_trigger, '$.type == "Succeeded"') - AND catalog_name IS NULL - AND schema_name IS NULL - AND view_name IS NULL - AND view_id IS NULL + AND catalog_name = 'greptime' + AND schema_name = 'public' + AND view_name = '{view}' + AND view_id IS NOT NULL AND json_is_null(payload)"#, ), )