WIP: Enforce topk output schema column names properly

Author: srh

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

@@ -2045,7 +2045,6 @@ pub mod tests {
         );
         let plan = choose_index(plan, &indices).await.unwrap().0;
 
-        // TODO upgrade DF: Isn't this projection wrong?
         assert_eq!(
             pretty_printers::pp_plan(&plan),
             "Projection, [s.orders.order_customer:order_customer, sum(s.orders.order_amount)]\

rust/cubestore/cubestore/src/queryplanner/topk/plan.rs

@@ -134,20 +134,21 @@ fn materialize_topk_under_limit_sort(fetch: usize, sort_expr: &Vec<SortExpr>, so
                             snapshots: cs.snapshots.clone(),
                         }),
                     });
+                    // TODO upgrade DF: Projection should always be Some, now.  We are also in some
+                    // cases (such as when there was no projection node) creating a spurious
+                    // projection.
                     if let Some(p) = projection {
-                        let in_schema = topk.schema();
                         let out_schema = p.schema;
                         let mut expr = Vec::with_capacity(p.input_columns.len());
                         for out_i in 0..p.input_columns.len() {
-                            let in_field = in_schema.field(p.input_columns[out_i]);
-                            let out_name: &str = out_schema.field(out_i).name();
+                            let (out_tr, out_field) = out_schema.qualified_field(out_i);
 
-                            //let mut e = Expr::Column(f.qualified_column());
                             let mut e =
                                 p.post_projection[p.input_columns[out_i]].clone();
-                            if out_name != in_field.name() {
-                                // TODO upgrade DF: Check if we might need relation -- there is a commented line f.qualified_column() above, too.
-                                e = Expr::Alias(Alias { expr: Box::new(e), relation: None, name: out_name.to_owned() });
+                            let (e_tr, e_name) = e.qualified_name();
+
+                            if out_tr != e_tr.as_ref() || out_field.name() != &e_name {
+                                e = Expr::Alias(Alias { expr: Box::new(e), relation: out_tr.cloned(), name: out_field.name().clone() });
                             }
                             expr.push(e);
                         }
@@ -239,7 +240,7 @@ fn extract_aggregate_fun(e: &Expr) -> Option<(TopKAggregateFunction, &Vec<Expr>)
 
 #[derive(Debug)]
 struct ColumnProjection<'a> {
-    // The (sole) column index within `input.schema()` that the post_projection expr uses.
+    // The (sole) column indexes within `input.schema()` that the post_projection expr uses.
     input_columns: Vec<usize>,
     input: &'a Arc<LogicalPlan>,
     // Output schema (after applying `having_expr` and then `post_projection` and then aliases).  In