feat(planner): support independent right join

Author: xudong963

src/query/service/src/pipelines/processors/transforms/hash_join/join_hash_table.rs

@@ -130,6 +130,7 @@ pub struct JoinHashTable {
     pub(crate) row_space: RowSpace,
     pub(crate) hash_join_desc: HashJoinDesc,
     pub(crate) row_ptrs: RwLock<Vec<RowPtr>>,
+    pub(crate) probe_schema: DataSchemaRef,
     finished_notify: Arc<Notify>,
 }
 
@@ -138,6 +139,7 @@ impl JoinHashTable {
         ctx: Arc<QueryContext>,
         build_keys: &[PhysicalScalar],
         build_schema: DataSchemaRef,
+        probe_schema: DataSchemaRef,
         hash_join_desc: HashJoinDesc,
     ) -> Result<Arc<JoinHashTable>> {
         let hash_key_types: Vec<DataTypeImpl> =
@@ -151,6 +153,7 @@ impl JoinHashTable {
                     hash_method: HashMethodSerializer::default(),
                 }),
                 build_schema,
+                probe_schema,
                 hash_join_desc,
             )?),
             HashMethodKind::KeysU8(hash_method) => Arc::new(JoinHashTable::try_create(
@@ -160,6 +163,7 @@ impl JoinHashTable {
                     hash_method,
                 }),
                 build_schema,
+                probe_schema,
                 hash_join_desc,
             )?),
             HashMethodKind::KeysU16(hash_method) => Arc::new(JoinHashTable::try_create(
@@ -169,6 +173,7 @@ impl JoinHashTable {
                     hash_method,
                 }),
                 build_schema,
+                probe_schema,
                 hash_join_desc,
             )?),
             HashMethodKind::KeysU32(hash_method) => Arc::new(JoinHashTable::try_create(
@@ -178,6 +183,7 @@ impl JoinHashTable {
                     hash_method,
                 }),
                 build_schema,
+                probe_schema,
                 hash_join_desc,
             )?),
             HashMethodKind::KeysU64(hash_method) => Arc::new(JoinHashTable::try_create(
@@ -187,6 +193,7 @@ impl JoinHashTable {
                     hash_method,
                 }),
                 build_schema,
+                probe_schema,
                 hash_join_desc,
             )?),
             HashMethodKind::KeysU128(hash_method) => Arc::new(JoinHashTable::try_create(
@@ -196,6 +203,7 @@ impl JoinHashTable {
                     hash_method,
                 }),
                 build_schema,
+                probe_schema,
                 hash_join_desc,
             )?),
             HashMethodKind::KeysU256(hash_method) => Arc::new(JoinHashTable::try_create(
@@ -205,6 +213,7 @@ impl JoinHashTable {
                     hash_method,
                 }),
                 build_schema,
+                probe_schema,
                 hash_join_desc,
             )?),
             HashMethodKind::KeysU512(hash_method) => Arc::new(JoinHashTable::try_create(
@@ -214,6 +223,7 @@ impl JoinHashTable {
                     hash_method,
                 }),
                 build_schema,
+                probe_schema,
                 hash_join_desc,
             )?),
         })
@@ -223,6 +233,7 @@ impl JoinHashTable {
         ctx: Arc<QueryContext>,
         hash_table: HashTable,
         mut build_data_schema: DataSchemaRef,
+        mut probe_data_schema: DataSchemaRef,
         hash_join_desc: HashJoinDesc,
     ) -> Result<Self> {
         if hash_join_desc.join_type == JoinType::Left
@@ -237,6 +248,16 @@ impl JoinHashTable {
             }
             build_data_schema = DataSchemaRefExt::create(nullable_field);
         };
+        if hash_join_desc.join_type == JoinType::Right {
+            let mut nullable_field = Vec::with_capacity(probe_data_schema.fields().len());
+            for field in probe_data_schema.fields().iter() {
+                nullable_field.push(DataField::new_nullable(
+                    field.name(),
+                    field.data_type().clone(),
+                ));
+            }
+            probe_data_schema = DataSchemaRefExt::create(nullable_field);
+        }
         Ok(Self {
             row_space: RowSpace::new(build_data_schema),
             ref_count: Mutex::new(0),
@@ -245,6 +266,7 @@ impl JoinHashTable {
             ctx,
             hash_table: RwLock::new(hash_table),
             row_ptrs: RwLock::new(vec![]),
+            probe_schema: probe_data_schema,
             finished_notify: Arc::new(Notify::new()),
         })
     }
@@ -429,7 +451,8 @@ impl HashJoinState for JoinHashTable {
             | JoinType::Anti
             | JoinType::Left
             | Mark
-            | JoinType::Single => self.probe_join(input, probe_state),
+            | JoinType::Single
+            | JoinType::Right => self.probe_join(input, probe_state),
             JoinType::Cross => self.probe_cross_join(input, probe_state),
             _ => unimplemented!("{} is unimplemented", self.hash_join_desc.join_type),
         }

src/query/service/src/pipelines/processors/transforms/hash_join/result_blocks.rs

@@ -158,6 +158,17 @@ impl JoinHashTable {
                     return Ok(vec![result]);
                 }
             }
+            JoinType::Right => {
+                if self.hash_join_desc.other_predicate.is_none() {
+                    let result =
+                        self.right_join::<false, _, _>(hash_table, probe_state, keys_iter, input)?;
+                    return Ok(vec![result]);
+                } else {
+                    let result =
+                        self.right_join::<true, _, _>(hash_table, probe_state, keys_iter, input)?;
+                    return Ok(vec![result]);
+                }
+            }
             Mark => {
                 results.push(DataBlock::empty());
                 // Three cases will produce Mark join:
@@ -520,6 +531,71 @@ impl JoinHashTable {
         DataBlock::filter_block(merged_block, &predicate)
     }
 
+    fn right_join<const WITH_OTHER_CONJUNCT: bool, Key, IT>(
+        &self,
+        hash_table: &HashMap<Key, Vec<RowPtr>>,
+        probe_state: &mut ProbeState,
+        keys_iter: IT,
+        input: &DataBlock,
+    ) -> Result<DataBlock>
+    where
+        Key: HashTableKeyable + Clone + 'static,
+        IT: Iterator<Item = Key> + TrustedLen,
+    {
+        let build_indexes = &mut probe_state.build_indexs;
+        let probe_indexes = &mut probe_state.probe_indexs;
+        let valids = &probe_state.valids;
+        let mut validity = MutableBitmap::new();
+        for (i, key) in keys_iter.enumerate() {
+            let probe_result_ptr = Self::probe_key(hash_table, key, valids, i);
+            if let Some(v) = probe_result_ptr {
+                let probe_result_ptrs = v.get_value();
+                build_indexes.extend_from_slice(probe_result_ptrs);
+                probe_indexes.extend(std::iter::repeat(i as u32).take(probe_result_ptrs.len()));
+                validity.extend_constant(probe_result_ptrs.len(), true);
+            }
+        }
+
+        // For right join, build side will always appear in the joined table
+        // Find the unmatched rows in build side
+        let mut unmatched_build_indexes = vec![];
+        for kv in hash_table.iter() {
+            for v in kv.get_value() {
+                if !build_indexes.contains(v) {
+                    unmatched_build_indexes.push(v.clone());
+                }
+            }
+        }
+        build_indexes.extend_from_slice(&unmatched_build_indexes);
+        let build_block = self.row_space.gather(build_indexes)?;
+        let probe_block = DataBlock::block_take_by_indices(input, probe_indexes)?;
+        let validity: Bitmap = validity.into();
+        let nullable_columns = probe_block
+            .columns()
+            .iter()
+            .map(|c| Self::set_validity(c, &validity))
+            .collect::<Result<Vec<_>>>()?;
+        let mut nullable_probe_block =
+            DataBlock::create(self.probe_schema.clone(), nullable_columns);
+        // Create null block for unmatched rows in probe side
+        let null_probe_block = DataBlock::create(
+            self.probe_schema.clone(),
+            nullable_probe_block
+                .columns()
+                .iter()
+                .map(|c| {
+                    c.data_type()
+                        .create_constant_column(&DataValue::Null, unmatched_build_indexes.len())
+                })
+                .collect::<Result<Vec<_>>>()?,
+        );
+
+        nullable_probe_block =
+            DataBlock::concat_blocks(&vec![nullable_probe_block, null_probe_block])?;
+        let merged_block = self.merge_eq_block(&nullable_probe_block, &build_block)?;
+        Ok(merged_block)
+    }
+
     // modify the bm by the value row_state
     // keep the index of the first positive state
     // bitmap: [1, 1, 1] with row_state [0, 0], probe_index: [0, 0, 0] (repeat the first element 3 times)

src/query/service/src/sql/executor/physical_plan.rs

@@ -212,6 +212,17 @@ impl HashJoin {
                 }
             }
 
+            JoinType::Right => {
+                fields.clear();
+                fields = self.build.output_schema()?.fields().clone();
+                for field in self.probe.output_schema()?.fields() {
+                    fields.push(DataField::new(
+                        field.name().as_str(),
+                        wrap_nullable(field.data_type()),
+                    ));
+                }
+            }
+
             JoinType::Semi | JoinType::Anti => {
                 // Do nothing
             }

src/query/service/src/sql/executor/pipeline_builder.rs

@@ -145,6 +145,7 @@ impl PipelineBuilder {
             self.ctx.clone(),
             &join.build_keys,
             join.build.output_schema()?,
+            join.probe.output_schema()?,
             HashJoinDesc::create(join)?,
         )
     }

src/query/service/src/sql/planner/binder/join.rs

@@ -71,15 +71,15 @@ impl<'a> Binder {
                 }
             }
             JoinOperator::RightOuter => {
-                for column in right_context.all_column_bindings().iter() {
-                    bind_context.add_column_binding(column.clone());
-                }
-
                 for column in left_context.all_column_bindings() {
                     let mut nullable_column = column.clone();
                     nullable_column.data_type = Box::new(wrap_nullable(&column.data_type));
                     bind_context.add_column_binding(nullable_column);
                 }
+
+                for column in right_context.all_column_bindings().iter() {
+                    bind_context.add_column_binding(column.clone());
+                }
             }
             _ => {
                 for column in left_context.all_column_bindings() {
@@ -145,12 +145,12 @@ impl<'a> Binder {
                 right_child,
             ),
             JoinOperator::RightOuter => self.bind_join_with_type(
-                JoinType::Left,
-                right_join_conditions,
+                JoinType::Right,
                 left_join_conditions,
+                right_join_conditions,
                 other_conditions,
-                right_child,
                 left_child,
+                right_child,
             ),
             JoinOperator::FullOuter => self.bind_join_with_type(
                 JoinType::Full,

tests/logictest/suites/base/15_query/join.test

@@ -129,12 +129,12 @@ insert into t2 values(1, 4), (2, 3), (6, 8);
 
 
 statement query IIII
-select * from t1 right join t2 on t1.a = t2.c;
+select * from t1 right join t2 on t1.a = t2.c order by t2.c;
 
 ----
-1 4 1 2
-2 3 NULL NULL
-6 8 NULL NULL
+1 2 1 4
+NULL NULL 2 3
+NULL NULL 6 8
 
 
 statement query IIII
@@ -322,105 +322,6 @@ select t.number from numbers(10000) as t inner join numbers(1000) as t1 on t.num
 statement ok
 drop table t;
 
-statement ok
-drop table t1;
-
-
-statement ok
-drop table t2;
-
-statement ok
-create table t1(a int, b int);
-
-
-statement ok
-create table t2(c int, d int);
-
-
-statement ok
-insert into t1 values(1, 2), (3 ,4), (7, 8);
-
-
-statement ok
-insert into t2 values(1, 4), (2, 3), (6, 8);
-
-
-statement query IIII
-select * from t1 right join t2 on t1.a = t2.c;
-
-----
-1 4 1 2
-2 3 NULL NULL
-6 8 NULL NULL
-
-
-statement query IIII
-select * from t1 left join t2 on t1.a = t2.c;
-
-----
-1 2 1 4
-3 4 NULL NULL
-7 8 NULL NULL
-
-
-statement query IIII
-select * from t1 left outer join t2 on t1.a = t2.c and t1.a > 3 order by a,b,c,d;
-
-----
-1 2 NULL NULL
-3 4 NULL NULL
-7 8 NULL NULL
-
-
-statement query IIII
-select * from t1 left outer join t2 on t1.a = t2.c and t2.c > 4 order by a,b,c,d;
-
-----
-1 2 NULL NULL
-3 4 NULL NULL
-7 8 NULL NULL
-
-
-statement query IIII
-select * from t1 left outer join t2 on t2.c > 4 and t1.a > 3 order by a,b,c,d;
-
-----
-1 2 NULL NULL
-3 4 NULL NULL
-7 8 6 8
-
-
-statement query IIII
-select * from t1 left outer join t2 on t1.a > 3 order by a,b,c,d;
-
-----
-1 2 NULL NULL
-3 4 NULL NULL
-7 8 1 4
-7 8 2 3
-7 8 6 8
-
-
-statement query IIII
-select * from t1 left outer join t2 on t2.c > 4 order by a,b,c,d;
-
-----
-1 2 6 8
-3 4 6 8
-7 8 6 8
-
-
-statement query IIII
-select * from t1 left outer join t2 on t1.a > t2.c order by a,b,c,d;
-
-----
-1 2 NULL NULL
-3 4 1 4
-3 4 2 3
-7 8 1 4
-7 8 2 3
-7 8 6 8
-
 statement ok
 CREATE TABLE t3(c0 BIGINT NULL, c1 DOUBLE NULL);