Merge pull request #9824 from RinChanNOWWW/parquet-topk-opt

feat(storage): `ParquetTable` support topk optimization.

src/query/catalog/src/plan/projection.rs

@@ -74,6 +74,34 @@ impl Projection {
         };
         Ok(column_nodes)
     }
+
+    pub fn add_col(&mut self, col: usize) {
+        match self {
+            Projection::Columns(indices) => {
+                if indices.contains(&col) {
+                    return;
+                }
+                indices.push(col);
+                indices.sort();
+            }
+            Projection::InnerColumns(path_indices) => {
+                path_indices.entry(col).or_insert(vec![col]);
+            }
+        }
+    }
+
+    pub fn remove_col(&mut self, col: usize) {
+        match self {
+            Projection::Columns(indices) => {
+                if let Some(pos) = indices.iter().position(|x| *x == col) {
+                    indices.remove(pos);
+                }
+            }
+            Projection::InnerColumns(path_indices) => {
+                path_indices.remove(&col);
+            }
+        }
+    }
 }
 
 impl core::fmt::Debug for Projection {

src/query/catalog/src/plan/pushdown.rs

@@ -72,6 +72,7 @@ impl PushDownInfo {
             }
 
             if let RemoteExpr::<String>::ColumnRef { id, .. } = &order.0 {
+                // TODO: support sub column of nested type.
                 let field = schema.field_with_name(id).unwrap();
                 let data_type: DataType = field.data_type().into();
                 if !support(&data_type) {

src/query/expression/src/kernels/topk.rs

@@ -25,6 +25,7 @@ use crate::with_number_mapped_type;
 use crate::Column;
 use crate::Scalar;
 
+#[derive(Clone)]
 pub struct TopKSorter {
     data: Vec<Scalar>,
     limit: usize,
@@ -152,6 +153,12 @@ impl TopKSorter {
 fn make_heap<T, F>(v: &mut [T], is_less: &mut F)
 where F: FnMut(&T, &T) -> bool {
     let len = v.len();
+
+    if len < 2 {
+        // no need to adjust heap
+        return;
+    }
+
     let mut parent = (len - 2) / 2;
 
     loop {

src/query/storages/parquet/Cargo.toml

@@ -21,11 +21,15 @@ common-functions = { path = "../../functions" }
 common-meta-app = { path = "../../../meta/app" }
 common-meta-types = { path = "../../../meta/types" }
 common-pipeline-core = { path = "../../pipeline/core" }
+common-pipeline-sources = { path = "../../pipeline/sources" }
 common-storage = { path = "../../../common/storage" }
+
+storages-common-index = { path = "../common/index" }
 storages-common-pruner = { path = "../common/pruner" }
 storages-common-table-meta = { path = "../common/table-meta" }
 
 async-trait = { version = "0.1.57", package = "async-trait-fn" }
+backon = "0.2"
 chrono = { workspace = true }
 futures = "0.3.24"
 opendal = { workspace = true }