chore(cubestore): Upgrade DF 46: Pretty-printing improvements

- TraceDataLoadedExec nodes are now pretty-printed, consistently with the original.
- CoalesceBatches now printed without the "Exec".

Author: srh

rust/cubestore/cubestore-sql-tests/src/tests.rs

@@ -3095,7 +3095,7 @@ async fn planning_inplace_aggregate(service: Box<dyn SqlClient>) {
         "PartiallySortedFinalAggregate, partitions: 1\
         \n  Worker, partitions: 1\
         \n    PartiallySortedPartialAggregate, partitions: 1\
-        \n      CoalesceBatchesExec, partitions: 1\
+        \n      CoalesceBatches, partitions: 1\
         \n        Filter, partitions: 1\
         \n          Scan, index: default:2:[2]:sort_on[url, segment, day], fields: *, partitions: 1\
         \n            Sort, partitions: 1\
@@ -3113,7 +3113,7 @@ async fn planning_inplace_aggregate(service: Box<dyn SqlClient>) {
         "PartiallySortedFinalAggregate, partitions: 1\
         \n  Worker, partitions: 1\
         \n    PartiallySortedPartialAggregate, partitions: 1\
-        \n      CoalesceBatchesExec, partitions: 1\
+        \n      CoalesceBatches, partitions: 1\
         \n        Filter, partitions: 1\
         \n          Scan, index: default:2:[2]:sort_on[url, segment, day], fields: *, partitions: 1\
         \n            Sort, partitions: 1\
@@ -3198,7 +3198,7 @@ async fn planning_hints(service: Box<dyn SqlClient>) {
         \n  Worker, single_vals: [1]\
         \n    CoalescePartitions, single_vals: [1]\
         \n      Projection, [id3, id2], single_vals: [1]\
-        \n        CoalesceBatchesExec, single_vals: [0]\
+        \n        CoalesceBatches, single_vals: [0]\
         \n          Filter, single_vals: [0]\
         \n            Scan, index: default:1:[1], fields: [id2, id3]\
         \n              Empty"
@@ -3212,7 +3212,7 @@ async fn planning_hints(service: Box<dyn SqlClient>) {
     assert_eq!(
         pp_phys_plan_ext(p.worker.as_ref(), &show_hints),
         "Worker, sort_order: [0]\
-        \n  CoalesceBatchesExec, sort_order: [0]\
+        \n  CoalesceBatches, sort_order: [0]\
         \n    Filter, sort_order: [0]\
         \n      Scan, index: default:1:[1]:sort_on[id1, id2], fields: *, sort_order: [0, 1, 2]\
         \n        Sort, sort_order: [0, 1, 2]\
@@ -3225,7 +3225,7 @@ async fn planning_hints(service: Box<dyn SqlClient>) {
     assert_eq!(
         pp_phys_plan_ext(p.worker.as_ref(), &show_hints),
         "Worker, sort_order: [0, 1]\
-        \n  CoalesceBatchesExec, sort_order: [0, 1]\
+        \n  CoalesceBatches, sort_order: [0, 1]\
         \n    Filter, sort_order: [0, 1]\
         \n      CoalescePartitions, sort_order: [0, 1, 2]\
         \n        Scan, index: default:1:[1], fields: *, sort_order: [0, 1, 2]\
@@ -3284,13 +3284,13 @@ async fn planning_inplace_aggregate2(service: Box<dyn SqlClient>) {
            \n          CoalescePartitions\
            \n            Union\
            \n              CoalescePartitions\
-           \n                CoalesceBatchesExec\
+           \n                CoalesceBatches\
            \n                  Filter\
            \n                    Scan, index: default:1:[1], fields: *, sort_order: [0, 1, 2, 3, 4]\
            \n                      Sort, by: [allowed@0, site_id@1, url@2, day@3, hits@4], sort_order: [0, 1, 2, 3, 4]\
            \n                        Empty\
            \n              CoalescePartitions\
-           \n                CoalesceBatchesExec\
+           \n                CoalesceBatches\
            \n                  Filter\
            \n                    Scan, index: default:2:[2], fields: *, sort_order: [0, 1, 2, 3, 4]\
            \n                      Sort, by: [allowed@0, site_id@1, url@2, day@3, hits@4], sort_order: [0, 1, 2, 3, 4]\
@@ -3547,7 +3547,7 @@ async fn planning_simple(service: Box<dyn SqlClient>) {
     assert_eq!(
         pp_phys_plan(p.worker.as_ref()),
         "Worker\
-        \n  CoalesceBatchesExec\
+        \n  CoalesceBatches\
         \n    Filter\
         \n      CoalescePartitions\
         \n        Scan, index: default:1:[1], fields: [id, amount]\
@@ -3573,7 +3573,7 @@ async fn planning_simple(service: Box<dyn SqlClient>) {
         pp_phys_plan(p.worker.as_ref()),
         "Sort\
         \n  Worker\
-        \n    CoalesceBatchesExec\
+        \n    CoalesceBatches\
         \n      Filter\
         \n        CoalescePartitions\
         \n          Scan, index: default:1:[1], fields: [id, amount]\
@@ -3599,7 +3599,7 @@ async fn planning_simple(service: Box<dyn SqlClient>) {
         pp_phys_plan(p.worker.as_ref()),
         "GlobalLimit, n: 10\
         \n  Worker\
-        \n    CoalesceBatchesExec\
+        \n    CoalesceBatches\
         \n      Filter\
         \n        CoalescePartitions\
         \n          Scan, index: default:1:[1], fields: [id, amount]\
@@ -3692,7 +3692,7 @@ async fn planning_filter_index_selection(service: Box<dyn SqlClient>) {
         "SortedFinalAggregate\
         \n  Worker\
         \n    SortedPartialAggregate\
-        \n      CoalesceBatchesExec\
+        \n      CoalesceBatches\
         \n        Filter\
         \n          Scan, index: cb:2:[2]:sort_on[c, b], fields: [b, c, amount]\
         \n            Sort\
@@ -3716,7 +3716,7 @@ async fn planning_filter_index_selection(service: Box<dyn SqlClient>) {
         \n    Worker\
         \n      CoalescePartitions\
         \n        LinearPartialAggregate\
-        \n          CoalesceBatchesExec\
+        \n          CoalesceBatches\
         \n            Filter\
         \n              Scan, index: cb:2:[2], fields: [b, c, amount]\
         \n                Sort\
@@ -3741,7 +3741,7 @@ async fn planning_filter_index_selection(service: Box<dyn SqlClient>) {
         "SortedFinalAggregate\
         \n  Worker\
         \n    SortedPartialAggregate\
-        \n      CoalesceBatchesExec\
+        \n      CoalesceBatches\
         \n        Filter\
         \n          Scan, index: cb:2:[2]:sort_on[c, b], fields: [a, b, c, amount]\
         \n            Sort\
@@ -3911,15 +3911,15 @@ async fn planning_3_table_joins(service: Box<dyn SqlClient>) {
             \n        MergeJoin, on: [product_id@1 = product_id@0]\
             \n          Projection, [order_id, product_id, customer_name]\
             \n            MergeJoin, on: [customer_id@1 = customer_id@0]\
-            \n              CoalesceBatchesExec\
+            \n              CoalesceBatches\
             \n                Filter, predicate: product_id@2 = 125\
             \n                  Scan, index: by_product_customer:3:[3]:sort_on[product_id, customer_id], fields: [order_id, customer_id, product_id], predicate: BinaryExpr(BinaryExpr { left: Column(Column { relation: None, name: \"product_id\" }), op: Eq, right: Literal(Int64(125)) })\
             \n                    Sort\
             \n                      Empty\
             \n              Scan, index: default:4:[4]:sort_on[customer_id], fields: *\
             \n                Sort\
             \n                  Empty\
-            \n          CoalesceBatchesExec\
+            \n          CoalesceBatches\
             \n            Filter, predicate: product_id@0 = 125\
             \n              Scan, index: default:5:[5]:sort_on[product_id], fields: *, predicate: BinaryExpr(BinaryExpr { left: Column(Column { relation: None, name: \"product_id\" }), op: Eq, right: Literal(Int64(125)) })\
             \n                Sort\
@@ -7530,7 +7530,7 @@ async fn planning_aggregate_index(service: Box<dyn SqlClient>) {
         "SortedFinalAggregate\
         \n  Worker\
         \n    SortedPartialAggregate\
-        \n      CoalesceBatchesExec\
+        \n      CoalesceBatches\
         \n        Filter\
         \n          Scan, index: default:3:[3]:sort_on[a, b, c], fields: *\
         \n            Sort\
@@ -7576,7 +7576,7 @@ async fn planning_aggregate_index(service: Box<dyn SqlClient>) {
         "SortedFinalAggregate\
         \n  Worker\
         \n    SortedPartialAggregate\
-        \n      CoalesceBatchesExec\
+        \n      CoalesceBatches\
         \n        Filter\
         \n          Scan, index: aggr_index:2:[2]:sort_on[a, b], fields: [a, b, a_sum]\
         \n            Sort\

rust/cubestore/cubestore/src/queryplanner/flatten_union.rs

@@ -7,6 +7,7 @@ use datafusion::optimizer::OptimizerConfig;
 use std::fmt::Debug;
 use std::sync::Arc;
 
+// TODO upgrade DF: Remove?  We have EliminateNestedUnion.
 #[derive(Debug)]
 pub struct FlattenUnion;
 

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

@@ -11,11 +11,13 @@ use datafusion::logical_expr::{
     Aggregate, EmptyRelation, Explain, Extension, FetchType, Filter, Join, Limit, LogicalPlan, Projection, Repartition, SkipType, Sort, TableScan, Union, Window
 };
 use datafusion::physical_expr::{AcrossPartitions, ConstExpr};
+use datafusion::physical_optimizer::pruning;
 use datafusion::physical_plan::aggregates::{AggregateExec, AggregateMode};
+use datafusion::physical_plan::coalesce_batches::CoalesceBatchesExec;
 use datafusion::physical_plan::coalesce_partitions::CoalescePartitionsExec;
 use datafusion::physical_plan::filter::FilterExec;
 use datafusion::physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
-use datafusion::physical_plan::{DefaultDisplay, DisplayAs, DisplayFormatType, ExecutionPlan, InputOrderMode, PlanProperties};
+use datafusion::physical_plan::{DefaultDisplay, ExecutionPlan, InputOrderMode, PlanProperties};
 use datafusion::prelude::Expr;
 use datafusion_datasource::file_scan_config::FileScanConfig;
 use datafusion_datasource::memory::MemoryExec;
@@ -37,7 +39,6 @@ use crate::queryplanner::serialized_plan::{IndexSnapshot, RowRange};
 use crate::queryplanner::tail_limit::TailLimitExec;
 use crate::queryplanner::topk::SortColumn;
 use crate::queryplanner::topk::{AggregateTopKExec, ClusterAggregateTopKUpper, ClusterAggregateTopKLower};
-use crate::queryplanner::trace_data_loaded::TraceDataLoadedExec;
 use crate::queryplanner::{CubeTableLogical, InfoSchemaTableProvider, QueryPlan};
 use crate::streaming::topic_table_provider::TopicTableProvider;
 use datafusion::physical_plan::empty::EmptyExec;
@@ -59,10 +60,12 @@ pub struct PPOptions {
     pub show_output_hints: bool,
     pub show_check_memory_nodes: bool,
     pub show_partitions: bool,
+    pub show_metrics: bool,
     pub traverse_past_clustersend: bool,
 }
 
 impl PPOptions {
+    // TODO upgrade DF: Rename
     #[allow(unused)]
     pub fn show_all() -> PPOptions {
         PPOptions {
@@ -73,6 +76,7 @@ impl PPOptions {
             show_output_hints: true,
             show_check_memory_nodes: true,
             show_partitions: true,
+            show_metrics: false,  // yeah
             traverse_past_clustersend: false,
         }
     }
@@ -470,8 +474,7 @@ pub fn pp_sort_columns(first_agg: usize, cs: &[SortColumn]) -> String {
 }
 
 fn pp_phys_plan_indented(p: &dyn ExecutionPlan, indent: usize, o: &PPOptions, out: &mut String) {
-    if (p.as_any().is::<CheckMemoryExec>() || p.as_any().is::<TraceDataLoadedExec>())
-        && !o.show_check_memory_nodes
+    if p.as_any().is::<CheckMemoryExec>() && !o.show_check_memory_nodes
     {
         //We don't show CheckMemoryExec in plan by default
         if let Some(child) = p.children().first() {
@@ -630,6 +633,8 @@ fn pp_phys_plan_indented(p: &dyn ExecutionPlan, indent: usize, o: &PPOptions, ou
             *out += "PanicWorker";
         } else if let Some(_) = a.downcast_ref::<WorkerExec>() {
             *out += &format!("Worker");
+        } else if let Some(_) = a.downcast_ref::<CoalesceBatchesExec>() {
+            *out += "CoalesceBatches";
         } else if let Some(_) = a.downcast_ref::<CoalescePartitionsExec>() {
             *out += "CoalescePartitions";
         } else if let Some(s) = a.downcast_ref::<SortPreservingMergeExec>() {
@@ -676,6 +681,23 @@ fn pp_phys_plan_indented(p: &dyn ExecutionPlan, indent: usize, o: &PPOptions, ou
                         "ParquetScan, files: {}",
                         fse.file_groups.iter().flatten().map(|p| p.object_meta.location.to_string()).join(","),
                     );
+                    if o.show_filters {
+                        if let Some(predicate) = p.predicate() {
+                            *out += &format!(", predicate: {}", predicate);
+                        }
+                        // pruning_predicate and page_pruning_predicate are derived from
+                        // p.predicate(), and they tend to be more verbose.  Note: because we have
+                        // configured the default pushdown_filters = false (default false as of DF
+                        // <= 46.0.1), p.predicate() is not directly used.
+
+                        // if let Some(pruning_predicate) = p.pruning_predicate() {
+                        //     *out += &format!(", pruning_predicate: {}", pruning_predicate.predicate_expr());
+                        // }
+                        // if let Some(page_pruning_predicate) = p.page_pruning_predicate() {
+                        //     // If this is uncommented, page_pruning_predicate.predicates() would need to be added to DF.
+                        //     *out += &format!(", page_pruning_predicates: [{}]", page_pruning_predicate.predicates().iter().map(|pred| pred.predicate_expr()).join(", "));
+                        // }
+                    }
                 } else {
                     *out += &format!("{}", DefaultDisplay(dse));
                 }
@@ -766,6 +788,12 @@ fn pp_phys_plan_indented(p: &dyn ExecutionPlan, indent: usize, o: &PPOptions, ou
         if o.show_partitions && !skip_show_partitions {
             *out += &format!(", partitions: {}", p.properties().output_partitioning().partition_count());
         }
+
+        if o.show_metrics {
+            if let Some(m) = p.metrics() {
+                *out += &format!(", metrics: {}", m);
+            }
+        }
     }
 }
 

rust/cubestore/cubestore/src/sql/mod.rs

@@ -3360,7 +3360,7 @@ mod tests {
                 \n    Worker\
                 \n      CoalescePartitions\
                 \n        LinearPartialAggregate\
-                \n          CoalesceBatchesExec\
+                \n          CoalesceBatches\
                 \n            Filter\
                 \n              MergeSort\
                 \n                Scan, index: default:1:[1]:sort_on[num], fields: *\
@@ -4430,7 +4430,7 @@ mod tests {
                     .values()[2] {
                         TableValue::String(pp_plan) => {
                             let regex = Regex::new(
-                                r"LinearPartialAggregate\s+CoalesceBatchesExec\s+Filter\s+Scan, index: default:1:\[1\], fields: \[platform, age, amount\]\s+ParquetScan, files: \S*\.chunk\.parquet"
+                                r"LinearPartialAggregate\s+CoalesceBatches\s+Filter\s+Scan, index: default:1:\[1\], fields: \[platform, age, amount\]\s+ParquetScan, files: \S*\.chunk\.parquet"
                             ).unwrap();
                             let matches = regex.captures_iter(&pp_plan).count();
                             assert_eq!(matches, 1, "pp_plan = {}", pp_plan);