feat(processing_engine): initial implementation of Processing Engine plugins and triggers

Cargo.toml

@@ -8,6 +8,7 @@ members = [
     "influxdb3_id",
     "influxdb3_load_generator",
     "influxdb3_process",
+    "influxdb3_py_api",
     "influxdb3_server",
     "influxdb3_telemetry",
     "influxdb3_test_helpers",

influxdb3/Cargo.toml

@@ -73,6 +73,7 @@ tokio_console = ["console-subscriber", "tokio/tracing", "observability_deps/rele
 
 # Use jemalloc as the default allocator.
 jemalloc_replacing_malloc = ["influxdb3_process/jemalloc_replacing_malloc"]
+system-py = ["influxdb3_write/system-py"]
 
 [dev-dependencies]
 # Core Crates

influxdb3/tests/server/flight.rs

@@ -108,20 +108,22 @@ async fn flight() -> Result<(), influxdb3_client::Error> {
 
         assert_batches_sorted_eq!(
             [
-                "+--------------+--------------------+---------------+------------+",
-                "| catalog_name | db_schema_name     | table_name    | table_type |",
-                "+--------------+--------------------+---------------+------------+",
-                "| public       | information_schema | columns       | VIEW       |",
-                "| public       | information_schema | df_settings   | VIEW       |",
-                "| public       | information_schema | schemata      | VIEW       |",
-                "| public       | information_schema | tables        | VIEW       |",
-                "| public       | information_schema | views         | VIEW       |",
-                "| public       | iox                | cpu           | BASE TABLE |",
-                "| public       | system             | last_caches   | BASE TABLE |",
-                "| public       | system             | meta_caches   | BASE TABLE |",
-                "| public       | system             | parquet_files | BASE TABLE |",
-                "| public       | system             | queries       | BASE TABLE |",
-                "+--------------+--------------------+---------------+------------+",
+                "+--------------+--------------------+----------------------------+------------+",
+                "| catalog_name | db_schema_name     | table_name                 | table_type |",
+                "+--------------+--------------------+----------------------------+------------+",
+                "| public       | information_schema | columns                    | VIEW       |",
+                "| public       | information_schema | df_settings                | VIEW       |",
+                "| public       | information_schema | schemata                   | VIEW       |",
+                "| public       | information_schema | tables                     | VIEW       |",
+                "| public       | information_schema | views                      | VIEW       |",
+                "| public       | iox                | cpu                        | BASE TABLE |",
+                "| public       | system             | last_caches                | BASE TABLE |",
+                "| public       | system             | meta_caches                | BASE TABLE |",
+                "| public       | system             | parquet_files              | BASE TABLE |",
+                "| public       | system             | processing_engine_plugins  | BASE TABLE |",
+                "| public       | system             | processing_engine_triggers | BASE TABLE |",
+                "| public       | system             | queries                    | BASE TABLE |",
+                "+--------------+--------------------+----------------------------+------------+",
             ],
             &batches
         );

influxdb3_cache/src/last_cache/mod.rs

@@ -1267,6 +1267,8 @@ mod tests {
                 map.insert(TableId::from(1), "test_table_2".into());
                 map
             },
+            processing_engine_plugins: Default::default(),
+            processing_engine_triggers: Default::default(),
             deleted: false,
         };
         let table_id = TableId::from(0);

influxdb3_catalog/src/catalog.rs

@@ -1,13 +1,16 @@
 //! Implementation of the Catalog that sits entirely in memory.
 
-use crate::catalog::Error::TableNotFound;
+use crate::catalog::Error::{
+    ProcessingEngineCallExists, ProcessingEngineTriggerExists, TableNotFound,
+};
 use bimap::BiHashMap;
 use hashbrown::HashMap;
 use indexmap::IndexMap;
 use influxdb3_id::{ColumnId, DbId, SerdeVecMap, TableId};
 use influxdb3_wal::{
     CatalogBatch, CatalogOp, DeleteDatabaseDefinition, DeleteTableDefinition, FieldAdditions,
     FieldDefinition, LastCacheDefinition, LastCacheDelete, MetaCacheDefinition, MetaCacheDelete,
+    PluginDefinition, TriggerDefinition,
 };
 use influxdb_line_protocol::FieldValue;
 use iox_time::Time;
@@ -74,6 +77,37 @@ pub enum Error {
         table_name: String,
         existing: String,
     },
+
+    #[error(
+        "Cannot overwrite Processing Engine Call {} in Database {}",
+        call_name,
+        database_name
+    )]
+    ProcessingEngineCallExists {
+        database_name: String,
+        call_name: String,
+    },
+    #[error(
+        "Cannot overwrite Processing Engine Trigger {} in Database {}",
+        trigger_name,
+        database_name
+    )]
+    ProcessingEngineTriggerExists {
+        database_name: String,
+        trigger_name: String,
+    },
+
+    #[error(
+        "Processing Engine Plugin {} not in DB schema for {}",
+        plugin_name,
+        database_name
+    )]
+    ProcessingEnginePluginNotFound {
+        plugin_name: String,
+        database_name: String,
+    },
+    #[error("Processing Engine Unimplemented: {}", feature_description)]
+    ProcessingEngineUnimplemented { feature_description: String },
 }
 
 pub type Result<T, E = Error> = std::result::Result<T, E>;
@@ -481,6 +515,9 @@ pub struct DatabaseSchema {
     /// The database is a map of tables
     pub tables: SerdeVecMap<TableId, Arc<TableDefinition>>,
     pub table_map: BiHashMap<TableId, Arc<str>>,
+    pub processing_engine_plugins: HashMap<String, PluginDefinition>,
+    // TODO: care about performance of triggers
+    pub processing_engine_triggers: HashMap<String, TriggerDefinition>,
     pub deleted: bool,
 }
 
@@ -491,6 +528,8 @@ impl DatabaseSchema {
             name,
             tables: Default::default(),
             table_map: BiHashMap::new(),
+            processing_engine_plugins: HashMap::new(),
+            processing_engine_triggers: HashMap::new(),
             deleted: false,
         }
     }
@@ -642,6 +681,8 @@ impl UpdateDatabaseSchema for CatalogOp {
             }
             CatalogOp::DeleteDatabase(delete_database) => delete_database.update_schema(schema),
             CatalogOp::DeleteTable(delete_table) => delete_table.update_schema(schema),
+            CatalogOp::CreatePlugin(create_plugin) => create_plugin.update_schema(schema),
+            CatalogOp::CreateTrigger(create_trigger) => create_trigger.update_schema(schema),
         }
     }
 }
@@ -708,6 +749,53 @@ impl UpdateDatabaseSchema for DeleteTableDefinition {
     }
 }
 
+impl UpdateDatabaseSchema for PluginDefinition {
+    fn update_schema<'a>(
+        &self,
+        mut schema: Cow<'a, DatabaseSchema>,
+    ) -> Result<Cow<'a, DatabaseSchema>> {
+        match schema.processing_engine_plugins.get(&self.plugin_name) {
+            Some(current) if self.eq(current) => {}
+            Some(_) => {
+                return Err(ProcessingEngineCallExists {
+                    database_name: schema.name.to_string(),
+                    call_name: self.plugin_name.to_string(),
+                })
+            }
+            None => {
+                schema
+                    .to_mut()
+                    .processing_engine_plugins
+                    .insert(self.plugin_name.to_string(), self.clone());
+            }
+        }
+
+        Ok(schema)
+    }
+}
+
+impl UpdateDatabaseSchema for TriggerDefinition {
+    fn update_schema<'a>(
+        &self,
+        mut schema: Cow<'a, DatabaseSchema>,
+    ) -> Result<Cow<'a, DatabaseSchema>> {
+        if let Some(current) = schema.processing_engine_triggers.get(&self.trigger_name) {
+            if current == self {
+                return Ok(schema);
+            }
+            return Err(ProcessingEngineTriggerExists {
+                database_name: schema.name.to_string(),
+                trigger_name: self.trigger_name.to_string(),
+            });
+        }
+        schema
+            .to_mut()
+            .processing_engine_triggers
+            .insert(self.trigger_name.to_string(), self.clone());
+        Ok(schema)
+    }
+}
+
 fn make_new_name_using_deleted_time(name: &str, deletion_time: Time) -> Arc<str> {
     Arc::from(format!(
         "{}-{}",
@@ -1187,6 +1275,8 @@ mod tests {
                 map.insert(TableId::from(2), "test_table_2".into());
                 map
             },
+            processing_engine_plugins: Default::default(),
+            processing_engine_triggers: Default::default(),
             deleted: false,
         };
         use InfluxColumnType::*;
@@ -1396,6 +1486,8 @@ mod tests {
             name: "test".into(),
             tables: SerdeVecMap::new(),
             table_map: BiHashMap::new(),
+            processing_engine_plugins: Default::default(),
+            processing_engine_triggers: Default::default(),
             deleted: false,
         };
         database.tables.insert(
@@ -1453,6 +1545,8 @@ mod tests {
                 map.insert(TableId::from(1), "test_table_1".into());
                 map
             },
+            processing_engine_plugins: Default::default(),
+            processing_engine_triggers: Default::default(),
             deleted: false,
         };
         use InfluxColumnType::*;
@@ -1508,6 +1602,8 @@ mod tests {
                 map.insert(TableId::from(0), "test".into());
                 map
             },
+            processing_engine_plugins: Default::default(),
+            processing_engine_triggers: Default::default(),
             deleted: false,
         };
         use InfluxColumnType::*;
@@ -1607,6 +1703,8 @@ mod tests {
             name: "test".into(),
             tables: SerdeVecMap::new(),
             table_map: BiHashMap::new(),
+            processing_engine_plugins: Default::default(),
+            processing_engine_triggers: Default::default(),
             deleted: false,
         };
         let deleted_table_id = TableId::new();