fix(batch): avoid empty chunk in source executor (#13046)

risingwavelabs · Oct 30, 2023 · 0648396 · 0648396
1 parent 2f9eb97
commit 0648396
Show file tree

Hide file tree

Showing 2 changed files with 10 additions and 2 deletions.
diff --git a/src/batch/src/executor/source.rs b/src/batch/src/executor/source.rs
@@ -162,7 +162,10 @@ impl SourceExecutor {
         for chunk in stream {
             match chunk {
                 Ok(chunk) => {
-                    yield covert_stream_chunk_to_batch_chunk(chunk.chunk)?;
+                    let data_chunk = covert_stream_chunk_to_batch_chunk(chunk.chunk)?;
+                    if data_chunk.capacity() > 0 {
+                        yield data_chunk;
+                    }
                 }
                 Err(e) => {
                     return Err(e);

diff --git a/src/common/src/util/chunk_coalesce.rs b/src/common/src/util/chunk_coalesce.rs
@@ -285,7 +285,12 @@ impl SlicedDataChunk {
     }
 
     pub fn with_offset_checked(data_chunk: DataChunk, offset: usize) -> Self {
-        assert!(offset < data_chunk.capacity());
+        assert!(
+            offset < data_chunk.capacity(),
+            "offset {}, data_chunk capacity {}",
+            offset,
+            data_chunk.capacity()
+        );
         Self { data_chunk, offset }
     }