refactor: Split connectors to separate crates

dozer-cli/src/errors.rs

@@ -17,7 +17,6 @@ use dozer_api::{
 use dozer_cache::dozer_log::storage;
 use dozer_cache::errors::CacheError;
 use dozer_core::errors::ExecutionError;
-use dozer_ingestion::errors::ConnectorError;
 use dozer_sql::errors::PipelineError;
 use dozer_types::{constants::LOCK_FILE, thiserror::Error};
 use dozer_types::{errors::internal::BoxedError, serde_json};
@@ -76,8 +75,6 @@ pub enum OrchestrationError {
     #[error(transparent)]
     ExecutionError(#[from] ExecutionError),
     #[error(transparent)]
-    ConnectorError(#[from] ConnectorError),
-    #[error(transparent)]
     PipelineError(#[from] PipelineError),
     #[error(transparent)]
     CliError(#[from] CliError),

dozer-cli/src/lib.rs

@@ -24,8 +24,8 @@ mod tests;
 mod utils;
 // Re-exports
 pub use dozer_ingestion::{
-    connectors::{get_connector, TableInfo},
     errors::ConnectorError,
+    {get_connector, TableInfo},
 };
 pub use dozer_sql::builder::QueryContext;
 pub fn wrapped_statement_to_pipeline(sql: &str) -> Result<QueryContext, PipelineError> {

dozer-cli/src/pipeline/builder.rs

@@ -9,7 +9,7 @@ use dozer_core::app::AppPipeline;
 use dozer_core::app::PipelineEntryPoint;
 use dozer_core::node::SinkFactory;
 use dozer_core::DEFAULT_PORT_HANDLE;
-use dozer_ingestion::connectors::{get_connector, get_connector_info_table};
+use dozer_ingestion::{get_connector, get_connector_info_table};
 use dozer_sql::builder::statement_to_pipeline;
 use dozer_sql::builder::{OutputNodeInfo, QueryContext};
 use dozer_tracing::LabelsAndProgress;
@@ -26,6 +26,7 @@ use tokio::sync::Mutex;
 use crate::pipeline::dummy_sink::DummySinkFactory;
 use crate::pipeline::LogSinkFactory;
 
+use super::connector_source::ConnectorSourceFactoryError;
 use super::source_builder::SourceBuilder;
 use crate::errors::OrchestrationError;
 use dozer_types::log::info;
@@ -91,13 +92,17 @@ impl<'a> PipelineBuilder<'a> {
 
         let mut connector_map = HashMap::new();
         for connection in self.connections {
-            let connector = get_connector(connection.clone())?;
+            let connector = get_connector(connection.clone())
+                .map_err(|e| ConnectorSourceFactoryError::Connector(e.into()))?;
 
             if let Some(info_table) = get_connector_info_table(connection) {
                 info!("[{}] Connection parameters\n{info_table}", connection.name);
             }
 
-            let connector_tables = connector.list_tables().await?;
+            let connector_tables = connector
+                .list_tables()
+                .await
+                .map_err(ConnectorSourceFactoryError::Connector)?;
 
             // override source name if specified
             let connector_tables: Vec<Source> = connector_tables

dozer-cli/src/pipeline/connector_source.rs

@@ -3,11 +3,10 @@ use dozer_core::channels::SourceChannelForwarder;
 use dozer_core::node::{
     OutputPortDef, OutputPortType, PortHandle, Source, SourceFactory, SourceState,
 };
-use dozer_ingestion::connectors::{
+use dozer_ingestion::{
     get_connector, CdcType, Connector, TableIdentifier, TableInfo, TableToIngest,
 };
-use dozer_ingestion::errors::ConnectorError;
-use dozer_ingestion::ingestion::{IngestionConfig, Ingestor};
+use dozer_ingestion::{IngestionConfig, Ingestor};
 
 use dozer_tracing::LabelsAndProgress;
 use dozer_types::errors::internal::BoxedError;
@@ -39,7 +38,7 @@ struct Table {
 #[derive(Debug, Error)]
 pub enum ConnectorSourceFactoryError {
     #[error("Connector error: {0}")]
-    Connector(#[from] ConnectorError),
+    Connector(#[source] BoxedError),
     #[error("Port not found for source: {0}")]
     PortNotFoundInSource(PortHandle),
     #[error("Schema not initialized")]
@@ -75,14 +74,18 @@ impl ConnectorSourceFactory {
     ) -> Result<Self, ConnectorSourceFactoryError> {
         let connection_name = connection.name.clone();
 
-        let connector = get_connector(connection)?;
+        let connector = get_connector(connection)
+            .map_err(|e| ConnectorSourceFactoryError::Connector(e.into()))?;
 
         // Fill column names if not provided.
         let table_identifiers = table_and_ports
             .iter()
             .map(|(table, _)| TableIdentifier::new(table.schema.clone(), table.name.clone()))
             .collect();
-        let all_columns = connector.list_columns(table_identifiers).await?;
+        let all_columns = connector
+            .list_columns(table_identifiers)
+            .await
+            .map_err(ConnectorSourceFactoryError::Connector)?;
         for ((table, _), columns) in table_and_ports.iter_mut().zip(all_columns) {
             if table.column_names.is_empty() {
                 table.column_names = columns.column_names;
@@ -93,13 +96,16 @@ impl ConnectorSourceFactory {
             .iter()
             .map(|(table, _)| table.clone())
             .collect();
-        let source_schemas = connector.get_schemas(&tables).await?;
+        let source_schemas = connector
+            .get_schemas(&tables)
+            .await
+            .map_err(ConnectorSourceFactoryError::Connector)?;
 
         let mut tables = vec![];
         for ((table, port), source_schema) in table_and_ports.into_iter().zip(source_schemas) {
             let name = table.name;
             let columns = table.column_names;
-            let source_schema = source_schema?;
+            let source_schema = source_schema.map_err(ConnectorSourceFactoryError::Connector)?;
             let schema = source_schema.schema;
             let cdc_type = source_schema.cdc_type;
 
@@ -280,9 +286,6 @@ impl Source for ConnectorSource {
                             .await;
                     match result {
                         Ok(Ok(_)) => {}
-                        // If we get a channel error, it means the source sender thread has quit.
-                        // Any error handling is done in that thread.
-                        Ok(Err(ConnectorError::IngestorError)) => {}
                         Ok(Err(e)) => std::panic::panic_any(e),
                         // Aborted means we are shutting down
                         Err(Aborted) => (),

dozer-cli/src/pipeline/source_builder.rs

@@ -2,7 +2,7 @@ use crate::pipeline::connector_source::ConnectorSourceFactory;
 use crate::OrchestrationError;
 use dozer_api::shutdown::ShutdownReceiver;
 use dozer_core::appsource::{AppSourceManager, AppSourceMappings};
-use dozer_ingestion::connectors::TableInfo;
+use dozer_ingestion::TableInfo;
 
 use dozer_tracing::LabelsAndProgress;
 use dozer_types::models::connection::Connection;

dozer-cli/src/simple/orchestrator.rs

@@ -1,6 +1,7 @@
 use super::executor::{run_dag_executor, Executor};
 use super::Contract;
 use crate::errors::OrchestrationError;
+use crate::pipeline::connector_source::ConnectorSourceFactoryError;
 use crate::pipeline::PipelineBuilder;
 use crate::simple::build;
 use crate::simple::helper::validate_config;
@@ -30,7 +31,7 @@ use crate::console_helper::GREEN;
 use crate::console_helper::PURPLE;
 use crate::console_helper::RED;
 use dozer_core::errors::ExecutionError;
-use dozer_ingestion::connectors::{get_connector, SourceSchema, TableInfo};
+use dozer_ingestion::{get_connector, SourceSchema, TableInfo};
 use dozer_sql::builder::statement_to_pipeline;
 use dozer_sql::errors::PipelineError;
 use dozer_types::log::info;
@@ -288,8 +289,12 @@ impl SimpleOrchestrator {
     ) -> Result<HashMap<String, (Vec<TableInfo>, Vec<SourceSchema>)>, OrchestrationError> {
         let mut schema_map = HashMap::new();
         for connection in &self.config.connections {
-            let connector = get_connector(connection.clone())?;
-            let schema_tuples = connector.list_all_schemas().await?;
+            let connector = get_connector(connection.clone())
+                .map_err(|e| ConnectorSourceFactoryError::Connector(e.into()))?;
+            let schema_tuples = connector
+                .list_all_schemas()
+                .await
+                .map_err(ConnectorSourceFactoryError::Connector)?;
             schema_map.insert(connection.name.clone(), schema_tuples);
         }