feat: BigQuery sink

dozer-cli/src/pipeline/builder.rs

@@ -10,12 +10,13 @@ use dozer_core::app::PipelineEntryPoint;
 use dozer_core::node::SinkFactory;
 use dozer_core::DEFAULT_PORT_HANDLE;
 use dozer_ingestion::{get_connector, get_connector_info_table};
+use dozer_sinks::bigquery::BigQuerySinkFactory;
 use dozer_sql::builder::statement_to_pipeline;
 use dozer_sql::builder::{OutputNodeInfo, QueryContext};
 use dozer_tracing::LabelsAndProgress;
 use dozer_types::log::debug;
 use dozer_types::models::connection::Connection;
-use dozer_types::models::endpoint::AerospikeSinkConfig;
+use dozer_types::models::endpoint::{AerospikeSinkConfig, BigQuery};
 use dozer_types::models::flags::Flags;
 use dozer_types::models::source::Source;
 use dozer_types::models::udf_config::UdfConfig;
@@ -60,6 +61,7 @@ pub enum EndpointLogKind {
     Api { log: Arc<Mutex<Log>> },
     Dummy,
     Aerospike { config: AerospikeSinkConfig },
+    BigQuery(BigQuery),
 }
 
 pub struct PipelineBuilder<'a> {
@@ -290,6 +292,7 @@ impl<'a> PipelineBuilder<'a> {
                 EndpointLogKind::Aerospike { config } => {
                     Box::new(AerospikeSinkFactory::new(config))
                 }
+                EndpointLogKind::BigQuery(config) => Box::new(BigQuerySinkFactory::new(config)),
             };
 
             match table_info {

dozer-cli/src/simple/executor.rs

@@ -5,7 +5,7 @@ use dozer_cache::dozer_log::home_dir::{BuildPath, HomeDir};
 use dozer_cache::dozer_log::replication::Log;
 use dozer_core::checkpoint::{CheckpointOptions, OptionCheckpoint};
 use dozer_tracing::LabelsAndProgress;
-use dozer_types::models::endpoint::{AerospikeSinkConfig, Endpoint, EndpointKind};
+use dozer_types::models::endpoint::{AerospikeSinkConfig, BigQuery, Endpoint, EndpointKind};
 use dozer_types::models::flags::Flags;
 use tokio::runtime::Runtime;
 use tokio::sync::Mutex;
@@ -45,6 +45,7 @@ enum ExecutorEndpointKind {
     Api { log_endpoint: LogEndpoint },
     Dummy,
     Aerospike { config: AerospikeSinkConfig },
+    BigQuery(BigQuery),
 }
 
 impl<'a> Executor<'a> {
@@ -89,6 +90,7 @@ impl<'a> Executor<'a> {
                 EndpointKind::Aerospike(config) => ExecutorEndpointKind::Aerospike {
                     config: config.clone(),
                 },
+                EndpointKind::BigQuery(config) => ExecutorEndpointKind::BigQuery(config.clone()),
             };
 
             executor_endpoints.push(ExecutorEndpoint {
@@ -147,6 +149,7 @@ impl<'a> Executor<'a> {
                         ExecutorEndpointKind::Aerospike { config } => {
                             EndpointLogKind::Aerospike { config }
                         }
+                        ExecutorEndpointKind::BigQuery(config) => EndpointLogKind::BigQuery(config),
                     };
                     EndpointLog {
                         table_name: endpoint.table_name,

dozer-sinks/Cargo.toml

@@ -7,6 +7,7 @@ edition = "2021"
 
 [dependencies]
 dozer-sinks-snowflake = { path = "./snowflake", optional = true }
+dozer-sinks-bigquery = { path = "./bigquery" }
 
 [features]
 snowflake = ["dep:dozer-sinks-snowflake"]

dozer-sinks/bigquery/Cargo.toml

@@ -0,0 +1,22 @@
+[package]
+name = "dozer-sinks-bigquery"
+version = "0.1.0"
+edition = "2021"
+
+# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
+
+[dependencies]
+dozer-api = { path = "../../dozer-api" }
+dozer-types = { path = "../../dozer-types" }
+dozer-log = { path = "../../dozer-log" }
+dozer-core = { path = "../../dozer-core" }
+dozer-recordstore = { path = "../../dozer-recordstore" }
+itertools = "0.10.5"
+futures-util = "0.3.28"
+chrono = "0.4.31"
+gcp-bigquery-client = "0.18.0"
+object_store = { version = "0.9.0", features = ["gcp"] }
+parquet = "48.0.1"
+yup-oauth2 = "8.3.2"
+bytes = "1.5.0"
+html-escape = "0.2.13"