Add basic support for flatten join

Author: osipovartem

datafusion/physical-expr/src/planner.rs

@@ -118,6 +118,10 @@ pub fn create_physical_expr(
             let idx = input_dfschema.index_of_column(c)?;
             Ok(Arc::new(Column::new(&c.name, idx)))
         }
+        Expr::OuterReferenceColumn(_datatype, c) => {
+            let idx = input_dfschema.index_of_column(c)?;
+            Ok(Arc::new(Column::new(&c.name, idx)))
+        }
         Expr::Literal(value) => Ok(Arc::new(Literal::new(value.clone()))),
         Expr::ScalarVariable(_, variable_names) => {
             if is_system_variables(variable_names) {

datafusion/sql/src/relation/mod.rs

@@ -400,6 +400,48 @@ impl<S: ContextProvider> SqlToRel<'_, S> {
 
                 (unpivot_plan, alias)
             }
+            TableFactor::Function {
+                name, args, alias, ..
+            } => {
+                let tbl_func_name = self.object_name_to_table_reference(name)?;
+                let schema = planner_context
+                    .outer_query_schema()
+                    .cloned()
+                    .unwrap_or_else(DFSchema::empty);
+                let func_args = args
+                    .into_iter()
+                    .map(|arg| match arg {
+                        FunctionArg::Unnamed(FunctionArgExpr::Expr(expr)) => {
+                            let expr = self.sql_expr_to_logical_expr(
+                                expr,
+                                &schema,
+                                planner_context,
+                            )?;
+                            Ok((expr, None))
+                        }
+                        FunctionArg::Named {
+                            name,
+                            arg: FunctionArgExpr::Expr(expr),
+                            ..
+                        } => {
+                            let expr = self.sql_expr_to_logical_expr(
+                                expr,
+                                &schema,
+                                planner_context,
+                            )?;
+                            Ok((expr, Some(name.value.clone())))
+                        }
+                        _ => plan_err!("Unsupported function argument: {arg:?}"),
+                    })
+                    .collect::<Result<Vec<(Expr, Option<String>)>>>()?;
+
+                let provider = self
+                    .context_provider
+                    .get_table_function_source(tbl_func_name.table(), func_args)?;
+                let plan =
+                    LogicalPlanBuilder::scan(tbl_func_name, provider, None)?.build()?;
+                (plan, alias)
+            }
             // @todo: Support TableFactory::TableFunction
             _ => {
                 return not_impl_err!(