fix: update icelake upsert query (#13232)

Co-authored-by: ZENOTME <[email protected]>
risingwavelabs · Nov 3, 2023 · fff5de4 · fff5de4
1 parent 624094c
commit fff5de4
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 3 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/Cargo.toml b/Cargo.toml
@@ -112,7 +112,7 @@ criterion = { version = "0.5", features = ["async_futures"] }
 tonic = { package = "madsim-tonic", version = "0.4.0" }
 tonic-build = { package = "madsim-tonic-build", version = "0.4.2" }
 prost = { version = "0.12" }
-icelake = { git = "https://github.com/icelake-io/icelake", rev = "186fde7663545d1d6a5856ce9fbbc541224eadfb" }
+icelake = { git = "https://github.com/icelake-io/icelake", rev = "5a5202d1f3502f0cf82041044b0427434da59adc" }
 arrow-array = "48"
 arrow-cast = "48"
 arrow-schema = "48"

diff --git a/src/connector/src/sink/iceberg.rs b/src/connector/src/sink/iceberg.rs
@@ -849,7 +849,10 @@ impl SinkCommitCoordinator for IcebergSinkCommitter {
             .iter()
             .map(|meta| WriteResult::try_from(meta, &self.partition_type))
             .collect::<Result<Vec<WriteResult>>>()?;
-
+        if write_results.is_empty() || write_results.iter().all(|r| r.data_files.is_empty()) {
+            tracing::debug!(?epoch, "no data to commit");
+            return Ok(());
+        }
         let mut txn = Transaction::new(&mut self.table);
         write_results.into_iter().for_each(|s| {
             txn.append_data_file(s.data_files);