Implement physical execution of uncorrelated scalar subqueries

[andygrove]

Is your feature request related to a problem or challenge? Please describe what you are trying to do.

We currently support uncorrelated scalar subqueries by translating them into a cross-join. It would likely be more efficient to execute the subquery and update the original plan with the scalar value.

We also need to do this so we can throw an error if more than 1 row is returned and there is no way to do this in the logical plan.

Describe the solution you'd like

The optimizer will need some kind of trait that execution engines (DataFusion, Ballista, Dask SQL, etc) can implement for resolving scalar values.

trait OptimizerContext {
  fn supports_scalar_subquery_execution() -> bool;
  fn execute_scalar_subquery(plan: &LogicalPlan) -> Result<ScalarValue>;
}

Describe alternatives you've considered

Additional context

[Dandandan]

We can also remove the restriction on the conversion that the subquery should contain an aggregate, and perform the check of being scalar in the physical node.

[andygrove]

We can also remove the restriction on the conversion that the subquery should contain an aggregate, and perform the check of being scalar in the physical node.

Related issue: #3725

[xudong963]

We can just change it to left join, and add logic to check if more than 1 row is returned in left join.

[Dandandan]

We can just change it to left join, and add logic to check if more than 1 row is returned in left join.

I don't think that's possible. A cross join is used as that doesn't require a join condition. The cross join is less efficient however, as it repeats the value for the scalar as much as there are values on the left side, and then uses that to filter on (which is less efficient than using a scalar in the filter).

[xudong963]

A cross join is used as that doesn't require a join condition

Oh, my mistake. The tile is uncorrelated scalar subquery.

For correlated scalar subquery, we can change it to left join.