WIP: Make field_index return Result

Author: srh

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

@@ -107,7 +107,7 @@ fn materialize_topk_under_limit_sort(fetch: usize, sort_expr: &Vec<SortExpr>, so
                 &sort_expr,
                 sort_input.schema(),
                 projection.as_ref().map(|c| c.input_columns.as_slice()),
-            ) {
+            )? {
                 sort_columns = sc;
             } else {
                 return Ok(None);
@@ -276,21 +276,15 @@ fn extract_projections_and_havings(p: &Arc<LogicalPlan>) -> Result<Option<Column
             for e in expr {
                 match e {
                     Expr::Alias(Alias { expr: box Expr::Column(c), relation: _, name: _ }) | Expr::Column(c) => {
-                        // TODO upgrade DF: These field_index errors should return Err (right?) as
-                        // we do in the Filter case only because it's in a transform_up with map_data.
-                        let Some(fi) = field_index(in_schema, c.relation.as_ref(), &c.name) else {
-                            return Ok(None);
-                        };
+                        let fi = field_index(in_schema, c.relation.as_ref(), &c.name)?;
                         input_columns.push(fi);
                     },
                     Expr::Alias(Alias { expr: box Expr::ScalarFunction(ScalarFunction { func, args }), relation: _, name: _})
                     | Expr::ScalarFunction(ScalarFunction { func, args }) => match func.name() {
                         // TODO upgrade DF: Use as_any() or something
                         "cardinality" => match &args[0] {
                             Expr::Column(c) => {
-                                let Some(fi) = field_index(in_schema, c.relation.as_ref(), &c.name) else {
-                                    return Ok(None);
-                                };
+                                let fi = field_index(in_schema, c.relation.as_ref(), &c.name)?;
                                 input_columns.push(fi);
                             }
                             _ => return Ok(None),
@@ -372,9 +366,7 @@ fn extract_projections_and_havings(p: &Arc<LogicalPlan>) -> Result<Option<Column
                 node.unalias_nested().transform_data(|node|
                     match node {
                         Expr::Column(c) => {
-                            let Some(fi) = field_index(in_schema, c.relation.as_ref(), &c.name) else {
-                                return Err(DataFusionError::Plan(format!("Field '{}' not found", &c.name)));  // TODO upgrade DF: spend brain cycles on error message
-                            };
+                            let fi = field_index(in_schema, c.relation.as_ref(), &c.name)?;
                             let replacement = inner_column_projection.post_projection[fi].clone();
                             // Transformed::yes/no doesn't matter here.
                             // let unequal = &replacement != &node;
@@ -424,7 +416,7 @@ fn extract_sort_columns(
     sort_expr: &[SortExpr],
     schema: &DFSchema,
     projection: Option<&[usize]>,
-) -> Option<Vec<SortColumn>> {
+) -> Result<Option<Vec<SortColumn>>, DataFusionError> {
     let mut sort_columns = Vec::with_capacity(sort_expr.len());
     for e in sort_expr {
         let SortExpr {expr, asc, nulls_first} = e;
@@ -435,27 +427,26 @@ fn extract_sort_columns(
                     index = p[index];
                 }
                 if index < group_key_len {
-                    return None;
+                    return Ok(None);
                 }
                 sort_columns.push(SortColumn {
                     agg_index: index - group_key_len,
                     asc: *asc,
                     nulls_first: *nulls_first,
                 })
             }
-            _ => return None,
+            _ => return Ok(None),
         }
     }
-    Some(sort_columns)
+    Ok(Some(sort_columns))
 }
 
-fn field_index(schema: &DFSchema, qualifier: Option<&TableReference>, name: &str) -> Option<usize> {
+// It is actually an error if expressions are nonsense expressions that don't evaluate on the given
+// schema.  So we return Result (instead of Option<_>) now.
+fn field_index(schema: &DFSchema, qualifier: Option<&TableReference>, name: &str) -> Result<usize, DataFusionError> {
+    // Calling field_not_found is exactly `schema.index_of_column(col: &Column)` behavior.
     schema.index_of_column_by_name(qualifier, name)
-
-    // TODO upgrade DF: Reconsider.
-    // schema
-    //     .iter()
-    //     .position(|f| f.qualifier().map(|s| s.as_str()) == qualifier && f.name() == name)
+        .ok_or_else(|| datafusion::common::field_not_found(qualifier.cloned(), name, schema))
 }
 
 pub fn plan_topk(