blob: 4ca01bba87ffbed1d0e65dadbc6aa106d00802cf [file]
use once_cell::sync::Lazy;
use regex::Regex;
/// Parsed operator info from the operator header line.
#[derive(Debug, Clone)]
pub struct OperatorInfo {
pub full_name: String,
pub operator_type: String,
pub id: i32,
pub nereids_id: Option<i32>,
pub table_name: Option<String>,
pub dest_id: Option<i32>,
pub is_sink: bool,
}
// Match any operator line containing (id=N):
// Captures: 1=everything before (id=, 2=id, 3=nereids_id, 4=dst_id
static OPERATOR_RE: Lazy<Regex> = Lazy::new(|| {
Regex::new(r"(.+?)\(?id=(-?\d+)(?:\s*,?\s*nereids_id=(\d+))?(?:\s*,?\s*dst_id=(\d+))?\)\s*:")
.unwrap()
});
// Extract table_name from Doris 4.0 operator header:
// "OLAP_SCAN_OPERATOR(nereids_id=556. table_name=orders(orders))(id=2):"
static TABLE_NAME_RE: Lazy<Regex> = Lazy::new(|| Regex::new(r"table_name=(\w+)").unwrap());
impl OperatorInfo {
/// Parse an operator header line.
pub fn parse(line: &str) -> Option<Self> {
let caps = OPERATOR_RE.captures(line)?;
let full_name = caps[1].trim().to_string();
let id: i32 = caps[2].parse().ok()?;
let nereids_id: Option<i32> = caps.get(3).and_then(|m| m.as_str().parse().ok());
let dest_id: Option<i32> = caps.get(4).and_then(|m| m.as_str().parse().ok());
// Extract the base operator type (without parenthesized qualifiers)
let operator_type = full_name
.split('(')
.next()
.unwrap_or(&full_name)
.trim()
.to_string();
let is_sink = operator_type.contains("SINK");
// Extract table_name from Doris 4.0 header format
let table_name = TABLE_NAME_RE.captures(line).map(|c| c[1].to_string());
Some(OperatorInfo {
full_name,
operator_type,
id,
nereids_id,
table_name,
dest_id,
is_sink,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_parse_simple() {
let info = OperatorInfo::parse("RESULT_SINK_OPERATOR (id=0):").unwrap();
assert_eq!(info.operator_type, "RESULT_SINK_OPERATOR");
assert_eq!(info.id, 0);
assert!(info.is_sink);
}
#[test]
fn test_parse_with_nereids() {
let info = OperatorInfo::parse("SORT_OPERATOR (id=13 , nereids_id=1617):").unwrap();
assert_eq!(info.id, 13);
assert_eq!(info.nereids_id, Some(1617));
}
#[test]
fn test_parse_with_local_exchange() {
let info =
OperatorInfo::parse("LOCAL_EXCHANGE_OPERATOR (LOCAL_MERGE_SORT) (id=-4):").unwrap();
assert_eq!(info.id, -4);
assert!(info.full_name.contains("LOCAL_MERGE_SORT"));
}
#[test]
fn test_parse_dst_id() {
let info = OperatorInfo::parse("DATA_STREAM_SINK_OPERATOR (id=14,dst_id=14):").unwrap();
assert_eq!(info.dest_id, Some(14));
}
#[test]
fn test_parse_doris4_scan_with_table() {
let info = OperatorInfo::parse(
"OLAP_SCAN_OPERATOR(nereids_id=556. table_name=orders(orders))(id=2):",
)
.unwrap();
assert_eq!(info.id, 2);
assert_eq!(info.table_name, Some("orders".to_string()));
assert!(info.operator_type.contains("OLAP_SCAN"));
}
}