[FEAT] Outer joins for native executor

src/arrow2/src/bitmap/mutable.rs

@@ -325,6 +325,25 @@
     pub(crate) fn bitchunks_exact_mut<T: BitChunk>(&mut self) -> BitChunksExactMut<T> {
         BitChunksExactMut::new(&mut self.buffer, self.length)
     }
+
+    pub fn or(&self, other: &MutableBitmap) -> MutableBitmap {
+        assert_eq!(
+            self.length, other.length,
+            "Bitmaps must have the same length"
+        );
+
+        let new_buffer: Vec<u8> = self
+            .buffer
+            .iter()
+            .zip(other.buffer.iter())
+            .map(|(&a, &b)| a | b) // Apply bitwise OR on each pair of bytes
+            .collect();
+
+        MutableBitmap {
+            buffer: new_buffer,
+            length: self.length,
+        }
+    }
 }
 
 impl From<MutableBitmap> for Bitmap {

src/daft-local-execution/Cargo.toml

@@ -1,4 +1,5 @@
 [dependencies]
+arrow2 = {workspace = true}
 common-daft-config = {path = "../common/daft-config", default-features = false}
 common-display = {path = "../common/display", default-features = false}
 common-error = {path = "../common/error", default-features = false}

src/daft-local-execution/src/intermediate_ops/anti_semi_hash_join_probe.rs

@@ -43,14 +43,14 @@ impl IntermediateOperatorState for AntiSemiProbeState {
 
 pub struct AntiSemiProbeOperator {
     probe_on: Vec<ExprRef>,
-    join_type: JoinType,
+    is_semi: bool,
 }
 
 impl AntiSemiProbeOperator {
-    pub fn new(probe_on: Vec<ExprRef>, join_type: JoinType) -> Self {
+    pub fn new(probe_on: Vec<ExprRef>, join_type: &JoinType) -> Self {
         Self {
             probe_on,
-            join_type,
+            is_semi: *join_type == JoinType::Semi,
         }
     }
 
@@ -76,7 +76,7 @@ impl AntiSemiProbeOperator {
                 let iter = probe_set.probe_exists(&join_keys)?;
 
                 for (probe_row_idx, matched) in iter.enumerate() {
-                    match (self.join_type == JoinType::Semi, matched) {
+                    match (self.is_semi, matched) {
                         (true, true) | (false, false) => {
                             probe_side_growable.extend(probe_side_table_idx, probe_row_idx, 1);
                         }
@@ -120,10 +120,7 @@ impl IntermediateOperator for AntiSemiProbeOperator {
                     .downcast_mut::<AntiSemiProbeState>()
                     .expect("AntiSemiProbeOperator state should be AntiSemiProbeState");
                 let input = input.as_data();
-                let out = match self.join_type {
-                    JoinType::Semi | JoinType::Anti => self.probe_anti_semi(input, state),
-                    _ => unreachable!("Only Semi and Anti joins are supported"),
-                }?;
+                let out = self.probe_anti_semi(input, state)?;
                 Ok(IntermediateOperatorResult::NeedMoreInput(Some(out)))
             }
         }