feat: Support PiecewiseMergeJoin to speed up single range predicate joins

datafusion/physical-plan/src/joins/hash_join.rs

@@ -1621,6 +1621,7 @@ impl HashJoinStream {
         let (left_side, right_side) = get_final_indices_from_shared_bitmap(
             build_side.left_data.visited_indices_bitmap(),
             self.join_type,
+            false,
         );
         let empty_right_batch = RecordBatch::new_empty(self.right.schema());
         // use the left and right indices to produce the batch result

datafusion/physical-plan/src/joins/mod.rs

@@ -23,12 +23,14 @@ use datafusion_physical_expr::PhysicalExprRef;
 pub use hash_join::HashJoinExec;
 pub use nested_loop_join::NestedLoopJoinExec;
 use parking_lot::Mutex;
+pub use piecewise_merge_join::PiecewiseMergeJoinExec;
 // Note: SortMergeJoin is not used in plans yet
 pub use sort_merge_join::SortMergeJoinExec;
 pub use symmetric_hash_join::SymmetricHashJoinExec;
 mod cross_join;
 mod hash_join;
 mod nested_loop_join;
+mod piecewise_merge_join;
 mod sort_merge_join;
 mod stream_join_utils;
 mod symmetric_hash_join;
@@ -67,5 +69,5 @@ pub enum StreamJoinPartitionMode {
     SinglePartition,
 }
 
-/// Shared bitmap for visited left-side indices
+/// Shared bitmap for visited indices
 type SharedBitmapBuilder = Mutex<BooleanBufferBuilder>;

datafusion/physical-plan/src/joins/nested_loop_join.rs

@@ -943,8 +943,11 @@ impl<T: BatchTransformer> NestedLoopJoinStream<T> {
             // Only setting up timer, input is exhausted
             let timer = self.join_metrics.join_time.timer();
             // use the global left bitmap to produce the left indices and right indices
-            let (left_side, right_side) =
-                get_final_indices_from_shared_bitmap(visited_left_side, self.join_type);
+            let (left_side, right_side) = get_final_indices_from_shared_bitmap(
+                visited_left_side,
+                self.join_type,
+                false,
+            );
             let empty_right_batch = RecordBatch::new_empty(self.outer_table.schema());
             // use the left and right indices to produce the batch result
             let result = build_batch_from_indices(