1use super::QueryPipelineManager;
2use crate::executor::base::ExecutionResult;
3use crate::executor::explain::physical_plan_explain::physical_plan_to_plan_description;
4use crate::executor::streaming::instance::{QueryBindings, QueryExecutionInstance, ResultSink};
5use crate::executor::streaming::transaction_scope::TransactionScope;
6use crate::parser::ast::stmt::{ExplainStmt, ProfileStmt};
7use crate::planning::plan::explain::ProfilingStats;
8use crate::storage::QueryStorage;
9use crate::QueryContext;
10use linkrs_core::error::{DBError, DBResult, QueryError};
11use linkrs_core::types::expr::expression_context::ExpressionAnalysisContext;
12use linkrs_core::Value;
13use std::sync::Arc;
14
15impl<S: QueryStorage + 'static> QueryPipelineManager<S> {
16 fn compile_diagnostic_target(
23 &mut self,
24 qctx: Arc<QueryContext>,
25 inner_stmt: &crate::parser::ast::Stmt,
26 ) -> DBResult<Arc<crate::executor::streaming::plan::PhysicalPlan>> {
27 let key_text = qctx.request_context().query.clone();
28 let expr_ctx = Arc::new(ExpressionAnalysisContext::new());
29 let ast = Arc::new(crate::parser::ast::stmt::Ast::new(
30 inner_stmt.clone(),
31 expr_ctx,
32 ));
33 let bound = self
34 .bind_parsed_statement(ast.clone(), qctx.clone())?
35 .ok_or_else(|| {
36 DBError::from(QueryError::execution(
37 "EXPLAIN/PROFILE target statement could not be bound".to_string(),
38 ))
39 })?;
40 self.compile_or_get_cached(&key_text, qctx, Some(&bound), inner_stmt, &ast, false)
41 }
42
43 pub fn execute_explain(
44 &mut self,
45 explain_stmt: &ExplainStmt,
46 qctx: Arc<QueryContext>,
47 ) -> DBResult<ExecutionResult> {
48 let inner_stmt = (*explain_stmt.statement).clone();
49 let physical_plan = self.compile_diagnostic_target(qctx, &inner_stmt)?;
50
51 let plan_desc =
52 crate::executor::explain::physical_plan_explain::physical_plan_to_plan_description(
53 &physical_plan,
54 );
55
56 let output = match explain_stmt.format {
57 crate::parser::ast::stmt::ExplainFormat::Table => {
58 crate::executor::explain::format::format_plan_as_table(&plan_desc)
59 }
60 crate::parser::ast::stmt::ExplainFormat::Dot => {
61 crate::executor::explain::format::format_plan_as_dot(&plan_desc)
62 }
63 };
64
65 let data_set = linkrs_core::DataSet::from_rows(
66 vec![vec![linkrs_core::Value::string(output)]],
67 vec!["plan".to_string()],
68 );
69 Ok(ExecutionResult::DataSet { data: data_set })
70 }
71
72 pub fn execute_explain_analyze(
73 &mut self,
74 explain_stmt: &ExplainStmt,
75 qctx: Arc<QueryContext>,
76 transaction_scope: TransactionScope,
77 ) -> DBResult<ExecutionResult> {
78 let inner_ast = &explain_stmt.statement;
79 let expr_ctx = Arc::new(ExpressionAnalysisContext::new());
80 let ast = Arc::new(crate::parser::ast::stmt::Ast::new(
81 (**inner_ast).clone(),
82 expr_ctx,
83 ));
84 let bound = self.bind_parsed_statement(ast.clone(), qctx.clone())?;
85
86 let physical_plan = if let Some(b) = bound {
87 self.compile_from_bound(qctx.clone(), &b, &ast)?
88 } else {
89 return Err(DBError::from(QueryError::execution(
90 "EXPLAIN target statement could not be bound".to_string(),
91 )));
92 };
93
94 let exec_ctx = self.build_execution_context(&qctx);
95 let mut bindings = QueryBindings::from_context(&exec_ctx, transaction_scope);
96 bindings.query_id = exec_ctx.query_id;
97
98 let mut instance = QueryExecutionInstance::instantiate_plan(
99 physical_plan.clone(),
100 bindings,
101 ResultSink::Materialize,
102 self.query_registry.clone(),
103 self.query_manager.clone(),
104 )
105 .map_err(|e| DBError::from(QueryError::execution(e.to_string())))?;
106
107 let _result = instance
108 .execute()
109 .map_err(|e| DBError::from(QueryError::execution(e.to_string())))?;
110
111 let mut plan_desc =
112 crate::executor::explain::physical_plan_explain::physical_plan_to_plan_description(
113 &physical_plan,
114 );
115 Self::overlay_execution_profile(&instance, &mut plan_desc, exec_ctx.max_workers);
116
117 let output = match explain_stmt.format {
118 crate::parser::ast::stmt::ExplainFormat::Table => {
119 crate::executor::explain::format::format_plan_as_table(&plan_desc)
120 }
121 crate::parser::ast::stmt::ExplainFormat::Dot => {
122 crate::executor::explain::format::format_plan_as_dot(&plan_desc)
123 }
124 };
125
126 let data_set = linkrs_core::DataSet::from_rows(
127 vec![vec![linkrs_core::Value::string(output)]],
128 vec!["plan".to_string()],
129 );
130 Ok(ExecutionResult::DataSet { data: data_set })
131 }
132
133 pub fn execute_profile(
134 &mut self,
135 profile_stmt: &ProfileStmt,
136 qctx: Arc<QueryContext>,
137 transaction_scope: TransactionScope,
138 ) -> DBResult<ExecutionResult> {
139 let inner_ast = &profile_stmt.statement;
140 let expr_ctx = Arc::new(ExpressionAnalysisContext::new());
141 let ast = Arc::new(crate::parser::ast::stmt::Ast::new(
142 (**inner_ast).clone(),
143 expr_ctx,
144 ));
145 let bound = self.bind_parsed_statement(ast.clone(), qctx.clone())?;
146
147 let physical_plan = if let Some(b) = bound {
148 self.compile_from_bound(qctx.clone(), &b, &ast)?
149 } else {
150 return Err(DBError::from(QueryError::execution(
151 "PROFILE target statement could not be bound".to_string(),
152 )));
153 };
154
155 let exec_ctx = self.build_execution_context(&qctx);
156 let mut bindings = QueryBindings::from_context(&exec_ctx, transaction_scope);
157 bindings.query_id = exec_ctx.query_id;
158
159 let mut instance = QueryExecutionInstance::instantiate_plan(
160 physical_plan.clone(),
161 bindings,
162 ResultSink::Materialize,
163 self.query_registry.clone(),
164 self.query_manager.clone(),
165 )
166 .map_err(|e| DBError::from(QueryError::execution(e.to_string())))?;
167
168 instance
169 .execute()
170 .map_err(|e| DBError::from(QueryError::execution(e.to_string())))?;
171
172 let mut plan_desc = physical_plan_to_plan_description(&physical_plan);
173 Self::overlay_execution_profile(&instance, &mut plan_desc, exec_ctx.max_workers);
174
175 if profile_stmt.format == crate::parser::ast::stmt::ExplainFormat::Dot {
178 let output = crate::executor::explain::format::format_plan_as_dot(&plan_desc);
179 let data_set = linkrs_core::DataSet::from_rows(
180 vec![vec![linkrs_core::Value::string(output)]],
181 vec!["plan".to_string()],
182 );
183 return Ok(ExecutionResult::DataSet { data: data_set });
184 }
185
186 let mut ids = Vec::new();
187 let mut names = Vec::new();
188 let mut dependencies = Vec::new();
189 let mut profiling_data = Vec::new();
190 let mut operator_info = Vec::new();
191
192 for node_desc in &plan_desc.plan_node_descs {
193 ids.push(Value::BigInt(node_desc.id));
194 names.push(Value::string_from_owned(node_desc.name.clone()));
195
196 let deps = node_desc
197 .dependencies
198 .as_ref()
199 .map(|d| {
200 d.iter()
201 .map(|id| id.to_string())
202 .collect::<Vec<_>>()
203 .join(", ")
204 })
205 .unwrap_or_default();
206 dependencies.push(Value::string(&deps));
207
208 let profile_str = if let Some(ref profiles) = node_desc.profiles {
209 profiles
210 .iter()
211 .map(|p| {
212 let base =
213 format!("rows: {}, exec_time: {}us", p.rows, p.exec_duration_in_us);
214 let spilled = p.other_stats.get("spilled_bytes").map(String::as_str);
215 let count = p.other_stats.get("spill_count").map(String::as_str);
216 let rows = p.other_stats.get("spilled_rows").map(String::as_str);
217 match (spilled, count, rows) {
218 (Some(b), Some(c), Some(r)) if b != "0" || c != "0" || r != "0" => {
219 format!(
220 "{}, spilled_rows: {}, spilled_bytes: {}, spill_count: {}",
221 base, r, b, c
222 )
223 }
224 _ => base,
225 }
226 })
227 .collect::<Vec<_>>()
228 .join("; ")
229 } else {
230 "N/A".to_string()
231 };
232 profiling_data.push(Value::string(&profile_str));
233
234 let info = node_desc
235 .description
236 .as_ref()
237 .map(|descs| {
238 descs
239 .iter()
240 .map(|p| format!("{}: {}", p.key, p.value))
241 .collect::<Vec<_>>()
242 .join(", ")
243 })
244 .unwrap_or_default();
245 operator_info.push(Value::string(&info));
246 }
247
248 let parallel_info = if plan_desc.actual_workers > 0 {
249 format!(
250 "requested_workers: {}, actual_workers: {}, wall_time: {}us, work_time: {}us, buffered_chunks_peak: {}, buffered_bytes_peak: {}",
251 plan_desc.requested_workers,
252 plan_desc.actual_workers,
253 plan_desc.parallel_wall_time_us,
254 plan_desc.parallel_work_time_us,
255 plan_desc.parallel_buffered_chunks_peak,
256 plan_desc.parallel_buffered_bytes_peak,
257 )
258 } else if !plan_desc.parallel_fallback_reason.is_empty() {
259 format!(
260 "requested_workers: {}, fallback: {}",
261 plan_desc.requested_workers, plan_desc.parallel_fallback_reason,
262 )
263 } else {
264 format!(
265 "requested_workers: {}, serial only",
266 plan_desc.requested_workers
267 )
268 };
269 let parallel_info = if plan_desc.cbo_notes.is_empty() {
270 parallel_info
271 } else {
272 format!("{}; cbo: {}", parallel_info, plan_desc.cbo_notes.join(", "))
273 };
274 let parallel_info = if plan_desc.columnar_summary.is_empty() {
275 parallel_info
276 } else {
277 format!("{}; {}", parallel_info, plan_desc.columnar_summary)
278 };
279 ids.push(Value::BigInt(-1));
280 names.push(Value::string("Parallel"));
281 dependencies.push(Value::string(""));
282 profiling_data.push(Value::string(""));
283 operator_info.push(Value::string(¶llel_info));
284
285 let total_rows = plan_desc.plan_node_descs.len() + 1;
286 let result_dataset = linkrs_core::DataSet {
287 col_names: vec![
288 "id".to_string(),
289 "name".to_string(),
290 "dependencies".to_string(),
291 "profiling_data".to_string(),
292 "operator_info".to_string(),
293 ],
294 rows: (0..total_rows)
295 .map(|i| {
296 use linkrs_core::value::NullType;
297 vec![
298 ids.get(i)
299 .cloned()
300 .unwrap_or_else(|| Value::Null(NullType::Null)),
301 names
302 .get(i)
303 .cloned()
304 .unwrap_or_else(|| Value::Null(NullType::Null)),
305 dependencies
306 .get(i)
307 .cloned()
308 .unwrap_or_else(|| Value::Null(NullType::Null)),
309 profiling_data
310 .get(i)
311 .cloned()
312 .unwrap_or_else(|| Value::Null(NullType::Null)),
313 operator_info
314 .get(i)
315 .cloned()
316 .unwrap_or_else(|| Value::Null(NullType::Null)),
317 ]
318 })
319 .collect(),
320 };
321 Ok(ExecutionResult::DataSet {
322 data: result_dataset,
323 })
324 }
325
326 fn overlay_execution_profile(
332 instance: &QueryExecutionInstance,
333 plan_desc: &mut crate::planning::plan::explain::PlanDescription,
334 max_workers: usize,
335 ) {
336 let ((wall_us, work_us, workers, chunks_peak, bytes_peak), columnar) = {
337 let profile = instance.runtime().profile().flush_to_collector();
338 for (key, op_profile) in &profile.operators {
339 let node_id = key.physical_operator_id.0 as i64;
340 if let Some(node_desc) = plan_desc.get_node_desc_mut(node_id) {
341 let profiling = ProfilingStats {
342 rows: op_profile.output_rows as i64,
343 exec_duration_in_us: (op_profile.open_time_us
344 + op_profile.next_time_us
345 + op_profile.close_time_us)
346 as i64,
347 total_duration_in_us: (op_profile.open_time_us
348 + op_profile.next_time_us
349 + op_profile.close_time_us)
350 as i64,
351 other_stats: {
352 let mut map = std::collections::HashMap::new();
353 map.insert(
354 "open_time_us".to_string(),
355 op_profile.open_time_us.to_string(),
356 );
357 map.insert(
358 "next_time_us".to_string(),
359 op_profile.next_time_us.to_string(),
360 );
361 map.insert(
362 "close_time_us".to_string(),
363 op_profile.close_time_us.to_string(),
364 );
365 map.insert(
366 "peak_memory_bytes".to_string(),
367 op_profile.peak_memory_bytes.to_string(),
368 );
369 map.insert(
370 "spilled_bytes".to_string(),
371 op_profile.spilled_bytes.to_string(),
372 );
373 map.insert(
374 "spill_count".to_string(),
375 op_profile.spill_count.to_string(),
376 );
377 map.insert(
378 "spilled_rows".to_string(),
379 op_profile.spilled_rows.to_string(),
380 );
381 map
382 },
383 };
384 node_desc.add_profile(profiling);
385 if op_profile.spilled_bytes > 0
388 || op_profile.spill_count > 0
389 || op_profile.spilled_rows > 0
390 {
391 node_desc
392 .add_description("spilled_rows", op_profile.spilled_rows.to_string());
393 node_desc
394 .add_description("spilled_bytes", op_profile.spilled_bytes.to_string());
395 node_desc
396 .add_description("spill_count", op_profile.spill_count.to_string());
397 }
398 }
399 }
400 (
401 profile.parallel_profile(),
402 crate::executor::streaming::runtime::ColumnarStatsSnapshot::from_stats(
403 &instance.runtime().columnar_stats(),
404 ),
405 )
406 };
407 plan_desc.requested_workers = max_workers;
408 if workers > 0 {
409 plan_desc.actual_workers = workers;
410 }
411 plan_desc.parallel_wall_time_us = wall_us;
412 plan_desc.parallel_work_time_us = work_us;
413 plan_desc.parallel_buffered_chunks_peak = chunks_peak;
414 plan_desc.parallel_buffered_bytes_peak = bytes_peak;
415 plan_desc.columnar_summary = columnar.summary();
416 if plan_desc.actual_workers == 0
417 && plan_desc.requested_workers > 1
418 && plan_desc.parallel_fallback_reason.is_empty()
419 {
420 plan_desc.parallel_fallback_reason =
421 "serial fallback (parallelism not activated)".to_string();
422 }
423 }
424}