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"));
    }
}