Projection pushdown removes unqualified column names even when they are used

[alamb]

Describe the bug

Problem:

I have a (user defined node) that looks like

ExtensionNode (expr uses col1, col2, col3) --> outputs new_col
  TableScan t1 of (col1, col2, col3)

Prior to #55 this worked

After that PR the projection pushdown logic decides that since the col1, col2 and col3 references don't have the table qualifier t1 on them, they are removed and the optimized plan looks like

ExtensionNode (expr uses col1, col2, col3) --> outputs new_col
  TableScan t1 of (col1)

Which then has issues because it expects col2 and col3 to be present but they have been "optimized" out

Expected behavior
The table scan should include col2 and col3 (they should not be optimized out), in addition to col1

[alamb]

FYI @houqp -- I believe I have found the problem, but now I need to sort out one other failure before I can make a PR

[alamb]

🤔 I think this may be more complicated than I thought - I am not sure if I am creating the plans incorrectly or if projection pushdown is doing the wrong thing. I need to study this some more

[alamb]

So at least initially the answer appears to be "I should be creating fully qualified Exprs (with relation name)"

Here is the change I needed to make in IOx:

         let exprs = input
             .schema()
             .fields()
             .iter()
-            .map(|field| logical_plan::col(field.name()))
+            .map(|field| Expr::Column(field.qualified_column()))
             .collect::<Vec<_>>();

This was a non obvious failure more however, and I can probably figure out a better way to detect such errors. Anyhow, closing for now

[houqp]

hmm... the projection push down should be able to handle unqualified columns without problem, i.e. the current design doesn't require clients to add qualifiers to all the fields. Do you have a reproducible example i can play with? I definitely didn't test the change with extension nodes.

[alamb]

@houqp I will get one

[alamb]

Here is a reproducer (in projection_push_down.rs) -- it can also be found on https://github.com/alamb/arrow-datafusion/tree/alamb/repro_projection_pruning

    #[test]
    fn table_scan_projected_schema_non_qualified_relation() -> Result<()> {
        let table_scan = test_table_scan()?;
        let input_schema = table_scan.schema();
        assert_eq!(3, input_schema.fields().len());
        assert_fields_eq(&table_scan, vec!["a", "b", "c"]);

        // Build the LogicalPlan directly (don't use PlanBuilder), so
        // that the Column references are unqualified (e.g. their
        // relation is `None`). PlanBuilder resolves the expressions
        let expr = vec![col("a"), col("b")];
        let projected_fields = exprlist_to_fields(&expr, input_schema).unwrap();
        let projected_schema = DFSchema::new(projected_fields).unwrap();
        let plan = LogicalPlan::Projection {
            expr,
            input: Arc::new(table_scan),
            schema: Arc::new(projected_schema),
        };

        assert_fields_eq(&plan, vec!["a", "b"]);

        let expected = "Projection: #a, #b\
        \n  TableScan: test projection=Some([0, 1])";

        assert_optimized_plan_eq(&plan, expected);

        Ok(())
    }

It fails in the following way:

---- optimizer::projection_push_down::tests::table_scan_projected_schema_non_qualified_relation stdout ----
thread 'optimizer::projection_push_down::tests::table_scan_projected_schema_non_qualified_relation' panicked at 'assertion failed: `(left == right)`
  left: `"Projection: #a, #b\n  TableScan: test projection=Some([0])"`,
 right: `"Projection: #test.a, #test.b\n  TableScan: test projection=Some([0, 1])"`', datafusion/src/optimizer/projection_push_down.rs:745:9
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace

The problem is that the TableScan is only scanning one column, even though the Projection is using two columns.

I will put up a PR with code to fix the problem

[alamb]

#619