WIP: AggregateTopKExec pretty printing

Author: srh

rust/cubestore/cubestore/src/queryplanner/planning.rs

@@ -2044,11 +2044,11 @@ pub mod tests {
             &indices,
         );
         let plan = choose_index(plan, &indices).await.unwrap().0;
+
         assert_eq!(
             pretty_printers::pp_plan(&plan),
-            "Projection, [s.Orders.order_customer, SUM(s.Orders.order_amount)]\
-           \n  ClusterAggregateTopK, limit: 10\
-           \n    Scan s.Orders, source: CubeTable(index: by_customer:3:[]:sort_on[order_customer]), fields: [order_customer, order_amount]"
+            "ClusterAggregateTopK, limit: 10\
+           \n  Scan s.orders, source: CubeTable(index: by_customer:3:[]:sort_on[order_customer]), fields: [order_customer, order_amount]"
         );
 
         // Projections should be handled properly.
@@ -2062,7 +2062,7 @@ pub mod tests {
             pretty_printers::pp_plan(&plan),
             "Projection, [customer, amount]\
            \n  ClusterAggregateTopK, limit: 10\
-           \n    Scan s.Orders, source: CubeTable(index: by_customer:3:[]:sort_on[order_customer]), fields: [order_customer, order_amount]"
+           \n    Scan s.orders, source: CubeTable(index: by_customer:3:[]:sort_on[order_customer]), fields: [order_customer, order_amount]"
         );
 
         let plan = initial_plan(
@@ -2077,7 +2077,7 @@ pub mod tests {
             pretty_printers::pp_plan_ext(&plan, &with_sort_by),
             "Projection, [amount, customer]\
            \n  ClusterAggregateTopK, limit: 10, sortBy: [2 desc null last]\
-           \n    Scan s.Orders, source: CubeTable(index: by_customer:3:[]:sort_on[order_customer]), fields: [order_customer, order_amount]"
+           \n    Scan s.orders, source: CubeTable(index: by_customer:3:[]:sort_on[order_customer]), fields: [order_customer, order_amount]"
         );
 
         // Ascending order is also ok.

rust/cubestore/cubestore/src/queryplanner/pretty_printers.rs

@@ -30,7 +30,7 @@ use crate::queryplanner::query_executor::{
 use crate::queryplanner::rolling::RollingWindowAggregate;
 use crate::queryplanner::serialized_plan::{IndexSnapshot, RowRange};
 use crate::queryplanner::tail_limit::TailLimitExec;
-use crate::queryplanner::topk::ClusterAggregateTopK;
+use crate::queryplanner::topk::{AggregateTopKExec, ClusterAggregateTopK};
 use crate::queryplanner::topk::SortColumn;
 use crate::queryplanner::trace_data_loaded::TraceDataLoadedExec;
 use crate::queryplanner::{CubeTableLogical, InfoSchemaTableProvider};
@@ -379,7 +379,6 @@ fn pp_source(t: Arc<dyn TableProvider>) -> String {
     }
 }
 
-// TODO upgrade DF: No pub
 pub fn pp_sort_columns(first_agg: usize, cs: &[SortColumn]) -> String {
     format!(
         "[{}]",
@@ -536,23 +535,22 @@ fn pp_phys_plan_indented(p: &dyn ExecutionPlan, indent: usize, o: &PPOptions, ou
                     })
                     .join(", ")
             );
-            // TODO upgrade DF <-
-            // } else if let Some(topk) = a.downcast_ref::<AggregateTopKExec>() {
-            //     *out += &format!("AggregateTopK, limit: {:?}", topk.limit);
-            //     if o.show_aggregations {
-            //         *out += &format!(", aggs: {:?}", topk.agg_expr);
-            //     }
-            //     if o.show_sort_by {
-            //         *out += &format!(
-            //             ", sortBy: {}",
-            //             pp_sort_columns(topk.key_len, &topk.order_by)
-            //         );
-            //     }
-            //     if o.show_filters {
-            //         if let Some(having) = &topk.having {
-            //             *out += &format!(", having: {}", having);
-            //         }
-            //     }
+        } else if let Some(topk) = a.downcast_ref::<AggregateTopKExec>() {
+            *out += &format!("AggregateTopK, limit: {:?}", topk.limit);
+            if o.show_aggregations {
+                *out += &format!(", aggs: {:?}", topk.agg_expr);
+            }
+            if o.show_sort_by {
+                *out += &format!(
+                    ", sortBy: {}",
+                    pp_sort_columns(topk.key_len, &topk.order_by)
+                );
+            }
+            if o.show_filters {
+                if let Some(having) = &topk.having {
+                    *out += &format!(", having: {}", having);
+                }
+            }
         } else if let Some(_) = a.downcast_ref::<PanicWorkerExec>() {
             *out += "PanicWorker";
         } else if let Some(_) = a.downcast_ref::<WorkerExec>() {