fix(query): parquet don't need sort by meta infos (#10982)

* fix(query): parquet don't need sort by meta infos

* fix(query): parquet don't need sort by meta infos

Author: sundy-li

src/query/storages/parquet/src/deserialize_transform.rs

@@ -77,8 +77,6 @@ pub struct ParquetDeserializeTransform {
 
     // Used for remain reading
     remain_reader: Arc<ParquetReader>,
-    // Used for top k optimization
-    top_k_finished: bool,
 }
 
 impl ParquetDeserializeTransform {
@@ -108,8 +106,6 @@ impl ParquetDeserializeTransform {
 
                 prewhere_info,
                 remain_reader,
-
-                top_k_finished: false,
             },
         )))
     }
@@ -127,26 +123,6 @@ impl ParquetDeserializeTransform {
         self.output_data = Some(data_block);
         Ok(())
     }
-
-    /// check topk should return finished or not
-    fn check_topn(&mut self) {
-        if let Some(ParquetPrewhereInfo {
-            top_k: Some((_, sorter)),
-            ..
-        }) = &mut self.prewhere_info.as_mut()
-        {
-            if let Some(next_part) = self.parts.front() {
-                let next_part = next_part
-                    .as_any()
-                    .downcast_ref::<ParquetRowGroupPart>()
-                    .unwrap();
-
-                if let Some(sort_min_max) = &next_part.sort_min_max {
-                    self.top_k_finished = sorter.never_match(sort_min_max);
-                }
-            }
-        }
-    }
 }
 
 impl Processor for ParquetDeserializeTransform {
@@ -181,26 +157,12 @@ impl Processor for ParquetDeserializeTransform {
             return Ok(Event::Sync);
         }
 
-        if self.top_k_finished {
-            self.input.finish();
-            self.output.finish();
-            return Ok(Event::Finished);
-        }
-
         if self.input.has_data() {
             let mut data_block = self.input.pull_data().unwrap()?;
             let source_meta = data_block.take_meta().unwrap();
             let source_meta = ParquetSourceMeta::downcast_from(source_meta).unwrap();
 
             self.parts = VecDeque::from(source_meta.parts);
-
-            self.check_topn();
-            if self.top_k_finished {
-                self.input.finish();
-                self.output.finish();
-                return Ok(Event::Finished);
-            }
-
             self.data_readers = VecDeque::from(source_meta.readers);
             return Ok(Event::Sync);
         }

src/query/storages/parquet/src/pruning.rs

@@ -121,8 +121,6 @@ impl PartitionPruner {
         // 2. Use file meta to prune row groups or pages.
 
         // If one row group does not have stats, we cannot use the stats for topk optimization.
-        let mut all_have_minmax = true;
-
         for (file_id, file_meta) in file_metas.iter().enumerate() {
             partitions_total += file_meta.row_groups.len();
             let mut row_group_pruned = vec![false; file_meta.row_groups.len()];
@@ -159,9 +157,6 @@ impl PartitionPruner {
                 None
             };
 
-            // If one row group does not have stats, we cannot use the stats for topk optimization.
-            all_have_minmax &= row_group_stats.is_some();
-
             for (rg_idx, rg) in file_meta.row_groups.iter().enumerate() {
                 if row_group_pruned[rg_idx] {
                     continue;
@@ -221,45 +216,7 @@ impl PartitionPruner {
             }
         }
 
-        // 3. Check if can conduct topk push down optimization.
-        // Only all row groups have min/max stats can we use topk optimization.
-        // If we can use topk optimization, we should use `PartitionsShuffleKind::Seq`.
-        let partition_kind = if let (Some((top_k, _)), true) = (top_k, all_have_minmax) {
-            partitions.sort_by(|a, b| {
-                let (a_min, a_max) = a
-                    .column_metas
-                    .get(&(top_k.column_id as usize))
-                    .unwrap()
-                    .min_max
-                    .as_ref()
-                    .unwrap();
-                let (b_min, b_max) = b
-                    .column_metas
-                    .get(&(top_k.column_id as usize))
-                    .unwrap()
-                    .min_max
-                    .as_ref()
-                    .unwrap();
-
-                if top_k.asc {
-                    (a_min.as_ref(), a_max.as_ref()).cmp(&(b_min.as_ref(), b_max.as_ref()))
-                } else {
-                    (b_max.as_ref(), b_min.as_ref()).cmp(&(a_max.as_ref(), a_min.as_ref()))
-                }
-            });
-            for part in partitions.iter_mut() {
-                part.sort_min_max = part
-                    .column_metas
-                    .get(&(top_k.column_id as usize))
-                    .unwrap()
-                    .min_max
-                    .clone();
-            }
-            PartitionsShuffleKind::Seq
-        } else {
-            PartitionsShuffleKind::Mod
-        };
-
+        let partition_kind = PartitionsShuffleKind::Mod;
         let partitions = partitions
             .into_iter()
             .map(|p| p.convert_to_part_info())