fix: fix empty ops case in iceberg sink (#12811)

Co-authored-by: ZENOTME <[email protected]>
risingwavelabs · Oct 12, 2023 · c0f9b80 · c0f9b80
1 parent a36d3c0
commit c0f9b80
Showing 1 changed file with 4 additions and 0 deletions.
diff --git a/src/connector/src/sink/iceberg.rs b/src/connector/src/sink/iceberg.rs
@@ -544,6 +544,7 @@ impl UpsertWriter {
     }
 
     fn partition_ops(ops: &[Op]) -> Vec<(usize, usize)> {
+        assert!(!ops.is_empty());
         let mut res = vec![];
         let mut start = 0;
         let mut prev_op = ops[0];
@@ -560,6 +561,9 @@ impl UpsertWriter {
 
     pub async fn write(&mut self, chunk: StreamChunk) -> Result<()> {
         let (chunk, ops) = chunk.compact().into_parts();
+        if ops.len() == 0 {
+            return Ok(());
+        }
         let chunk = to_record_batch_with_schema(self.schema.clone(), &chunk.compact())
             .map_err(|err| SinkError::Iceberg(anyhow!(err)))?;
         let ranges = Self::partition_ops(&ops);