Skip to content

Commit 4852889

Browse files
KKouldKKould
committed
refactor: optimizing HashJoin and RadixSort 
1 parent f1215c6 commit 4852889

28 files changed

+899
-339
lines changed

‎README.md‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -67,13 +67,13 @@ run `cargo run -p tpcc --release` to run tpcc
6767
- Tips: TPC-C currently only supports single thread
6868
```shell
6969
<90th Percentile RT (MaxRT)>
70-
New-Order : 0.002 (0.025)
71-
Payment : 0.001 (0.013)
72-
Order-Status : 0.054 (0.159)
73-
Delivery : 0.020 (0.034)
74-
Stock-Level : 0.003 (0.004)
70+
New-Order : 0.002 (0.018)
71+
Payment : 0.001 (0.024)
72+
Order-Status : 0.050 (0.067)
73+
Delivery : 0.021 (0.030)
74+
Stock-Level : 0.003 (0.005)
7575
<TpmC>
76-
7892 Tpmc
76+
8101 Tpmc
7777
```
7878
#### 👉[check more](tpcc/README.md)
7979

‎src/catalog/column.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,7 @@
11
use crate::catalog::TableName;
22
use crate::errors::DatabaseError;
33
use crate::expression::ScalarExpression;
4+
use crate::types::tuple::Tuple;
45
use crate::types::value::DataValue;
56
use crate::types::{ColumnId, LogicalType};
67
use kite_sql_serde_macros::ReferenceSerialization;
@@ -169,7 +170,7 @@ impl ColumnCatalog {
169170
self.desc
170171
.default
171172
.as_ref()
172-
.map(|expr| expr.eval(None))
173+
.map(|expr| expr.eval::<&Tuple>(None))
173174
.transpose()
174175
}
175176

‎src/execution/dql/join/hash/full_join.rs‎

Lines changed: 107 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,107 @@
1+
use crate::execution::dql::join::hash::{filter, FilterArgs, JoinProbeState, ProbeArgs};
2+
use crate::execution::dql::join::hash_join::BuildState;
3+
use crate::execution::dql::sort::BumpVec;
4+
use crate::execution::Executor;
5+
use crate::throw;
6+
use crate::types::tuple::Tuple;
7+
use crate::types::value::DataValue;
8+
use ahash::HashMap;
9+
use fixedbitset::FixedBitSet;
10+
11+
pub(crate) struct FullJoinState {
12+
pub(crate) left_schema_len: usize,
13+
pub(crate) right_schema_len: usize,
14+
pub(crate) bits: FixedBitSet,
15+
}
16+
17+
impl<'a> JoinProbeState<'a> for FullJoinState {
18+
fn probe(
19+
&mut self,
20+
probe_args: ProbeArgs<'a>,
21+
filter_args: Option<&'a FilterArgs>,
22+
) -> Executor<'a> {
23+
let left_schema_len = self.left_schema_len;
24+
let bits_ptr: *mut FixedBitSet = &mut self.bits;
25+
26+
Box::new(
27+
#[coroutine]
28+
move || {
29+
let ProbeArgs { probe_tuple, .. } = probe_args;
30+
31+
if let ProbeArgs {
32+
is_keys_has_null: false,
33+
build_state: Some(build_state),
34+
..
35+
} = probe_args
36+
{
37+
let mut has_filtered = false;
38+
for (i, Tuple { values, pk }) in build_state.tuples.iter() {
39+
let full_values =
40+
Vec::from_iter(values.iter().chain(probe_tuple.values.iter()).cloned());
41+
42+
match &filter_args {
43+
None => (),
44+
Some(filter_args) => {
45+
if !throw!(filter(&full_values, filter_args)) {
46+
has_filtered = true;
47+
unsafe {
48+
(*bits_ptr).set(*i, true);
49+
}
50+
yield Ok(Self::full_right_row(left_schema_len, &probe_tuple));
51+
continue;
52+
}
53+
}
54+
}
55+
yield Ok(Tuple::new(pk.clone(), full_values));
56+
}
57+
build_state.is_used = !has_filtered;
58+
build_state.has_filted = has_filtered;
59+
return;
60+
}
61+
62+
yield Ok(Self::full_right_row(left_schema_len, &probe_tuple));
63+
},
64+
)
65+
}
66+
67+
fn left_drop(
68+
&mut self,
69+
_build_map: HashMap<BumpVec<'a, DataValue>, BuildState>,
70+
_filter_args: Option<&'a FilterArgs>,
71+
) -> Option<Executor<'a>> {
72+
let full_schema_len = self.right_schema_len + self.left_schema_len;
73+
let bits_ptr: *mut FixedBitSet = &mut self.bits;
74+
75+
Some(Box::new(
76+
#[coroutine]
77+
move || {
78+
for (_, state) in _build_map {
79+
if state.is_used {
80+
continue;
81+
}
82+
for (i, mut left_tuple) in state.tuples {
83+
unsafe {
84+
if !(*bits_ptr).contains(i) && state.has_filted {
85+
continue;
86+
}
87+
}
88+
left_tuple.values.resize(full_schema_len, DataValue::Null);
89+
yield Ok(left_tuple);
90+
}
91+
}
92+
},
93+
))
94+
}
95+
}
96+
97+
impl FullJoinState {
98+
pub(crate) fn full_right_row(left_schema_len: usize, probe_tuple: &Tuple) -> Tuple {
99+
let full_values = Vec::from_iter(
100+
(0..left_schema_len)
101+
.map(|_| DataValue::Null)
102+
.chain(probe_tuple.values.iter().cloned()),
103+
);
104+
105+
Tuple::new(probe_tuple.pk.clone(), full_values)
106+
}
107+
}

‎src/execution/dql/join/hash/inner_join.rs‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
use crate::execution::dql::join::hash::{filter, FilterArgs, JoinProbeState, ProbeArgs};
2+
use crate::execution::Executor;
3+
use crate::throw;
4+
use crate::types::tuple::Tuple;
5+
6+
pub(crate) struct InnerJoinState;
7+
8+
impl<'a> JoinProbeState<'a> for InnerJoinState {
9+
fn probe(
10+
&mut self,
11+
probe_args: ProbeArgs<'a>,
12+
filter_args: Option<&'a FilterArgs>,
13+
) -> Executor<'a> {
14+
Box::new(
15+
#[coroutine]
16+
move || {
17+
let ProbeArgs {
18+
is_keys_has_null: false,
19+
probe_tuple,
20+
build_state: Some(build_state),
21+
..
22+
} = probe_args
23+
else {
24+
return;
25+
};
26+
27+
build_state.is_used = true;
28+
for (_, Tuple { values, pk }) in build_state.tuples.iter() {
29+
let full_values =
30+
Vec::from_iter(values.iter().chain(probe_tuple.values.iter()).cloned());
31+
32+
match &filter_args {
33+
None => (),
34+
Some(filter_args) => {
35+
if !throw!(filter(&full_values, filter_args)) {
36+
continue;
37+
}
38+
}
39+
}
40+
yield Ok(Tuple::new(pk.clone(), full_values));
41+
}
42+
},
43+
)
44+
}
45+
}

‎src/execution/dql/join/hash/left_anti_join.rs‎

Lines changed: 71 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,71 @@
1+
use crate::execution::dql::join::hash::left_semi_join::LeftSemiJoinState;
2+
use crate::execution::dql::join::hash::{filter, FilterArgs, JoinProbeState, ProbeArgs};
3+
use crate::execution::dql::join::hash_join::BuildState;
4+
use crate::execution::dql::sort::BumpVec;
5+
use crate::execution::Executor;
6+
use crate::throw;
7+
use crate::types::value::DataValue;
8+
use ahash::HashMap;
9+
use fixedbitset::FixedBitSet;
10+
11+
pub(crate) struct LeftAntiJoinState {
12+
pub(crate) right_schema_len: usize,
13+
pub(crate) inner: LeftSemiJoinState,
14+
}
15+
16+
impl<'a> JoinProbeState<'a> for LeftAntiJoinState {
17+
fn probe(
18+
&mut self,
19+
probe_args: ProbeArgs<'a>,
20+
filter_args: Option<&'a FilterArgs>,
21+
) -> Executor<'a> {
22+
self.inner.probe(probe_args, filter_args)
23+
}
24+
25+
fn left_drop(
26+
&mut self,
27+
_build_map: HashMap<BumpVec<'a, DataValue>, BuildState>,
28+
filter_args: Option<&'a FilterArgs>,
29+
) -> Option<Executor<'a>> {
30+
let bits_ptr: *mut FixedBitSet = &mut self.inner.bits;
31+
let right_schema_len = self.right_schema_len;
32+
Some(Box::new(
33+
#[coroutine]
34+
move || {
35+
for (
36+
_,
37+
BuildState {
38+
tuples,
39+
is_used,
40+
has_filted,
41+
},
42+
) in _build_map
43+
{
44+
if is_used {
45+
continue;
46+
}
47+
for (i, tuple) in tuples {
48+
unsafe {
49+
if (*bits_ptr).contains(i) && has_filted {
50+
continue;
51+
}
52+
}
53+
if let Some(filter_args) = filter_args {
54+
let full_values = Vec::from_iter(
55+
tuple
56+
.values
57+
.iter()
58+
.cloned()
59+
.chain((0..right_schema_len).map(|_| DataValue::Null)),
60+
);
61+
if !throw!(filter(&full_values, filter_args)) {
62+
continue;
63+
}
64+
}
65+
yield Ok(tuple);
66+
}
67+
}
68+
},
69+
))
70+
}
71+
}

‎src/execution/dql/join/hash/left_join.rs‎

Lines changed: 90 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,90 @@
1+
use crate::execution::dql::join::hash::{filter, FilterArgs, JoinProbeState, ProbeArgs};
2+
use crate::execution::dql::join::hash_join::BuildState;
3+
use crate::execution::dql::sort::BumpVec;
4+
use crate::execution::Executor;
5+
use crate::throw;
6+
use crate::types::tuple::Tuple;
7+
use crate::types::value::DataValue;
8+
use ahash::HashMap;
9+
use fixedbitset::FixedBitSet;
10+
11+
pub(crate) struct LeftJoinState {
12+
pub(crate) left_schema_len: usize,
13+
pub(crate) right_schema_len: usize,
14+
pub(crate) bits: FixedBitSet,
15+
}
16+
17+
impl<'a> JoinProbeState<'a> for LeftJoinState {
18+
fn probe(
19+
&mut self,
20+
probe_args: ProbeArgs<'a>,
21+
filter_args: Option<&'a FilterArgs>,
22+
) -> Executor<'a> {
23+
let bits_ptr: *mut FixedBitSet = &mut self.bits;
24+
Box::new(
25+
#[coroutine]
26+
move || {
27+
let ProbeArgs {
28+
is_keys_has_null: false,
29+
probe_tuple,
30+
build_state: Some(build_state),
31+
..
32+
} = probe_args
33+
else {
34+
return;
35+
};
36+
37+
let mut has_filted = false;
38+
for (i, Tuple { values, pk }) in build_state.tuples.iter() {
39+
let full_values =
40+
Vec::from_iter(values.iter().chain(probe_tuple.values.iter()).cloned());
41+
42+
match &filter_args {
43+
None => (),
44+
Some(filter_args) => {
45+
if !throw!(filter(&full_values, filter_args)) {
46+
has_filted = true;
47+
unsafe {
48+
(*bits_ptr).set(*i, true);
49+
}
50+
continue;
51+
}
52+
}
53+
}
54+
yield Ok(Tuple::new(pk.clone(), full_values));
55+
}
56+
build_state.is_used = !has_filted;
57+
build_state.has_filted = has_filted;
58+
},
59+
)
60+
}
61+
62+
fn left_drop(
63+
&mut self,
64+
_build_map: HashMap<BumpVec<'a, DataValue>, BuildState>,
65+
_filter_args: Option<&'a FilterArgs>,
66+
) -> Option<Executor<'a>> {
67+
let full_schema_len = self.right_schema_len + self.left_schema_len;
68+
let bits_ptr: *mut FixedBitSet = &mut self.bits;
69+
70+
Some(Box::new(
71+
#[coroutine]
72+
move || {
73+
for (_, state) in _build_map {
74+
if state.is_used {
75+
continue;
76+
}
77+
for (i, mut left_tuple) in state.tuples {
78+
unsafe {
79+
if !(*bits_ptr).contains(i) && state.has_filted {
80+
continue;
81+
}
82+
}
83+
left_tuple.values.resize(full_schema_len, DataValue::Null);
84+
yield Ok(left_tuple);
85+
}
86+
}
87+
},
88+
))
89+
}
90+
}

0 commit comments

Comments
 (0)