fix(batch): fix a bug of exchange and project. (#2545)

fix(batch): fix the bug of exchange.
risingwavelabs · May 16, 2022 · 523b669 · 523b669
1 parent f880b26
commit 523b669
Show file tree

Hide file tree

Showing 3 changed files with 7 additions and 2 deletions.
diff --git a/e2e_test/v2/batch/tpch.slt b/e2e_test/v2/batch/tpch.slt
@@ -32,6 +32,6 @@ include ../../batch/tpch/q17.slt.part
 include ../../batch/tpch/q18.slt.part
 include ../../batch/tpch/q19.slt.part
 include ../../batch/tpch/q20.slt.part
-# include ../../batch/tpch/q21.slt.part
+include ../../batch/tpch/q21.slt.part
 
 include ../../tpch/drop_tables.slt.part
diff --git a/src/batch/src/executor2/project.rs b/src/batch/src/executor2/project.rs
@@ -56,7 +56,11 @@ impl ProjectExecutor2 {
                 .iter_mut()
                 .map(|expr| expr.eval(&data_chunk).map(Column::new))
                 .collect::<Result<Vec<_>>>()?;
-            let ret = DataChunk::builder().columns(arrays).build();
+            let ret = if arrays.is_empty() {
+                DataChunk::new_dummy(data_chunk.cardinality())
+            } else {
+                DataChunk::builder().columns(arrays).build()
+            };
             yield ret
         }
     }

diff --git a/src/batch/src/task/hash_shuffle_channel.rs b/src/batch/src/task/hash_shuffle_channel.rs
@@ -101,6 +101,7 @@ impl ChanSender for HashShuffleSender {
 
 impl HashShuffleSender {
     async fn send_chunk(&mut self, chunk: DataChunk) -> Result<()> {
+        let chunk = chunk.compact()?;
         let hash_values = generate_hash_values(&chunk, &self.hash_info)?;
         let new_data_chunks = generate_new_data_chunks(&chunk, &self.hash_info, &hash_values)?;