| use crate::models::profile::*; |
| use crate::parser::{fragment_parser, section_parser, value_parser}; |
| |
| /// Parse a raw Doris profile text into a structured DorisProfile. |
| pub fn parse(raw_text: &str) -> DorisProfile { |
| let normalized = section_parser::normalize_text(raw_text); |
| let sections = section_parser::split_sections(&normalized); |
| |
| // Parse Summary |
| let summary_text = sections.get("Summary").map(|s| s.as_str()).unwrap_or(""); |
| let summary_kv = section_parser::parse_kv_section(summary_text); |
| let total_time_str = summary_kv.get("Total").cloned().unwrap_or_default(); |
| let total_time_ms = value_parser::parse_duration_ms(&total_time_str); |
| |
| let summary = ProfileSummary { |
| query_id: summary_kv.get("Profile ID").cloned().unwrap_or_default(), |
| total_time: total_time_str, |
| total_time_ms, |
| state: summary_kv.get("Task State").cloned().unwrap_or_default(), |
| sql: summary_kv.get("Sql Statement").cloned().unwrap_or_default(), |
| doris_version: summary_kv.get("Doris Version").cloned(), |
| is_nereids: summary_kv.get("Is Nereids").map(|v| v == "Yes"), |
| is_cached: summary_kv.get("Is Cached").map(|v| v == "Yes"), |
| total_instances: summary_kv |
| .get("Total Instances Num") |
| .and_then(|v| v.parse().ok()), |
| user: summary_kv.get("User").cloned(), |
| start_time: summary_kv.get("Start Time").cloned(), |
| end_time: summary_kv.get("End Time").cloned(), |
| default_db: summary_kv.get("Default Db").cloned(), |
| }; |
| |
| // Parse Execution Summary |
| let exec_text = sections |
| .get("Execution Summary") |
| .map(|s| s.as_str()) |
| .unwrap_or(""); |
| let exec_kv = section_parser::parse_kv_section(exec_text); |
| |
| let execution_summary = ExecutionSummary { |
| parse_sql_time: exec_kv.get("Parse SQL Time").cloned(), |
| plan_time: exec_kv.get("Plan Time").cloned(), |
| schedule_time: exec_kv.get("Schedule Time").cloned(), |
| wait_fetch_result_time: exec_kv.get("Wait and Fetch Result Time").cloned(), |
| fetch_result_time: exec_kv.get("Fetch Result Time").cloned(), |
| write_result_time: exec_kv.get("Write Result Time").cloned(), |
| nereids_analysis_time: exec_kv.get("Nereids Analysis Time").cloned(), |
| nereids_rewrite_time: exec_kv.get("Nereids Rewrite Time").cloned(), |
| nereids_optimize_time: exec_kv.get("Nereids Optimize Time").cloned(), |
| workload_group: exec_kv.get("Workload Group").cloned(), |
| }; |
| |
| // Parse Changed Session Variables |
| let vars_text = sections |
| .get("Changed Session Variables") |
| .map(|s| s.as_str()) |
| .unwrap_or(""); |
| let changed_session_vars = parse_session_vars(vars_text); |
| |
| // Parse MergedProfile |
| let merged_text = sections |
| .get("MergedProfile") |
| .map(|s| s.as_str()) |
| .unwrap_or(""); |
| let fragments = fragment_parser::parse_merged_profile(merged_text); |
| |
| DorisProfile { |
| summary, |
| execution_summary, |
| changed_session_vars, |
| fragments, |
| } |
| } |
| |
| fn parse_session_vars(text: &str) -> Vec<SessionVar> { |
| let trimmed = text.trim(); |
| |
| // Newer Doris emits this section as a JSON array of |
| // [{VarName, CurrentValue, DefaultValue}, ...]; older versions use a |
| // pipe-delimited table. Try JSON first, then fall back to the table parser. |
| if trimmed.starts_with('[') { |
| if let Ok(serde_json::Value::Array(items)) = |
| serde_json::from_str::<serde_json::Value>(trimmed) |
| { |
| let vars: Vec<SessionVar> = items |
| .iter() |
| .filter_map(|item| { |
| let name = item.get("VarName").and_then(|v| v.as_str())?; |
| if name.is_empty() { |
| return None; |
| } |
| Some(SessionVar { |
| name: name.to_string(), |
| current_value: item |
| .get("CurrentValue") |
| .and_then(|v| v.as_str()) |
| .unwrap_or_default() |
| .to_string(), |
| default_value: item |
| .get("DefaultValue") |
| .and_then(|v| v.as_str()) |
| .unwrap_or_default() |
| .to_string(), |
| }) |
| }) |
| .collect(); |
| if !vars.is_empty() { |
| return vars; |
| } |
| } |
| } |
| |
| // Legacy pipe-delimited table: "VarName | CurrentValue | DefaultValue". |
| let mut vars = Vec::new(); |
| for line in text.lines() { |
| let line = line.trim(); |
| // Skip header and separator lines |
| if line.is_empty() |
| || line.starts_with("VarName") |
| || line.starts_with("---") |
| || line.starts_with('-') |
| { |
| continue; |
| } |
| |
| let parts: Vec<&str> = line.split('|').map(|s| s.trim()).collect(); |
| if parts.len() >= 3 { |
| vars.push(SessionVar { |
| name: parts[0].to_string(), |
| current_value: parts[1].to_string(), |
| default_value: parts[2].to_string(), |
| }); |
| } |
| } |
| vars |
| } |
| |
| /// Extract all operators from a parsed profile, flattened. |
| pub fn flatten_operators(profile: &DorisProfile) -> Vec<FlatOperator> { |
| let total_time_ms = profile.summary.total_time_ms.unwrap_or(1.0).max(0.001); |
| let mut result = Vec::new(); |
| |
| for fragment in &profile.fragments { |
| for pipeline in &fragment.pipelines { |
| for op in &pipeline.operators { |
| let exec_time_avg_ms = op |
| .metrics |
| .exec_time |
| .as_ref() |
| .and_then(|v| v.avg) |
| .unwrap_or(0.0); |
| |
| let input_rows = op.metrics.input_rows.as_ref().and_then(|v| v.sum.or(v.avg)); |
| |
| let output_rows = op |
| .metrics |
| .rows_produced |
| .as_ref() |
| .and_then(|v| v.sum.or(v.avg)); |
| |
| let skew_ratio = op.metrics.exec_time.as_ref().and_then(|v| v.skew_ratio()); |
| |
| let time_pct = (exec_time_avg_ms / total_time_ms) * 100.0; |
| |
| // Selectivity: input / output ratio |
| let selectivity = match (input_rows, output_rows) { |
| (Some(inp), Some(out)) if out > 0.0 => Some(round2(inp / out)), |
| _ => None, |
| }; |
| |
| // Spill detection |
| let spill_write = op |
| .all_counters |
| .get("SpillWriteBytes") |
| .or_else(|| op.all_counters.get("SpillWriteBytesToLocalStorage")) |
| .and_then(|v| v.sum.or(v.avg)) |
| .unwrap_or(0.0); |
| let spilled = if spill_write > 0.0 { Some(true) } else { None }; |
| |
| // Peak memory |
| let peak_mem_bytes = op |
| .metrics |
| .memory_usage_peak |
| .as_ref() |
| .and_then(|v| v.sum.or(v.avg)) |
| .filter(|&v| v > 0.0); |
| |
| // Wait dependency analysis: blocked if wait_time > exec_time * 2 |
| let wait_time_ms = op |
| .metrics |
| .wait_for_dependency_time |
| .as_ref() |
| .and_then(|v| v.avg) |
| .filter(|&v| v > 0.0); |
| |
| let blocked_on_upstream = match (wait_time_ms, exec_time_avg_ms) { |
| (Some(wait), exec) if exec > 0.0 && wait > exec * 2.0 => Some(true), |
| _ => None, |
| }; |
| |
| // Runtime filter extraction from plan_info AND all_counters |
| let mut runtime_filters: Vec<String> = Vec::new(); |
| |
| // From plan_info: keys containing RF references |
| for (k, v) in &op.plan_info { |
| if k.contains("RF") || k.contains("runtime") || k.contains("RuntimeFilter") { |
| runtime_filters.push(format!("{k}: {v}")); |
| } |
| } |
| |
| // From all_counters: RuntimeFilter-related counters |
| for (k, v) in &op.all_counters { |
| if k.contains("RuntimeFilter") || k.starts_with("RF") { |
| let val = v.avg.or(v.sum).unwrap_or(0.0); |
| if val > 0.0 { |
| runtime_filters.push(format!("{k}: {val}")); |
| } |
| } |
| } |
| |
| // From all_counters: check for runtime filter specific metrics |
| let rf_input = op |
| .all_counters |
| .get("RuntimeFilterInput") |
| .or_else(|| op.all_counters.get("RuntimeFilterNum")) |
| .and_then(|v| v.sum.or(v.avg)); |
| if let Some(rf_count) = rf_input { |
| if rf_count > 0.0 |
| && !runtime_filters.iter().any(|r| r.contains("RuntimeFilter")) |
| { |
| runtime_filters.push(format!("RuntimeFiltersApplied: {rf_count}")); |
| } |
| } |
| |
| let runtime_filters = if runtime_filters.is_empty() { |
| None |
| } else { |
| Some(runtime_filters) |
| }; |
| |
| // Shuffle bytes (for exchange/data_stream operators) |
| let shuffle_bytes = op |
| .all_counters |
| .get("ShuffleSendBytes") |
| .or_else(|| op.all_counters.get("BytesSent")) |
| .or_else(|| op.all_counters.get("OverallThroughput")) |
| .and_then(|v| v.sum.or(v.avg)) |
| .filter(|&v| v > 0.0); |
| |
| // Cache hit percentage (for scan operators) |
| let cache_hit_pct = { |
| let hit = op |
| .all_counters |
| .get("FileCacheHitBytes") |
| .or_else(|| op.all_counters.get("CacheHitBytes")) |
| .and_then(|v| v.sum.or(v.avg)) |
| .unwrap_or(0.0); |
| let miss = op |
| .all_counters |
| .get("FileCacheMissBytes") |
| .or_else(|| op.all_counters.get("CacheMissBytes")) |
| .and_then(|v| v.sum.or(v.avg)) |
| .unwrap_or(0.0); |
| let total = hit + miss; |
| if total > 0.0 { |
| Some(round2((hit / total) * 100.0)) |
| } else { |
| None |
| } |
| }; |
| |
| // Rows filtered (scan: output < input means filtering happened) |
| let rows_filtered = match (input_rows, output_rows) { |
| (Some(inp), Some(out)) if inp > out && inp > 0.0 => Some(inp - out), |
| _ => None, |
| }; |
| |
| // Join type from PlanInfo |
| let join_type = op |
| .plan_info |
| .iter() |
| .find_map(|(k, v)| { |
| let combined = format!("{k} {v}").to_lowercase(); |
| if combined.contains("broadcast") { |
| Some("broadcast".to_string()) |
| } else if combined.contains("bucket_shuffle") |
| || combined.contains("bucketshuffle") |
| { |
| Some("bucket_shuffle".to_string()) |
| } else if combined.contains("colocate") { |
| Some("colocated".to_string()) |
| } else if combined.contains("shuffle") || combined.contains("hash") { |
| Some("shuffle".to_string()) |
| } else { |
| None |
| } |
| }) |
| // Also try extracting from operator name |
| .or_else(|| { |
| let name_lower = op.info.full_name.to_lowercase(); |
| if name_lower.contains("broadcast") { |
| Some("broadcast".to_string()) |
| } else if name_lower.contains("colocate") { |
| Some("colocated".to_string()) |
| } else { |
| None |
| } |
| }); |
| |
| result.push(FlatOperator { |
| name: op.info.full_name.clone(), |
| table: op.info.table_name.clone(), |
| frag: fragment.id, |
| pipeline: pipeline.id, |
| exec_time_avg_ms: round2(exec_time_avg_ms), |
| input_rows, |
| output_rows, |
| skew_ratio: skew_ratio.map(round2), |
| time_pct: round2(time_pct), |
| selectivity, |
| spilled, |
| peak_mem_bytes, |
| blocked_on_upstream, |
| wait_time_ms: wait_time_ms.map(round2), |
| runtime_filters, |
| shuffle_bytes, |
| cache_hit_pct, |
| rows_filtered, |
| join_type, |
| }); |
| } |
| } |
| } |
| |
| // Sort by exec_time descending |
| result.sort_by(|a, b| { |
| b.exec_time_avg_ms |
| .partial_cmp(&a.exec_time_avg_ms) |
| .unwrap_or(std::cmp::Ordering::Equal) |
| }); |
| result |
| } |
| |
| fn round2(v: f64) -> f64 { |
| (v * 100.0).round() / 100.0 |
| } |
| |
| #[cfg(test)] |
| mod real_profile_tests { |
| use super::*; |
| |
| /// A REAL Doris profile captured by the e2e harness: `suite_profile.sh` writes |
| /// it on a successful `profile get --raw`. Committing that file turns this into |
| /// an offline regression guard for the whole parse pipeline; until it exists the |
| /// test is a visible no-op (it prints how to produce the fixture and returns). |
| fn fixture_path() -> std::path::PathBuf { |
| std::path::Path::new(env!("CARGO_MANIFEST_DIR")) |
| .join("tests/e2e/fixtures/sample_profile.txt") |
| } |
| |
| /// WHY this test exists: the parser's contract is to turn an opaque profile |
| /// text into a structured tree the diagnosis commands can rely on. A unit test |
| /// over a hand-written snippet can't catch a real-profile-only regression (a |
| /// section header that moved, a counter format that changed). This parses an |
| /// actual captured profile and asserts the load-bearing invariants every |
| /// downstream command depends on — so it FAILS if any of them break. |
| #[test] |
| fn parses_a_real_captured_profile() { |
| let path = fixture_path(); |
| let text = match std::fs::read_to_string(&path) { |
| Ok(t) if !t.trim().is_empty() => t, |
| _ => { |
| eprintln!( |
| "SKIP parses_a_real_captured_profile: no fixture at {}.\n \ |
| Run ./start-testing.sh against a cluster, then commit\n \ |
| tests/e2e/fixtures/sample_profile.txt to activate this test.", |
| path.display() |
| ); |
| return; |
| } |
| }; |
| |
| let profile = parse(&text); |
| |
| // Summary: a real profile must yield a non-empty Profile ID and a positive |
| // total time (proves the Summary section + duration parsing both worked). |
| assert!( |
| !profile.summary.query_id.is_empty(), |
| "summary.query_id must be parsed from a real profile" |
| ); |
| assert!( |
| profile.summary.total_time_ms.unwrap_or(0.0) > 0.0, |
| "summary.total_time_ms must parse to a positive number, got {:?}", |
| profile.summary.total_time_ms |
| ); |
| |
| // Structure: at least one fragment with at least one pipeline of operators |
| // (proves MergedProfile -> fragment -> pipeline -> operator parsing worked). |
| assert!( |
| !profile.fragments.is_empty(), |
| "a real profile must yield at least one fragment" |
| ); |
| assert!( |
| profile |
| .fragments |
| .iter() |
| .any(|f| f.pipelines.iter().any(|p| !p.operators.is_empty())), |
| "at least one fragment must contain operators" |
| ); |
| |
| // Flattening: the diagnosis surface must be non-empty, every operator named, |
| // and sorted by exec_time descending (the head is the slowest). |
| let flat = flatten_operators(&profile); |
| assert!( |
| !flat.is_empty(), |
| "flatten_operators must yield operators for a real profile" |
| ); |
| assert!( |
| flat.iter().all(|op| !op.name.is_empty()), |
| "every flattened operator must carry a name" |
| ); |
| assert!( |
| flat.windows(2) |
| .all(|w| w[0].exec_time_avg_ms >= w[1].exec_time_avg_ms), |
| "flattened operators must be sorted by exec_time_avg_ms descending" |
| ); |
| |
| // Regression (PlanInfo-swallows-counters): an operator that carries a |
| // "- PlanInfo" block must still surface its counters. The bug was that the |
| // PlanInfo block never terminated, so every following counter line (which |
| // shares the "- key: value" shape of a plan-info item) was parsed as |
| // plan-info and the operator lost RowsProduced -> output_rows -> the |
| // downstream total_scan_rows degraded to null. |
| let with_plan_info: Vec<_> = profile |
| .fragments |
| .iter() |
| .flat_map(|f| &f.pipelines) |
| .flat_map(|p| &p.operators) |
| .filter(|op| !op.plan_info.is_empty()) |
| .collect(); |
| assert!( |
| !with_plan_info.is_empty(), |
| "fixture must contain operators with a PlanInfo block to exercise this path" |
| ); |
| for op in &with_plan_info { |
| // Every Doris operator emits at least ExecTime, so a PlanInfo operator |
| // with an empty counter map means the block absorbed the counters. |
| assert!( |
| !op.all_counters.is_empty(), |
| "operator {:?} carries a PlanInfo block but has no counters — the \ |
| PlanInfo parser swallowed them", |
| op.info.full_name |
| ); |
| } |
| assert!( |
| with_plan_info |
| .iter() |
| .any(|op| op.metrics.rows_produced.is_some()), |
| "at least one PlanInfo-bearing operator must surface RowsProduced (output_rows)" |
| ); |
| |
| // Regression: DetailProfile / Appendix must be section boundaries, not bleed |
| // into MergedProfile and fabricate empty fragments — every parsed fragment has |
| // real pipelines. |
| assert!( |
| profile.fragments.iter().all(|f| !f.pipelines.is_empty()), |
| "no parsed fragment may be empty (DetailProfile/Appendix must terminate MergedProfile)" |
| ); |
| |
| // Regression: the Changed Session Variables section is parsed (JSON-array or |
| // pipe-table form). A profile captured by the e2e harness ran with --profile, |
| // which always changes at least `enable_profile`. |
| assert!( |
| !profile.changed_session_vars.is_empty(), |
| "changed_session_vars must be parsed from a real profile" |
| ); |
| |
| // Regression: the Physical Plan section is recognized regardless of spelling |
| // ("PhysicalPlan" vs "Physical Plan"). |
| let sections = section_parser::split_sections(§ion_parser::normalize_text(&text)); |
| assert!( |
| sections.contains_key("Physical Plan"), |
| "the Physical Plan section must be extracted" |
| ); |
| } |
| } |
