risingwave_stream/from_proto/
hash_join.rs1use std::cmp::min;
16use std::sync::Arc;
17
18use risingwave_common::hash::{HashKey, HashKeyDispatcher};
19use risingwave_common::types::DataType;
20use risingwave_expr::expr::{
21 InputRefExpression, NonStrictExpression, build_func_non_strict, build_non_strict_from_prost,
22};
23use risingwave_pb::plan_common::JoinType as JoinTypeProto;
24use risingwave_pb::stream_plan::{HashJoinNode, JoinEncodingType as JoinEncodingTypeProto};
25
26use super::*;
27use crate::common::table::state_table::{StateTable, StateTableBuilder};
28use crate::executor::hash_join::*;
29use crate::executor::monitor::StreamingMetrics;
30use crate::executor::{ActorContextRef, CpuEncoding, JoinType, MemoryEncoding};
31use crate::task::AtomicU64Ref;
32
33pub struct HashJoinExecutorBuilder;
34
35impl ExecutorBuilder for HashJoinExecutorBuilder {
36 type Node = HashJoinNode;
37
38 async fn new_boxed_executor(
39 params: ExecutorParams,
40 node: &Self::Node,
41 store: impl StateStore,
42 ) -> StreamResult<Executor> {
43 let is_append_only = node.is_append_only;
44 let vnodes = Arc::new(params.vnode_bitmap.expect("vnodes not set for hash join"));
45
46 let [source_l, source_r]: [_; 2] = params.input.try_into().unwrap();
47
48 let table_l = node.get_left_table()?;
49 let degree_table_l = node.get_left_degree_table()?;
50
51 let table_r = node.get_right_table()?;
52 let degree_table_r = node.get_right_degree_table()?;
53
54 let params_l = JoinParams::new(
55 node.get_left_key()
56 .iter()
57 .map(|key| *key as usize)
58 .collect_vec(),
59 node.get_left_deduped_input_pk_indices()
60 .iter()
61 .map(|key| *key as usize)
62 .collect_vec(),
63 );
64 let params_r = JoinParams::new(
65 node.get_right_key()
66 .iter()
67 .map(|key| *key as usize)
68 .collect_vec(),
69 node.get_right_deduped_input_pk_indices()
70 .iter()
71 .map(|key| *key as usize)
72 .collect_vec(),
73 );
74 let null_safe = node.get_null_safe().to_vec();
75 let output_indices = node
76 .get_output_indices()
77 .iter()
78 .map(|&x| x as usize)
79 .collect_vec();
80
81 let condition = match node.get_condition() {
82 Ok(cond_prost) => Some(build_non_strict_from_prost(
83 cond_prost,
84 params.eval_error_report.clone(),
85 )?),
86 Err(_) => None,
87 };
88 trace!("Join non-equi condition: {:?}", condition);
89 let mut inequality_pairs = Vec::with_capacity(node.get_inequality_pairs().len());
90 for inequality_pair in node.get_inequality_pairs() {
91 let key_required_larger = inequality_pair.get_key_required_larger() as usize;
92 let key_required_smaller = inequality_pair.get_key_required_smaller() as usize;
93 inequality_pairs.push((
94 key_required_larger,
95 key_required_smaller,
96 inequality_pair.get_clean_state(),
97 if let Some(delta_expression) = inequality_pair.delta_expression.as_ref() {
98 let data_type = source_l.schema().fields
99 [min(key_required_larger, key_required_smaller)]
100 .data_type();
101 Some(build_func_non_strict(
102 delta_expression.delta_type(),
103 data_type.clone(),
104 vec![
105 Box::new(InputRefExpression::new(data_type, 0)),
106 build_non_strict_from_prost(
107 delta_expression.delta.as_ref().unwrap(),
108 params.eval_error_report.clone(),
109 )?
110 .into_inner(),
111 ],
112 params.eval_error_report.clone(),
113 )?)
114 } else {
115 None
116 },
117 ));
118 }
119
120 let join_key_data_types = params_l
121 .join_key_indices
122 .iter()
123 .map(|idx| source_l.schema().fields[*idx].data_type())
124 .collect_vec();
125
126 let state_table_l = StateTableBuilder::new(table_l, store.clone(), Some(vnodes.clone()))
127 .enable_preload_all_rows_by_config(¶ms.actor_context.streaming_config)
128 .build()
129 .await;
130 let degree_state_table_l =
131 StateTableBuilder::new(degree_table_l, store.clone(), Some(vnodes.clone()))
132 .enable_preload_all_rows_by_config(¶ms.actor_context.streaming_config)
133 .build()
134 .await;
135
136 let state_table_r = StateTableBuilder::new(table_r, store.clone(), Some(vnodes.clone()))
137 .enable_preload_all_rows_by_config(¶ms.actor_context.streaming_config)
138 .build()
139 .await;
140 let degree_state_table_r = StateTableBuilder::new(degree_table_r, store, Some(vnodes))
141 .enable_preload_all_rows_by_config(¶ms.actor_context.streaming_config)
142 .build()
143 .await;
144
145 let join_encoding_type = node
146 .get_join_encoding_type()
147 .unwrap_or(JoinEncodingTypeProto::MemoryOptimized);
148
149 let args = HashJoinExecutorDispatcherArgs {
150 ctx: params.actor_context,
151 info: params.info.clone(),
152 source_l,
153 source_r,
154 params_l,
155 params_r,
156 null_safe,
157 output_indices,
158 cond: condition,
159 inequality_pairs,
160 state_table_l,
161 degree_state_table_l,
162 state_table_r,
163 degree_state_table_r,
164 lru_manager: params.watermark_epoch,
165 is_append_only,
166 metrics: params.executor_stats,
167 join_type_proto: node.get_join_type()?,
168 join_key_data_types,
169 chunk_size: params.env.config().developer.chunk_size,
170 high_join_amplification_threshold: params
171 .env
172 .config()
173 .developer
174 .high_join_amplification_threshold,
175 join_encoding_type,
176 };
177
178 let exec = args.dispatch()?;
179 Ok((params.info, exec).into())
180 }
181}
182
183struct HashJoinExecutorDispatcherArgs<S: StateStore> {
184 ctx: ActorContextRef,
185 info: ExecutorInfo,
186 source_l: Executor,
187 source_r: Executor,
188 params_l: JoinParams,
189 params_r: JoinParams,
190 null_safe: Vec<bool>,
191 output_indices: Vec<usize>,
192 cond: Option<NonStrictExpression>,
193 inequality_pairs: Vec<(usize, usize, bool, Option<NonStrictExpression>)>,
194 state_table_l: StateTable<S>,
195 degree_state_table_l: StateTable<S>,
196 state_table_r: StateTable<S>,
197 degree_state_table_r: StateTable<S>,
198 lru_manager: AtomicU64Ref,
199 is_append_only: bool,
200 metrics: Arc<StreamingMetrics>,
201 join_type_proto: JoinTypeProto,
202 join_key_data_types: Vec<DataType>,
203 chunk_size: usize,
204 high_join_amplification_threshold: usize,
205 join_encoding_type: JoinEncodingTypeProto,
206}
207
208impl<S: StateStore> HashKeyDispatcher for HashJoinExecutorDispatcherArgs<S> {
209 type Output = StreamResult<Box<dyn Execute>>;
210
211 fn dispatch_impl<K: HashKey>(self) -> Self::Output {
212 macro_rules! build {
214 ($join_type:ident, $join_encoding:ident) => {
215 Ok(
216 HashJoinExecutor::<K, S, { JoinType::$join_type }, $join_encoding>::new(
217 self.ctx,
218 self.info,
219 self.source_l,
220 self.source_r,
221 self.params_l,
222 self.params_r,
223 self.null_safe,
224 self.output_indices,
225 self.cond,
226 self.inequality_pairs,
227 self.state_table_l,
228 self.degree_state_table_l,
229 self.state_table_r,
230 self.degree_state_table_r,
231 self.lru_manager,
232 self.is_append_only,
233 self.metrics,
234 self.chunk_size,
235 self.high_join_amplification_threshold,
236 )
237 .boxed(),
238 )
239 };
240 }
241
242 macro_rules! build_match {
243 ($($join_type:ident),*) => {
244 match (self.join_type_proto, self.join_encoding_type) {
245 (JoinTypeProto::AsofInner, _)
246 | (JoinTypeProto::AsofLeftOuter, _)
247 | (JoinTypeProto::Unspecified, _)
248 | (_, JoinEncodingTypeProto::Unspecified ) => unreachable!(),
249 $(
250 (JoinTypeProto::$join_type, JoinEncodingTypeProto::MemoryOptimized) => build!($join_type, MemoryEncoding),
251 (JoinTypeProto::$join_type, JoinEncodingTypeProto::CpuOptimized) => build!($join_type, CpuEncoding),
252 )*
253 }
254 };
255 }
256 build_match! {
257 Inner,
258 LeftOuter,
259 RightOuter,
260 FullOuter,
261 LeftSemi,
262 LeftAnti,
263 RightSemi,
264 RightAnti
265 }
266 }
267
268 fn data_types(&self) -> &[DataType] {
269 &self.join_key_data_types
270 }
271}