feat: udf onnx feature expression

Cargo.toml

@@ -14,6 +14,7 @@ members = [
   "dozer-log-js",
   "dozer-log-python",
   "dozer-utils",
+  "dozer-recordstore",
 ]
 resolver = "2"
 

dozer-cli/Cargo.toml

@@ -18,6 +18,7 @@ dozer-sql = { path = "../dozer-sql" }
 dozer-types = { path = "../dozer-types" }
 dozer-tracing = { path = "../dozer-tracing" }
 dozer-storage = { path = "../dozer-storage" }
+dozer-recordstore = { path = "../dozer-recordstore" }
 
 uuid = { version = "1.3.0", features = ["v4", "serde"] }
 tokio = { version = "1", features = ["full"] }

dozer-cli/src/pipeline/dummy_sink.rs

@@ -5,9 +5,9 @@ use dozer_core::{
     epoch::Epoch,
     executor_operation::ProcessorOperation,
     node::{PortHandle, Sink, SinkFactory},
-    processor_record::ProcessorRecordStore,
     DEFAULT_PORT_HANDLE,
 };
+use dozer_recordstore::ProcessorRecordStore;
 use dozer_types::{errors::internal::BoxedError, types::Schema};
 
 #[derive(Debug)]

dozer-cli/src/pipeline/log_sink.rs

@@ -8,9 +8,9 @@ use dozer_core::{
     epoch::Epoch,
     executor_operation::ProcessorOperation,
     node::{PortHandle, Sink, SinkFactory},
-    processor_record::ProcessorRecordStore,
     DEFAULT_PORT_HANDLE,
 };
+use dozer_recordstore::ProcessorRecordStore;
 use dozer_tracing::LabelsAndProgress;
 use dozer_types::indicatif::ProgressBar;
 use dozer_types::types::Schema;
@@ -105,9 +105,7 @@ impl Sink for LogSink {
         self.log
             .lock()
             .write(dozer_cache::dozer_log::replication::LogOperation::Op {
-                op: record_store
-                    .load_operation(&op)
-                    .map_err(Into::<BoxedError>::into)?,
+                op: op.load(record_store)?,
             });
         self.update_counter();
         Ok(())

dozer-core/Cargo.toml

@@ -11,6 +11,7 @@ dozer-log = { path = "../dozer-log/" }
 dozer-storage = {path = "../dozer-storage/"}
 dozer-types = {path = "../dozer-types/"}
 dozer-tracing = {path = "../dozer-tracing/"}
+dozer-recordstore = {path = "../dozer-recordstore/"}
 
 uuid = {version = "1.3.0", features = ["v1", "v4", "fast-rng"]}
 crossbeam = "0.8.2"

dozer-core/src/checkpoint/mod.rs

@@ -7,6 +7,7 @@ use dozer_log::{
     storage::{self, Object, Queue, Storage},
     tokio::task::JoinHandle,
 };
+use dozer_recordstore::ProcessorRecordStore;
 use dozer_types::{
     bincode,
     log::{error, info},
@@ -18,7 +19,7 @@ use dozer_types::{
 };
 use tempdir::TempDir;
 
-use crate::{errors::ExecutionError, processor_record::ProcessorRecordStore};
+use crate::errors::ExecutionError;
 
 #[derive(Debug)]
 pub struct CheckpointFactory {

dozer-core/src/epoch/manager.rs

@@ -1,3 +1,4 @@
+use dozer_recordstore::ProcessorRecordStore;
 use dozer_types::node::{NodeHandle, SourceStates, TableState};
 use dozer_types::parking_lot::Mutex;
 use std::collections::HashMap;
@@ -7,7 +8,6 @@ use std::thread::sleep;
 use std::time::{Duration, SystemTime};
 
 use crate::checkpoint::{CheckpointFactory, CheckpointWriter};
-use crate::processor_record::ProcessorRecordStore;
 
 use super::EpochCommonInfo;
 

dozer-core/src/errors.rs

@@ -1,7 +1,7 @@
 use std::path::PathBuf;
 
 use crate::node::PortHandle;
-use dozer_storage::errors::StorageError;
+use dozer_recordstore::RecordStoreError;
 use dozer_types::errors::internal::BoxedError;
 use dozer_types::node::NodeHandle;
 use dozer_types::thiserror::Error;
@@ -35,8 +35,8 @@ pub enum ExecutionError {
     Source(#[source] BoxedError),
     #[error("File system error {0:?}: {1}")]
     FileSystemError(PathBuf, #[source] std::io::Error),
-    #[error("Storage error: {0}")]
-    Storage(#[from] StorageError),
+    #[error("Recordstore error: {0}")]
+    RecordStore(#[from] RecordStoreError),
     #[error("Object storage error: {0}")]
     ObjectStorage(#[from] dozer_log::storage::Error),
     #[error("Checkpoint writer thread panicked")]

dozer-core/src/executor/execution_dag.rs

@@ -15,14 +15,14 @@ use crate::{
     executor_operation::ExecutorOperation,
     hash_map_to_vec::insert_vec_element,
     node::PortHandle,
-    processor_record::ProcessorRecordStore,
     record_store::{create_record_writer, RecordWriter},
 };
 use crossbeam::channel::{bounded, Receiver, Sender};
 use daggy::petgraph::{
     visit::{EdgeRef, IntoEdges, IntoEdgesDirected, IntoNodeIdentifiers},
     Direction,
 };
+use dozer_recordstore::ProcessorRecordStore;
 use dozer_tracing::LabelsAndProgress;
 
 pub type SharedRecordWriter = Rc<RefCell<Option<Box<dyn RecordWriter>>>>;

dozer-core/src/executor/processor_node.rs

@@ -8,13 +8,13 @@ use dozer_types::node::NodeHandle;
 use crate::epoch::Epoch;
 use crate::error_manager::ErrorManager;
 use crate::executor_operation::{ExecutorOperation, ProcessorOperation};
-use crate::processor_record::ProcessorRecordStore;
 use crate::{
     builder_dag::NodeKind,
     errors::ExecutionError,
     forwarder::ProcessorChannelManager,
     node::{PortHandle, Processor},
 };
+use dozer_recordstore::ProcessorRecordStore;
 
 use super::{execution_dag::ExecutionDag, name::Name, receiver_loop::ReceiverLoop};
 

dozer-core/src/executor/receiver_loop.rs

@@ -105,7 +105,7 @@ mod tests {
         types::{Field, Record},
     };
 
-    use crate::processor_record::{ProcessorRecord, ProcessorRecordStore};
+    use dozer_recordstore::{ProcessorRecord, ProcessorRecordStore};
 
     use super::*;
 

dozer-core/src/executor_operation.rs

@@ -1,4 +1,7 @@
-use crate::{epoch::Epoch, processor_record::ProcessorRecord};
+use dozer_recordstore::{ProcessorRecord, ProcessorRecordStore};
+use dozer_types::types::Operation;
+
+use crate::{epoch::Epoch, errors::ExecutionError};
 
 #[derive(Clone, Debug, PartialEq, Eq)]
 /// A CDC event.
@@ -15,6 +18,41 @@ pub enum ProcessorOperation {
     },
 }
 
+impl ProcessorOperation {
+    pub fn new(
+        op: &Operation,
+        record_store: &ProcessorRecordStore,
+    ) -> Result<Self, ExecutionError> {
+        Ok(match op {
+            Operation::Delete { old } => ProcessorOperation::Delete {
+                old: record_store.create_record(old)?,
+            },
+            Operation::Insert { new } => ProcessorOperation::Insert {
+                new: record_store.create_record(new)?,
+            },
+            Operation::Update { old, new } => ProcessorOperation::Update {
+                old: record_store.create_record(old)?,
+                new: record_store.create_record(new)?,
+            },
+        })
+    }
+
+    pub fn load(&self, record_store: &ProcessorRecordStore) -> Result<Operation, ExecutionError> {
+        Ok(match self {
+            ProcessorOperation::Delete { old } => Operation::Delete {
+                old: record_store.load_record(old)?,
+            },
+            ProcessorOperation::Insert { new } => Operation::Insert {
+                new: record_store.load_record(new)?,
+            },
+            ProcessorOperation::Update { old, new } => Operation::Update {
+                old: record_store.load_record(old)?,
+                new: record_store.load_record(new)?,
+            },
+        })
+    }
+}
+
 #[derive(Clone, Debug)]
 pub enum ExecutorOperation {
     Op { op: ProcessorOperation },

dozer-core/src/forwarder.rs

@@ -234,7 +234,7 @@ impl SourceChannelManager {
                 );
 
                 self.manager.send_op(
-                    self.epoch_manager.record_store().create_operation(&op)?,
+                    ProcessorOperation::new(&op, self.epoch_manager.record_store())?,
                     port,
                 )?;
                 self.num_uncommitted_ops += 1;

dozer-core/src/lib.rs

@@ -14,7 +14,6 @@ pub mod executor_operation;
 pub mod forwarder;
 mod hash_map_to_vec;
 pub mod node;
-pub mod processor_record;
 pub mod record_store;
 
 #[cfg(test)]

dozer-core/src/node.rs

@@ -1,7 +1,7 @@
 use crate::channels::{ProcessorChannelForwarder, SourceChannelForwarder};
 use crate::epoch::Epoch;
 use crate::executor_operation::ProcessorOperation;
-use crate::processor_record::ProcessorRecordStore;
+use dozer_recordstore::ProcessorRecordStore;
 
 use dozer_log::storage::{Object, Queue};
 use dozer_types::errors::internal::BoxedError;