chore(cubestore): Upgrade DF 46: Make Kafka plan error messages display plan

Author: srh

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

@@ -57,11 +57,12 @@ pub struct PPOptions {
     pub show_output_hints: bool,
     pub show_check_memory_nodes: bool,
     pub show_partitions: bool,
+    pub traverse_past_clustersend: bool,
 }
 
 impl PPOptions {
     #[allow(unused)]
-    pub fn everything() -> PPOptions {
+    pub fn show_all() -> PPOptions {
         PPOptions {
             show_filters: true,
             show_sort_by: true,
@@ -70,6 +71,7 @@ impl PPOptions {
             show_output_hints: true,
             show_check_memory_nodes: true,
             show_partitions: true,
+            traverse_past_clustersend: false,
         }
     }
 
@@ -476,7 +478,7 @@ fn pp_phys_plan_indented(p: &dyn ExecutionPlan, indent: usize, o: &PPOptions, ou
         return;
     }
     pp_instance(p, indent, o, out);
-    if p.as_any().is::<ClusterSendExec>() {
+    if !o.traverse_past_clustersend && p.as_any().is::<ClusterSendExec>() {
         // Do not show children of ClusterSend. This is a hack to avoid rewriting all tests.
         return;
     }

rust/cubestore/cubestore/src/streaming/kafka_post_processing.rs

@@ -1,5 +1,6 @@
 use crate::metastore::Column;
 use crate::queryplanner::metadata_cache::MetadataCacheFactory;
+use crate::queryplanner::pretty_printers::{pp_plan_ext, PPOptions};
 use crate::queryplanner::{sql_to_rel_options, QueryPlannerImpl};
 use crate::sql::MySqlDialectWithBackTicks;
 use crate::streaming::topic_table_provider::TopicTableProvider;
@@ -425,6 +426,12 @@ impl KafkaPostProcessPlanner {
         &self,
         plan: &LogicalPlan,
     ) -> Result<(Arc<dyn ExecutionPlan>, Option<Arc<dyn ExecutionPlan>>), CubeError> {
+        fn only_certain_plans_allowed_error(plan: &LogicalPlan) -> CubeError {
+            CubeError::user(
+                format!("Only Projection > [Filter] > TableScan plans are allowed for streaming; got plan {}", pp_plan_ext(plan, &PPOptions::show_all())),
+            )
+        }
+
         let source_schema = Arc::new(Schema::new(
             self.source_columns
                 .iter()
@@ -465,10 +472,7 @@ impl KafkaPostProcessPlanner {
 
                         Ok((projection_phys_plan.clone(), Some(filter_phys_plan)))
                     }
-                    _ => Err(CubeError::user(
-                        "Only Projection > [Filter] > TableScan plans are allowed for streaming"
-                            .to_string(),
-                    )),
+                    _ => Err(only_certain_plans_allowed_error(plan)),
                 },
                 LogicalPlan::TableScan { .. } => {
                     let projection_plan =
@@ -484,15 +488,9 @@ impl KafkaPostProcessPlanner {
                         .with_new_children(vec![empty_exec.clone()])?;
                     Ok((projection_phys_plan, None))
                 }
-                _ => Err(CubeError::user(
-                    "Only Projection > [Filter] > TableScan plans are allowed for streaming"
-                        .to_string(),
-                )),
+                _ => Err(only_certain_plans_allowed_error(plan)),
             },
-            _ => Err(CubeError::user(
-                "Only Projection > [Filter] > TableScan plans are allowed for streaming"
-                    .to_string(),
-            )),
+            _ => Err(only_certain_plans_allowed_error(plan)),
         }
     }