feat(processing_engine): Store triggers by DbId and TriggerId instead of name

Author: kumarlokesh

Cargo.lock

@@ -2516,7 +2516,7 @@ def process_scheduled_call(influxdb3_local, schedule_time, args=None):
         .unwrap();
 
     // Wait for trigger to run several times
-    tokio::time::sleep(std::time::Duration::from_millis(3100)).await;
+    tokio::time::sleep(std::time::Duration::from_millis(5000)).await;
 
     // Query to see what values were written before disabling
     let first_query_result = server
@@ -2535,7 +2535,7 @@ def process_scheduled_call(influxdb3_local, schedule_time, args=None):
     server.enable_trigger(db_name, trigger_name).run().unwrap();
 
     // Wait for trigger to run again
-    tokio::time::sleep(std::time::Duration::from_millis(1100)).await;
+    tokio::time::sleep(std::time::Duration::from_millis(5000)).await;
 
     // Query results after re-enabling
     let second_query_result = server

influxdb3/tests/cli/mod.rs

@@ -300,9 +300,8 @@ impl Catalog {
         self.inner.read().db_exists(db_id)
     }
 
-    /// Get active triggers by database and trigger name
-    // NOTE: this could be id-based in future
-    pub fn active_triggers(&self) -> Vec<(Arc<str>, Arc<str>)> {
+    /// Get active triggers by database ID and trigger ID
+    pub fn active_triggers(&self) -> Vec<(DbId, TriggerId)> {
         let inner = self.inner.read();
         let result = inner
             .databases
@@ -314,7 +313,7 @@ impl Catalog {
                         if trigger.disabled {
                             None
                         } else {
-                            Some((Arc::clone(&db.name), Arc::clone(&trigger.trigger_name)))
+                            Some((db.id, trigger.trigger_id))
                         }
                     })
             })

influxdb3_catalog/src/catalog.rs

@@ -539,6 +539,7 @@ impl Catalog {
                 db.id,
                 db.name(),
                 vec![DatabaseCatalogOp::CreateTrigger(TriggerDefinition {
+                    database_id: db.id,
                     trigger_id,
                     trigger_name: trigger_name.into(),
                     plugin_filename: plugin_filename.to_string(),

influxdb3_catalog/src/catalog/update.rs

@@ -642,6 +642,7 @@ pub struct TriggerDefinition {
     pub trigger_id: TriggerId,
     pub trigger_name: Arc<str>,
     pub plugin_filename: String,
+    pub database_id: DbId,
     pub database_name: Arc<str>,
     pub node_id: Arc<str>,
     pub trigger: TriggerSpecificationDefinition,

influxdb3_catalog/src/log.rs

@@ -211,6 +211,7 @@ impl Snapshot for TableDefinition {
 
 #[derive(Debug, Serialize, Deserialize)]
 pub(crate) struct ProcessingEngineTriggerSnapshot {
+    pub database_id: DbId,
     pub trigger_id: TriggerId,
     pub trigger_name: Arc<str>,
     pub node_id: Arc<str>,
@@ -227,6 +228,7 @@ impl Snapshot for TriggerDefinition {
 
     fn snapshot(&self) -> Self::Serialized {
         Self::Serialized {
+            database_id: self.database_id,
             trigger_id: self.trigger_id,
             trigger_name: Arc::clone(&self.trigger_name),
             node_id: Arc::clone(&self.node_id),
@@ -241,6 +243,7 @@ impl Snapshot for TriggerDefinition {
 
     fn from_snapshot(snap: Self::Serialized) -> Self {
         Self {
+            database_id: snap.database_id,
             trigger_id: snap.trigger_id,
             trigger_name: snap.trigger_name,
             node_id: snap.node_id,

influxdb3_catalog/src/snapshot.rs

@@ -19,6 +19,7 @@ hyper.workspace = true
 iox_time.workspace = true
 influxdb3_catalog = { path = "../influxdb3_catalog" }
 influxdb3_client = { path = "../influxdb3_client" }
+influxdb3_id = { path = "../influxdb3_id" }
 influxdb3_internal_api = { path = "../influxdb3_internal_api" }
 influxdb3_py_api = { path = "../influxdb3_py_api" }
 influxdb3_types = { path = "../influxdb3_types" }