Implement MERGE INTO (#147)

* # This is a combination of 14 commits.
# This is the 1st commit message:

Upload CSV to table

# This is the commit message #2:

Upload CSV to table

# This is the commit message #3:

Fix linter

# This is the commit message #4:

Remove unused error

# This is the commit message #5:

Fix clippy

# This is the commit message #6:

Fix clippy

# This is the commit message #7:

Fix linter

# This is the commit message #8:

Fix clippy

# This is the commit message #9:

Fix clippy

# This is the commit message #10:

Fix clippy

# This is the commit message #11:

Fix fmt

# This is the commit message #12:

Add pre-commit

# This is the commit message #13:

Parse merge into

# This is the commit message #14:

Parse merge into

* Merge

* Add full message text into the error

* Fix

* Update MERGE to insert only not matched records

* Update deps

* Update MERGE to insert only not matched records

Author: osipovartem

.pre-commit-config.yaml

@@ -0,0 +1,10 @@
+repos:
+  - repo: https://github.com/doublify/pre-commit-rust
+    rev: v1.0
+    hooks:
+      - id: cargo-check
+        args: [ "--workspace" ]
+      - id: fmt
+        args: [ "--", "--check" ]
+      - id: clippy
+        args: [ "--", "-D", "warnings" ]

Cargo.toml

@@ -35,7 +35,7 @@ datafusion = { version = "43" }
 snafu = { version = "0.8.5", features = ["futures"] }
 
 [patch.crates-io]
-datafusion = { git="https://github.com/Embucket/datafusion.git", rev = "bc5a978a4102391b2b7427dfdf94dd4e2667be49" }
+datafusion = { git="https://github.com/Embucket/datafusion.git", rev = "cc37c5920463b0cb0b224fc7f567fd4ae0368ffe" }
 
 
 [workspace.lints.clippy]

crates/control_plane/Cargo.toml

@@ -18,13 +18,13 @@ flatbuffers = { version = "24.3.25" }
 #iceberg-rest-catalog = { git = "https://github.com/JanKaul/iceberg-rust.git", rev = "836f11f" }
 #datafusion_iceberg = { git = "https://github.com/JanKaul/iceberg-rust.git", rev = "836f11f" }
 
-datafusion = { git="https://github.com/Embucket/datafusion.git", rev = "bc5a978a4102391b2b7427dfdf94dd4e2667be49" }
-datafusion-common = { git="https://github.com/Embucket/datafusion.git", rev = "bc5a978a4102391b2b7427dfdf94dd4e2667be49" }
-datafusion-expr = { git="https://github.com/Embucket/datafusion.git", rev = "bc5a978a4102391b2b7427dfdf94dd4e2667be49" }
+datafusion = { git="https://github.com/Embucket/datafusion.git", rev = "cc37c5920463b0cb0b224fc7f567fd4ae0368ffe" }
+datafusion-common = { git="https://github.com/Embucket/datafusion.git", rev = "cc37c5920463b0cb0b224fc7f567fd4ae0368ffe" }
+datafusion-expr = { git="https://github.com/Embucket/datafusion.git", rev = "cc37c5920463b0cb0b224fc7f567fd4ae0368ffe" }
 
-iceberg-rust = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "5d4211521085722de35b14c444da087c52309771" }
-iceberg-rest-catalog = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "5d4211521085722de35b14c444da087c52309771" }
-datafusion_iceberg = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "5d4211521085722de35b14c444da087c52309771" }
+iceberg-rust = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "68577441273eda894f1eb6d87b1c3e87dee0fdf6" }
+iceberg-rest-catalog = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "68577441273eda894f1eb6d87b1c3e87dee0fdf6" }
+datafusion_iceberg = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "68577441273eda894f1eb6d87b1c3e87dee0fdf6" }
 
 arrow = { version = "53" }
 arrow-json = { version = "53" }

crates/nexus/src/http/dbt/error.rs

@@ -57,9 +57,24 @@ impl IntoResponse for DbtError {
             Self::NotImplemented => http::StatusCode::NOT_IMPLEMENTED,
         };
 
+        let message = match &self {
+            Self::GZipDecompress { source } => format!("failed to decompress GZip body: {source}"),
+            Self::LoginRequestParse { source } => {
+                format!("failed to parse login request: {source}")
+            }
+            Self::QueryBodyParse { source } => format!("failed to parse query body: {source}"),
+            Self::InvalidWarehouseIdFormat { source } => format!("invalid warehouse_id: {source}"),
+            Self::ControlService { source } => source.to_string(),
+            Self::RowParse { source } => format!("failed to parse row JSON: {source}"),
+            Self::MissingAuthToken | Self::MissingDbtSession | Self::InvalidAuthData => {
+                "session error".to_string()
+            }
+            Self::NotImplemented => "feature not implemented".to_string(),
+        };
+
         let body = Json(JsonResponse {
             success: false,
-            message: Some(self.to_string()),
+            message: Some(message),
             data: None,
             code: Some(status_code.as_u16().to_string()),
         });

crates/nexus/src/http/dbt/handlers.rs

@@ -32,7 +32,7 @@ pub async fn login(
 
     //println!("Received login request: {:?}", query);
     //println!("Body data parameters: {:?}", body_json);
-    let token = uuid::Uuid::new_v4().to_string();
+    let token = Uuid::new_v4().to_string();
 
     let warehouses = state
         .control_svc

crates/nexus/src/main.rs

@@ -199,7 +199,7 @@ async fn buffer_and_print<B>(
     body: B,
 ) -> Result<Bytes, (StatusCode, String)>
 where
-    B: axum::body::HttpBody<Data = Bytes>,
+    B: axum::body::HttpBody<Data = Bytes> + Send,
     B::Error: std::fmt::Display,
 {
     let bytes = match body.collect().await {

crates/runtime/Cargo.toml

@@ -14,13 +14,13 @@ serde = { workspace = true }
 serde_json = { workspace = true }
 object_store = { workspace = true }
 
-datafusion = { git="https://github.com/Embucket/datafusion.git", rev = "bc5a978a4102391b2b7427dfdf94dd4e2667be49" }
-datafusion-common = { git="https://github.com/Embucket/datafusion.git", rev = "bc5a978a4102391b2b7427dfdf94dd4e2667be49" }
-datafusion-expr = { git="https://github.com/Embucket/datafusion.git", rev = "bc5a978a4102391b2b7427dfdf94dd4e2667be49" }
+datafusion = { git="https://github.com/Embucket/datafusion.git", rev = "cc37c5920463b0cb0b224fc7f567fd4ae0368ffe" }
+datafusion-common = { git="https://github.com/Embucket/datafusion.git", rev = "cc37c5920463b0cb0b224fc7f567fd4ae0368ffe" }
+datafusion-expr = { git="https://github.com/Embucket/datafusion.git", rev = "cc37c5920463b0cb0b224fc7f567fd4ae0368ffe" }
 
-iceberg-rust = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "5d4211521085722de35b14c444da087c52309771" }
-iceberg-rest-catalog = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "5d4211521085722de35b14c444da087c52309771" }
-datafusion_iceberg = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "5d4211521085722de35b14c444da087c52309771" }
+iceberg-rust = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "68577441273eda894f1eb6d87b1c3e87dee0fdf6" }
+iceberg-rest-catalog = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "68577441273eda894f1eb6d87b1c3e87dee0fdf6" }
+datafusion_iceberg = { git = "https://github.com/Embucket/iceberg-rust.git", rev = "68577441273eda894f1eb6d87b1c3e87dee0fdf6" }
 
 arrow = { version = "53" }
 arrow-json = { version = "53" }