Merge branch 'main' into feat-Add-SQL-CHR-function

getdozer · Feb 28, 2024 · 63bc485 · 63bc485
2 parents caaa216 + 1b88cb8
commit 63bc485
Show file tree

Hide file tree

Showing 46 changed files with 450 additions and 588 deletions.
diff --git a/.github/workflows/integration/test-dozer.sh b/.github/workflows/integration/test-dozer.sh
@@ -2,9 +2,3 @@ set -e
 
 # Check if dozer version matches `DOZER_VERSION`
 dozer -V | grep "$DOZER_VERSION"
-
-# Install Rust
-curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y
-
-# Run grpc ingest test because it doesn't need docker or ETH secrets.
-CARGO_TARGET_DIR=../ DOZER_BIN=dozer RUST_LOG=info "$HOME/.cargo/bin/cargo" run -p dozer-tests --bin dozer-tests -- grpc_ingest
diff --git a/.github/workflows/unit.yaml b/.github/workflows/unit.yaml
@@ -80,8 +80,11 @@ jobs:
           SN_DRIVER: ${{ secrets.SN_DRIVER }}
         shell: bash
         run: |
-          cargo test -p 'dozer-ingestion-*' --lib --no-fail-fast -- --ignored
-
+          cargo test \
+          -p dozer-ingestion-postgres \
+          -p dozer-ingestion-kafka \
+          -p dozer-ingestion-mysql \
+          --lib --no-fail-fast -- --ignored
       - name: Run tests
         shell: bash
         run: |

diff --git a/Cargo.lock b/Cargo.lock
diff --git a/dozer-cli/Cargo.toml b/dozer-cli/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "dozer-cli"
-version = "0.3.0"
+version = "0.4.0"
 edition = "2021"
 default-run = "dozer"
 authors = ["getdozer/dozer-dev"]

diff --git a/dozer-cli/src/pipeline/builder.rs b/dozer-cli/src/pipeline/builder.rs
@@ -365,23 +365,24 @@ fn add_sink_to_pipeline(
     id: &str,
     table_infos: Vec<(&OutputTableInfo, PortHandle)>,
 ) {
-    let mut connections = vec![];
-    let mut entry_points = vec![];
+    pipeline.add_sink(sink, id.to_string());
 
     for (table_info, port) in table_infos {
         match table_info {
             OutputTableInfo::Original(table_info) => {
-                entry_points.push(PipelineEntryPoint::new(table_info.table_name.clone(), port))
+                pipeline.add_entry_point(
+                    id.to_string(),
+                    PipelineEntryPoint::new(table_info.table_name.clone(), port),
+                );
             }
             OutputTableInfo::Transformed(table_info) => {
-                connections.push((table_info, port));
+                pipeline.connect_nodes(
+                    table_info.node.clone(),
+                    table_info.port,
+                    id.to_string(),
+                    port,
+                );
             }
         }
     }
-
-    pipeline.add_sink(sink, id, entry_points);
-
-    for (table_info, port) in connections {
-        pipeline.connect_nodes(&table_info.node, table_info.port, id, port);
-    }
 }
diff --git a/dozer-core/Cargo.toml b/dozer-core/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "dozer-core"
-version = "0.3.0"
+version = "0.4.0"
 edition = "2021"
 authors = ["getdozer/dozer-dev"]
 

diff --git a/dozer-core/src/app.rs b/dozer-core/src/app.rs
@@ -33,44 +33,33 @@ pub struct AppPipeline {
 }
 
 impl AppPipeline {
-    pub fn add_processor(
-        &mut self,
-        proc: Box<dyn ProcessorFactory>,
-        id: &str,
-        entry_points: Vec<PipelineEntryPoint>,
-    ) {
-        let handle = NodeHandle::new(None, id.to_string());
-        self.processors.push((handle.clone(), proc));
+    fn create_handle(id: String) -> NodeHandle {
+        NodeHandle::new(None, id)
+    }
 
-        for entry_point in entry_points {
-            self.entry_points.push((handle.clone(), entry_point));
-        }
+    pub fn add_processor(&mut self, proc: Box<dyn ProcessorFactory>, id: String) {
+        self.processors.push((Self::create_handle(id), proc));
     }
 
-    pub fn add_sink(
-        &mut self,
-        sink: Box<dyn SinkFactory>,
-        id: &str,
-        entry_points: Vec<PipelineEntryPoint>,
-    ) {
-        let handle = NodeHandle::new(None, id.to_string());
-        self.sinks.push((handle.clone(), sink));
+    pub fn add_sink(&mut self, sink: Box<dyn SinkFactory>, id: String) {
+        self.sinks.push((Self::create_handle(id), sink));
+    }
 
-        for entry_point in entry_points {
-            self.entry_points.push((handle.clone(), entry_point));
-        }
+    pub fn add_entry_point(&mut self, id: String, entry_point: PipelineEntryPoint) {
+        self.entry_points
+            .push((Self::create_handle(id), entry_point));
     }
 
     pub fn connect_nodes(
         &mut self,
-        from: &str,
+        from: String,
         from_port: PortHandle,
-        to: &str,
+        to: String,
         to_port: PortHandle,
     ) {
         let edge = Edge::new(
-            Endpoint::new(NodeHandle::new(None, from.to_string()), from_port),
-            Endpoint::new(NodeHandle::new(None, to.to_string()), to_port),
+            Endpoint::new(NodeHandle::new(None, from), from_port),
+            Endpoint::new(NodeHandle::new(None, to), to_port),
         );
         self.edges.push(edge);
     }