Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

chore(query): refine hash join bitmap and support fast path for datablock take #13213

Merged
merged 3 commits into from
Oct 12, 2023
Merged

chore(query): refine hash join bitmap and support fast path for datablock take #13213

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e10f8fe
add set_true_validity
Dousir9 Oct 11, 2023
4ee0970
fast path for take
Dousir9 Oct 11, 2023
ded5eb5
Merge branch 'main' into improve_hash_join_bitmap
sundy-li Oct 12, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions src/query/expression/src/kernels/take.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -262,6 +262,15 @@ impl Column {
pub fn take_boolean_types<I>(col: &Bitmap, indices: &[I]) -> Bitmap
where I: common_arrow::arrow::types::Index {
let num_rows = indices.len();
// Fast path: avoid iterating column to generate a new bitmap.
// If this [`Bitmap`] is all true or all false and `num_rows <= bitmap.len()``,
// we can just slice it.
if num_rows <= col.len() && (col.unset_bits() == 0 || col.unset_bits() == col.len()) {
let mut bitmap = col.clone();
bitmap.slice(0, num_rows);
return bitmap;
}

let capacity = num_rows.saturating_add(7) / 8;
let mut builder: Vec<u8> = Vec::with_capacity(capacity);
let mut builder_len = 0;
Expand Down
13 changes: 13 additions & 0 deletions src/query/expression/src/kernels/take_chunks.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -771,6 +771,19 @@ impl Column {

pub fn take_block_vec_boolean_types(col: &[Bitmap], indices: &[RowPtr]) -> Bitmap {
let num_rows = indices.len();
// Fast path: avoid iterating column to generate a new bitmap.
for bitmap in col.iter() {
// If this [`Bitmap`] is all true or all false and `num_rows <= bitmap.len()``,
// we can just slice it.
if num_rows <= bitmap.len()
&& (bitmap.unset_bits() == 0 || bitmap.unset_bits() == bitmap.len())
{
let mut bitmap = bitmap.clone();
bitmap.slice(0, num_rows);
return bitmap;
}
}

let capacity = num_rows.saturating_add(7) / 8;
let mut builder: Vec<u8> = Vec::with_capacity(capacity);
let mut builder_len = 0;
Expand Down
32 changes: 10 additions & 22 deletions src/query/service/src/pipelines/processors/transforms/hash_join/common.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,6 @@ use common_expression::DataBlock;
use common_expression::Evaluator;
use common_expression::Expr;
use common_expression::FunctionContext;
use common_expression::Scalar;
use common_expression::Value;
use common_functions::BUILTIN_FUNCTIONS;
use common_sql::executor::cast_expr_to_non_null_boolean;
Expand Down Expand Up @@ -187,32 +186,21 @@ impl HashJoinState {
}
}

pub(crate) fn set_validity(column: &BlockEntry, num_rows: usize, validity: &Bitmap) -> BlockEntry {
pub(crate) fn set_true_validity(
column: &BlockEntry,
num_rows: usize,
true_validity: &Bitmap,
) -> BlockEntry {
let (value, data_type) = (&column.value, &column.data_type);
let col = value.convert_to_full_column(data_type, num_rows);

if matches!(col, Column::Null { .. }) {
if matches!(col, Column::Null { .. }) || col.as_nullable().is_some() {
column.clone()
} else if let Some(col) = col.as_nullable() {
if col.len() == 0 {
return BlockEntry::new(data_type.clone(), Value::Scalar(Scalar::Null));
}
// It's possible validity is longer than col.
let diff_len = validity.len() - col.validity.len();
let mut new_validity = MutableBitmap::with_capacity(validity.len());
for (b1, b2) in validity.iter().zip(col.validity.iter()) {
new_validity.push(b1 & b2);
}
new_validity.extend_constant(diff_len, false);
let col = Column::Nullable(Box::new(NullableColumn {
column: col.column.clone(),
validity: new_validity.into(),
}));
BlockEntry::new(data_type.clone(), Value::Column(col))
} else {
let mut validity = true_validity.clone();
validity.slice(0, num_rows);
let col = Column::Nullable(Box::new(NullableColumn {
column: col.clone(),
validity: validity.clone(),
column: col,
validity,
}));
BlockEntry::new(data_type.wrap_nullable(), Value::Column(col))
}
Expand Down
5 changes: 2 additions & 3 deletions src/query/service/src/pipelines/processors/transforms/hash_join/hash_join_build_state.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,7 +54,7 @@ use log::info;
use parking_lot::Mutex;
use parking_lot::RwLock;

use crate::pipelines::processors::transforms::hash_join::common::set_validity;
use crate::pipelines::processors::transforms::hash_join::common::set_true_validity;
use crate::pipelines::processors::transforms::hash_join::desc::MARKER_KIND_FALSE;
use crate::pipelines::processors::transforms::hash_join::hash_join_state::FixedKeyHashJoinHashTable;
use crate::pipelines::processors::transforms::hash_join::hash_join_state::HashJoinHashTable;
Expand Down Expand Up @@ -171,11 +171,10 @@ impl HashJoinBuildState {
let mut validity = MutableBitmap::new();
validity.extend_constant(data_block.num_rows(), true);
let validity: Bitmap = validity.into();

let nullable_columns = data_block
.columns()
.iter()
.map(|c| set_validity(c, validity.len(), &validity))
.map(|c| set_true_validity(c, validity.len(), &validity))
.collect::<Vec<_>>();
data_block = DataBlock::new(nullable_columns, data_block.num_rows());
}
Expand Down
31 changes: 7 additions & 24 deletions src/query/service/src/pipelines/processors/transforms/hash_join/hash_join_probe_state.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@ use std::sync::atomic::AtomicUsize;
use std::sync::atomic::Ordering;
use std::sync::Arc;

use common_arrow::arrow::bitmap::Bitmap;
use common_arrow::arrow::bitmap::MutableBitmap;
use common_base::base::tokio::sync::Barrier;
use common_catalog::table_context::TableContext;
Expand Down Expand Up @@ -46,7 +45,7 @@ use parking_lot::Mutex;
use parking_lot::RwLock;

use super::ProbeState;
use crate::pipelines::processors::transforms::hash_join::common::set_validity;
use crate::pipelines::processors::transforms::hash_join::common::set_true_validity;
use crate::pipelines::processors::transforms::hash_join::desc::MARKER_KIND_FALSE;
use crate::pipelines::processors::transforms::hash_join::desc::MARKER_KIND_NULL;
use crate::pipelines::processors::transforms::hash_join::desc::MARKER_KIND_TRUE;
Expand Down Expand Up @@ -169,12 +168,7 @@ impl HashJoinProbeState {
let nullable_columns = input
.columns()
.iter()
.map(|c| {
let mut validity = MutableBitmap::new();
validity.extend_constant(input.num_rows(), true);
let validity: Bitmap = validity.into();
set_validity(c, validity.len(), &validity)
})
.map(|c| set_true_validity(c, input.num_rows(), &probe_state.true_validity))
.collect::<Vec<_>>();
input = DataBlock::new(nullable_columns, input.num_rows());
}
Expand Down Expand Up @@ -456,22 +450,11 @@ impl HashJoinProbeState {

if self.hash_join_state.hash_join_desc.join_type == JoinType::Full {
let num_rows = unmatched_build_block.num_rows();
let nullable_unmatched_build_columns = if num_rows == max_block_size {
unmatched_build_block
.columns()
.iter()
.map(|c| set_validity(c, num_rows, true_validity))
.collect::<Vec<_>>()
} else {
let mut validity = MutableBitmap::new();
validity.extend_constant(num_rows, true);
let validity: Bitmap = validity.into();
unmatched_build_block
.columns()
.iter()
.map(|c| set_validity(c, num_rows, &validity))
.collect::<Vec<_>>()
};
let nullable_unmatched_build_columns = unmatched_build_block
.columns()
.iter()
.map(|c| set_true_validity(c, num_rows, true_validity))
.collect::<Vec<_>>();
unmatched_build_block =
DataBlock::new(nullable_unmatched_build_columns, num_rows);
};
Expand Down
34 changes: 11 additions & 23 deletions src/query/service/src/pipelines/processors/transforms/hash_join/probe_join/inner_join.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,8 +15,6 @@
use std::iter::TrustedLen;
use std::sync::atomic::Ordering;

use common_arrow::arrow::bitmap::Bitmap;
use common_arrow::arrow::bitmap::MutableBitmap;
use common_exception::ErrorCode;
use common_exception::Result;
use common_expression::types::BooleanType;
Expand All @@ -27,7 +25,7 @@ use common_functions::BUILTIN_FUNCTIONS;
use common_hashtable::HashJoinHashtableLike;
use common_sql::executor::cast_expr_to_non_null_boolean;

use crate::pipelines::processors::transforms::hash_join::common::set_validity;
use crate::pipelines::processors::transforms::hash_join::common::set_true_validity;
use crate::pipelines::processors::transforms::hash_join::HashJoinProbeState;
use crate::pipelines::processors::transforms::hash_join::ProbeState;

Expand Down Expand Up @@ -120,16 +118,11 @@ impl HashJoinProbeState {
(true, false) => {
result_block.get_by_offset(*index).clone().remove_nullable()
}
(false, true) => {
let mut validity = MutableBitmap::new();
validity.extend_constant(result_block.num_rows(), true);
let validity: Bitmap = validity.into();
set_validity(
result_block.get_by_offset(*index),
validity.len(),
&validity,
)
}
(false, true) => set_true_validity(
result_block.get_by_offset(*index),
result_block.num_rows(),
&probe_state.true_validity,
),
};
result_block.add_column(entry);
}
Expand Down Expand Up @@ -195,16 +188,11 @@ impl HashJoinProbeState {
(true, false) => {
result_block.get_by_offset(*index).clone().remove_nullable()
}
(false, true) => {
let mut validity = MutableBitmap::new();
validity.extend_constant(result_block.num_rows(), true);
let validity: Bitmap = validity.into();
set_validity(
result_block.get_by_offset(*index),
validity.len(),
&validity,
)
}
(false, true) => set_true_validity(
result_block.get_by_offset(*index),
result_block.num_rows(),
&probe_state.true_validity,
),
};
result_block.add_column(entry);
}
Expand Down
Loading
Loading