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

fix(event): preserve procedure lifecycle locators (#8787)

Browse Source 
* fix(event): preserve procedure lifecycle locators

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

* fix(event): preserve dropped table lifecycle locators

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

* test(event): cover lifecycle locators

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

* test(event): fix lifecycle context expectations

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

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
Weny Xu
2026-08-11 15:12:45 +08:00
committed by GitHub
parent 4eeb4052d6
commit 32e215cad0
31 changed files with 562 additions and 315 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/meta/src/ddl/alter_database.rs
+1 -1
View File
@@ -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))
}
src/common/meta/src/ddl/alter_logical_tables.rs
+9 -4
View File
@@ -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();
}
src/common/meta/src/ddl/alter_table.rs
+4 -5
View File
@@ -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))
src/common/meta/src/ddl/create_database.rs
+1 -1
View File
@@ -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))
}
src/common/meta/src/ddl/create_flow.rs
+9 -2
View File
@@ -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))
src/common/meta/src/ddl/create_logical_tables.rs
+34 -32
View File
@@ -212,6 +212,19 @@ impl CreateLogicalTablesProcedure {
}
}
impl CreateLogicalTablesProcedure {
fn event_locators(&self) -> impl Iterator<Item = TableDdlLocator> + '_ {
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::<Vec<TableId>>()
.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))
src/common/meta/src/ddl/create_table.rs
+10 -7
View File
@@ -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::<TableId>()
.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))
src/common/meta/src/ddl/create_view.rs
+25 -4
View File
@@ -297,12 +297,33 @@ impl Procedure for CreateViewProcedure {
ProcedureState::Done {
output: Some(output),
} => output.downcast_ref::<TableId>().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))
src/common/meta/src/ddl/drop_database.rs
+1 -1
View File
@@ -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))
}
src/common/meta/src/ddl/drop_flow.rs
+5 -1
View File
@@ -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))
src/common/meta/src/ddl/drop_table.rs
+9 -11
View File
@@ -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))
src/common/meta/src/ddl/drop_view.rs
+9 -1
View File
@@ -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))
src/common/meta/src/ddl/event/database.rs
+9 -9
View File
@@ -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,
}
src/common/meta/src/ddl/event/flow.rs
+19 -12
View File
@@ -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<u32>) -> 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<u32>,
) -> 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,
src/common/meta/src/ddl/event/table.rs
+14 -17
View File
@@ -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<Item = TableDdlLocator>,
) -> 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<Item = TableDdlLocator>,
) -> Self {
Self {
event_type: TableDdlEventType::CreateLogicalTables,
locators: locators.into_iter().collect(),
payload: None,
event_context: None,
}
Self::lifecycle(TableDdlEventType::CreateLogicalTables, locators)
}
fn submitted(
src/common/meta/src/ddl/event/view.rs
+47 -17
View File
@@ -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,
src/common/meta/src/ddl/purge_dropped_table.rs
+19 -10
View File
@@ -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))
}
}
src/common/meta/src/ddl/tests/event/database.rs
+17 -12
View File
@@ -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 }
src/common/meta/src/ddl/tests/event/flow.rs
+30 -13
View File
@@ -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 }
src/common/meta/src/ddl/tests/event/table.rs
+36 -8
View File
@@ -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::<Vec<_>>();
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::<Vec<_>>(),
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(
src/common/meta/src/ddl/tests/event/view.rs
+81 -33
View File
@@ -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<ColumnSchema> {
}
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 }]);
}
src/common/meta/src/ddl/truncate_table.rs
+9 -11
View File
@@ -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))
src/common/meta/src/ddl/undrop_table.rs
+19 -14
View File
@@ -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))
src/meta-srv/src/event/repartition.rs
+28 -26
View File
@@ -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(),
}
}
}
src/meta-srv/src/procedure/repartition.rs
+1 -1
View File
@@ -809,7 +809,7 @@ impl Procedure for RepartitionProcedure {
let start = self.state.as_any().downcast_ref::<RepartitionStart>()?;
RepartitionEvent::submitted(&self.context.persistent_ctx, start)
} else {
RepartitionEvent::lifecycle()
RepartitionEvent::lifecycle(&self.context.persistent_ctx)
};
Some(Box::new(event))
}
src/meta-srv/src/procedure/repartition/group.rs
+1 -1
View File
@@ -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))
}
tests-integration/tests/database_ddl_event.rs
+2 -2
View File
@@ -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;
tests-integration/tests/flow_ddl_event.rs
+5 -5
View File
@@ -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)"#,
),
)
tests-integration/tests/repartition_event.rs
+49
View File
@@ -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<frontend::instance::Instance>,
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<frontend::instance::Instance>,
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<frontend::instance::Instance>,
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;
}
tests-integration/tests/table_ddl_event.rs
+52 -47
View File
@@ -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<Instance>,
event_type: &str,
procedure_id: &str,
table_id: u32,
terminal_trigger: &str,
expected: &str,
table_name: Option<&str>,
table_id: Option<u32>,
) {
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<Instance>,
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;
}
tests-integration/tests/view_ddl_event.rs
+7 -7
View File
@@ -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)"#,
),
)