[WIP] Implement INTERSECT & INTERSECT DISTINCT

ballista/rust/core/proto/ballista.proto

@@ -435,6 +435,7 @@ message JoinNode {
   JoinConstraint join_constraint = 4;
   repeated Column left_join_column = 5;
   repeated Column right_join_column = 6;
+  bool null_equal_null = 7;
 }
 
 message CrossJoinNode {
@@ -649,6 +650,7 @@ message HashJoinExecNode {
   repeated JoinOn on = 3;
   JoinType join_type = 4;
   PartitionMode partition_mode = 6;
+  bool null_equal_null = 7;
 }
 
 message CrossJoinExecNode {

ballista/rust/core/src/serde/logical_plan/to_proto.rs

@@ -850,6 +850,7 @@ impl TryInto<protobuf::LogicalPlanNode> for &LogicalPlan {
                 on,
                 join_type,
                 join_constraint,
+                null_equal_null,
                 ..
             } => {
                 let left: protobuf::LogicalPlanNode = left.as_ref().try_into()?;
@@ -868,6 +869,7 @@ impl TryInto<protobuf::LogicalPlanNode> for &LogicalPlan {
                             join_constraint: join_constraint.into(),
                             left_join_column,
                             right_join_column,
+                            null_equal_null: *null_equal_null,
                         },
                     ))),
                 })

ballista/rust/core/src/serde/physical_plan/from_proto.rs

@@ -365,6 +365,7 @@ impl TryInto<Arc<dyn ExecutionPlan>> for &protobuf::PhysicalPlanNode {
                     on,
                     &join_type.into(),
                     partition_mode,
+                    &hashjoin.null_equal_null,
                 )?))
             }
             PhysicalPlanType::CrossJoin(crossjoin) => {

ballista/rust/core/src/serde/physical_plan/mod.rs

@@ -107,6 +107,7 @@ mod roundtrip_tests {
                     on.clone(),
                     join_type,
                     *partition_mode,
+                    &false,
                 )?))?;
             }
         }

ballista/rust/core/src/serde/physical_plan/to_proto.rs

@@ -162,6 +162,7 @@ impl TryInto<protobuf::PhysicalPlanNode> for Arc<dyn ExecutionPlan> {
                         on,
                         join_type: join_type.into(),
                         partition_mode: partition_mode.into(),
+                        null_equal_null: *exec.null_equal_null(),
                     },
                 ))),
             })

datafusion/src/logical_plan/builder.rs

@@ -492,6 +492,18 @@ impl LogicalPlanBuilder {
         right: &LogicalPlan,
         join_type: JoinType,
         join_keys: (Vec<impl Into<Column>>, Vec<impl Into<Column>>),
+    ) -> Result<Self> {
+        self.join_detailed(right, join_type, join_keys, false)
+    }
+
+    /// Apply a join with on constraint and specified null equality
+    /// If null_equal_null is true then null == null, else null != null
+    pub fn join_detailed(
+        &self,
+        right: &LogicalPlan,
+        join_type: JoinType,
+        join_keys: (Vec<impl Into<Column>>, Vec<impl Into<Column>>),
+        null_equal_null: bool,
     ) -> Result<Self> {
         if join_keys.0.len() != join_keys.1.len() {
             return Err(DataFusionError::Plan(
@@ -580,6 +592,7 @@ impl LogicalPlanBuilder {
             join_type,
             join_constraint: JoinConstraint::On,
             schema: DFSchemaRef::new(join_schema),
+            null_equal_null,
         }))
     }
 
@@ -611,6 +624,7 @@ impl LogicalPlanBuilder {
             join_type,
             join_constraint: JoinConstraint::Using,
             schema: DFSchemaRef::new(join_schema),
+            null_equal_null: false,
         }))
     }
 

datafusion/src/logical_plan/plan.rs

@@ -135,6 +135,8 @@ pub enum LogicalPlan {
         join_constraint: JoinConstraint,
         /// The output schema, containing fields from the left and right inputs
         schema: DFSchemaRef,
+        /// If null_equal_null is true, null == null else null != null
+        null_equal_null: bool,
     },
     /// Apply Cross Join to two logical plans
     CrossJoin {

datafusion/src/optimizer/projection_push_down.rs

@@ -195,6 +195,7 @@ fn optimize_plan(
             on,
             join_type,
             join_constraint,
+            null_equal_null,
             ..
         } => {
             for (l, r) in on {
@@ -231,6 +232,7 @@ fn optimize_plan(
                 join_constraint: *join_constraint,
                 on: on.clone(),
                 schema: DFSchemaRef::new(schema),
+                null_equal_null: *null_equal_null,
             })
         }
         LogicalPlan::Window {

datafusion/src/optimizer/utils.rs

@@ -198,6 +198,7 @@ pub fn from_plan(
             join_type,
             join_constraint,
             on,
+            null_equal_null,
             ..
         } => {
             let schema =
@@ -209,6 +210,7 @@ pub fn from_plan(
                 join_constraint: *join_constraint,
                 on: on.clone(),
                 schema: DFSchemaRef::new(schema),
+                null_equal_null: *null_equal_null,
             })
         }
         LogicalPlan::CrossJoin { .. } => {

datafusion/src/physical_optimizer/hash_build_probe_order.rs

@@ -123,6 +123,7 @@ impl PhysicalOptimizerRule for HashBuildProbeOrder {
                         .collect(),
                     &swap_join_type(*hash_join.join_type()),
                     *hash_join.partition_mode(),
+                    hash_join.null_equal_null(),
                 )?;
                 let proj = ProjectionExec::try_new(
                     swap_reverting_projection(&*left.schema(), &*right.schema()),
@@ -195,6 +196,7 @@ mod tests {
             )],
             &JoinType::Left,
             PartitionMode::CollectLeft,
+            &false,
         )
         .unwrap();
 
@@ -238,6 +240,7 @@ mod tests {
             )],
             &JoinType::Left,
             PartitionMode::CollectLeft,
+            &false,
         )
         .unwrap();