fix: make snowflake connector async

dozer-ingestion/src/connectors/delta_lake/connector.rs

@@ -2,8 +2,8 @@ use crate::connectors::delta_lake::reader::DeltaLakeReader;
 use crate::connectors::delta_lake::schema_helper::SchemaHelper;
 use crate::connectors::delta_lake::ConnectorResult;
 use crate::connectors::{
-    table_name, ConnectorMeta, ConnectorStart, ListOrFilterColumns, SourceSchemaResult,
-    TableIdentifier, TableInfo, TableToIngest,
+    table_name, Connector, ListOrFilterColumns, SourceSchemaResult, TableIdentifier, TableInfo,
+    TableToIngest,
 };
 use crate::errors::ConnectorError;
 use crate::ingestion::Ingestor;
@@ -22,7 +22,7 @@ impl DeltaLakeConnector {
 }
 
 #[async_trait]
-impl ConnectorMeta for DeltaLakeConnector {
+impl Connector for DeltaLakeConnector {
     fn types_mapping() -> Vec<(String, Option<dozer_types::types::FieldType>)>
     where
         Self: Sized,
@@ -107,10 +107,7 @@ impl ConnectorMeta for DeltaLakeConnector {
         let schema_helper = SchemaHelper::new(self.config.clone());
         schema_helper.get_schemas(&table_infos).await
     }
-}
 
-#[async_trait(?Send)]
-impl ConnectorStart for DeltaLakeConnector {
     async fn start(&self, ingestor: &Ingestor, tables: Vec<TableToIngest>) -> ConnectorResult<()> {
         let reader = DeltaLakeReader::new(self.config.clone());
         reader.read(&tables, ingestor).await

dozer-ingestion/src/connectors/delta_lake/test/deltalake_test.rs

@@ -1,5 +1,5 @@
 use crate::connectors::delta_lake::DeltaLakeConnector;
-use crate::connectors::ConnectorMeta;
+use crate::connectors::Connector;
 use crate::test_util::create_runtime_and_spawn_connector_all_tables;
 use dozer_types::ingestion_types::IngestionMessage;
 use dozer_types::ingestion_types::{DeltaLakeConfig, DeltaTable};

dozer-ingestion/src/connectors/dozer/connector.rs

@@ -25,8 +25,8 @@ use tonic::{async_trait, transport::Channel};
 
 use crate::{
     connectors::{
-        warn_dropped_primary_index, CdcType, ConnectorMeta, ConnectorStart, SourceSchema,
-        SourceSchemaResult, TableIdentifier, TableInfo, TableToIngest,
+        warn_dropped_primary_index, CdcType, Connector, SourceSchema, SourceSchemaResult,
+        TableIdentifier, TableInfo, TableToIngest,
     },
     errors::{ConnectorError, NestedDozerConnectorError},
     ingestion::Ingestor,
@@ -38,7 +38,7 @@ pub struct NestedDozerConnector {
 }
 
 #[async_trait]
-impl ConnectorMeta for NestedDozerConnector {
+impl Connector for NestedDozerConnector {
     fn types_mapping() -> Vec<(String, Option<dozer_types::types::FieldType>)>
     where
         Self: Sized,
@@ -123,10 +123,7 @@ impl ConnectorMeta for NestedDozerConnector {
 
         Ok(schemas)
     }
-}
 
-#[async_trait(?Send)]
-impl ConnectorStart for NestedDozerConnector {
     async fn start(
         &self,
         ingestor: &Ingestor,

dozer-ingestion/src/connectors/ethereum/log/connector.rs

@@ -2,8 +2,8 @@ use std::collections::HashMap;
 use std::{str::FromStr, sync::Arc};
 
 use crate::connectors::{
-    table_name, CdcType, ConnectorMeta, ConnectorStart, SourceSchema, SourceSchemaResult,
-    TableIdentifier, TableToIngest,
+    table_name, CdcType, Connector, SourceSchema, SourceSchemaResult, TableIdentifier,
+    TableToIngest,
 };
 use crate::ingestion::Ingestor;
 use crate::{connectors::TableInfo, errors::ConnectorError};
@@ -123,7 +123,7 @@ impl EthLogConnector {
 }
 
 #[async_trait]
-impl ConnectorMeta for EthLogConnector {
+impl Connector for EthLogConnector {
     fn types_mapping() -> Vec<(String, Option<dozer_types::types::FieldType>)>
     where
         Self: Sized,
@@ -232,10 +232,7 @@ impl ConnectorMeta for EthLogConnector {
 
         Ok(result)
     }
-}
 
-#[async_trait(?Send)]
-impl ConnectorStart for EthLogConnector {
     async fn start(
         &self,
         ingestor: &Ingestor,

dozer-ingestion/src/connectors/ethereum/log/tests/helper.rs

@@ -3,7 +3,7 @@ use std::{sync::Arc, time::Duration};
 use crate::{
     connectors::{
         ethereum::{helper, EthLogConnector},
-        ConnectorMeta, TableInfo,
+        Connector, TableInfo,
     },
     errors::ConnectorError,
     test_util::spawn_connector,

dozer-ingestion/src/connectors/ethereum/trace/connector.rs

@@ -1,8 +1,8 @@
 use super::super::helper as conn_helper;
 use super::helper::{self, get_block_traces, map_trace_to_ops};
 use crate::connectors::{
-    table_name, CdcType, ConnectorMeta, ConnectorStart, SourceSchema, SourceSchemaResult,
-    TableIdentifier, TableToIngest,
+    table_name, CdcType, Connector, SourceSchema, SourceSchemaResult, TableIdentifier,
+    TableToIngest,
 };
 use crate::{connectors::TableInfo, errors::ConnectorError, ingestion::Ingestor};
 use dozer_types::ingestion_types::{EthTraceConfig, IngestionMessage};
@@ -25,7 +25,7 @@ impl EthTraceConnector {
 }
 
 #[async_trait]
-impl ConnectorMeta for EthTraceConnector {
+impl Connector for EthTraceConnector {
     fn types_mapping() -> Vec<(String, Option<dozer_types::types::FieldType>)>
     where
         Self: Sized,
@@ -91,10 +91,7 @@ impl ConnectorMeta for EthTraceConnector {
             CdcType::Nothing,
         ))])
     }
-}
 
-#[async_trait(?Send)]
-impl ConnectorStart for EthTraceConnector {
     async fn start(
         &self,
         ingestor: &Ingestor,

dozer-ingestion/src/connectors/ethereum/trace/tests.rs

@@ -9,7 +9,7 @@ use dozer_types::{
 use crate::{
     connectors::{
         ethereum::{helper, trace::helper::get_block_traces, EthTraceConnector},
-        ConnectorMeta,
+        Connector,
     },
     test_util::{create_test_runtime, spawn_connector},
 };

dozer-ingestion/src/connectors/grpc/connector.rs

@@ -4,8 +4,7 @@ use std::path::Path;
 use super::adapter::{GrpcIngestor, IngestAdapter};
 use super::ingest::IngestorServiceImpl;
 use crate::connectors::{
-    table_name, ConnectorMeta, ConnectorStart, SourceSchema, SourceSchemaResult, TableIdentifier,
-    TableToIngest,
+    table_name, Connector, SourceSchema, SourceSchemaResult, TableIdentifier, TableToIngest,
 };
 use crate::{connectors::TableInfo, errors::ConnectorError, ingestion::Ingestor};
 use dozer_types::grpc_types::ingest::ingest_service_server::IngestServiceServer;
@@ -119,7 +118,7 @@ impl<T: IngestAdapter> GrpcConnector<T> {
 }
 
 #[async_trait]
-impl<T> ConnectorMeta for GrpcConnector<T>
+impl<T> Connector for GrpcConnector<T>
 where
     T: IngestAdapter,
 {
@@ -217,10 +216,7 @@ where
 
         Ok(result)
     }
-}
 
-#[async_trait(?Send)]
-impl<T: IngestAdapter> ConnectorStart for GrpcConnector<T> {
     async fn start(
         &self,
         ingestor: &Ingestor,

dozer-ingestion/src/connectors/kafka/connector.rs

@@ -2,7 +2,7 @@ use rdkafka::consumer::BaseConsumer;
 use rdkafka::ClientConfig;
 
 use crate::connectors::{
-    ConnectorMeta, ConnectorStart, SourceSchema, SourceSchemaResult, TableIdentifier, TableToIngest,
+    Connector, SourceSchema, SourceSchemaResult, TableIdentifier, TableToIngest,
 };
 use crate::ingestion::Ingestor;
 use crate::{connectors::TableInfo, errors::ConnectorError};
@@ -42,7 +42,7 @@ impl KafkaConnector {
 }
 
 #[async_trait]
-impl ConnectorMeta for KafkaConnector {
+impl Connector for KafkaConnector {
     fn types_mapping() -> Vec<(String, Option<dozer_types::types::FieldType>)>
     where
         Self: Sized,
@@ -126,10 +126,7 @@ impl ConnectorMeta for KafkaConnector {
             .map(Ok)
             .collect())
     }
-}
 
-#[async_trait(?Send)]
-impl ConnectorStart for KafkaConnector {
     async fn start(
         &self,
         ingestor: &Ingestor,

dozer-ingestion/src/connectors/mod.rs

@@ -86,7 +86,7 @@ impl SourceSchema {
 pub type SourceSchemaResult = Result<SourceSchema, ConnectorError>;
 
 #[async_trait]
-pub trait ConnectorMeta {
+pub trait Connector: Send + Sync + Debug {
     /// Returns all the external types and their corresponding Dozer types.
     /// If the external type is not supported, None should be returned.
     fn types_mapping() -> Vec<(String, Option<FieldType>)>
@@ -132,12 +132,7 @@ pub trait ConnectorMeta {
             .collect::<Result<Vec<_>, _>>()?;
         Ok((table_infos, schemas))
     }
-}
 
-/// We split `Connector` trait into two because snowflake's (using odbc) `start` future is `!Send`.
-/// Once we switch to a better client, we should merge them.
-#[async_trait(?Send)]
-pub trait ConnectorStart {
     /// Starts outputting data from `tables` to `ingestor`. This method should never return unless there is an unrecoverable error.
     async fn start(
         &self,
@@ -146,10 +141,6 @@ pub trait ConnectorStart {
     ) -> Result<(), ConnectorError>;
 }
 
-pub trait Connector: ConnectorMeta + ConnectorStart + Send + Sync + Debug {}
-
-impl<T: ConnectorMeta + ConnectorStart + Send + Sync + Debug> Connector for T {}
-
 #[derive(Debug, Clone, PartialEq, Eq, Hash)]
 /// Unique identifier of a source table. A source table must have a `name`, optionally under a `schema` scope.
 pub struct TableIdentifier {

dozer-ingestion/src/connectors/mongodb/mod.rs

@@ -20,8 +20,7 @@ use dozer_types::{
 };
 
 use super::{
-    ConnectorMeta, ConnectorStart, SourceSchema, SourceSchemaResult, TableIdentifier, TableInfo,
-    TableToIngest,
+    Connector, SourceSchema, SourceSchemaResult, TableIdentifier, TableInfo, TableToIngest,
 };
 
 #[derive(Error, Debug)]
@@ -454,7 +453,7 @@ impl MongodbConnector {
 }
 
 #[async_trait]
-impl ConnectorMeta for MongodbConnector {
+impl Connector for MongodbConnector {
     async fn validate_connection(&self) -> Result<(), ConnectorError> {
         let client = self.client().await?;
         let server_info = self.identify_server(&client).await?;
@@ -588,10 +587,7 @@ impl ConnectorMeta for MongodbConnector {
         }
         Ok(())
     }
-}
 
-#[async_trait(?Send)]
-impl ConnectorStart for MongodbConnector {
     async fn start(
         &self,
         ingestor: &Ingestor,

dozer-ingestion/src/connectors/mysql/connector.rs

@@ -7,8 +7,8 @@ use super::{
 };
 use crate::{
     connectors::{
-        CdcType, ConnectorMeta, ConnectorStart, SourceSchema, SourceSchemaResult, TableIdentifier,
-        TableInfo, TableToIngest,
+        CdcType, Connector, SourceSchema, SourceSchemaResult, TableIdentifier, TableInfo,
+        TableToIngest,
     },
     errors::MySQLConnectorError,
 };
@@ -43,7 +43,7 @@ impl MySQLConnector {
 }
 
 #[async_trait]
-impl ConnectorMeta for MySQLConnector {
+impl Connector for MySQLConnector {
     fn types_mapping() -> Vec<(String, Option<dozer_types::types::FieldType>)>
     where
         Self: Sized,
@@ -186,10 +186,7 @@ impl ConnectorMeta for MySQLConnector {
 
         Ok(schemas)
     }
-}
 
-#[async_trait(?Send)]
-impl ConnectorStart for MySQLConnector {
     async fn start(
         &self,
         ingestor: &Ingestor,
@@ -405,7 +402,7 @@ mod tests {
                 connection::Conn,
                 tests::{create_test_table, mariadb_test_config, mysql_test_config, TestConfig},
             },
-            CdcType, ConnectorMeta, SourceSchema, TableIdentifier,
+            CdcType, Connector, SourceSchema, TableIdentifier,
         },
         ingestion::{IngestionIterator, Ingestor},
     };

dozer-ingestion/src/connectors/object_store/connector.rs

@@ -8,8 +8,7 @@ use tonic::async_trait;
 use crate::connectors::object_store::adapters::DozerObjectStore;
 use crate::connectors::object_store::schema_mapper;
 use crate::connectors::{
-    ConnectorMeta, ConnectorStart, ListOrFilterColumns, SourceSchemaResult, TableIdentifier,
-    TableInfo, TableToIngest,
+    Connector, ListOrFilterColumns, SourceSchemaResult, TableIdentifier, TableInfo, TableToIngest,
 };
 use crate::errors::{ConnectorError, ObjectStoreConnectorError};
 use crate::ingestion::Ingestor;
@@ -36,7 +35,7 @@ impl<T: DozerObjectStore + 'static> ObjectStoreConnector<T> {
 }
 
 #[async_trait]
-impl<T: DozerObjectStore> ConnectorMeta for ObjectStoreConnector<T> {
+impl<T: DozerObjectStore> Connector for ObjectStoreConnector<T> {
     fn types_mapping() -> Vec<(String, Option<dozer_types::types::FieldType>)>
     where
         Self: Sized,
@@ -99,10 +98,7 @@ impl<T: DozerObjectStore> ConnectorMeta for ObjectStoreConnector<T> {
             .collect::<Vec<_>>();
         schema_mapper::get_schema(&self.config, &list_or_filter_columns).await
     }
-}
 
-#[async_trait(?Send)]
-impl<T: DozerObjectStore> ConnectorStart for ObjectStoreConnector<T> {
     async fn start(&self, ingestor: &Ingestor, tables: Vec<TableToIngest>) -> ConnectorResult<()> {
         let (sender, mut receiver) =
             channel::<Result<Option<IngestionMessage>, ObjectStoreConnectorError>>(100); // todo: increase buffer siz

dozer-ingestion/src/connectors/object_store/tests/local_storage_tests.rs

@@ -1,5 +1,5 @@
 use crate::connectors::object_store::connector::ObjectStoreConnector;
-use crate::connectors::ConnectorMeta;
+use crate::connectors::Connector;
 use crate::test_util::create_runtime_and_spawn_connector_all_tables;
 use dozer_types::ingestion_types::IngestionMessage;
 use dozer_types::ingestion_types::LocalDetails;

dozer-ingestion/src/connectors/postgres/connector.rs

@@ -1,8 +1,7 @@
 use crate::connectors::postgres::connection::validator::validate_connection;
 use crate::connectors::postgres::iterator::PostgresIterator;
 use crate::connectors::{
-    ConnectorMeta, ConnectorStart, ListOrFilterColumns, SourceSchemaResult, TableIdentifier,
-    TableInfo, TableToIngest,
+    Connector, ListOrFilterColumns, SourceSchemaResult, TableIdentifier, TableInfo, TableToIngest,
 };
 use crate::errors::ConnectorError;
 use crate::ingestion::Ingestor;
@@ -82,7 +81,7 @@ impl PostgresConnector {
 }
 
 #[async_trait]
-impl ConnectorMeta for PostgresConnector {
+impl Connector for PostgresConnector {
     fn types_mapping() -> Vec<(String, Option<dozer_types::types::FieldType>)>
     where
         Self: Sized,
@@ -162,10 +161,7 @@ impl ConnectorMeta for PostgresConnector {
             .await
             .map_err(Into::into)
     }
-}
 
-#[async_trait(?Send)]
-impl ConnectorStart for PostgresConnector {
     async fn start(
         &self,
         ingestor: &Ingestor,