diff --git a/src/common/event-recorder/src/event_table.rs b/src/common/event-recorder/src/event_table.rs index 14768946cc..40d6625b5d 100644 --- a/src/common/event-recorder/src/event_table.rs +++ b/src/common/event-recorder/src/event_table.rs @@ -101,9 +101,9 @@ pub const PROCEDURE_ERROR_COLUMN: EventTableColumn = EventTableColumn::new( SemanticType::Field, ); /// The canonical procedure trigger envelope column. -pub const PROCEDURE_TRIGGER_COLUMN: EventTableColumn = EventTableColumn::new( +pub const PROCEDURE_TRIGGER_COLUMN: EventTableColumn = EventTableColumn::json_binary( "procedure_trigger", - ColumnDataType::String, + ColumnDataType::Binary, SemanticType::Field, ); /// The canonical catalog name dimension. @@ -281,6 +281,11 @@ pub fn nullable_json(value: Option<&serde_json::Value>) -> Value { nullable_value(value.map(|value| ValueData::BinaryValue(jsonb::Value::from(value).to_vec()))) } +/// Builds a JSONB API value. +pub fn jsonb_value(value: &serde_json::Value) -> Value { + ValueData::BinaryValue(jsonb::Value::from(value).to_vec()).into() +} + #[cfg(test)] mod tests { use super::*; @@ -319,18 +324,35 @@ mod tests { fn procedure_envelope_schema_preserves_names_types_semantics_and_order() { assert_eq!( procedure_event_column_schemas(), - [ - "procedure_id", - "procedure_state", - "procedure_error", - "procedure_trigger", + vec![ + ColumnSchema { + column_name: "procedure_id".to_string(), + datatype: ColumnDataType::String.into(), + semantic_type: SemanticType::Field.into(), + ..Default::default() + }, + ColumnSchema { + column_name: "procedure_state".to_string(), + datatype: ColumnDataType::String.into(), + semantic_type: SemanticType::Field.into(), + ..Default::default() + }, + ColumnSchema { + column_name: "procedure_error".to_string(), + datatype: ColumnDataType::String.into(), + semantic_type: SemanticType::Field.into(), + ..Default::default() + }, + ColumnSchema { + column_name: "procedure_trigger".to_string(), + datatype: ColumnDataType::Binary.into(), + semantic_type: SemanticType::Field.into(), + datatype_extension: Some(ColumnDataTypeExtension { + type_ext: Some(TypeExt::JsonType(JsonTypeExtension::JsonBinary.into())), + }), + ..Default::default() + }, ] - .map(|column_name| ColumnSchema { - column_name: column_name.to_string(), - datatype: ColumnDataType::String.into(), - semantic_type: SemanticType::Field.into(), - ..Default::default() - }) ); } diff --git a/src/common/event-recorder/src/recorder.rs b/src/common/event-recorder/src/recorder.rs index d7da4eca9b..b6c18ed08d 100644 --- a/src/common/event-recorder/src/recorder.rs +++ b/src/common/event-recorder/src/recorder.rs @@ -34,7 +34,9 @@ use tokio::time::sleep; use tokio_util::sync::CancellationToken; use crate::error::{MismatchedSchemaSnafu, Result}; -use crate::event_table::{PAYLOAD_COLUMN, TIMESTAMP_COLUMN, TYPE_COLUMN, base_column_schemas}; +use crate::event_table::{ + PAYLOAD_COLUMN, TIMESTAMP_COLUMN, TYPE_COLUMN, base_column_schemas, jsonb_value, +}; /// The default table name for storing the events. pub const DEFAULT_EVENTS_TABLE_NAME: &str = "events"; @@ -194,7 +196,7 @@ pub fn build_row_inserts_request(events: &[&Box]) -> Result format!("{error:?}"), _ => String::new(), }; - let trigger = self.trigger.to_string(); + let trigger = serde_json::to_value(&self.trigger).context(SerializeEventSnafu)?; for internal_event_extra_row in internal_event_extra_rows.iter_mut() { let mut values = Vec::with_capacity(4 + internal_event_extra_row.values.len()); @@ -103,7 +104,7 @@ impl Event for ProcedureEvent { ValueData::StringValue(procedure_id.clone()).into(), ValueData::StringValue(state.clone()).into(), ValueData::StringValue(error.clone()).into(), - ValueData::StringValue(trigger.clone()).into(), + jsonb_value(&trigger), ]); values.append(&mut internal_event_extra_row.values); rows.push(Row { values }); @@ -126,8 +127,13 @@ mod tests { use common_error::mock::MockError; use common_error::status_code::StatusCode; use common_event_recorder::Event; + use common_event_recorder::event_table::{PROCEDURE_TRIGGER_COLUMN, jsonb_value}; + use serde_json::json; - use crate::{Error, EventTrigger, ProcedureEvent, ProcedureId, ProcedureState}; + use crate::{ + ChildSubmissionOutcome, Error, EventTrigger, ProcedureEvent, ProcedureId, ProcedureState, + RetryPhase, + }; #[derive(Debug)] struct TestEvent; @@ -184,7 +190,7 @@ mod tests { ValueData::StringValue(procedure_id.to_string()).into(), ValueData::StringValue("Running".to_string()).into(), ValueData::StringValue(String::new()).into(), - ValueData::StringValue("Submitted".to_string()).into(), + jsonb_value(&json!({"type": "Submitted"})), ValueData::StringValue("test_event1".to_string()).into(), ], }, @@ -193,7 +199,7 @@ mod tests { ValueData::StringValue(procedure_id.to_string()).into(), ValueData::StringValue("Running".to_string()).into(), ValueData::StringValue(String::new()).into(), - ValueData::StringValue("Submitted".to_string()).into(), + jsonb_value(&json!({"type": "Submitted"})), ValueData::StringValue("test_event2".to_string()).into(), ], }, @@ -202,7 +208,7 @@ mod tests { ValueData::StringValue(procedure_id.to_string()).into(), ValueData::StringValue("Running".to_string()).into(), ValueData::StringValue(String::new()).into(), - ValueData::StringValue("Submitted".to_string()).into(), + jsonb_value(&json!({"type": "Submitted"})), Value { value_data: None }, ], }, @@ -233,7 +239,7 @@ mod tests { ); assert_eq!( procedure_event_extra_rows[0].values[3], - ValueData::StringValue("Failed".to_string()).into() + jsonb_value(&json!({"type": "Failed"})) ); } @@ -248,44 +254,72 @@ mod tests { assert_eq!( procedure_event.extra_schema(), - [ - "procedure_id", - "procedure_state", - "procedure_error", - "procedure_trigger", - "test_event_column", + vec![ + ColumnSchema { + column_name: "procedure_id".to_string(), + datatype: ColumnDataType::String.into(), + semantic_type: SemanticType::Field.into(), + ..Default::default() + }, + ColumnSchema { + column_name: "procedure_state".to_string(), + datatype: ColumnDataType::String.into(), + semantic_type: SemanticType::Field.into(), + ..Default::default() + }, + ColumnSchema { + column_name: "procedure_error".to_string(), + datatype: ColumnDataType::String.into(), + semantic_type: SemanticType::Field.into(), + ..Default::default() + }, + PROCEDURE_TRIGGER_COLUMN.column_schema(), + ColumnSchema { + column_name: "test_event_column".to_string(), + datatype: ColumnDataType::String.into(), + semantic_type: SemanticType::Field.into(), + ..Default::default() + }, ] - .map(|column_name| ColumnSchema { - column_name: column_name.to_string(), - datatype: ColumnDataType::String.into(), - semantic_type: SemanticType::Field.into(), - ..Default::default() - }) ); } #[test] - fn test_event_trigger_display() { + fn test_event_trigger_serialization() { let procedure_id = ProcedureId::parse_str("00000000-0000-0000-0000-000000000001").unwrap(); - assert_eq!(EventTrigger::Submitted.to_string(), "Submitted"); - assert_eq!(EventTrigger::Recovered.to_string(), "Recovered"); + for (trigger, name) in [ + (EventTrigger::Submitted, "Submitted"), + (EventTrigger::Recovered, "Recovered"), + (EventTrigger::RollingBack, "RollingBack"), + (EventTrigger::Succeeded, "Succeeded"), + (EventTrigger::Failed, "Failed"), + (EventTrigger::Poisoned, "Poisoned"), + ] { + assert_eq!( + serde_json::to_value(trigger).unwrap(), + json!({"type": name}) + ); + } assert_eq!( - EventTrigger::ChildSubmitted { + serde_json::to_value(EventTrigger::ChildSubmitted { procedure_id, - outcome: crate::ChildSubmissionOutcome::Accepted, - } - .to_string(), - "ChildSubmitted(procedure_id=00000000-0000-0000-0000-000000000001, outcome=Accepted)" + outcome: ChildSubmissionOutcome::Accepted, + }) + .unwrap(), + json!({ + "type": "ChildSubmitted", + "procedure_id": "00000000-0000-0000-0000-000000000001", + "outcome": "Accepted", + }) ); assert_eq!( - EventTrigger::Retrying { - phase: crate::RetryPhase::Execute, + serde_json::to_value(EventTrigger::Retrying { + phase: RetryPhase::Execute, attempt: 2, - } - .to_string(), - "Retrying(Execute, 2)" + }) + .unwrap(), + json!({"type": "Retrying", "phase": "Execute", "attempt": 2}) ); - assert_eq!(EventTrigger::RollingBack.to_string(), "RollingBack"); } } diff --git a/src/common/procedure/src/procedure.rs b/src/common/procedure/src/procedure.rs index 84b7298895..c37e1b58bc 100644 --- a/src/common/procedure/src/procedure.rs +++ b/src/common/procedure/src/procedure.rs @@ -255,7 +255,10 @@ pub struct EventContext<'a> { } /// Lifecycle action that causes the framework to invoke [`Procedure::event`]. -#[derive(Debug, Clone, PartialEq, Eq)] +/// +/// It is recorded as a tagged JSON object in the `procedure_trigger` event envelope column. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(tag = "type")] pub enum EventTrigger { /// The procedure was submitted to the manager. Submitted, @@ -286,7 +289,7 @@ pub enum EventTrigger { } /// Phase of a procedure retry. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] pub enum RetryPhase { /// Retrying procedure execution. Execute, @@ -295,7 +298,7 @@ pub enum RetryPhase { } /// Outcome of submitting a child procedure. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub enum ChildSubmissionOutcome { Accepted, AlreadyAccepted, @@ -303,47 +306,6 @@ pub enum ChildSubmissionOutcome { SpawnFailed, } -impl Display for EventTrigger { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - Self::Submitted => write!(f, "Submitted"), - Self::Recovered => write!(f, "Recovered"), - Self::ChildSubmitted { - procedure_id, - outcome, - } => write!( - f, - "ChildSubmitted(procedure_id={procedure_id}, outcome={outcome})" - ), - Self::Retrying { phase, attempt } => write!(f, "Retrying({phase}, {attempt})"), - Self::RollingBack => write!(f, "RollingBack"), - Self::Succeeded => write!(f, "Succeeded"), - Self::Failed => write!(f, "Failed"), - Self::Poisoned => write!(f, "Poisoned"), - } - } -} - -impl Display for RetryPhase { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - Self::Execute => write!(f, "Execute"), - Self::Rollback => write!(f, "Rollback"), - } - } -} - -impl Display for ChildSubmissionOutcome { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - match self { - Self::Accepted => write!(f, "Accepted"), - Self::AlreadyAccepted => write!(f, "AlreadyAccepted"), - Self::ManagerStopped => write!(f, "ManagerStopped"), - Self::SpawnFailed => write!(f, "SpawnFailed"), - } - } -} - #[async_trait] impl Procedure for Box { fn type_name(&self) -> &str { diff --git a/src/meta-srv/src/event/gc.rs b/src/meta-srv/src/event/gc.rs index 8043f52dd7..dd417ca77f 100644 --- a/src/meta-srv/src/event/gc.rs +++ b/src/meta-srv/src/event/gc.rs @@ -214,7 +214,7 @@ mod tests { use common_event_recorder::EventTypeFilter; use common_event_recorder::event_table::{ PROCEDURE_ERROR_COLUMN, PROCEDURE_ID_COLUMN, PROCEDURE_STATE_COLUMN, - PROCEDURE_TRIGGER_COLUMN, + PROCEDURE_TRIGGER_COLUMN, jsonb_value, }; use common_event_recorder::testing::assert_event_contract; use common_meta::key::TableMetadataManager; @@ -526,7 +526,7 @@ mod tests { ValueData::StringValue(procedure_id.to_string()).into(), ValueData::StringValue("Done".to_string()).into(), ValueData::StringValue(String::new()).into(), - ValueData::StringValue("Succeeded".to_string()).into(), + jsonb_value(&serde_json::json!({"type": "Succeeded"})), ]; values.extend( region_row( diff --git a/src/meta-srv/src/event/repartition.rs b/src/meta-srv/src/event/repartition.rs index 2fa6989b90..d266b581f8 100644 --- a/src/meta-srv/src/event/repartition.rs +++ b/src/meta-srv/src/event/repartition.rs @@ -365,7 +365,7 @@ mod tests { use common_event_recorder::Event; use common_event_recorder::event_table::{ PROCEDURE_ERROR_COLUMN, PROCEDURE_ID_COLUMN, PROCEDURE_STATE_COLUMN, - PROCEDURE_TRIGGER_COLUMN, + PROCEDURE_TRIGGER_COLUMN, jsonb_value, }; use common_event_recorder::testing::assert_event_contract; use common_procedure::{EventTrigger, ProcedureEvent, ProcedureId, ProcedureState}; @@ -669,7 +669,7 @@ mod tests { ValueData::StringValue(procedure_id.to_string()).into(), ValueData::StringValue("Running".to_string()).into(), ValueData::StringValue(String::new()).into(), - ValueData::StringValue("Submitted".to_string()).into(), + jsonb_value(&serde_json::json!({"type": "Submitted"})), ValueData::StringValue("greptime".to_string()).into(), ValueData::StringValue("public".to_string()).into(), ValueData::StringValue("repartition_events".to_string()).into(), diff --git a/tests-integration/tests/database_ddl_event.rs b/tests-integration/tests/database_ddl_event.rs index fdb17e69f0..ab8d3ee025 100644 --- a/tests-integration/tests/database_ddl_event.rs +++ b/tests-integration/tests/database_ddl_event.rs @@ -116,7 +116,7 @@ async fn assert_database_event( FROM greptime_private.events WHERE type = '{event_type}' AND procedure_state = 'Running' - AND procedure_trigger = 'Submitted' + AND json_path_match(procedure_trigger, '$.type == "Submitted"') AND catalog_name = 'greptime' AND schema_name = '{DATABASE_NAME}' AND {submitted_payload_predicate}"# @@ -128,7 +128,7 @@ WHERE type = '{event_type}' FROM greptime_private.events WHERE type = '{event_type}' AND procedure_state = 'Done' - AND procedure_trigger = 'Succeeded' + AND json_path_match(procedure_trigger, '$.type == "Succeeded"') AND catalog_name IS NULL AND schema_name IS NULL AND json_is_null(payload)"# diff --git a/tests-integration/tests/flow_ddl_event.rs b/tests-integration/tests/flow_ddl_event.rs index 206389150e..13ca36898a 100644 --- a/tests-integration/tests/flow_ddl_event.rs +++ b/tests-integration/tests/flow_ddl_event.rs @@ -76,7 +76,7 @@ FROM greptime_private.events WHERE type = '{CREATE_FLOW_EVENT_TYPE}' AND procedure_id = '{procedure_id}' AND procedure_state = 'Running' - AND procedure_trigger = 'Submitted' + AND json_path_match(procedure_trigger, '$.type == "Submitted"') AND catalog_name = 'greptime' AND flow_name = '{flow}' AND flow_id IS NULL @@ -96,7 +96,7 @@ FROM greptime_private.events WHERE type = '{CREATE_FLOW_EVENT_TYPE}' AND procedure_id = '{procedure_id}' AND procedure_state = 'Done' - AND procedure_trigger = 'Succeeded' + AND json_path_match(procedure_trigger, '$.type == "Succeeded"') AND catalog_name IS NULL AND flow_name IS NULL AND flow_id IS NOT NULL @@ -116,7 +116,7 @@ FROM greptime_private.events WHERE type = '{DROP_FLOW_EVENT_TYPE}' AND procedure_id = '{procedure_id}' AND procedure_state = 'Running' - AND procedure_trigger = 'Submitted' + AND json_path_match(procedure_trigger, '$.type == "Submitted"') AND catalog_name = 'greptime' AND flow_name = '{flow}' AND flow_id IS NOT NULL @@ -133,7 +133,7 @@ FROM greptime_private.events WHERE type = '{DROP_FLOW_EVENT_TYPE}' AND procedure_id = '{procedure_id}' AND procedure_state = 'Done' - AND procedure_trigger = 'Succeeded' + AND json_path_match(procedure_trigger, '$.type == "Succeeded"') AND catalog_name IS NULL AND flow_name IS NULL AND flow_id IS NULL @@ -155,7 +155,7 @@ async fn find_submitted_procedure_id( FROM greptime_private.events WHERE type = '{event_type}' AND flow_name = '{flow_name}' - AND procedure_trigger = 'Submitted' + AND json_path_match(procedure_trigger, '$.type == "Submitted"') ORDER BY timestamp DESC LIMIT 1"#, ), diff --git a/tests-integration/tests/gc_event.rs b/tests-integration/tests/gc_event.rs index dcfae42be1..3cea55bca7 100644 --- a/tests-integration/tests/gc_event.rs +++ b/tests-integration/tests/gc_event.rs @@ -148,7 +148,7 @@ FROM greptime_private.events WHERE type = 'batch_gc' AND procedure_id = '{procedure_id}' AND procedure_state = 'Done' - AND procedure_trigger = 'Succeeded' + AND json_path_match(procedure_trigger, '$.type == "Succeeded"') AND json_is_null(payload)"#, ); let actual_report: serde_json::Value = diff --git a/tests-integration/tests/region_migration.rs b/tests-integration/tests/region_migration.rs index fff7e6e9d2..2d11966dc7 100644 --- a/tests-integration/tests/region_migration.rs +++ b/tests-integration/tests/region_migration.rs @@ -29,7 +29,6 @@ use common_meta::key::{RegionDistribution, RegionRoleSet, TableMetadataManagerRe use common_meta::peer::Peer; use common_procedure::event::{ EVENTS_TABLE_PROCEDURE_ID_COLUMN_NAME, EVENTS_TABLE_PROCEDURE_STATE_COLUMN_NAME, - EVENTS_TABLE_PROCEDURE_TRIGGER_COLUMN_NAME, }; use common_query::Output; use common_recordbatch::RecordBatches; @@ -51,7 +50,7 @@ use meta_srv::procedure::region_migration::{ RegionMigrationProcedureTask, RegionMigrationTriggerReason, }; use meta_srv::selector::{Selector, SelectorOptions}; -use sea_query::{Expr, Iden, Order, PostgresQueryBuilder, Query}; +use sea_query::{Alias, Expr, Iden, Order, PostgresQueryBuilder, Query}; use servers::error::Result as ServerResult; use servers::query_handler::sql::SqlQueryHandler; use session::context::{QueryContext, QueryContextRef}; @@ -1289,7 +1288,6 @@ enum RegionMigrationEvents { ProcedureId, Timestamp, ProcedureState, - ProcedureTrigger, Schema, Table, EventType, @@ -1308,7 +1306,6 @@ impl Iden for RegionMigrationEvents { Self::ProcedureId => EVENTS_TABLE_PROCEDURE_ID_COLUMN_NAME, Self::Timestamp => EVENTS_TABLE_TIMESTAMP_COLUMN_NAME, Self::ProcedureState => EVENTS_TABLE_PROCEDURE_STATE_COLUMN_NAME, - Self::ProcedureTrigger => EVENTS_TABLE_PROCEDURE_TRIGGER_COLUMN_NAME, Self::Schema => DEFAULT_PRIVATE_SCHEMA_NAME, Self::Table => DEFAULT_EVENTS_TABLE_NAME, Self::EventType => EVENTS_TABLE_TYPE_COLUMN_NAME, @@ -1333,7 +1330,9 @@ async fn check_region_migration_events_system_table( tokio::time::sleep(DEFAULT_FLUSH_INTERVAL_SECONDS * 2).await; // The query is equivalent to the following SQL: - // SELECT region_migration_trigger_reason, procedure_state FROM greptime_private.events WHERE + // SELECT region_migration_trigger_reason, procedure_state, + // json_get_string(procedure_trigger, 'type') AS procedure_trigger + // FROM greptime_private.events WHERE // type = 'region_migration' AND // procedure_id = '${procedure_id}' AND // table_id = ${table_id} AND @@ -1344,7 +1343,10 @@ async fn check_region_migration_events_system_table( let query = Query::select() .column(RegionMigrationEvents::RegionMigrationTriggerReason) .column(RegionMigrationEvents::ProcedureState) - .column(RegionMigrationEvents::ProcedureTrigger) + .expr_as( + Expr::cust("json_get_string(procedure_trigger, 'type')"), + Alias::new("procedure_trigger"), + ) .from((RegionMigrationEvents::Schema, RegionMigrationEvents::Table)) .and_where(Expr::col(RegionMigrationEvents::EventType).eq(REGION_MIGRATION_EVENT_TYPE)) .and_where(Expr::col(RegionMigrationEvents::ProcedureId).eq(procedure_id)) diff --git a/tests-integration/tests/repartition_event.rs b/tests-integration/tests/repartition_event.rs index 86ac8c5f33..3871406cfa 100644 --- a/tests-integration/tests/repartition_event.rs +++ b/tests-integration/tests/repartition_event.rs @@ -118,7 +118,7 @@ async fn assert_repartition_event(instance: &Arc) FROM greptime_private.events WHERE type = 'repartition' AND procedure_id = '{procedure_id}' - AND procedure_trigger = 'Submitted' + AND json_path_match(procedure_trigger, '$.type == "Submitted"') AND table_name = '{TABLE_NAME}'"# ); let expected = "\ @@ -158,7 +158,7 @@ async fn assert_repartition_group_event( FROM greptime_private.events WHERE type = 'repartition_group' AND procedure_id = '{procedure_id}' - AND procedure_trigger = 'Submitted' + AND json_path_match(procedure_trigger, '$.type == "Submitted"') AND table_name = '{TABLE_NAME}' ORDER BY target_region_id"# ); @@ -199,7 +199,7 @@ async fn assert_merge_repartition_group_event( FROM greptime_private.events WHERE type = 'repartition_group' AND procedure_id = '{procedure_id}' - AND procedure_trigger = 'Submitted' + AND json_path_match(procedure_trigger, '$.type == "Submitted"') AND table_name = '{TABLE_NAME}' AND source_partition_expr IS NOT NULL ORDER BY source_region_id"# @@ -226,7 +226,7 @@ async fn find_submitted_procedure_id( r#"SELECT procedure_id FROM greptime_private.events WHERE type = '{event_type}' - AND procedure_trigger = 'Submitted' + AND json_path_match(procedure_trigger, '$.type == "Submitted"') AND {predicate} ORDER BY timestamp DESC LIMIT 1"# diff --git a/tests-integration/tests/table_ddl_event.rs b/tests-integration/tests/table_ddl_event.rs index 26f4afd1a0..6b332dffc3 100644 --- a/tests-integration/tests/table_ddl_event.rs +++ b/tests-integration/tests/table_ddl_event.rs @@ -340,7 +340,7 @@ async fn submitted_procedure_id( table_name: &str, ) -> String { let query = format!( - "SELECT {} FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{table_name}' AND {} = 'Submitted' AND json_path_match({}, '$.version == 1') ORDER BY timestamp DESC LIMIT 1", + "SELECT {} FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{table_name}' AND json_path_match({}, '$.type == \"Submitted\"') AND json_path_match({}, '$.version == 1') ORDER BY timestamp DESC LIMIT 1", PROCEDURE_ID_COLUMN.name(), TYPE_COLUMN.name(), TABLE_NAME_COLUMN.name(), @@ -357,7 +357,7 @@ async fn assert_named_submitted_event( expected: &str, ) { let query = format!( - "SELECT {}, {} AS name FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{procedure_id}' AND {} = 'Submitted' AND json_path_match({}, '$.version == 1')", + "SELECT {}, {} AS name FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{procedure_id}' AND json_path_match({}, '$.type == \"Submitted\"') AND json_path_match({}, '$.version == 1')", TYPE_COLUMN.name(), TABLE_NAME_COLUMN.name(), TYPE_COLUMN.name(), @@ -375,7 +375,7 @@ async fn assert_id_submitted_event( expected: &str, ) { let query = format!( - "SELECT {} FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{procedure_id}' AND {} = 'Submitted' AND json_path_match({}, '$.version == 1')", + "SELECT {} FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{procedure_id}' AND json_path_match({}, '$.type == \"Submitted\"') AND json_path_match({}, '$.version == 1')", TYPE_COLUMN.name(), TYPE_COLUMN.name(), PROCEDURE_ID_COLUMN.name(), @@ -394,7 +394,7 @@ async fn assert_id_terminal_event( expected: &str, ) { let query = format!( - "SELECT {} FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{procedure_id}' AND {} = '{terminal_trigger}' AND {} = {table_id} AND json_is_null({}) AND {} IS NULL AND {} IS NULL AND {} IS NULL", + "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(), TYPE_COLUMN.name(), PROCEDURE_ID_COLUMN.name(), @@ -418,7 +418,7 @@ async fn assert_logical_terminal_event( expected: &str, ) { let query = format!( - "SELECT {}, {} AS name FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{procedure_id}' AND {} = '{terminal_trigger}' AND {} = {table_id} AND {} = {physical_table_id} AND json_is_null({})", + "SELECT {}, {} AS name FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{procedure_id}' AND json_path_match({}, '$.type == \"{terminal_trigger}\"') AND {} = {table_id} AND {} = {physical_table_id} AND json_is_null({})", TYPE_COLUMN.name(), TABLE_NAME_COLUMN.name(), TYPE_COLUMN.name(), @@ -438,7 +438,7 @@ async fn assert_lightweight_terminal_event( terminal_trigger: &str, ) { let query = format!( - "SELECT count(*) AS event_count FROM {EVENTS_TABLE} WHERE {} = '{event_type}' AND {} = '{procedure_id}' AND {} = '{terminal_trigger}' AND json_is_null({}) AND {} IS NULL AND {} IS NULL AND {} IS NULL AND {} IS NULL", + "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(), diff --git a/tests-integration/tests/view_ddl_event.rs b/tests-integration/tests/view_ddl_event.rs index cfc18c4e8b..2f362c256f 100644 --- a/tests-integration/tests/view_ddl_event.rs +++ b/tests-integration/tests/view_ddl_event.rs @@ -111,7 +111,7 @@ FROM greptime_private.events WHERE type = '{CREATE_VIEW_EVENT_TYPE}' AND procedure_id = '{procedure_id}' AND procedure_state = 'Running' - AND procedure_trigger = 'Submitted' + AND json_path_match(procedure_trigger, '$.type == "Submitted"') AND catalog_name = 'greptime' AND schema_name = 'public' AND view_name = '{view}' @@ -132,7 +132,7 @@ FROM greptime_private.events WHERE type = '{CREATE_VIEW_EVENT_TYPE}' AND procedure_id = '{procedure_id}' AND procedure_state = 'Done' - AND procedure_trigger = 'Succeeded' + AND json_path_match(procedure_trigger, '$.type == "Succeeded"') AND catalog_name IS NULL AND schema_name IS NULL AND view_name IS NULL @@ -153,7 +153,7 @@ FROM greptime_private.events WHERE type = '{DROP_VIEW_EVENT_TYPE}' AND procedure_id = '{procedure_id}' AND procedure_state = 'Running' - AND procedure_trigger = 'Submitted' + AND json_path_match(procedure_trigger, '$.type == "Submitted"') AND catalog_name = 'greptime' AND schema_name = 'public' AND view_name = '{view}' @@ -171,7 +171,7 @@ FROM greptime_private.events WHERE type = '{DROP_VIEW_EVENT_TYPE}' AND procedure_id = '{procedure_id}' AND procedure_state = 'Done' - AND procedure_trigger = 'Succeeded' + AND json_path_match(procedure_trigger, '$.type == "Succeeded"') AND catalog_name IS NULL AND schema_name IS NULL AND view_name IS NULL @@ -194,7 +194,7 @@ async fn find_submitted_procedure_id( FROM greptime_private.events WHERE type = '{event_type}' AND view_name = '{view_name}' - AND procedure_trigger = 'Submitted' + AND json_path_match(procedure_trigger, '$.type == "Submitted"') ORDER BY timestamp DESC LIMIT 1"#, ), diff --git a/tests-integration/tests/wal_prune_event.rs b/tests-integration/tests/wal_prune_event.rs index ca595ed757..05ceb1b92d 100644 --- a/tests-integration/tests/wal_prune_event.rs +++ b/tests-integration/tests/wal_prune_event.rs @@ -99,7 +99,7 @@ FROM greptime_private.events WHERE type = 'wal_prune' AND procedure_id = '{procedure_id}' AND procedure_state = 'Done' - AND procedure_trigger = 'Succeeded' + AND json_path_match(procedure_trigger, '$.type == "Succeeded"') AND topic_name = '{topic_name}' AND prunable_entry_id = {pruned_entry_id} AND latest_offset = 3