fix(query): fix WindowPartitionTopNExchange painc with an empty block (#17453)

* fix

* add top_n to format tree

* fix test

* skip empty data block

Author: forsaken628

src/query/pipeline/core/src/processors/shuffle_processor.rs

@@ -32,6 +32,7 @@ pub enum MultiwayStrategy {
 
 pub trait Exchange: Send + Sync + 'static {
     const NAME: &'static str;
+    const SKIP_EMPTY_DATA_BLOCK: bool = false;
     const STRATEGY: MultiwayStrategy = MultiwayStrategy::Random;
 
     fn partition(&self, data_block: DataBlock, n: usize) -> Result<Vec<DataBlock>>;
@@ -247,6 +248,10 @@ impl<T: Exchange> Processor for PartitionProcessor<T> {
 
     fn process(&mut self) -> Result<()> {
         if let Some(block) = self.input_data.take() {
+            if T::SKIP_EMPTY_DATA_BLOCK && block.is_empty() {
+                return Ok(());
+            }
+
             let partitioned = self.exchange.partition(block, self.outputs.len())?;
 
             for (index, block) in partitioned.into_iter().enumerate() {

src/query/service/src/pipelines/processors/transforms/window/partition/window_partition_partial_top_n_exchange.rs

@@ -67,6 +67,8 @@ impl WindowPartitionTopNExchange {
 
 impl Exchange for WindowPartitionTopNExchange {
     const NAME: &'static str = "WindowTopN";
+    const SKIP_EMPTY_DATA_BLOCK: bool = true;
+
     fn partition(&self, block: DataBlock, n: usize) -> Result<Vec<DataBlock>> {
         let partition_permutation = self.partition_permutation(&block);
 

src/query/sql/src/executor/format.rs

@@ -1262,6 +1262,10 @@ fn window_partition_to_format_tree(
         FormatTreeNode::new(format!("hash keys: [{partition_by}]")),
     ];
 
+    if let Some(top_n) = &plan.top_n {
+        children.push(FormatTreeNode::new(format!("top: {}", top_n.top)));
+    }
+
     if let Some(info) = &plan.stat_info {
         let items = plan_stats_info_to_format_tree(info);
         children.extend(items);

tests/sqllogictests/suites/mode/cluster/shuffle.test

@@ -50,6 +50,7 @@ Sort
             │       └── WindowPartition
             │           ├── output columns: [t2.number (#1)]
             │           ├── hash keys: [number]
+            │           ├── top: 1           
             │           ├── estimated rows: 14.00
             │           └── Exchange
             │               ├── output columns: [t2.number (#1)]

tests/sqllogictests/suites/mode/standalone/explain/window.test

@@ -109,6 +109,7 @@ Filter
     └── WindowPartition
         ├── output columns: [k (#4), v (#5)]
         ├── hash keys: [k]
+        ├── top: 1
         ├── estimated rows: 0.00
         └── UnionAll
             ├── output columns: [k (#4), v (#5)]
@@ -157,6 +158,7 @@ Filter
     └── WindowPartition
         ├── output columns: [k (#4), v (#5)]
         ├── hash keys: [v]
+        ├── top: 1
         ├── estimated rows: 0.00
         └── UnionAll
             ├── output columns: [k (#4), v (#5)]
@@ -793,6 +795,7 @@ AggregateFinal
             └── WindowPartition
                 ├── output columns: [numbers.number (#0), id (#1)]
                 ├── hash keys: [id]
+                ├── top: 1
                 ├── estimated rows: 1000.00
                 └── EvalScalar
                     ├── output columns: [numbers.number (#0), id (#1)]