init group join implementation

datafusion/common/src/functional_dependencies.rs

@@ -368,6 +368,13 @@ impl FunctionalDependencies {
                 left_func_dependencies.extend(right_func_dependencies);
                 left_func_dependencies
             }
+            JoinType::LeftGroup => {
+                right_func_dependencies.add_offset(left_cols_len);
+                left_func_dependencies =
+                    left_func_dependencies.with_dependency(Dependency::Single);
+                right_func_dependencies.downgrade_dependencies();
+                left_func_dependencies
+            }
             JoinType::LeftSemi | JoinType::LeftAnti | JoinType::LeftMark => {
                 // These joins preserve functional dependencies of the left side:
                 left_func_dependencies

datafusion/common/src/join_type.rs

@@ -67,6 +67,10 @@ pub enum JoinType {
     ///
     /// [1]: http://btw2017.informatik.uni-stuttgart.de/slidesandpapers/F1-10-37/paper_web.pdf
     LeftMark,
+    /// Left Group join
+    ///
+    /// [2]: https://www.vldb.org/pvldb/vol4/p843-moerkotte.pdf
+    LeftGroup,
 }
 
 impl JoinType {
@@ -90,6 +94,9 @@ impl JoinType {
             JoinType::LeftMark => {
                 unreachable!("LeftMark join type does not support swapping")
             }
+            JoinType::LeftGroup => {
+                unreachable!("LeftGroup join type does not support swapping")
+            }
         }
     }
 
@@ -121,6 +128,7 @@ impl Display for JoinType {
             JoinType::LeftAnti => "LeftAnti",
             JoinType::RightAnti => "RightAnti",
             JoinType::LeftMark => "LeftMark",
+            JoinType::LeftGroup => "LeftGroup",
         };
         write!(f, "{join_type}")
     }
@@ -141,6 +149,7 @@ impl FromStr for JoinType {
             "LEFTANTI" => Ok(JoinType::LeftAnti),
             "RIGHTANTI" => Ok(JoinType::RightAnti),
             "LEFTMARK" => Ok(JoinType::LeftMark),
+            "LEFTGROUP" => Ok(JoinType::LeftGroup),
             _ => _not_impl_err!("The join type {s} does not exist or is not implemented"),
         }
     }

datafusion/expr/src/logical_plan/builder.rs

@@ -870,6 +870,10 @@ impl LogicalPlanBuilder {
             return plan_err!("left_keys and right_keys were not the same length");
         }
 
+        if join_type == JoinType::LeftGroup {
+            return plan_err!("Cannot create LefGroup Join manualy.");
+        }
+
         let filter = if let Some(expr) = filter {
             let filter = normalize_col_with_schemas_and_ambiguity_check(
                 expr,
@@ -977,6 +981,8 @@ impl LogicalPlanBuilder {
             join_constraint: JoinConstraint::On,
             schema: DFSchemaRef::new(join_schema),
             null_equals_null,
+            group_expr: None,
+            aggr_expr: None,
         })))
     }
 
@@ -1041,6 +1047,8 @@ impl LogicalPlanBuilder {
                 join_constraint: JoinConstraint::Using,
                 schema: DFSchemaRef::new(join_schema),
                 null_equals_null: false,
+                group_expr: None,
+                aggr_expr: None,
             })))
         }
     }
@@ -1058,6 +1066,8 @@ impl LogicalPlanBuilder {
             join_constraint: JoinConstraint::On,
             null_equals_null: false,
             schema: DFSchemaRef::new(join_schema),
+            group_expr: None,
+            aggr_expr: None,
         })))
     }
 
@@ -1276,6 +1286,8 @@ impl LogicalPlanBuilder {
             join_constraint: JoinConstraint::On,
             schema: DFSchemaRef::new(join_schema),
             null_equals_null: false,
+            group_expr: None,
+            aggr_expr: None,
         })))
     }
 
@@ -1389,7 +1401,7 @@ pub fn build_join_schema(
                 .collect::<Vec<_>>();
             left_fields.into_iter().chain(right_fields).collect()
         }
-        JoinType::Left => {
+        JoinType::Left | JoinType::LeftGroup => {
             // left then right, right set to nullable in case of not matched scenario
             let left_fields = left_fields
                 .map(|(q, f)| (q.cloned(), Arc::clone(f)))

datafusion/expr/src/logical_plan/invariants.rs

@@ -271,7 +271,8 @@ fn check_inner_plan(inner_plan: &LogicalPlan, can_contain_outer_ref: bool) -> Re
             JoinType::Left
             | JoinType::LeftSemi
             | JoinType::LeftAnti
-            | JoinType::LeftMark => {
+            | JoinType::LeftMark
+            | JoinType::LeftGroup => {
                 check_inner_plan(left, can_contain_outer_ref)?;
                 check_inner_plan(right, false)
             }

datafusion/expr/src/logical_plan/plan.rs

@@ -545,7 +545,11 @@ impl LogicalPlan {
                 join_type,
                 ..
             }) => match join_type {
-                JoinType::Inner | JoinType::Left | JoinType::Right | JoinType::Full => {
+                JoinType::Inner
+                | JoinType::Left
+                | JoinType::Right
+                | JoinType::Full
+                | JoinType::LeftGroup => {
                     if left.schema().fields().is_empty() {
                         right.head_output_expr()
                     } else {
@@ -658,6 +662,8 @@ impl LogicalPlan {
                 on,
                 schema: _,
                 null_equals_null,
+                group_expr,
+                aggr_expr,
             }) => {
                 let schema =
                     build_join_schema(left.schema(), right.schema(), &join_type)?;
@@ -679,6 +685,8 @@ impl LogicalPlan {
                     filter,
                     schema: DFSchemaRef::new(schema),
                     null_equals_null,
+                    group_expr,
+                    aggr_expr,
                 }))
             }
             LogicalPlan::Subquery(_) => Ok(self),
@@ -932,6 +940,8 @@ impl LogicalPlan {
                     filter: filter_expr,
                     schema: DFSchemaRef::new(schema),
                     null_equals_null: *null_equals_null,
+                    group_expr: None,
+                    aggr_expr: None,
                 }))
             }
             LogicalPlan::Subquery(Subquery {
@@ -1330,9 +1340,10 @@ impl LogicalPlan {
                         (left_max, right_max, _) => Some(left_max * right_max),
                     }
                 }
-                JoinType::LeftSemi | JoinType::LeftAnti | JoinType::LeftMark => {
-                    left.max_rows()
-                }
+                JoinType::LeftSemi
+                | JoinType::LeftAnti
+                | JoinType::LeftMark
+                | JoinType::LeftGroup => left.max_rows(),
                 JoinType::RightSemi | JoinType::RightAnti => right.max_rows(),
             },
             LogicalPlan::Repartition(Repartition { input, .. }) => input.max_rows(),
@@ -3295,6 +3306,9 @@ pub struct Join {
     pub schema: DFSchemaRef,
     /// If null_equals_null is true, null == null else null != null
     pub null_equals_null: bool,
+    /// Only for GroupJoin
+    pub group_expr: Option<Vec<Expr>>,
+    pub aggr_expr: Option<Vec<Expr>>,
 }
 
 impl Join {
@@ -3328,6 +3342,8 @@ impl Join {
             join_constraint: original_join.join_constraint,
             schema: Arc::new(join_schema),
             null_equals_null: original_join.null_equals_null,
+            group_expr: original_join.group_expr.clone(),
+            aggr_expr: original_join.aggr_expr.clone(),
         })
     }
 }

datafusion/expr/src/logical_plan/tree_node.rs

@@ -141,6 +141,8 @@ impl TreeNode for LogicalPlan {
                 join_constraint,
                 schema,
                 null_equals_null,
+                group_expr,
+                aggr_expr,
             }) => (left, right).map_elements(f)?.update_data(|(left, right)| {
                 LogicalPlan::Join(Join {
                     left,
@@ -151,6 +153,8 @@ impl TreeNode for LogicalPlan {
                     join_constraint,
                     schema,
                     null_equals_null,
+                    group_expr,
+                    aggr_expr,
                 })
             }),
             LogicalPlan::Limit(Limit { skip, fetch, input }) => input
@@ -573,6 +577,8 @@ impl LogicalPlan {
                 join_constraint,
                 schema,
                 null_equals_null,
+                group_expr,
+                aggr_expr,
             }) => (on, filter).map_elements(f)?.update_data(|(on, filter)| {
                 LogicalPlan::Join(Join {
                     left,
@@ -583,6 +589,8 @@ impl LogicalPlan {
                     join_constraint,
                     schema,
                     null_equals_null,
+                    group_expr,
+                    aggr_expr,
                 })
             }),
             LogicalPlan::Sort(Sort { expr, input, fetch }) => expr

datafusion/optimizer/src/eliminate_cross_join.rs

@@ -329,6 +329,8 @@ fn find_inner_join(
                 filter: None,
                 schema: join_schema,
                 null_equals_null: false,
+                group_expr: None,
+                aggr_expr: None,
             }));
         }
     }
@@ -351,6 +353,8 @@ fn find_inner_join(
         join_type: JoinType::Inner,
         join_constraint: JoinConstraint::On,
         null_equals_null: false,
+        aggr_expr: None,
+        group_expr: None,
     }))
 }
 

datafusion/optimizer/src/eliminate_outer_join.rs

@@ -111,14 +111,8 @@ impl OptimizerRule for EliminateOuterJoin {
                     };
 
                     let new_join = Arc::new(LogicalPlan::Join(Join {
-                        left: join.left,
-                        right: join.right,
                         join_type: new_join_type,
-                        join_constraint: join.join_constraint,
-                        on: join.on.clone(),
-                        filter: join.filter.clone(),
-                        schema: Arc::clone(&join.schema),
-                        null_equals_null: join.null_equals_null,
+                        ..join
                     }));
                     Filter::try_new(filter.predicate, new_join)
                         .map(|f| Transformed::yes(LogicalPlan::Filter(f)))

datafusion/optimizer/src/extract_equijoin_predicate.rs

@@ -76,6 +76,8 @@ impl OptimizerRule for ExtractEquijoinPredicate {
                 join_constraint,
                 schema,
                 null_equals_null,
+                group_expr,
+                aggr_expr,
             }) => {
                 let left_schema = left.schema();
                 let right_schema = right.schema();
@@ -93,6 +95,8 @@ impl OptimizerRule for ExtractEquijoinPredicate {
                         join_constraint,
                         schema,
                         null_equals_null,
+                        group_expr,
+                        aggr_expr,
                     })))
                 } else {
                     Ok(Transformed::no(LogicalPlan::Join(Join {
@@ -104,6 +108,8 @@ impl OptimizerRule for ExtractEquijoinPredicate {
                         join_constraint,
                         schema,
                         null_equals_null,
+                        group_expr,
+                        aggr_expr,
                     })))
                 }
             }