feat(meta): record catalog and database reconciliation events (#8896)

* feat(meta): record catalog and database reconciliation events

Signed-off-by: dhruvxvaishnav <dhruvvaishnav687@gmail.com>

* test(meta): join catalog and database reconciliation events

Signed-off-by: dhruvxvaishnav <dhruvvaishnav687@gmail.com>

---------

Signed-off-by: dhruvxvaishnav <dhruvvaishnav687@gmail.com>

config/config.md

@@ -249,7 +249,7 @@
 | `pipeline.cache_ttl` | String | `10s` | Time to live of the local pipeline cache. Default is `10s`. |
 | `event_recorder` | -- | -- | Configuration options for the event recorder. |
 | `event_recorder.ttl` | String | `90d` | TTL for the events table that will be used to store the events. Default is `90d`. |
-| `event_recorder.event_types` | Array | -- | Event types to record. Current available event types: `create_database`,<br/>`alter_database`, `drop_database`, `create_flow`, `drop_flow`,<br/>`create_table`, `create_logical_tables`, `alter_table`, `alter_logical_tables`,<br/>`drop_table`, `undrop_table`, `purge_dropped_table`, `truncate_table`,<br/>`create_view`, `drop_view`, `admin_function`, `reconcile_table`,<br/>`reconcile_logical_tables`.<br/>When omitted, all current and future event types are recorded.<br/>Set to an empty array to disable event recording. |
+| `event_recorder.event_types` | Array | -- | Event types to record. Current available event types: `create_database`,<br/>`alter_database`, `drop_database`, `create_flow`, `drop_flow`,<br/>`create_table`, `create_logical_tables`, `alter_table`, `alter_logical_tables`,<br/>`drop_table`, `undrop_table`, `purge_dropped_table`, `truncate_table`,<br/>`create_view`, `drop_view`, `admin_function`, `reconcile_catalog`,<br/>`reconcile_database`, `reconcile_table`, `reconcile_logical_tables`.<br/>When omitted, all current and future event types are recorded.<br/>Set to an empty array to disable event recording. |
 | `memory` | -- | -- | The memory options. |
 | `memory.enable_heap_profiling` | Bool | `true` | Whether to enable heap profiling activation during startup.<br/>When enabled, heap profiling will be activated if the `MALLOC_CONF` environment variable<br/>is set to "prof:true,prof_active:false". The official image adds this env variable.<br/>Default is true. |
 
@@ -477,7 +477,7 @@
 | `wal.create_topic_timeout` | String | `30s` | The timeout for creating a Kafka topic.<br/>**It's only used when the provider is `kafka`**. |
 | `event_recorder` | -- | -- | Configuration options for the event recorder. |
 | `event_recorder.ttl` | String | `90d` | TTL for the events table that will be used to store the events. Default is `90d`. |
-| `event_recorder.event_types` | Array | -- | Event types to record. Current available event types: `region_migration`,<br/>`create_database`, `alter_database`, `drop_database`, `create_flow`,<br/>`drop_flow`, `create_table`, `create_logical_tables`, `alter_table`,<br/>`alter_logical_tables`, `drop_table`, `undrop_table`, `purge_dropped_table`,<br/>`truncate_table`, `create_view`, `drop_view`, `repartition`,<br/>`repartition_group`, `wal_prune`, `batch_gc`, `reconcile_table`,<br/>`reconcile_logical_tables`.<br/>When omitted, all current and future event types are recorded.<br/>Set to an empty array to disable event recording. |
+| `event_recorder.event_types` | Array | -- | Event types to record. Current available event types: `region_migration`,<br/>`create_database`, `alter_database`, `drop_database`, `create_flow`,<br/>`drop_flow`, `create_table`, `create_logical_tables`, `alter_table`,<br/>`alter_logical_tables`, `drop_table`, `undrop_table`, `purge_dropped_table`,<br/>`truncate_table`, `create_view`, `drop_view`, `repartition`,<br/>`repartition_group`, `wal_prune`, `batch_gc`, `reconcile_catalog`,<br/>`reconcile_database`, `reconcile_table`, `reconcile_logical_tables`.<br/>When omitted, all current and future event types are recorded.<br/>Set to an empty array to disable event recording. |
 | `stats_persistence` | -- | -- | Configuration options for the stats persistence. |
 | `stats_persistence.ttl` | String | `0s` | TTL for the stats table that will be used to store the stats.<br/>Set to `0s` to disable stats persistence.<br/>Default is `0s`.<br/>If you want to enable stats persistence, set the TTL to a value greater than 0.<br/>It is recommended to set a small value, e.g., `3h`. |
 | `stats_persistence.interval` | String | `10m` | The interval to persist the stats. Default is `10m`.<br/>The minimum value is `10m`, if the value is less than `10m`, it will be overridden to `10m`. |

config/metasrv.example.toml

@@ -328,8 +328,8 @@ ttl = "90d"
 ## `drop_flow`, `create_table`, `create_logical_tables`, `alter_table`,
 ## `alter_logical_tables`, `drop_table`, `undrop_table`, `purge_dropped_table`,
 ## `truncate_table`, `create_view`, `drop_view`, `repartition`,
-## `repartition_group`, `wal_prune`, `batch_gc`, `reconcile_table`,
-## `reconcile_logical_tables`.
+## `repartition_group`, `wal_prune`, `batch_gc`, `reconcile_catalog`,
+## `reconcile_database`, `reconcile_table`, `reconcile_logical_tables`.
 ## When omitted, all current and future event types are recorded.
 ## Set to an empty array to disable event recording.
 #+ event_types = ["region_migration"]

config/standalone.example.toml

@@ -973,8 +973,8 @@ ttl = "90d"
 ## `alter_database`, `drop_database`, `create_flow`, `drop_flow`,
 ## `create_table`, `create_logical_tables`, `alter_table`, `alter_logical_tables`,
 ## `drop_table`, `undrop_table`, `purge_dropped_table`, `truncate_table`,
-## `create_view`, `drop_view`, `admin_function`, `reconcile_table`,
-## `reconcile_logical_tables`.
+## `create_view`, `drop_view`, `admin_function`, `reconcile_catalog`,
+## `reconcile_database`, `reconcile_table`, `reconcile_logical_tables`.
 ## When omitted, all current and future event types are recorded.
 ## Set to an empty array to disable event recording.
 #+ event_types = ["create_database"]

src/common/meta/src/reconciliation/event.rs

@@ -28,6 +28,10 @@ use store_api::storage::TableId;
 
 use crate::reconciliation::ResolveStrategy;
 
+/// Stable event type stored for catalog reconciliation procedures.
+pub(crate) const RECONCILE_CATALOG_EVENT_TYPE: &str = "reconcile_catalog";
+/// Stable event type stored for database reconciliation procedures.
+pub(crate) const RECONCILE_DATABASE_EVENT_TYPE: &str = "reconcile_database";
 /// Stable event type stored for logical table reconciliation procedures.
 pub(crate) const RECONCILE_LOGICAL_TABLES_EVENT_TYPE: &str = "reconcile_logical_tables";
 /// Stable event type stored for physical table reconciliation procedures.
@@ -45,6 +49,23 @@ pub(crate) struct ReconciliationLocator {
 }
 
 impl ReconciliationLocator {
+    /// Creates a locator for a catalog.
+    pub(crate) fn catalog(catalog_name: &str) -> Self {
+        Self {
+            catalog_name: Some(catalog_name.to_string()),
+            ..Default::default()
+        }
+    }
+
+    /// Creates a locator for a database.
+    pub(crate) fn database(catalog_name: &str, schema_name: &str) -> Self {
+        Self {
+            catalog_name: Some(catalog_name.to_string()),
+            schema_name: Some(schema_name.to_string()),
+            ..Default::default()
+        }
+    }
+
     /// Creates a locator for a physical table with its fully qualified name and ID.
     pub(crate) fn physical_table(
         catalog_name: &str,
@@ -101,6 +122,229 @@ impl ReconciliationLocator {
     }
 }
 
+#[derive(Debug, Serialize)]
+#[serde(untagged)]
+enum ReconcileCatalogPayload {
+    Submitted(CatalogSubmittedPayload),
+    Result(CatalogResultPayload),
+}
+
+#[derive(Debug, Serialize)]
+struct CatalogSubmittedPayload {
+    version: u8,
+    resolve_strategy: &'static str,
+    fail_fast: bool,
+    parallelism: usize,
+}
+
+#[derive(Debug, Serialize)]
+struct CatalogResultPayload {
+    version: u8,
+    complete: bool,
+    processed_database_count: usize,
+    succeeded_database_count: usize,
+    failed_database_count: usize,
+}
+
+/// Event representation for catalog reconciliation.
+#[derive(Debug)]
+pub(crate) struct ReconcileCatalogEvent {
+    locator: ReconciliationLocator,
+    payload: Option<ReconcileCatalogPayload>,
+}
+
+impl ReconcileCatalogEvent {
+    /// Builds the bounded intent event emitted when catalog reconciliation is submitted.
+    pub(crate) fn submitted(
+        locator: ReconciliationLocator,
+        resolve_strategy: ResolveStrategy,
+        fail_fast: bool,
+        parallelism: usize,
+    ) -> Self {
+        Self {
+            locator,
+            payload: Some(ReconcileCatalogPayload::Submitted(
+                CatalogSubmittedPayload {
+                    version: PAYLOAD_VERSION,
+                    resolve_strategy: resolve_strategy_name(resolve_strategy),
+                    fail_fast,
+                    parallelism,
+                },
+            )),
+        }
+    }
+
+    /// Builds a terminal event from the existing volatile reconciliation metrics.
+    ///
+    /// Metrics are best-effort observations from the current process and reset on recovery.
+    pub(crate) fn result(
+        locator: ReconciliationLocator,
+        complete: bool,
+        succeeded_database_count: usize,
+        failed_database_count: usize,
+    ) -> Self {
+        Self {
+            locator,
+            payload: Some(ReconcileCatalogPayload::Result(CatalogResultPayload {
+                version: PAYLOAD_VERSION,
+                complete,
+                processed_database_count: succeeded_database_count + failed_database_count,
+                succeeded_database_count,
+                failed_database_count,
+            })),
+        }
+    }
+
+    /// Builds a catalog lifecycle event whose reconciliation payload is null.
+    pub(crate) fn lifecycle(locator: ReconciliationLocator) -> Self {
+        Self {
+            locator,
+            payload: None,
+        }
+    }
+}
+
+impl Event for ReconcileCatalogEvent {
+    fn event_type(&self) -> &str {
+        RECONCILE_CATALOG_EVENT_TYPE
+    }
+
+    fn json_payload(&self) -> Result<serde_json::Value> {
+        match &self.payload {
+            Some(payload) => serde_json::to_value(payload).context(SerializeEventSnafu),
+            None => Ok(serde_json::Value::Null),
+        }
+    }
+
+    fn extra_schema(&self) -> Vec<ColumnSchema> {
+        ReconciliationLocator::schema()
+    }
+
+    fn extra_rows(&self) -> Result<Vec<Row>> {
+        Ok(vec![self.locator.row()])
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+}
+
+#[derive(Debug, Serialize)]
+#[serde(untagged)]
+enum ReconcileDatabasePayload {
+    Submitted(DatabaseSubmittedPayload),
+    Result(DatabaseResultPayload),
+}
+
+#[derive(Debug, Serialize)]
+struct DatabaseSubmittedPayload {
+    version: u8,
+    resolve_strategy: &'static str,
+    fail_fast: bool,
+    parallelism: usize,
+    is_subprocedure: bool,
+}
+
+#[derive(Debug, Serialize)]
+struct DatabaseResultPayload {
+    version: u8,
+    complete: bool,
+    processed_table_count: usize,
+    succeeded_table_count: usize,
+    failed_table_count: usize,
+    succeeded_subprocedure_count: usize,
+    failed_subprocedure_count: usize,
+}
+
+/// Event representation for database reconciliation.
+#[derive(Debug)]
+pub(crate) struct ReconcileDatabaseEvent {
+    locator: ReconciliationLocator,
+    payload: Option<ReconcileDatabasePayload>,
+}
+
+impl ReconcileDatabaseEvent {
+    /// Builds the bounded intent event emitted when database reconciliation is submitted.
+    pub(crate) fn submitted(
+        locator: ReconciliationLocator,
+        resolve_strategy: ResolveStrategy,
+        fail_fast: bool,
+        parallelism: usize,
+        is_subprocedure: bool,
+    ) -> Self {
+        Self {
+            locator,
+            payload: Some(ReconcileDatabasePayload::Submitted(
+                DatabaseSubmittedPayload {
+                    version: PAYLOAD_VERSION,
+                    resolve_strategy: resolve_strategy_name(resolve_strategy),
+                    fail_fast,
+                    parallelism,
+                    is_subprocedure,
+                },
+            )),
+        }
+    }
+
+    /// Builds a terminal event from the existing volatile reconciliation metrics.
+    ///
+    /// Metrics are best-effort observations from the current process and reset on recovery.
+    pub(crate) fn result(
+        locator: ReconciliationLocator,
+        complete: bool,
+        succeeded_table_count: usize,
+        failed_table_count: usize,
+        succeeded_subprocedure_count: usize,
+        failed_subprocedure_count: usize,
+    ) -> Self {
+        Self {
+            locator,
+            payload: Some(ReconcileDatabasePayload::Result(DatabaseResultPayload {
+                version: PAYLOAD_VERSION,
+                complete,
+                processed_table_count: succeeded_table_count + failed_table_count,
+                succeeded_table_count,
+                failed_table_count,
+                succeeded_subprocedure_count,
+                failed_subprocedure_count,
+            })),
+        }
+    }
+
+    /// Builds a database lifecycle event whose reconciliation payload is null.
+    pub(crate) fn lifecycle(locator: ReconciliationLocator) -> Self {
+        Self {
+            locator,
+            payload: None,
+        }
+    }
+}
+
+impl Event for ReconcileDatabaseEvent {
+    fn event_type(&self) -> &str {
+        RECONCILE_DATABASE_EVENT_TYPE
+    }
+
+    fn json_payload(&self) -> Result<serde_json::Value> {
+        match &self.payload {
+            Some(payload) => serde_json::to_value(payload).context(SerializeEventSnafu),
+            None => Ok(serde_json::Value::Null),
+        }
+    }
+
+    fn extra_schema(&self) -> Vec<ColumnSchema> {
+        ReconciliationLocator::schema()
+    }
+
+    fn extra_rows(&self) -> Result<Vec<Row>> {
+        Ok(vec![self.locator.row()])
+    }
+
+    fn as_any(&self) -> &dyn Any {
+        self
+    }
+}
+
 #[derive(Debug, Serialize)]
 #[serde(untagged)]
 enum ReconcileTablePayload {
@@ -355,12 +599,14 @@ mod tests {
 
     #[test]
     fn reconciliation_events_use_the_shared_locator_contract() {
+        let catalog = ReconcileCatalogEvent::lifecycle(ReconciliationLocator::catalog("greptime"));
+        assert_eq!(catalog.event_type(), RECONCILE_CATALOG_EVENT_TYPE);
         let table = ReconcileTableEvent::table_lifecycle(ReconciliationLocator::physical_table(
             "greptime", "public", "metrics", 42,
         ));
         assert_eq!(table.event_type(), RECONCILE_TABLE_EVENT_TYPE);
         assert_eq!(
-            table
+            catalog
                 .extra_schema()
                 .into_iter()
                 .map(|column| {
@@ -375,22 +621,22 @@ mod tests {
                 (
                     "catalog_name".to_string(),
                     ColumnDataType::String,
-                    SemanticType::Field
+                    SemanticType::Field,
                 ),
                 (
                     "schema_name".to_string(),
                     ColumnDataType::String,
-                    SemanticType::Field
+                    SemanticType::Field,
                 ),
                 (
                     "table_name".to_string(),
                     ColumnDataType::String,
-                    SemanticType::Field
+                    SemanticType::Field,
                 ),
                 (
                     "table_id".to_string(),
                     ColumnDataType::Uint32,
-                    SemanticType::Field
+                    SemanticType::Field,
                 ),
                 (
                     "physical_table_id".to_string(),
@@ -399,6 +645,40 @@ mod tests {
                 ),
             ]
         );
+        assert_eq!(
+            catalog.extra_rows().unwrap(),
+            vec![Row {
+                values: vec![
+                    ValueData::StringValue("greptime".to_string()).into(),
+                    Value::default(),
+                    Value::default(),
+                    Value::default(),
+                    Value::default(),
+                ],
+            }]
+        );
+        assert_eq!(catalog.json_payload().unwrap(), serde_json::Value::Null);
+
+        let database = ReconcileDatabaseEvent::lifecycle(ReconciliationLocator::database(
+            "greptime", "public",
+        ));
+        assert_eq!(database.event_type(), RECONCILE_DATABASE_EVENT_TYPE);
+        assert_eq!(database.extra_schema(), catalog.extra_schema());
+        assert_eq!(
+            database.extra_rows().unwrap(),
+            vec![Row {
+                values: vec![
+                    ValueData::StringValue("greptime".to_string()).into(),
+                    ValueData::StringValue("public".to_string()).into(),
+                    Value::default(),
+                    Value::default(),
+                    Value::default(),
+                ],
+            }]
+        );
+        assert_eq!(database.json_payload().unwrap(), serde_json::Value::Null);
+
+        assert_eq!(table.extra_schema(), catalog.extra_schema());
         assert_eq!(
             table.extra_rows().unwrap(),
             vec![Row {
@@ -458,6 +738,22 @@ mod tests {
             (ResolveStrategy::UseMetasrv, "use_metasrv"),
             (ResolveStrategy::AbortOnConflict, "abort_on_conflict"),
         ] {
+            let catalog = ReconcileCatalogEvent::submitted(
+                ReconciliationLocator::catalog("greptime"),
+                strategy,
+                true,
+                16,
+            );
+            assert_eq!(
+                catalog.json_payload().unwrap(),
+                json!({
+                    "version": 1,
+                    "resolve_strategy": expected,
+                    "fail_fast": true,
+                    "parallelism": 16,
+                })
+            );
+
             let table = ReconcileTableEvent::table_submitted(
                 ReconciliationLocator::physical_table("greptime", "public", "metrics", 42),
                 strategy,
@@ -473,6 +769,24 @@ mod tests {
             );
         }
 
+        let database = ReconcileDatabaseEvent::submitted(
+            ReconciliationLocator::database("greptime", "public"),
+            ResolveStrategy::UseMetasrv,
+            false,
+            64,
+            true,
+        );
+        assert_eq!(
+            database.json_payload().unwrap(),
+            json!({
+                "version": 1,
+                "resolve_strategy": "use_metasrv",
+                "fail_fast": false,
+                "parallelism": 64,
+                "is_subprocedure": true,
+            })
+        );
+
         let logical_tables = ReconcileLogicalTablesEvent::submitted(
             vec![
                 ReconciliationLocator::logical_table("greptime", "public", "cpu", 43, 42),
@@ -492,6 +806,40 @@ mod tests {
 
     #[test]
     fn terminal_payloads_distinguish_complete_and_partial_results() {
+        let catalog =
+            ReconcileCatalogEvent::result(ReconciliationLocator::catalog("greptime"), true, 3, 1);
+        assert_eq!(
+            catalog.json_payload().unwrap(),
+            json!({
+                "version": 1,
+                "complete": true,
+                "processed_database_count": 4,
+                "succeeded_database_count": 3,
+                "failed_database_count": 1,
+            })
+        );
+
+        let database = ReconcileDatabaseEvent::result(
+            ReconciliationLocator::database("greptime", "public"),
+            false,
+            5,
+            2,
+            4,
+            1,
+        );
+        assert_eq!(
+            database.json_payload().unwrap(),
+            json!({
+                "version": 1,
+                "complete": false,
+                "processed_table_count": 7,
+                "succeeded_table_count": 5,
+                "failed_table_count": 2,
+                "succeeded_subprocedure_count": 4,
+                "failed_subprocedure_count": 1,
+            })
+        );
+
         let table = ReconcileTableEvent::table_result(
             ReconciliationLocator::physical_table("greptime", "public", "metrics", 42),
             false,

src/common/meta/src/reconciliation/reconcile_catalog.rs

@@ -18,8 +18,8 @@ use std::time::Instant;
 
 use common_procedure::error::FromJsonSnafu;
 use common_procedure::{
-    Context as ProcedureContext, Error as ProcedureError, LockKey, Procedure,
-    Result as ProcedureResult, Status,
+    Context as ProcedureContext, Error as ProcedureError, EventContext, EventTrigger, LockKey,
+    Procedure, Result as ProcedureResult, Status,
 };
 use futures::stream::BoxStream;
 use serde::{Deserialize, Serialize};
@@ -31,6 +31,9 @@ use crate::key::TableMetadataManagerRef;
 use crate::lock_key::CatalogLock;
 use crate::metrics;
 use crate::node_manager::NodeManagerRef;
+use crate::reconciliation::event::{
+    RECONCILE_CATALOG_EVENT_TYPE, ReconcileCatalogEvent, ReconciliationLocator,
+};
 use crate::reconciliation::reconcile_catalog::start::ReconcileCatalogStart;
 use crate::reconciliation::reconcile_table::resolve_column_metadata::ResolveStrategy;
 use crate::reconciliation::utils::{
@@ -216,6 +219,43 @@ impl Procedure for ReconcileCatalogProcedure {
 
         LockKey::new(vec![CatalogLock::Write(catalog).into()])
     }
+
+    fn event(&self, ctx: &EventContext<'_>) -> Option<Box<dyn common_event_recorder::Event>> {
+        if !ctx.event_type_filter.allows(RECONCILE_CATALOG_EVENT_TYPE) {
+            return None;
+        }
+
+        let persistent_ctx = &self.context.persistent_ctx;
+        let locator = ReconciliationLocator::catalog(&persistent_ctx.catalog);
+        let event = match ctx.trigger {
+            EventTrigger::Submitted => ReconcileCatalogEvent::submitted(
+                locator,
+                persistent_ctx.resolve_strategy,
+                persistent_ctx.fast_fail,
+                persistent_ctx.parallelism,
+            ),
+            EventTrigger::Succeeded => self.result_event(locator, true),
+            EventTrigger::Failed | EventTrigger::Poisoned => self.result_event(locator, false),
+            _ => ReconcileCatalogEvent::lifecycle(locator),
+        };
+        Some(Box::new(event))
+    }
+}
+
+impl ReconcileCatalogProcedure {
+    fn result_event(
+        &self,
+        locator: ReconciliationLocator,
+        complete: bool,
+    ) -> ReconcileCatalogEvent {
+        let metrics = &self.context.volatile_ctx.metrics;
+        ReconcileCatalogEvent::result(
+            locator,
+            complete,
+            metrics.succeeded_databases,
+            metrics.failed_databases,
+        )
+    }
 }
 
 #[async_trait::async_trait]
@@ -235,3 +275,217 @@ pub(crate) trait State: Sync + Send + Debug {
 
     fn as_any(&self) -> &dyn Any;
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use common_event_recorder::{EventTypeFilter, EventTypeFilterRef};
+    use common_procedure::{
+        ChildSubmissionOutcome, EventContext, EventTrigger, Procedure, ProcedureId, ProcedureState,
+        RetryPhase,
+    };
+    use serde_json::{Value, json};
+
+    use super::*;
+    use crate::reconciliation::event::RECONCILE_DATABASE_EVENT_TYPE;
+    use crate::test_util::{MockDatanodeManager, new_ddl_context};
+
+    struct CatalogEventHarness {
+        procedure_id: ProcedureId,
+        lifecycle_state: ProcedureState,
+        event_type_filter: EventTypeFilterRef,
+    }
+
+    impl CatalogEventHarness {
+        fn all() -> Self {
+            Self {
+                procedure_id: ProcedureId::random(),
+                lifecycle_state: ProcedureState::Running,
+                event_type_filter: Arc::new(EventTypeFilter::All),
+            }
+        }
+
+        fn selected(event_types: impl IntoIterator<Item = &'static str>) -> Self {
+            Self {
+                event_type_filter: Arc::new(EventTypeFilter::Only(
+                    event_types.into_iter().map(str::to_string).collect(),
+                )),
+                ..Self::all()
+            }
+        }
+
+        fn event(
+            &self,
+            procedure: &dyn Procedure,
+            trigger: EventTrigger,
+        ) -> Option<Box<dyn common_event_recorder::Event>> {
+            procedure.event(&EventContext {
+                procedure_id: self.procedure_id,
+                lifecycle_state: &self.lifecycle_state,
+                trigger,
+                event_type_filter: self.event_type_filter.clone(),
+                event_context: None,
+            })
+        }
+    }
+
+    #[test]
+    fn catalog_submitted_event_reports_intent() {
+        let submitted = CatalogEventHarness::all()
+            .event(&test_procedure(), EventTrigger::Submitted)
+            .unwrap();
+        assert_eq!(submitted.event_type(), RECONCILE_CATALOG_EVENT_TYPE);
+        assert_eq!(
+            submitted.json_payload().unwrap(),
+            json!({
+                "version": 1,
+                "resolve_strategy": "use_latest",
+                "fail_fast": false,
+                "parallelism": 8,
+            })
+        );
+    }
+
+    #[test]
+    fn catalog_non_terminal_lifecycle_events_have_null_payloads() {
+        let events = CatalogEventHarness::all();
+        let mut procedure = test_procedure();
+        procedure.context.volatile_ctx.metrics = populated_metrics();
+
+        for trigger in [
+            EventTrigger::Recovered,
+            EventTrigger::ChildSubmitted {
+                procedure_id: ProcedureId::random(),
+                outcome: ChildSubmissionOutcome::Accepted,
+            },
+            EventTrigger::Retrying {
+                phase: RetryPhase::Execute,
+                attempt: 1,
+            },
+            EventTrigger::RollingBack,
+        ] {
+            assert_eq!(
+                events
+                    .event(&procedure, trigger)
+                    .unwrap()
+                    .json_payload()
+                    .unwrap(),
+                Value::Null
+            );
+        }
+    }
+
+    #[test]
+    fn catalog_terminal_events_report_existing_metrics() {
+        let events = CatalogEventHarness::all();
+        let mut procedure = test_procedure();
+        procedure.context.volatile_ctx.metrics = populated_metrics();
+
+        for (trigger, complete) in [
+            (EventTrigger::Succeeded, true),
+            (EventTrigger::Failed, false),
+            (EventTrigger::Poisoned, false),
+        ] {
+            assert_eq!(
+                events
+                    .event(&procedure, trigger)
+                    .unwrap()
+                    .json_payload()
+                    .unwrap(),
+                json!({
+                    "version": 1,
+                    "complete": complete,
+                    "processed_database_count": 4,
+                    "succeeded_database_count": 3,
+                    "failed_database_count": 1,
+                })
+            );
+        }
+    }
+
+    #[test]
+    fn catalog_event_filtering_uses_the_catalog_event_type() {
+        let procedure = test_procedure();
+        assert!(
+            CatalogEventHarness::selected([RECONCILE_CATALOG_EVENT_TYPE])
+                .event(&procedure, EventTrigger::Submitted)
+                .is_some()
+        );
+        assert!(
+            CatalogEventHarness::selected([RECONCILE_DATABASE_EVENT_TYPE])
+                .event(&procedure, EventTrigger::Submitted)
+                .is_none()
+        );
+        assert!(
+            CatalogEventHarness::selected([])
+                .event(&procedure, EventTrigger::Submitted)
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn catalog_recovery_preserves_locator_and_resets_metrics() {
+        let events = CatalogEventHarness::all();
+        let mut procedure = test_procedure();
+        procedure.context.volatile_ctx.metrics = populated_metrics();
+        let original_dump = procedure.dump().unwrap();
+        procedure.context.volatile_ctx.metrics = ReconcileCatalogMetrics::default();
+        assert_eq!(procedure.dump().unwrap(), original_dump);
+
+        let loaded = ReconcileCatalogProcedure::from_json(test_context(), &original_dump).unwrap();
+        assert_eq!(loaded.dump().unwrap(), original_dump);
+        assert_eq!(
+            events
+                .event(&loaded, EventTrigger::Recovered)
+                .unwrap()
+                .extra_rows()
+                .unwrap(),
+            events
+                .event(&procedure, EventTrigger::Submitted)
+                .unwrap()
+                .extra_rows()
+                .unwrap(),
+        );
+        assert_eq!(
+            events
+                .event(&loaded, EventTrigger::Succeeded)
+                .unwrap()
+                .json_payload()
+                .unwrap(),
+            json!({
+                "version": 1,
+                "complete": true,
+                "processed_database_count": 0,
+                "succeeded_database_count": 0,
+                "failed_database_count": 0,
+            })
+        );
+    }
+
+    fn populated_metrics() -> ReconcileCatalogMetrics {
+        ReconcileCatalogMetrics {
+            succeeded_databases: 3,
+            failed_databases: 1,
+        }
+    }
+
+    fn test_procedure() -> ReconcileCatalogProcedure {
+        ReconcileCatalogProcedure::new(
+            test_context(),
+            "greptime".to_string(),
+            false,
+            ResolveStrategy::UseLatest,
+            8,
+        )
+    }
+
+    fn test_context() -> Context {
+        let ddl_context = new_ddl_context(Arc::new(MockDatanodeManager::new(())));
+        Context {
+            node_manager: ddl_context.node_manager,
+            table_metadata_manager: ddl_context.table_metadata_manager,
+            cache_invalidator: ddl_context.cache_invalidator,
+        }
+    }
+}

src/common/meta/src/reconciliation/reconcile_database.rs

@@ -25,8 +25,8 @@ use std::time::Instant;
 use async_trait::async_trait;
 use common_procedure::error::{FromJsonSnafu, ToJsonSnafu};
 use common_procedure::{
-    Context as ProcedureContext, Error as ProcedureError, LockKey, Procedure,
-    Result as ProcedureResult, Status,
+    Context as ProcedureContext, Error as ProcedureError, EventContext, EventTrigger, LockKey,
+    Procedure, Result as ProcedureResult, Status,
 };
 use futures::stream::BoxStream;
 use serde::{Deserialize, Serialize};
@@ -41,6 +41,9 @@ use crate::key::table_name::TableNameValue;
 use crate::lock_key::{CatalogLock, SchemaLock};
 use crate::metrics;
 use crate::node_manager::NodeManagerRef;
+use crate::reconciliation::event::{
+    RECONCILE_DATABASE_EVENT_TYPE, ReconcileDatabaseEvent, ReconciliationLocator,
+};
 use crate::reconciliation::reconcile_database::start::ReconcileDatabaseStart;
 use crate::reconciliation::reconcile_table::resolve_column_metadata::ResolveStrategy;
 use crate::reconciliation::utils::{
@@ -264,6 +267,47 @@ impl Procedure for ReconcileDatabaseProcedure {
             SchemaLock::write(catalog, schema).into(),
         ])
     }
+
+    fn event(&self, ctx: &EventContext<'_>) -> Option<Box<dyn common_event_recorder::Event>> {
+        if !ctx.event_type_filter.allows(RECONCILE_DATABASE_EVENT_TYPE) {
+            return None;
+        }
+
+        let persistent_ctx = &self.context.persistent_ctx;
+        let locator =
+            ReconciliationLocator::database(&persistent_ctx.catalog, &persistent_ctx.schema);
+        let event = match ctx.trigger {
+            EventTrigger::Submitted => ReconcileDatabaseEvent::submitted(
+                locator,
+                persistent_ctx.resolve_strategy,
+                persistent_ctx.fail_fast,
+                persistent_ctx.parallelism,
+                persistent_ctx.is_subprocedure,
+            ),
+            EventTrigger::Succeeded => self.result_event(locator, true),
+            EventTrigger::Failed | EventTrigger::Poisoned => self.result_event(locator, false),
+            _ => ReconcileDatabaseEvent::lifecycle(locator),
+        };
+        Some(Box::new(event))
+    }
+}
+
+impl ReconcileDatabaseProcedure {
+    fn result_event(
+        &self,
+        locator: ReconciliationLocator,
+        complete: bool,
+    ) -> ReconcileDatabaseEvent {
+        let metrics = self.context.metrics();
+        ReconcileDatabaseEvent::result(
+            locator,
+            complete,
+            metrics.succeeded_tables,
+            metrics.failed_tables,
+            metrics.succeeded_procedures,
+            metrics.failed_procedures,
+        )
+    }
 }
 
 #[async_trait::async_trait]
@@ -283,3 +327,229 @@ pub(crate) trait State: Sync + Send + Debug {
 
     fn as_any(&self) -> &dyn Any;
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+
+    use common_event_recorder::{EventTypeFilter, EventTypeFilterRef};
+    use common_procedure::{
+        ChildSubmissionOutcome, EventContext, EventTrigger, Procedure, ProcedureId, ProcedureState,
+        RetryPhase,
+    };
+    use serde_json::{Value, json};
+
+    use super::*;
+    use crate::reconciliation::event::RECONCILE_CATALOG_EVENT_TYPE;
+    use crate::test_util::{MockDatanodeManager, new_ddl_context};
+
+    struct DatabaseEventHarness {
+        procedure_id: ProcedureId,
+        lifecycle_state: ProcedureState,
+        event_type_filter: EventTypeFilterRef,
+    }
+
+    impl DatabaseEventHarness {
+        fn all() -> Self {
+            Self {
+                procedure_id: ProcedureId::random(),
+                lifecycle_state: ProcedureState::Running,
+                event_type_filter: Arc::new(EventTypeFilter::All),
+            }
+        }
+
+        fn selected(event_types: impl IntoIterator<Item = &'static str>) -> Self {
+            Self {
+                event_type_filter: Arc::new(EventTypeFilter::Only(
+                    event_types.into_iter().map(str::to_string).collect(),
+                )),
+                ..Self::all()
+            }
+        }
+
+        fn event(
+            &self,
+            procedure: &dyn Procedure,
+            trigger: EventTrigger,
+        ) -> Option<Box<dyn common_event_recorder::Event>> {
+            procedure.event(&EventContext {
+                procedure_id: self.procedure_id,
+                lifecycle_state: &self.lifecycle_state,
+                trigger,
+                event_type_filter: self.event_type_filter.clone(),
+                event_context: None,
+            })
+        }
+    }
+
+    #[test]
+    fn database_submitted_events_cover_root_and_child_intent() {
+        let events = DatabaseEventHarness::all();
+        let root = test_procedure(false);
+        let child = test_procedure(true);
+        for (procedure, is_subprocedure) in [(&root, false), (&child, true)] {
+            let submitted = events.event(procedure, EventTrigger::Submitted).unwrap();
+            assert_eq!(submitted.event_type(), RECONCILE_DATABASE_EVENT_TYPE);
+            assert_eq!(
+                submitted.json_payload().unwrap(),
+                json!({
+                    "version": 1,
+                    "resolve_strategy": "use_metasrv",
+                    "fail_fast": false,
+                    "parallelism": 8,
+                    "is_subprocedure": is_subprocedure,
+                })
+            );
+        }
+    }
+
+    #[test]
+    fn database_non_terminal_lifecycle_events_have_null_payloads() {
+        let events = DatabaseEventHarness::all();
+        let mut procedure = test_procedure(true);
+        procedure.context.volatile_ctx.metrics = populated_metrics();
+
+        for trigger in [
+            EventTrigger::Recovered,
+            EventTrigger::ChildSubmitted {
+                procedure_id: ProcedureId::random(),
+                outcome: ChildSubmissionOutcome::Accepted,
+            },
+            EventTrigger::Retrying {
+                phase: RetryPhase::Execute,
+                attempt: 2,
+            },
+            EventTrigger::RollingBack,
+        ] {
+            assert_eq!(
+                events
+                    .event(&procedure, trigger)
+                    .unwrap()
+                    .json_payload()
+                    .unwrap(),
+                Value::Null
+            );
+        }
+    }
+
+    #[test]
+    fn database_terminal_events_report_existing_metrics() {
+        let events = DatabaseEventHarness::all();
+        let mut procedure = test_procedure(true);
+        procedure.context.volatile_ctx.metrics = populated_metrics();
+
+        for (trigger, complete) in [
+            (EventTrigger::Succeeded, true),
+            (EventTrigger::Failed, false),
+            (EventTrigger::Poisoned, false),
+        ] {
+            assert_eq!(
+                events
+                    .event(&procedure, trigger)
+                    .unwrap()
+                    .json_payload()
+                    .unwrap(),
+                json!({
+                    "version": 1,
+                    "complete": complete,
+                    "processed_table_count": 8,
+                    "succeeded_table_count": 5,
+                    "failed_table_count": 3,
+                    "succeeded_subprocedure_count": 3,
+                    "failed_subprocedure_count": 2,
+                })
+            );
+        }
+    }
+
+    #[test]
+    fn database_event_filtering_uses_the_database_event_type() {
+        let procedure = test_procedure(false);
+        assert!(
+            DatabaseEventHarness::selected([RECONCILE_DATABASE_EVENT_TYPE])
+                .event(&procedure, EventTrigger::Submitted)
+                .is_some()
+        );
+        assert!(
+            DatabaseEventHarness::selected([RECONCILE_CATALOG_EVENT_TYPE])
+                .event(&procedure, EventTrigger::Submitted)
+                .is_none()
+        );
+        assert!(
+            DatabaseEventHarness::selected([])
+                .event(&procedure, EventTrigger::Submitted)
+                .is_none()
+        );
+    }
+
+    #[test]
+    fn database_recovery_preserves_locator_and_resets_metrics() {
+        let events = DatabaseEventHarness::all();
+        let mut procedure = test_procedure(false);
+        procedure.context.volatile_ctx.metrics = populated_metrics();
+        let original_dump = procedure.dump().unwrap();
+        procedure.context.volatile_ctx.metrics = ReconcileDatabaseMetrics::default();
+        assert_eq!(procedure.dump().unwrap(), original_dump);
+
+        let loaded = ReconcileDatabaseProcedure::from_json(test_context(), &original_dump).unwrap();
+        assert_eq!(loaded.dump().unwrap(), original_dump);
+        assert_eq!(
+            events
+                .event(&loaded, EventTrigger::Recovered)
+                .unwrap()
+                .extra_rows()
+                .unwrap(),
+            events
+                .event(&procedure, EventTrigger::Submitted)
+                .unwrap()
+                .extra_rows()
+                .unwrap(),
+        );
+        assert_eq!(
+            events
+                .event(&loaded, EventTrigger::Succeeded)
+                .unwrap()
+                .json_payload()
+                .unwrap(),
+            json!({
+                "version": 1,
+                "complete": true,
+                "processed_table_count": 0,
+                "succeeded_table_count": 0,
+                "failed_table_count": 0,
+                "succeeded_subprocedure_count": 0,
+                "failed_subprocedure_count": 0,
+            })
+        );
+    }
+
+    fn populated_metrics() -> ReconcileDatabaseMetrics {
+        ReconcileDatabaseMetrics {
+            succeeded_tables: 5,
+            failed_tables: 3,
+            succeeded_procedures: 3,
+            failed_procedures: 2,
+        }
+    }
+
+    fn test_procedure(is_subprocedure: bool) -> ReconcileDatabaseProcedure {
+        ReconcileDatabaseProcedure::new(
+            test_context(),
+            "greptime".to_string(),
+            "public".to_string(),
+            false,
+            8,
+            ResolveStrategy::UseMetasrv,
+            is_subprocedure,
+        )
+    }
+
+    fn test_context() -> Context {
+        let ddl_context = new_ddl_context(Arc::new(MockDatanodeManager::new(())));
+        Context {
+            node_manager: ddl_context.node_manager,
+            table_metadata_manager: ddl_context.table_metadata_manager,
+            cache_invalidator: ddl_context.cache_invalidator,
+        }
+    }
+}

tests-integration/tests/reconciliation_event.rs

@@ -27,6 +27,216 @@ const EVENTS_TABLE: &str = "greptime_private.events";
 const CATALOG: &str = "greptime";
 const DATABASE: &str = "reconciliation_event_database";
 
+#[tokio::test(flavor = "multi_thread")]
+async fn test_catalog_and_database_reconciliation_events() {
+    common_telemetry::init_default_ut_logging();
+
+    let cluster = GreptimeDbClusterBuilder::new("catalog_database_reconciliation_events")
+        .await
+        .with_datanodes(1)
+        .build(true)
+        .await;
+    let frontend = cluster.fe_instance().clone();
+    run_sql(frontend.as_ref(), &format!("CREATE DATABASE {DATABASE}")).await;
+    run_sql(
+        frontend.as_ref(),
+        &format!("CREATE TABLE {DATABASE}.metrics (ts TIMESTAMP TIME INDEX)"),
+    )
+    .await;
+
+    let catalog_procedure_id = cluster
+        .metasrv
+        .reconciliation_manager()
+        .reconcile_catalog(CATALOG.to_string(), ResolveStrategy::UseLatest, 1)
+        .await
+        .unwrap();
+    let mut procedure_watcher = cluster
+        .metasrv
+        .procedure_manager()
+        .procedure_watcher(catalog_procedure_id)
+        .unwrap();
+    watcher::wait(&mut procedure_watcher).await.unwrap();
+    let catalog_procedure_id = catalog_procedure_id.to_string();
+
+    assert_eventually_eq(
+        &frontend,
+        &format!(
+            "SELECT count(*) AS event_count, \
+                 json_get_int(payload, 'version') AS version, \
+                 json_get_string(payload, 'resolve_strategy') AS resolve_strategy, \
+                 json_get_bool(payload, 'fail_fast') AS fail_fast, \
+                 json_get_int(payload, 'parallelism') AS parallelism \
+             FROM {EVENTS_TABLE} \
+             WHERE type = 'reconcile_catalog' AND procedure_id = '{catalog_procedure_id}' \
+             AND json_get_string(procedure_trigger, 'type') = 'Submitted' \
+             AND catalog_name = '{CATALOG}' AND schema_name IS NULL AND table_name IS NULL \
+             AND table_id IS NULL AND physical_table_id IS NULL \
+             GROUP BY json_get_int(payload, 'version'), \
+                 json_get_string(payload, 'resolve_strategy'), \
+                 json_get_bool(payload, 'fail_fast'), \
+                 json_get_int(payload, 'parallelism')"
+        ),
+        "\
++-------------+---------+------------------+-----------+-------------+
+| event_count | version | resolve_strategy | fail_fast | parallelism |
++-------------+---------+------------------+-----------+-------------+
+| 1           | 1       | use_latest       | false     | 1           |
++-------------+---------+------------------+-----------+-------------+",
+    )
+    .await;
+
+    let catalog_result_payload = find_eventually_string(
+        &frontend,
+        &format!(
+            "SELECT json_to_string(payload) AS payload FROM {EVENTS_TABLE} \
+             WHERE type = 'reconcile_catalog' AND procedure_id = '{catalog_procedure_id}' \
+             AND json_get_string(procedure_trigger, 'type') = 'Succeeded' \
+             AND catalog_name = '{CATALOG}' AND schema_name IS NULL AND table_name IS NULL \
+             AND table_id IS NULL AND physical_table_id IS NULL LIMIT 1"
+        ),
+        "payload",
+    )
+    .await;
+    let catalog_result: serde_json::Value = serde_json::from_str(&catalog_result_payload).unwrap();
+    assert_eq!(catalog_result["version"], serde_json::json!(1));
+    assert_eq!(catalog_result["complete"], serde_json::json!(true));
+    let processed_database_count = catalog_result["processed_database_count"].as_u64().unwrap();
+    let succeeded_database_count = catalog_result["succeeded_database_count"].as_u64().unwrap();
+    let failed_database_count = catalog_result["failed_database_count"].as_u64().unwrap();
+    assert_eq!(
+        processed_database_count,
+        succeeded_database_count + failed_database_count
+    );
+    assert!(processed_database_count > 0);
+    assert_eq!(failed_database_count, 0);
+
+    let database_procedure_id = find_eventually_string(
+        &frontend,
+        &format!(
+            "SELECT procedure_id FROM {EVENTS_TABLE} \
+             WHERE type = 'reconcile_database' AND catalog_name = '{CATALOG}' \
+             AND schema_name = '{DATABASE}' \
+             AND json_get_string(procedure_trigger, 'type') = 'Submitted' LIMIT 1"
+        ),
+        "procedure_id",
+    )
+    .await;
+
+    assert_eventually_eq(
+        &frontend,
+        &format!(
+            "SELECT parent.procedure_id AS parent_procedure_id, \
+                 child.procedure_id AS child_procedure_id, \
+                 parent.type AS parent_event_type, \
+                 child.type AS child_event_type, \
+                 parent.catalog_name AS parent_catalog_name, \
+                 parent.schema_name AS parent_schema_name, \
+                 child.catalog_name AS child_catalog_name, \
+                 child.schema_name AS child_schema_name \
+             FROM {EVENTS_TABLE} AS parent \
+             JOIN {EVENTS_TABLE} AS child \
+               ON json_get_string(parent.procedure_trigger, 'procedure_id') = child.procedure_id \
+             WHERE parent.procedure_id = '{catalog_procedure_id}' \
+             AND json_get_string(parent.procedure_trigger, 'type') = 'ChildSubmitted' \
+             AND child.procedure_id = '{database_procedure_id}' \
+             AND json_get_string(child.procedure_trigger, 'type') = 'Submitted'"
+        ),
+        &format!(
+            "\
++--------------------------------------+--------------------------------------+-------------------+--------------------+---------------------+--------------------+--------------------+-------------------------------+
+| parent_procedure_id                  | child_procedure_id                   | parent_event_type | child_event_type   | parent_catalog_name | parent_schema_name | child_catalog_name | child_schema_name             |
++--------------------------------------+--------------------------------------+-------------------+--------------------+---------------------+--------------------+--------------------+-------------------------------+
+| {catalog_procedure_id} | {database_procedure_id} | reconcile_catalog | reconcile_database | greptime            |                    | greptime           | reconciliation_event_database |
++--------------------------------------+--------------------------------------+-------------------+--------------------+---------------------+--------------------+--------------------+-------------------------------+"
+        ),
+    )
+    .await;
+
+    assert_eventually_eq(
+        &frontend,
+        &format!(
+            "SELECT count(*) AS event_count, \
+                 json_get_string(procedure_trigger, 'outcome') AS outcome \
+             FROM {EVENTS_TABLE} \
+             WHERE type = 'reconcile_catalog' AND procedure_id = '{catalog_procedure_id}' \
+             AND json_get_string(procedure_trigger, 'type') = 'ChildSubmitted' \
+             AND json_get_string(procedure_trigger, 'procedure_id') = '{database_procedure_id}' \
+             AND catalog_name = '{CATALOG}' AND schema_name IS NULL \
+             AND json_is_null(payload) \
+             GROUP BY json_get_string(procedure_trigger, 'outcome')"
+        ),
+        "\
++-------------+----------+
+| event_count | outcome  |
++-------------+----------+
+| 1           | Accepted |
++-------------+----------+",
+    )
+    .await;
+
+    assert_eventually_eq(
+        &frontend,
+        &format!(
+            "SELECT count(*) AS event_count, \
+                 json_get_int(payload, 'version') AS version, \
+                 json_get_string(payload, 'resolve_strategy') AS resolve_strategy, \
+                 json_get_bool(payload, 'fail_fast') AS fail_fast, \
+                 json_get_int(payload, 'parallelism') AS parallelism, \
+                 json_get_bool(payload, 'is_subprocedure') AS is_subprocedure \
+             FROM {EVENTS_TABLE} \
+             WHERE type = 'reconcile_database' AND procedure_id = '{database_procedure_id}' \
+             AND json_get_string(procedure_trigger, 'type') = 'Submitted' \
+             AND catalog_name = '{CATALOG}' AND schema_name = '{DATABASE}' \
+             AND table_name IS NULL AND table_id IS NULL AND physical_table_id IS NULL \
+             GROUP BY json_get_int(payload, 'version'), \
+                 json_get_string(payload, 'resolve_strategy'), \
+                 json_get_bool(payload, 'fail_fast'), \
+                 json_get_int(payload, 'parallelism'), \
+                 json_get_bool(payload, 'is_subprocedure')"
+        ),
+        "\
++-------------+---------+------------------+-----------+-------------+-----------------+
+| event_count | version | resolve_strategy | fail_fast | parallelism | is_subprocedure |
++-------------+---------+------------------+-----------+-------------+-----------------+
+| 1           | 1       | use_latest       | false     | 1           | true            |
++-------------+---------+------------------+-----------+-------------+-----------------+",
+    )
+    .await;
+
+    assert_eventually_eq(
+        &frontend,
+        &format!(
+            "SELECT count(*) AS event_count, \
+                 json_get_int(payload, 'version') AS version, \
+                 json_get_bool(payload, 'complete') AS complete, \
+                 json_get_int(payload, 'processed_table_count') AS processed_count, \
+                 json_get_int(payload, 'succeeded_table_count') AS succeeded_count, \
+                 json_get_int(payload, 'failed_table_count') AS failed_count, \
+                 json_get_int(payload, 'succeeded_subprocedure_count') AS succeeded_child_count, \
+                 json_get_int(payload, 'failed_subprocedure_count') AS failed_child_count \
+             FROM {EVENTS_TABLE} \
+             WHERE type = 'reconcile_database' AND procedure_id = '{database_procedure_id}' \
+             AND json_get_string(procedure_trigger, 'type') = 'Succeeded' \
+             AND catalog_name = '{CATALOG}' AND schema_name = '{DATABASE}' \
+             AND table_name IS NULL AND table_id IS NULL AND physical_table_id IS NULL \
+             GROUP BY json_get_int(payload, 'version'), \
+                 json_get_bool(payload, 'complete'), \
+                 json_get_int(payload, 'processed_table_count'), \
+                 json_get_int(payload, 'succeeded_table_count'), \
+                 json_get_int(payload, 'failed_table_count'), \
+                 json_get_int(payload, 'succeeded_subprocedure_count'), \
+                 json_get_int(payload, 'failed_subprocedure_count')"
+        ),
+        "\
++-------------+---------+----------+-----------------+-----------------+--------------+-----------------------+--------------------+
+| event_count | version | complete | processed_count | succeeded_count | failed_count | succeeded_child_count | failed_child_count |
++-------------+---------+----------+-----------------+-----------------+--------------+-----------------------+--------------------+
+| 1           | 1       | true     | 1               | 1               | 0            | 1                     | 0                  |
++-------------+---------+----------+-----------------+-----------------+--------------+-----------------------+--------------------+",
+    )
+    .await;
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn test_table_reconciliation_events() {
     common_telemetry::init_default_ut_logging();