LogicalPlan Optimizer: Push <col> = <col> predicates into Joins, if possible #1612

Is your feature request related to a problem or challenge? Please describe what you are trying to do.
We want to bring some extra performance for certain predicates to the dataframe API users as well.

Logically, a sql query like SELECT * from A, B where A.a = B.b can be planned as a cross join and filter:

Filter(a = b)
  Join (Cross)
    TableScan A
    TableScan B

Almost no database actually does this because of how expensive a CROSS join is, and instead uses a specialized "Inner Join" operator.

In DataFusion, The SQL planner has code to recognize certain predicates such as a = b as "join predicates"

However, this logic is not applied to the DataFrame API where one makes a plan like

let df = ctx.read_table(A)
    .join(ctx.read_table(B), JoinType::Inner, col("a"), col("b"))

Which is somewhat less than ergonomic

It would be cool if it were possible to make a plan like

let df = ctx.read_table(A)
    .join(ctx.read_table(B), JoinType::Cross)
    .filter(col("a").eq(col("b")))

And have the DataFusion optimizer rewrite the CrossJoin into a InnerJoin

Describe the solution you'd like

1. The sql planner can just naively plan these types of queries as cross joins + Filter
2. Add a new optimizer pass that rewrites them into inner joins whenever applicable.
3. This would also allow dataframe users to write simple joins + filters and run quickly too

In terms of the optimizer pass, it might be possible to extend FilterPushdown in a way that would try and push down column = column type expressions from Filter nodes into Join nodes. So for example

Rewrite

Filter(a = b)
  Join (Cross)
    TableScan A
    TableScan B

To

  Join (Inner) exprs: {a = b}
    TableScan A
    TableScan B

And rewrite

Filter(a = b)
  Join (Inner) exprs: {a2 = b2}
    TableScan A
    TableScan B

to

Filter(a = b)
  Join (Inner) exprs: {a2 = b2 AND a = b}
    TableScan A
    TableScan B

Describe alternatives you've considered
Maybe this should be its own optimizer pass

Additional context
Suggested by @houqp . See #1566 and #1566 (comment)