feat(processing_engine): respond to PR feedback, put python code behi…

…nd system-py feature flag

Cargo.toml

@@ -8,7 +8,6 @@ members = [
     "influxdb3_id",
     "influxdb3_load_generator",
     "influxdb3_process",
-    "influxdb3_processing_engine",
     "influxdb3_py_api",
     "influxdb3_server",
     "influxdb3_telemetry",
@@ -40,7 +39,7 @@ license = "MIT OR Apache-2.0"
 
 [workspace.dependencies]
 anyhow = "1.0"
-arrow = { version = "53.0.0", features = ["prettyprint", "chrono-tz", "pyarrow"] }
+arrow = { version = "53.0.0", features = ["prettyprint", "chrono-tz"] }
 arrow-array = "53.0.0"
 arrow-buffer = "53.0.0"
 arrow-csv = "53.0.0"

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_catalog/Cargo.toml

@@ -15,7 +15,6 @@ iox_time.workspace = true
 # Local deps
 influxdb3_id = { path = "../influxdb3_id" }
 influxdb3_wal = { path = "../influxdb3_wal" }
-influxdb3_processing_engine = { path = "../influxdb3_processing_engine" }
 
 # crates.io dependencies
 arrow.workspace = true

influxdb3_catalog/src/catalog.rs

@@ -1,16 +1,12 @@
 //! Implementation of the Catalog that sits entirely in memory.
 
 use crate::catalog::Error::{
-    ProcessingEngineCallExists, ProcessingEnginePluginNotFound, ProcessingEngineTriggerExists,
-    TableNotFound,
+    ProcessingEngineCallExists, ProcessingEngineTriggerExists, TableNotFound,
 };
 use bimap::BiHashMap;
 use hashbrown::HashMap;
 use indexmap::IndexMap;
 use influxdb3_id::{ColumnId, DbId, SerdeVecMap, TableId};
-use influxdb3_processing_engine::processing_engine_plugins::{
-    ProcessingEnginePlugin, ProcessingEngineTrigger,
-};
 use influxdb3_wal::{
     CatalogBatch, CatalogOp, DeleteDatabaseDefinition, DeleteTableDefinition, FieldAdditions,
     FieldDefinition, LastCacheDefinition, LastCacheDelete, MetaCacheDefinition, MetaCacheDelete,
@@ -519,9 +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, ProcessingEnginePlugin>,
+    pub processing_engine_plugins: HashMap<String, PluginDefinition>,
     // TODO: care about performance of triggers
-    pub processing_engine_triggers: HashMap<String, ProcessingEngineTrigger>,
+    pub processing_engine_triggers: HashMap<String, TriggerDefinition>,
     pub deleted: bool,
 }
 
@@ -758,21 +754,19 @@ impl UpdateDatabaseSchema for PluginDefinition {
         &self,
         mut schema: Cow<'a, DatabaseSchema>,
     ) -> Result<Cow<'a, DatabaseSchema>> {
-        let plugin: ProcessingEnginePlugin = self.into();
-
         match schema.processing_engine_plugins.get(&self.plugin_name) {
-            Some(current) if plugin.eq(current) => {}
+            Some(current) if self.eq(current) => {}
             Some(_) => {
                 return Err(ProcessingEngineCallExists {
                     database_name: schema.name.to_string(),
-                    call_name: plugin.plugin_name.to_string(),
+                    call_name: self.plugin_name.to_string(),
                 })
             }
             None => {
                 schema
                     .to_mut()
                     .processing_engine_plugins
-                    .insert(self.plugin_name.to_string(), plugin);
+                    .insert(self.plugin_name.to_string(), self.clone());
             }
         }
 
@@ -785,35 +779,19 @@ impl UpdateDatabaseSchema for TriggerDefinition {
         &self,
         mut schema: Cow<'a, DatabaseSchema>,
     ) -> Result<Cow<'a, DatabaseSchema>> {
-        let trigger_name = self.trigger_name.to_string();
-        let Some(plugin) = schema
-            .processing_engine_plugins
-            .get(&self.plugin_name)
-            .cloned()
-        else {
-            return Err(ProcessingEnginePluginNotFound {
-                plugin_name: self.plugin_name.to_string(),
-                database_name: schema.name.to_string(),
-            });
-        };
-        let trigger = ProcessingEngineTrigger {
-            trigger_name: trigger_name.to_string(),
-            plugin,
-            trigger: (&self.trigger).into(),
-        };
-        if let Some(current) = schema.processing_engine_triggers.get(&trigger_name) {
-            if current == &trigger {
+        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,
+                trigger_name: self.trigger_name.to_string(),
             });
         }
         schema
             .to_mut()
             .processing_engine_triggers
-            .insert(trigger_name, trigger);
+            .insert(self.trigger_name.to_string(), self.clone());
         Ok(schema)
     }
 }

influxdb3_catalog/src/serialize.rs

@@ -8,10 +8,9 @@ use influxdb3_id::ColumnId;
 use influxdb3_id::DbId;
 use influxdb3_id::SerdeVecMap;
 use influxdb3_id::TableId;
-use influxdb3_processing_engine::processing_engine_plugins::{
-    ProcessingEnginePlugin, ProcessingEngineTrigger,
+use influxdb3_wal::{
+    LastCacheDefinition, LastCacheValueColumnsDef, PluginDefinition, TriggerDefinition,
 };
-use influxdb3_wal::{LastCacheDefinition, LastCacheValueColumnsDef};
 use schema::InfluxColumnType;
 use schema::InfluxFieldType;
 use schema::TIME_DATA_TIMEZONE;
@@ -95,14 +94,15 @@ impl From<DatabaseSnapshot> for DatabaseSchema {
             .into_iter()
             .map(|(name, trigger)| {
                 // TODO: Decide whether to handle errors
-                let plugin = processing_engine_plugins
+                let plugin: PluginDefinition = processing_engine_plugins
                     .get(&trigger.plugin_name)
                     .cloned()
                     .expect("should have plugin");
                 (
                     name,
-                    ProcessingEngineTrigger {
+                    TriggerDefinition {
                         trigger_name: trigger.trigger_name,
+                        plugin_name: plugin.plugin_name.to_string(),
                         plugin,
                         trigger: serde_json::from_str(&trigger.trigger_specification).unwrap(),
                     },
@@ -405,8 +405,8 @@ impl From<TableSnapshot> for TableDefinition {
     }
 }
 
-impl From<&ProcessingEnginePlugin> for ProcessingEnginePluginSnapshot {
-    fn from(plugin: &ProcessingEnginePlugin) -> Self {
+impl From<&PluginDefinition> for ProcessingEnginePluginSnapshot {
+    fn from(plugin: &PluginDefinition) -> Self {
         Self {
             plugin_name: plugin.plugin_name.to_string(),
             code: plugin.code.to_string(),
@@ -416,7 +416,7 @@ impl From<&ProcessingEnginePlugin> for ProcessingEnginePluginSnapshot {
     }
 }
 
-impl From<ProcessingEnginePluginSnapshot> for ProcessingEnginePlugin {
+impl From<ProcessingEnginePluginSnapshot> for PluginDefinition {
     fn from(plugin: ProcessingEnginePluginSnapshot) -> Self {
         Self {
             plugin_name: plugin.plugin_type.to_string(),
@@ -427,11 +427,11 @@ impl From<ProcessingEnginePluginSnapshot> for ProcessingEnginePlugin {
     }
 }
 
-impl From<&ProcessingEngineTrigger> for ProcessingEngineTriggerSnapshot {
-    fn from(trigger: &ProcessingEngineTrigger) -> Self {
+impl From<&TriggerDefinition> for ProcessingEngineTriggerSnapshot {
+    fn from(trigger: &TriggerDefinition) -> Self {
         ProcessingEngineTriggerSnapshot {
             trigger_name: trigger.trigger_name.to_string(),
-            plugin_name: trigger.plugin.plugin_name.to_string(),
+            plugin_name: trigger.plugin_name.to_string(),
             trigger_specification: serde_json::to_string(&trigger.trigger)
                 .expect("should be able to serialize trigger specification"),
         }

influxdb3_py_api/Cargo.toml

@@ -5,16 +5,20 @@ authors.workspace = true
 edition.workspace = true
 license.workspace = true
 
+
+[features]
+system-py = ["pyo3"]
 [dependencies]
 influxdb3_wal = { path = "../influxdb3_wal" }
 influxdb3_catalog = {path = "../influxdb3_catalog"}
-influxdb-line-protocol = { workspace = true }
 schema = { workspace = true }
 
 [dependencies.pyo3]
 version = "0.23.3"
 # this is necessary to automatically initialize the Python interpreter
 features = ["auto-initialize"]
+optional = true
+
 
 [lints]
 workspace = true