[AURON #2539] Support residual conditions in native broadcast hash joins

native-engine/auron-planner/src/planner.rs

@@ -421,11 +421,6 @@ impl PhysicalPlanner {
                     .try_into()
                     .map_err(|_| proto_error("invalid JoinType"))?;
                 let join_filter = self.parse_join_filter(broadcast_join.filter.as_ref())?;
-                if join_filter.is_some() && join_type != JoinType::Inner {
-                    return Err(proto_error(
-                        "broadcast join filter is only supported for inner join",
-                    ));
-                }
 
                 let broadcast_side = protobuf::JoinSide::try_from(broadcast_join.broadcast_side)
                     .expect("invalid BroadcastSide");

native-engine/datafusion-ext-plans/src/broadcast_join_exec.rs

@@ -71,6 +71,8 @@ use crate::{
                 RProbedRightSemiJoiner,
             },
         },
+        filter_read_columns,
+        join_filter::JoinFilter as PreparedJoinFilter,
         join_hash_map::{JoinHashMap, join_data_schema, join_hash_map_schema},
         join_utils::{JoinType, JoinType::*},
     },
@@ -107,9 +109,6 @@ impl BroadcastJoinExec {
         is_null_aware_anti_join: bool,
         join_filter: Option<JoinFilter>,
     ) -> Result<Self> {
-        if join_filter.is_some() && join_type != JoinType::Inner {
-            df_execution_err!("join filter is only supported for inner hash join")?;
-        }
         Ok(Self {
             left,
             right,
@@ -138,6 +137,24 @@ impl BroadcastJoinExec {
         self.broadcast_side
     }
 
+    // Built hash maps append a private table column, absent from join input rows.
+    fn data_schemas(&self) -> (SchemaRef, SchemaRef) {
+        let left = self.left.schema();
+        let right = self.right.schema();
+        (
+            if self.is_built && self.broadcast_side == JoinSide::Left {
+                join_data_schema(&left)
+            } else {
+                left
+            },
+            if self.is_built && self.broadcast_side == JoinSide::Right {
+                join_data_schema(&right)
+            } else {
+                right
+            },
+        )
+    }
+
     fn create_join_params(&self, projection: &[usize]) -> Result<JoinParams> {
         let left_schema = self.left.schema();
         let right_schema = self.right.schema();
@@ -198,13 +215,26 @@ impl BroadcastJoinExec {
         context: Arc<TaskContext>,
         projection: Vec<usize>,
     ) -> Result<SendableRecordBatchStream> {
-        let join_params = self.create_join_params(&projection)?;
+        let mut join_params = self.create_join_params(&projection)?;
         let exec_ctx = ExecutionContext::new(
             context,
             partition,
             join_params.projection.schema.clone(),
             &self.metrics,
         );
+        if let Some(filter) = &self.join_filter {
+            let (left_schema, right_schema) = self.data_schemas();
+            let expr = filter.physical_expr(left_schema.fields().len())?;
+            let (left_read, right_read) = filter_read_columns(&expr, left_schema.fields().len());
+            join_params.residual_filter = Some(Arc::new(PreparedJoinFilter::try_new_remapped(
+                expr,
+                &left_schema,
+                &right_schema,
+                &left_read,
+                &right_read,
+                exec_ctx.register_timer_metric("post_filter_time"),
+            )?));
+        }
         let left = self.left.clone();
         let right = self.right.clone();
         let broadcast_side = self.broadcast_side;
@@ -466,7 +496,28 @@ async fn execute_join_with_smj_fallback(
         join_params.join_filter.clone(),
         vec![SortOptions::default(); join_params.left_keys.len()],
     )?);
-    let mut join_output = smj_exec.execute(exec_ctx.partition_id(), exec_ctx.task_ctx())?;
+    let mut projection = match join_params.join_type {
+        Inner | Left | Right | Full => join_params
+            .projection
+            .left
+            .iter()
+            .copied()
+            .chain(
+                join_params
+                    .projection
+                    .right
+                    .iter()
+                    .map(|&i| i + left_exec.schema().fields().len()),
+            )
+            .collect::<Vec<_>>(),
+        RightSemi | RightAnti => join_params.projection.right.clone(),
+        _ => join_params.projection.left.clone(),
+    };
+    if let Some(pos) = join_params.projection.existence_output {
+        projection.insert(pos, left_exec.schema().fields().len());
+    }
+    let mut join_output =
+        smj_exec.execute_projected(exec_ctx.partition_id(), exec_ctx.task_ctx(), &projection)?;
 
     // send all outputs
     while let Some(batch) = join_output.next().await.transpose()? {