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

fix(planner): InSubquery returns error result #6641

Merged
merged 4 commits into from
Jul 15, 2022
Merged

fix(planner): InSubquery returns error result #6641

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
e67ceb0
fix(planner): InSubquery returns error result
xudong963 Jul 15, 2022
928303d
fix and add test
xudong963 Jul 15, 2022
8b7e784
fix clippy
xudong963 Jul 15, 2022
f3728b3
Merge branch 'main' into fix_mark
mergify[bot] Jul 15, 2022
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
220 changes: 93 additions & 127 deletions query/src/pipelines/processors/transforms/hash_join/join_hash_table.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,15 +108,14 @@ pub enum HashTable {
KeyU512HashTable(KeyU512HashTable),
}

#[derive(Clone, Eq, PartialEq, Debug)]
#[derive(Clone, Copy, Eq, PartialEq, Debug)]
pub enum MarkerKind {
True,
False,
Null,
}

pub struct MarkJoinDesc {
pub(crate) marker: RwLock<Vec<MarkerKind>>,
pub(crate) marker_index: Option<IndexType>,
pub(crate) has_null: RwLock<bool>,
}
Expand All @@ -138,6 +137,7 @@ pub struct JoinHashTable {
pub(crate) hash_table: RwLock<HashTable>,
pub(crate) row_space: RowSpace,
pub(crate) hash_join_desc: HashJoinDesc,
pub(crate) row_ptrs: RwLock<Vec<RowPtr>>,
}

impl JoinHashTable {
Expand Down Expand Up @@ -167,7 +167,6 @@ impl JoinHashTable {
.map(Evaluator::eval_physical_scalar)
.transpose()?,
marker_join_desc: MarkJoinDesc {
marker: RwLock::new(vec![]),
marker_index,
has_null: RwLock::new(false),
},
Expand Down Expand Up @@ -271,6 +270,7 @@ impl JoinHashTable {
hash_join_desc,
ctx,
hash_table: RwLock::new(hash_table),
row_ptrs: RwLock::new(vec![]),
})
}

Expand Down Expand Up @@ -428,27 +428,6 @@ impl JoinHashTable {
}
}

macro_rules! insert_key {
($table: expr, $method: expr, $chunk: expr, $columns: expr, $chunk_index: expr, ) => {{
let keys_state = $method.build_keys_state(&$columns, $chunk.num_rows())?;
let build_keys_iter = $method.build_keys_iter(&keys_state)?;

for (row_index, key) in build_keys_iter.enumerate().take($chunk.num_rows()) {
let mut inserted = true;
let ptr = RowPtr {
chunk_index: $chunk_index as u32,
row_index: row_index as u32,
};
let entity = $table.insert_key(&key, &mut inserted);
if inserted {
entity.set_value(vec![ptr]);
} else {
entity.get_mut_value().push(ptr);
}
}
}};
}

impl HashJoinState for JoinHashTable {
fn build(&self, input: DataBlock) -> Result<()> {
let func_ctx = self.ctx.try_get_function_context()?;
Expand All @@ -458,20 +437,7 @@ impl HashJoinState for JoinHashTable {
.iter()
.map(|expr| Ok(expr.eval(&func_ctx, &input)?.vector().clone()))
.collect::<Result<Vec<ColumnRef>>>()?;

match &*self.hash_table.read() {
HashTable::SerializerHashTable(table) => {
let mut build_cols_ref = Vec::with_capacity(build_cols.len());
for build_col in build_cols.iter() {
build_cols_ref.push(build_col);
}
let keys_state = table
.hash_method
.build_keys_state(&build_cols_ref, input.num_rows())?;
self.row_space.push_keys_state(input, keys_state)
}
_ => self.row_space.push_cols(input, build_cols),
}
self.row_space.push_cols(input, build_cols)
}

fn probe(&self, input: &DataBlock, probe_state: &mut ProbeState) -> Result<Vec<DataBlock>> {
Expand Down Expand Up @@ -508,123 +474,166 @@ impl HashJoinState for JoinHashTable {
}

fn finish(&self) -> Result<()> {
let chunks = self.row_space.chunks.read().unwrap();
let mut marker = self.hash_join_desc.marker_join_desc.marker.write();
if self.hash_join_desc.join_type == Mark && self.hash_join_desc.other_predicate.is_some() {
let row_nums = chunks.iter().fold(0, |acc, chunk| acc + chunk.num_rows());
marker.append(&mut vec![MarkerKind::False; row_nums])
macro_rules! insert_key {
($table: expr, $markers: expr, $method: expr, $chunk: expr, $columns: expr, $chunk_index: expr, ) => {{
let keys_state = $method.build_keys_state(&$columns, $chunk.num_rows())?;
let build_keys_iter = $method.build_keys_iter(&keys_state)?;

for (row_index, key) in build_keys_iter.enumerate().take($chunk.num_rows()) {
let mut inserted = true;
let ptr = RowPtr {
chunk_index: $chunk_index as u32,
row_index: row_index as u32,
marker: $markers[row_index],
};
{
let mut self_row_ptrs = self.row_ptrs.write();
self_row_ptrs.push(ptr.clone());
}
let entity = $table.insert_key(&key, &mut inserted);
if inserted {
entity.set_value(vec![ptr]);
} else {
entity.get_mut_value().push(ptr);
}
}
}};
}

let mut chunks = self.row_space.chunks.write().unwrap();
for chunk_index in 0..chunks.len() {
let chunk = &chunks[chunk_index];
let mut columns = vec![];
if let Some(cols) = chunk.cols.as_ref() {
columns = Vec::with_capacity(cols.len());
for col in cols.iter() {
if self.hash_join_desc.join_type == JoinType::Mark {
assert_eq!(cols.len(), 1);
for row_idx in 0..col.len() {
if col.get(row_idx) == DataValue::Null {
marker.push(MarkerKind::Null);
} else {
marker.push(MarkerKind::False);
}
let chunk = &mut chunks[chunk_index];
let mut columns = Vec::with_capacity(chunk.cols.len());
let mut markers = if self.hash_join_desc.join_type == Mark
&& self.hash_join_desc.other_predicate.is_some()
{
vec![Some(MarkerKind::False); chunk.num_rows()]
} else {
vec![None; chunk.num_rows()]
};
for col in chunk.cols.iter() {
if self.hash_join_desc.join_type == Mark {
assert_eq!(col.len(), markers.len());
for (row_index, marker) in markers.iter_mut().enumerate() {
if col.get(row_index) == DataValue::Null {
*marker = Some(MarkerKind::Null);
} else {
*marker = Some(MarkerKind::False);
}
}
columns.push(col);
}
columns.push(col);
}
match (*self.hash_table.write()).borrow_mut() {
HashTable::SerializerHashTable(table) => {
if let Some(keys_state) = chunk.keys_state.as_ref() {
let build_keys_iter = table.hash_method.build_keys_iter(keys_state)?;

for (row_index, key) in build_keys_iter.enumerate().take(chunk.num_rows()) {
let mut inserted = true;
let ptr = RowPtr {
chunk_index: chunk_index as u32,
row_index: row_index as u32,
};
let keys_ref = KeysRef::create(key.as_ptr() as usize, key.len());
let entity = table.hash_table.insert_key(&keys_ref, &mut inserted);
if inserted {
entity.set_value(vec![ptr]);
} else {
entity.get_mut_value().push(ptr);
}
let mut build_cols_ref = Vec::with_capacity(chunk.cols.len());
for build_col in chunk.cols.iter() {
build_cols_ref.push(build_col);
}
let keys_state = table
.hash_method
.build_keys_state(&build_cols_ref, chunk.num_rows())?;
chunk.keys_state = Some(keys_state);
let build_keys_iter = table
.hash_method
.build_keys_iter(chunk.keys_state.as_ref().unwrap())?;
for (row_index, key) in build_keys_iter.enumerate().take(chunk.num_rows()) {
let mut inserted = true;
let ptr = RowPtr {
chunk_index: chunk_index as u32,
row_index: row_index as u32,
marker: markers[row_index],
};
{
let mut self_row_ptrs = self.row_ptrs.write();
self_row_ptrs.push(ptr);
}
let keys_ref = KeysRef::create(key.as_ptr() as usize, key.len());
let entity = table.hash_table.insert_key(&keys_ref, &mut inserted);
if inserted {
entity.set_value(vec![ptr]);
} else {
entity.get_mut_value().push(ptr);
}
}
}
HashTable::KeyU8HashTable(table) => insert_key! {
&mut table.hash_table,
&markers,
&table.hash_method,
chunk,
columns,
chunk_index,
},
HashTable::KeyU16HashTable(table) => insert_key! {
&mut table.hash_table,
&markers,
&table.hash_method,
chunk,
columns,
chunk_index,
},
HashTable::KeyU32HashTable(table) => insert_key! {
&mut table.hash_table,
&markers,
&table.hash_method,
chunk,
columns,
chunk_index,
},
HashTable::KeyU64HashTable(table) => insert_key! {
&mut table.hash_table,
&markers,
&table.hash_method,
chunk,
columns,
chunk_index,
},
HashTable::KeyU128HashTable(table) => insert_key! {
&mut table.hash_table,
&markers,
&table.hash_method,
chunk,
columns,
chunk_index,
},
HashTable::KeyU256HashTable(table) => insert_key! {
&mut table.hash_table,
&markers,
&table.hash_method,
chunk,
columns,
chunk_index,
},
HashTable::KeyU512HashTable(table) => insert_key! {
&mut table.hash_table,
&markers,
&table.hash_method,
chunk,
columns,
chunk_index,
},
}
}

Ok(())
}

fn mark_join_blocks(&self) -> Result<Vec<DataBlock>> {
let hash_table = self.hash_table.read();
let mut marker = self.hash_join_desc.marker_join_desc.marker.write();
let mut row_ptrs = self.row_ptrs.write();
let has_null = self.hash_join_desc.marker_join_desc.has_null.read();
let mut validity = MutableBitmap::with_capacity(marker.len());
let mut boolean_bit_map = MutableBitmap::with_capacity(marker.len());
for m in marker.iter_mut() {
if m == &mut MarkerKind::False && *has_null {
*m = MarkerKind::Null;
let mut validity = MutableBitmap::with_capacity(row_ptrs.len());
let mut boolean_bit_map = MutableBitmap::with_capacity(row_ptrs.len());
for row_ptr in row_ptrs.iter_mut() {
let marker = row_ptr.marker.unwrap();
if marker == MarkerKind::False && *has_null {
row_ptr.marker = Some(MarkerKind::Null);
}
if m == &mut MarkerKind::Null {
if marker == MarkerKind::Null {
validity.push(false);
} else {
validity.push(true);
}
if m == &mut MarkerKind::True {
if marker == MarkerKind::True {
boolean_bit_map.push(true);
} else {
boolean_bit_map.push(false);
Expand All @@ -644,50 +653,7 @@ impl HashJoinState for JoinHashTable {
)]);
let marker_block =
DataBlock::create(DataSchemaRef::from(marker_schema), vec![marker_column]);
let mut build_indexes = Vec::new();
match *hash_table {
HashTable::SerializerHashTable(ref hash_table) => {
for entity in hash_table.hash_table.iter() {
build_indexes.extend_from_slice(entity.get_value());
}
}
HashTable::KeyU8HashTable(ref hash_table) => {
for entity in hash_table.hash_table.iter() {
build_indexes.extend_from_slice(entity.get_value());
}
}
HashTable::KeyU16HashTable(ref hash_table) => {
for entity in hash_table.hash_table.iter() {
build_indexes.extend_from_slice(entity.get_value());
}
}
HashTable::KeyU32HashTable(ref hash_table) => {
for entity in hash_table.hash_table.iter() {
build_indexes.extend_from_slice(entity.get_value());
}
}
HashTable::KeyU64HashTable(ref hash_table) => {
for entity in hash_table.hash_table.iter() {
build_indexes.extend_from_slice(entity.get_value());
}
}
HashTable::KeyU128HashTable(ref hash_table) => {
for entity in hash_table.hash_table.iter() {
build_indexes.extend_from_slice(entity.get_value());
}
}
HashTable::KeyU256HashTable(ref hash_table) => {
for entity in hash_table.hash_table.iter() {
build_indexes.extend_from_slice(entity.get_value());
}
}
HashTable::KeyU512HashTable(ref hash_table) => {
for entity in hash_table.hash_table.iter() {
build_indexes.extend_from_slice(entity.get_value());
}
}
};
let build_block = self.row_space.gather(&build_indexes)?;
let build_block = self.row_space.gather(&row_ptrs)?;
Ok(vec![self.merge_eq_block(&marker_block, &build_block)?])
}
}
Loading