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

refactor(event): store procedure trigger as JSONB (#8700)

Browse Source 
* refactor(event): store procedure trigger as JSONB

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

* test(event): fix JSONB procedure trigger assertions

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

* refactor(event): remove procedure trigger display

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

* test(event): update batch GC trigger assertions

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

---------

Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
Weny Xu
2026-07-31 21:22:26 +08:00
committed by GitHub
parent 4e04f92591
commit 1aa35716cb
17 changed files with 156 additions and 136 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/event-recorder/src/event_table.rs
+35 -13
View File
@@ -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()
})
);
}
src/common/event-recorder/src/recorder.rs
+4 -2
View File
@@ -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<dyn Event>]) -> Result<RowInsert
let mut values = Vec::with_capacity(3 + extra_row.values.len());
values.extend([
ValueData::StringValue(event.event_type().to_string()).into(),
ValueData::BinaryValue(jsonb::Value::from(&event.json_payload()?).to_vec()).into(),
jsonb_value(&event.json_payload()?),
ValueData::TimestampNanosecondValue(event.timestamp().value()).into(),
]);
values.extend(extra_row.values);
src/common/meta/src/ddl/tests/event/database.rs
+2 -4
View File
@@ -25,7 +25,7 @@ use common_event_recorder::event_table::{
PROCEDURE_ID_COLUMN as EVENT_TABLE_PROCEDURE_ID_COLUMN,
PROCEDURE_STATE_COLUMN as EVENT_TABLE_PROCEDURE_STATE_COLUMN,
PROCEDURE_TRIGGER_COLUMN as EVENT_TABLE_PROCEDURE_TRIGGER_COLUMN,
SCHEMA_NAME_COLUMN as EVENT_TABLE_SCHEMA_NAME_COLUMN,
SCHEMA_NAME_COLUMN as EVENT_TABLE_SCHEMA_NAME_COLUMN, jsonb_value,
};
use common_event_recorder::testing::assert_event_contract;
use common_procedure::{EventTrigger, ProcedureEvent, ProcedureId, ProcedureState};
@@ -314,9 +314,7 @@ fn assert_procedure_event_contract(
Value {
value_data: Some(ValueData::StringValue(String::new())),
},
Value {
value_data: Some(ValueData::StringValue(procedure_trigger.to_string())),
},
jsonb_value(&serde_json::json!({"type": procedure_trigger})),
Value {
value_data: catalog_name.map(|value| ValueData::StringValue(value.to_string())),
},
src/common/meta/src/ddl/tests/event/flow.rs
+2 -2
View File
@@ -20,7 +20,7 @@ use common_catalog::consts::{DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME};
use common_event_recorder::Event;
use common_event_recorder::event_table::{
CATALOG_NAME_COLUMN, FLOW_ID_COLUMN, FLOW_NAME_COLUMN, PROCEDURE_ERROR_COLUMN,
PROCEDURE_ID_COLUMN, PROCEDURE_STATE_COLUMN, PROCEDURE_TRIGGER_COLUMN,
PROCEDURE_ID_COLUMN, PROCEDURE_STATE_COLUMN, PROCEDURE_TRIGGER_COLUMN, jsonb_value,
};
use common_event_recorder::testing::assert_event_contract;
use common_procedure::{EventTrigger, ProcedureEvent, ProcedureId, ProcedureState};
@@ -238,7 +238,7 @@ fn assert_procedure_event_contract(
ValueData::StringValue(event.procedure_id.to_string()).into(),
ValueData::StringValue(state.to_string()).into(),
ValueData::StringValue(String::new()).into(),
ValueData::StringValue(trigger.to_string()).into(),
jsonb_value(&serde_json::json!({"type": trigger})),
optional_string(locator.catalog_name),
optional_string(locator.flow_name),
locator
src/common/meta/src/ddl/tests/event/view.rs
+2 -2
View File
@@ -18,7 +18,7 @@ use api::v1::value::ValueData;
use api::v1::{ColumnSchema, Row, Value};
use common_event_recorder::event_table::{
CATALOG_NAME_COLUMN, PROCEDURE_ERROR_COLUMN, PROCEDURE_ID_COLUMN, PROCEDURE_STATE_COLUMN,
PROCEDURE_TRIGGER_COLUMN, SCHEMA_NAME_COLUMN, VIEW_ID_COLUMN, VIEW_NAME_COLUMN,
PROCEDURE_TRIGGER_COLUMN, SCHEMA_NAME_COLUMN, VIEW_ID_COLUMN, VIEW_NAME_COLUMN, jsonb_value,
};
use common_event_recorder::testing::assert_event_contract;
use common_event_recorder::{Event, EventTypeFilter};
@@ -302,7 +302,7 @@ fn assert_procedure_event_contract(
ValueData::StringValue(event.procedure_id.to_string()).into(),
ValueData::StringValue(state.to_string()).into(),
ValueData::StringValue(String::new()).into(),
ValueData::StringValue(trigger.to_string()).into(),
jsonb_value(&serde_json::json!({"type": trigger})),
];
values.extend(locator.values());
src/common/procedure/src/event.rs
+69 -35
View File
@@ -17,12 +17,13 @@ use std::any::Any;
use api::v1::value::ValueData;
use api::v1::{ColumnSchema, Row};
use common_event_recorder::Event;
use common_event_recorder::error::Result;
use common_event_recorder::error::{Result, SerializeEventSnafu};
use common_event_recorder::event_table::{
PROCEDURE_ERROR_COLUMN, PROCEDURE_ID_COLUMN, PROCEDURE_STATE_COLUMN, PROCEDURE_TRIGGER_COLUMN,
procedure_event_column_schemas,
jsonb_value, procedure_event_column_schemas,
};
use common_time::timestamp::{TimeUnit, Timestamp};
use snafu::ResultExt;
use crate::{EventTrigger, ProcedureId, ProcedureState};
@@ -95,7 +96,7 @@ impl Event for ProcedureEvent {
| ProcedureState::Poisoned { error, .. } => 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");
}
}
src/common/procedure/src/procedure.rs
+6 -44
View File
@@ -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<T: Procedure + ?Sized> Procedure for Box<T> {
fn type_name(&self) -> &str {
src/meta-srv/src/event/gc.rs
+2 -2
View File
@@ -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(
src/meta-srv/src/event/repartition.rs
+2 -2
View File
@@ -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(),
tests-integration/tests/database_ddl_event.rs
+2 -2
View File
@@ -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)"#
tests-integration/tests/flow_ddl_event.rs
+5 -5
View File
@@ -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"#,
),
tests-integration/tests/gc_event.rs
+1 -1
View File
@@ -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 =
tests-integration/tests/region_migration.rs
+8 -6
View File
@@ -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))
tests-integration/tests/repartition_event.rs
+4 -4
View File
@@ -118,7 +118,7 @@ async fn assert_repartition_event(instance: &Arc<frontend::instance::Instance>)
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"#
tests-integration/tests/table_ddl_event.rs
+6 -6
View File
@@ -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(),
tests-integration/tests/view_ddl_event.rs
+5 -5
View File
@@ -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"#,
),
tests-integration/tests/wal_prune_event.rs
+1 -1
View File
@@ -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