chore(query): fix topk in native (#17003)

* chore(query): fix topk in native

* chore(query): fix topk in native

Author: sundy-li

scripts/setup/rust-tools.txt

@@ -1,5 +1,5 @@
 [email protected]
 [email protected]
-[email protected].67
-taplo-cli@0.8.1
-typos-cli@1.17.2
+[email protected].85
+taplo-cli@0.9.3
+typos-cli@1.28.2

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

@@ -224,21 +224,17 @@ impl PushDownInfo {
                 if !TopK::support_type(data_type) {
                     return None;
                 }
-
                 let leaf_fields = schema.leaf_fields();
-                let (leaf_id, f) = leaf_fields
+                leaf_fields
                     .iter()
                     .enumerate()
                     .find(|&(_, p)| p.name() == id)
-                    .unwrap();
-
-                let top_k = TopK {
-                    limit: self.limit.unwrap(),
-                    field: f.clone(),
-                    asc: order.1,
-                    leaf_id,
-                };
-                Some(top_k)
+                    .map(|(leaf_id, f)| TopK {
+                        limit: self.limit.unwrap(),
+                        field: f.clone(),
+                        asc: order.1,
+                        leaf_id,
+                    })
             } else {
                 None
             }

src/query/sql/src/executor/physical_plans/physical_table_scan.rs

@@ -478,51 +478,47 @@ impl PhysicalPlanBuilder {
             })
             .transpose()?;
 
-        let order_by = scan
-            .order_by
-            .clone()
-            .map(|items| {
-                items
-                    .into_iter()
-                    .map(|item| {
-                        let metadata = self.metadata.read();
-                        let column = metadata.column(item.index);
-                        let (name, data_type) = match column {
-                            ColumnEntry::BaseTableColumn(BaseTableColumn {
-                                column_name,
-                                data_type,
-                                ..
-                            }) => (column_name.clone(), DataType::from(data_type)),
-                            ColumnEntry::DerivedColumn(DerivedColumn {
-                                alias, data_type, ..
-                            }) => (alias.clone(), data_type.clone()),
-                            ColumnEntry::InternalColumn(TableInternalColumn {
-                                internal_column,
-                                ..
-                            }) => (
-                                internal_column.column_name().to_owned(),
-                                internal_column.data_type(),
-                            ),
-                            ColumnEntry::VirtualColumn(VirtualColumn {
-                                column_name,
-                                data_type,
-                                ..
-                            }) => (column_name.clone(), DataType::from(data_type)),
-                        };
-
-                        // sort item is already a column
-                        let scalar = RemoteExpr::ColumnRef {
-                            span: None,
-                            id: name.clone(),
+        let order_by = scan.order_by.clone().map(|items| {
+            items
+                .into_iter()
+                .filter_map(|item| {
+                    let metadata = self.metadata.read();
+                    let column = metadata.column(item.index);
+                    let (name, data_type) = match column {
+                        ColumnEntry::BaseTableColumn(BaseTableColumn {
+                            column_name,
+                            data_type,
+                            ..
+                        }) => (column_name.clone(), DataType::from(data_type)),
+                        ColumnEntry::InternalColumn(TableInternalColumn {
+                            internal_column,
+                            ..
+                        }) => (
+                            internal_column.column_name().to_owned(),
+                            internal_column.data_type(),
+                        ),
+                        ColumnEntry::VirtualColumn(VirtualColumn {
+                            column_name,
                             data_type,
-                            display_name: name,
-                        };
+                            ..
+                        }) => (column_name.clone(), DataType::from(data_type)),
+                        ColumnEntry::DerivedColumn(_) => {
+                            return None;
+                        }
+                    };
 
-                        Ok((scalar, item.asc, item.nulls_first))
-                    })
-                    .collect::<Result<Vec<_>>>()
-            })
-            .transpose()?;
+                    // sort item is already a column
+                    let scalar = RemoteExpr::ColumnRef {
+                        span: None,
+                        id: name.clone(),
+                        data_type,
+                        display_name: name,
+                    };
+
+                    Some((scalar, item.asc, item.nulls_first))
+                })
+                .collect::<Vec<_>>()
+        });
 
         let virtual_column = self.build_virtual_column(&scan.columns);
 

tests/sqllogictests/suites/base/09_fuse_engine/09_0039_topk_sorter.test

@@ -39,5 +39,18 @@ select * from test where url like '%google%' order by event_time limit 10;
 0 2024-01-03 00:00:00.000000 http://www.google.com/xxx/
 0 2024-01-04 00:00:00.000000 http://www.google.com/xxx/
 
+
+statement ok
+create table test_k storage_format = 'native' as select * from test;
+
+
+query ITT
+select * from test_k where url like '%google%' order by event_time limit 1;
+----
+0 2024-01-01 00:00:00.000000 http://www.google.com/xxx/
+
 statement ok
 drop table test
+
+statement ok
+drop table test_k