dm_database_parser_sqllog/
parser.rs

1/// 解析后的日志记录
2///
3/// 该结构体包含从日志文本中解析出的所有字段。所有字符串字段都是对原始输入文本的引用，
4/// 因此不会产生额外的内存分配。
5///
6/// # 字段说明
7///
8/// - `ts`: 时间戳字符串（格式：`YYYY-MM-DD HH:MM:SS.mmm`）
9/// - `meta_raw`: 原始元信息字符串（括号内的内容）
10/// - `ep`: 执行计划标识符
11/// - `sess`: 会话标识符
12/// - `thrd`: 线程标识符
13/// - `user`: 用户名
14/// - `trxid`: 事务ID
15/// - `stmt`: 语句标识符
16/// - `appname`: 应用程序名称
17/// - `ip`: 客户端IP地址（可选）
18/// - `body`: 记录主体内容（SQL语句等）
19/// - `execute_time_ms`: 执行时间（毫秒，可选）
20/// - `row_count`: 影响的行数（可选）
21/// - `execute_id`: 执行ID（可选）
22///
23/// # 示例
24///
25/// ```rust
26/// use dm_database_parser_sqllog::parse_record;
27///
28/// let log_text = "2025-08-12 10:57:09.562 (EP[0] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp) SELECT 1";
29/// let parsed = parse_record(log_text);
30/// println!("用户: {}, 事务ID: {}", parsed.user, parsed.trxid);
31/// ```
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub struct ParsedRecord<'a> {
34    pub ts: &'a str,
35    pub meta_raw: &'a str,
36    pub ep: &'a str,
37    pub sess: &'a str,
38    pub thrd: &'a str,
39    pub user: &'a str,
40    pub trxid: &'a str,
41    pub stmt: &'a str,
42    pub appname: &'a str,
43    pub ip: Option<&'a str>,
44    pub body: &'a str,
45    pub execute_time_ms: Option<u64>,
46    pub row_count: Option<u64>,
47    pub execute_id: Option<u64>,
48}
49
50/// 迭代器，从输入日志文本中产生记录切片（`&str`），不进行额外分配。
51///
52/// `RecordSplitter` 旨在以最小分配（零分配或极少分配）的方式，从整个日志文本中
53/// 按"记录"（record）边界切分并逐条返回。这里的"记录"由如下格式决定：每条记录
54/// 都以固定长度的时间戳开始（23 字符，格式 `YYYY-MM-DD HH:MM:SS.mmm`），且时间戳位于
55/// 行首（紧贴换行或文件开头），时间戳之后通常跟随一个空格和一个以圆括号包围的元信息，
56/// 然后是记录主体（body），记录主体可能跨多行。
57///
58/// # 设计目标
59///
60/// - 尽量避免对每条记录进行拷贝；返回的是对原始输入字符串的切片（`&str`）
61/// - 通过只扫描字节数组并使用简单的索引运算来保持最高性能
62/// - 保持内部不变式以便 `next()` 实现更简单且安全
63///
64/// # 使用示例
65///
66/// ```rust
67/// use dm_database_parser_sqllog::RecordSplitter;
68///
69/// let log_text = r#"
70/// 2025-08-12 10:57:09.562 (EP[0] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp) SELECT 1
71/// 2025-08-12 10:57:10.123 (EP[0] sess:2 thrd:2 user:guest trxid:0 stmt:2 appname:MyApp) SELECT 2
72/// "#;
73///
74/// let splitter = RecordSplitter::new(log_text);
75/// for record in splitter {
76///     println!("记录: {}", record.lines().next().unwrap_or(""));
77/// }
78/// ```
79pub struct RecordSplitter<'a> {
80    text: &'a str,
81    bytes: &'a [u8],
82    n: usize,
83    // 扫描位置：始终单调不减
84    scan_pos: usize,
85    // 下一个要返回的记录的起始索引
86    next_start: Option<usize>,
87    // 是否已返回最后一条记录
88    finished: bool,
89    // 缓存的前缀（前导错误）结束索引
90    first_start: Option<usize>,
91}
92
93///
94/// 构造一个 RecordSplitter，用于将日志文本拆分为记录。
95/// 拆分的核心规则：
96/// 1. 每条记录以固定长度的时间戳开始（23 字符，`YYYY-MM-DD HH:MM:SS.mmm`），时间戳必须位于行首（紧贴换行或文件开头）。
97/// 2. 紧随时间戳后，必须跟着一个空格和以括号包围的元信息字段，格式和顺序如下：
98///    EP[xxx] sess:xxx thrd:xxx user:xxx trxid:xxx stmt:xxx appname:xxx
99///    所有字段缺一不可，且顺序必须一致。
100/// 3. 记录主体（body）内容可跨多行，直至遇到下一条合法记录的时间戳或文件结尾。
101/// 4. 每条记录内部，最终必定包含一行如下 "END" 信息：
102///    EXECTIME: xxxms ROWCOUNT: xxx EXEC_ID: xxx
103///    该行用于判定记录结束。
104///
105/// Splitter 工作流程：
106/// - 在整个文本中线性查找，基于上述规则判定记录起始和结束位置。
107/// - 避免不必要的分配，每条记录直接作为对原始日志输入的切片返回（&str）。
108///
109impl<'a> RecordSplitter<'a> {
110    pub fn new(text: &'a str) -> Self {
111        let bytes = text.as_bytes();
112        let n = text.len();
113        let first_start = Self::find_first_record_start(bytes, n);
114
115        let scan_pos = first_start.unwrap_or(0).saturating_add(1);
116        RecordSplitter {
117            text,
118            bytes,
119            n,
120            scan_pos,
121            next_start: first_start,
122            finished: false,
123            first_start,
124        }
125    }
126
127    /// 查找第一个合法记录的起始位置
128    fn find_first_record_start(bytes: &[u8], n: usize) -> Option<usize> {
129        const TS_LEN: usize = 23;
130        const FIELDS: [&[u8]; 7] = [
131            b"EP[",
132            b"sess:",
133            b"thrd:",
134            b"user:",
135            b"trxid:",
136            b"stmt:",
137            b"appname:",
138        ];
139
140        if n < TS_LEN {
141            return None;
142        }
143
144        let limit = n.saturating_sub(TS_LEN);
145        (0..=limit).find(|&pos| {
146            Self::is_line_start_with_timestamp(bytes, n, pos, TS_LEN)
147                && Self::validate_meta_fields(bytes, n, pos + TS_LEN, &FIELDS)
148        })
149    }
150
151    /// 检查位置是否为行首且后跟合法时间戳
152    fn is_line_start_with_timestamp(bytes: &[u8], n: usize, pos: usize, ts_len: usize) -> bool {
153        if pos + ts_len > n {
154            return false;
155        }
156        let is_line_start = pos == 0 || bytes[pos - 1] == b'\n';
157        let has_valid_timestamp = crate::tools::is_ts_millis_bytes(&bytes[pos..pos + ts_len]);
158        is_line_start && has_valid_timestamp
159    }
160
161    /// 验证元信息字段是否按顺序匹配（所有字段必须存在）
162    fn validate_meta_fields(bytes: &[u8], n: usize, mut pos: usize, fields: &[&[u8]]) -> bool {
163        // 检查时间戳后是否有空格
164        if pos >= n || bytes[pos] != b' ' {
165            return false;
166        }
167        pos += 1; // 跳过空格
168
169        // 跳过可选的左括号
170        if pos < n && bytes[pos] == b'(' {
171            pos += 1;
172        }
173
174        // 必须匹配所有字段
175        for &pat in fields {
176            // 检查字段前缀是否匹配
177            let pat_len = pat.len();
178            if pos + pat_len > n || &bytes[pos..pos + pat_len] != pat {
179                return false;
180            }
181            pos += pat_len;
182
183            // 跳过字段值
184            if pat == b"EP[" {
185                // EP[ 后直到 ]
186                while pos < n && bytes[pos] != b']' {
187                    pos += 1;
188                }
189                if pos >= n || bytes[pos] != b']' {
190                    return false;
191                }
192                pos += 1; // 跳过 ]
193            } else {
194                // 其他字段：跳过字段值直到空格
195                if pos >= n {
196                    return false;
197                }
198                while pos < n && bytes[pos] != b' ' {
199                    pos += 1;
200                }
201            }
202            // 跳过分隔的空格
203            while pos < n && bytes[pos] == b' ' {
204                pos += 1;
205            }
206        }
207
208        // 所有字段都已匹配
209        true
210    }
211
212    /// 返回完整的前导错误文本切片（第一条记录之前的所有内容）
213    ///
214    /// 说明：当日志文件开头存在非记录内容（例如垃圾行或日志碎片）时，`first_start` 会
215    /// 指向第一个合法记录的起始位置；`leading_errors_slice()` 返回从文件开始到该位置的全部文本，
216    /// 便于调用者单独处理这部分错误/告警信息。
217    pub fn leading_errors_slice(&self) -> Option<&'a str> {
218        self.first_start.map(|s| &self.text[..s])
219    }
220}
221
222impl<'a> Iterator for RecordSplitter<'a> {
223    type Item = &'a str;
224
225    fn next(&mut self) -> Option<Self::Item> {
226        if self.finished {
227            return None;
228        }
229        let start = match self.next_start {
230            Some(s) => s,
231            None => {
232                self.finished = true;
233                return None;
234            }
235        };
236
237        // 扫描下一个记录的起始位置
238        // 逻辑：从当前 scan_pos 向前搜索下一个满足“行首+时间戳”的位置。
239        // 如果找到，则当前记录的结束位置为该 timestamp 的起始位置（end = pos），并把 next_start 设置为该 pos，
240        // 以便下一次调用返回后续记录；如果搜索到末尾未找到，则把剩余文本作为最后一条记录返回。
241        if self.scan_pos > self.n {
242            // 没有足够空间容纳另一个时间戳，返回剩余内容
243            self.finished = true;
244            return Some(&self.text[start..self.n]);
245        }
246        let limit = self.n.saturating_sub(23);
247        let mut pos = self.scan_pos;
248        while pos <= limit {
249            if (pos == 0 || self.bytes[pos - 1] == b'\n')
250                && crate::tools::is_ts_millis_bytes(&self.bytes[pos..pos + 23])
251            {
252                // 找到下一个起始位置
253                let end = pos;
254                // 为下一次调用做准备
255                self.next_start = Some(pos);
256                self.scan_pos = pos + 1;
257                return Some(&self.text[start..end]);
258            }
259            pos += 1;
260        }
261
262        // 没有下一个起始位置 => 返回最后一条记录
263        self.finished = true;
264        Some(&self.text[start..self.n])
265    }
266}
267
268/// 使用时间戳检测将完整日志文本拆分为记录。
269///
270/// 该函数会扫描整个文本，识别所有合法的日志记录，并将它们拆分为独立的切片。
271/// 同时会收集文件开头的前导错误行（在第一个合法记录之前的内容）。
272///
273/// # 参数
274///
275/// * `text` - 完整的日志文本
276///
277/// # 返回值
278///
279/// 返回一个元组 `(records, errors)`：
280/// - `records`: 所有合法记录的切片向量，每个元素都是对原始文本的引用
281/// - `errors`: 前导错误行的向量（在第一个合法记录之前的所有行）
282///
283/// # 示例
284///
285/// ```rust
286/// use dm_database_parser_sqllog::split_by_ts_records_with_errors;
287///
288/// let log_text = r#"
289/// garbage line
290/// 2025-08-12 10:57:09.562 (EP[0] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp) SELECT 1
291/// 2025-08-12 10:57:10.123 (EP[0] sess:2 thrd:2 user:guest trxid:0 stmt:2 appname:MyApp) SELECT 2
292/// "#;
293///
294/// let (records, errors) = split_by_ts_records_with_errors(log_text);
295/// println!("找到 {} 条记录，{} 条前导错误", records.len(), errors.len());
296/// ```
297pub fn split_by_ts_records_with_errors<'a>(text: &'a str) -> (Vec<&'a str>, Vec<&'a str>) {
298    let mut records: Vec<&'a str> = Vec::new();
299    let mut errors: Vec<&'a str> = Vec::new();
300
301    let splitter = RecordSplitter::new(text);
302    if let Some(prefix) = splitter.leading_errors_slice() {
303        for line in prefix.lines() {
304            errors.push(line);
305        }
306    }
307    for rec in splitter {
308        records.push(rec);
309    }
310    (records, errors)
311}
312
313/// 拆分到调用者提供的容器以避免每次调用分配。
314///
315/// 该函数会清空并填充 `records` 和 `errors`。如果调用者在重复调用中重用这些
316/// 向量（例如在循环中），则可以避免每次调用分配新的 `Vec`。
317///
318/// # 参数
319///
320/// * `text` - 完整的日志文本
321/// * `records` - 用于存储记录切片的向量（会被清空后填充）
322/// * `errors` - 用于存储前导错误行的向量（会被清空后填充）
323///
324/// # 示例
325///
326/// ```rust
327/// use dm_database_parser_sqllog::split_into;
328///
329/// let mut records = Vec::new();
330/// let mut errors = Vec::new();
331///
332/// let log_text = r#"2025-08-12 10:57:09.562 (EP[0] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp) SELECT 1"#;
333/// split_into(log_text, &mut records, &mut errors);
334/// // 处理 records 和 errors...
335/// ```
336pub fn split_into<'a>(text: &'a str, records: &mut Vec<&'a str>, errors: &mut Vec<&'a str>) {
337    records.clear();
338    errors.clear();
339
340    let splitter = RecordSplitter::new(text);
341    if let Some(prefix) = splitter.leading_errors_slice() {
342        for line in prefix.lines() {
343            errors.push(line);
344        }
345    }
346    for rec in splitter {
347        records.push(rec);
348    }
349}
350
351/// 对记录进行流式处理，并对每条记录调用回调而不分配 Vec。
352///
353/// 这是处理日志文本时分配最少的方式。该函数会遍历所有记录，对每条记录调用回调函数，
354/// 但不分配任何 Vec 来存储记录。
355///
356/// # 参数
357///
358/// * `text` - 完整的日志文本
359/// * `f` - 对每条记录调用的回调函数
360///
361/// # 示例
362///
363/// ```rust
364/// use dm_database_parser_sqllog::for_each_record;
365///
366/// let log_text = r#"..."#;
367/// for_each_record(log_text, |rec| {
368///     println!("记录: {}", rec.lines().next().unwrap_or(""));
369/// });
370/// ```
371pub fn for_each_record<F>(text: &str, mut f: F)
372where
373    F: FnMut(&str),
374{
375    let splitter = RecordSplitter::new(text);
376    // 对流式 API 忽略前导错误；如果需要，调用者可以通过 RecordSplitter::leading_errors_slice 检查它们。
377    if let Some(_prefix) = splitter.leading_errors_slice() {
378        // 在迭代之前释放前缀借用
379    }
380    for rec in splitter {
381        f(rec);
382    }
383}
384
385/// 解析每条记录并用 ParsedRecord 调用回调；与流式 Splitter 一起使用时实现零分配。
386///
387/// 该函数会遍历所有记录，解析每条记录为 `ParsedRecord`，然后调用回调函数。
388/// 与 `for_each_record` 类似，这是零分配的处理方式。
389///
390/// # 参数
391///
392/// * `text` - 完整的日志文本
393/// * `f` - 对每条解析后的记录调用的回调函数
394///
395/// # 示例
396///
397/// ```rust
398/// use dm_database_parser_sqllog::parse_records_with;
399///
400/// let log_text = r#"..."#;
401/// parse_records_with(log_text, |parsed| {
402///     println!("用户: {}, 事务ID: {}", parsed.user, parsed.trxid);
403/// });
404/// ```
405pub fn parse_records_with<F>(text: &str, mut f: F)
406where
407    F: for<'r> FnMut(ParsedRecord<'r>),
408{
409    for_each_record(text, |rec| {
410        let parsed = parse_record(rec);
411        f(parsed);
412    });
413}
414
415/// 解析到调用方提供的 Vec 中以避免每次调用分配新的 Vec。
416pub fn parse_into<'a>(text: &'a str, out: &mut Vec<ParsedRecord<'a>>) {
417    out.clear();
418    let splitter = RecordSplitter::new(text);
419    for rec in splitter {
420        out.push(parse_record(rec));
421    }
422}
423
424/// 顺序解析所有记录并返回 ParsedRecord 的 Vec。
425///
426/// 这是最简单的解析方式，会分配一个新的 Vec 来存储所有解析后的记录。
427/// 如果需要避免分配，请使用 `parse_into` 或 `parse_records_with`。
428///
429/// # 参数
430///
431/// * `text` - 完整的日志文本
432///
433/// # 返回值
434///
435/// 返回包含所有解析后记录的向量。
436///
437/// # 示例
438///
439/// ```rust
440/// use dm_database_parser_sqllog::parse_all;
441///
442/// let log_text = r#"..."#;
443/// let records = parse_all(log_text);
444/// for record in records {
445///     println!("用户: {}", record.user);
446/// }
447/// ```
448pub fn parse_all(text: &str) -> Vec<ParsedRecord<'_>> {
449    let splitter = RecordSplitter::new(text);
450    splitter.map(|r| parse_record(r)).collect()
451}
452
453fn parse_digits_forward(s: &str, mut i: usize) -> Option<(u64, usize)> {
454    let bytes = s.as_bytes();
455    let n = bytes.len();
456    // 跳过非数字
457    while i < n && !bytes[i].is_ascii_digit() {
458        i += 1;
459    }
460    if i >= n || !bytes[i].is_ascii_digit() {
461        return None;
462    }
463    let mut val: u64 = 0;
464    while i < n && bytes[i].is_ascii_digit() {
465        val = val
466            .saturating_mul(10)
467            .saturating_add((bytes[i] - b'0') as u64);
468        i += 1;
469    }
470    Some((val, i))
471}
472
473// 辅助：将记录分割成 (ts, meta_raw, body)，均为 &str（借用）
474fn split_ts_meta_body<'a>(rec: &'a str) -> (&'a str, &'a str, &'a str) {
475    let ts: &'a str = if rec.len() >= 23 { &rec[..23] } else { "" };
476    let after_ts: &'a str = if rec.len() > 23 { &rec[23..] } else { "" };
477    let mut meta_raw: &'a str = "";
478    let mut body: &'a str = "";
479
480    if let Some(open_idx) = after_ts.find('(') {
481        if let Some(close_rel) = after_ts[open_idx..].find(')') {
482            meta_raw = &after_ts[open_idx + 1..open_idx + close_rel];
483            let body_start = 23 + open_idx + close_rel + 1;
484            if body_start < rec.len() {
485                body = rec[body_start..].trim_start();
486            }
487        } else {
488            // 没有闭合括号：将剩余部分视为 body
489            body = after_ts;
490        }
491    } else {
492        // 没有元数据括号：时间戳之后的全部内容都是 body
493        body = after_ts;
494    }
495
496    (ts, meta_raw, body)
497}
498
499// 辅助：解析 meta_raw 中的各个字段，返回一个小结构
500#[derive(Debug)]
501struct MetaParts<'a> {
502    ep: &'a str,
503    sess: &'a str,
504    thrd: &'a str,
505    user: &'a str,
506    trxid: &'a str,
507    stmt: &'a str,
508    appname: &'a str,
509    ip: Option<&'a str>,
510}
511
512fn parse_meta(meta_raw: &str) -> MetaParts<'_> {
513    let mut parts = MetaParts {
514        ep: "",
515        sess: "",
516        thrd: "",
517        user: "",
518        trxid: "",
519        stmt: "",
520        appname: "",
521        ip: None,
522    };
523
524    let mut iter = meta_raw.split_whitespace().peekable();
525    while let Some(tok) = iter.next() {
526        if tok.starts_with("EP[") {
527            parts.ep = tok;
528        } else if let Some(val) = tok.strip_prefix("sess:") {
529            parts.sess = val;
530        } else if let Some(val) = tok.strip_prefix("thrd:") {
531            parts.thrd = val;
532        } else if let Some(val) = tok.strip_prefix("user:") {
533            parts.user = val;
534        } else if let Some(val) = tok.strip_prefix("trxid:") {
535            parts.trxid = val;
536        } else if let Some(val) = tok.strip_prefix("stmt:") {
537            parts.stmt = val;
538        } else if tok == "appname:" {
539            if let Some(next) = iter.peek() {
540                if (*next).starts_with("ip:::") {
541                    let nexttok = iter.next().unwrap();
542                    let ippart = nexttok.trim_start_matches("ip:::");
543                    let ipclean = ippart.trim_start_matches("ffff:");
544                    parts.ip = Some(ipclean);
545                    parts.appname = "";
546                } else {
547                    let val = iter.next().unwrap();
548                    parts.appname = val;
549                }
550            } else {
551                parts.appname = "";
552            }
553        } else if let Some(val) = tok.strip_prefix("appname:") {
554            if val.starts_with("ip:::") {
555                let ippart = val.trim_start_matches("ip:::");
556                let ipclean = ippart.trim_start_matches("ffff:");
557                parts.ip = Some(ipclean);
558                parts.appname = "";
559            } else {
560                parts.appname = val;
561            }
562        }
563    }
564
565    parts
566}
567
568// 辅助：从 body 末尾反向提取数值指标（EXEC_ID, ROWCOUNT, EXECTIME）
569fn parse_body_metrics(body: &str) -> (Option<u64>, Option<u64>, Option<u64>) {
570    let mut execute_id: Option<u64> = None;
571    let mut row_count: Option<u64> = None;
572    let mut execute_time_ms: Option<u64> = None;
573
574    let body_str = body;
575    let mut search_end = body_str.len();
576
577    if let Some(pos) = body_str[..search_end].rfind("EXEC_ID:") {
578        let start = pos + "EXEC_ID:".len();
579        if let Some((v, _)) = parse_digits_forward(body_str, start) {
580            execute_id = Some(v);
581        }
582        search_end = pos;
583    }
584
585    if let Some(pos) = body_str[..search_end].rfind("ROWCOUNT:") {
586        let start = pos + "ROWCOUNT:".len();
587        if let Some((v, _)) = parse_digits_forward(body_str, start) {
588            row_count = Some(v);
589        }
590        search_end = pos;
591    }
592
593    if let Some(pos) = body_str[..search_end].rfind("EXECTIME:") {
594        let start = pos + "EXECTIME:".len();
595        if let Some((v, _)) = parse_digits_forward(body_str, start) {
596            execute_time_ms = Some(v);
597        }
598    }
599
600    (execute_time_ms, row_count, execute_id)
601}
602
603/// 解析单条记录。
604///
605/// 该函数将一条日志记录文本解析为 `ParsedRecord` 结构体。
606/// 返回的结构体中的所有字符串字段都是对输入文本的引用，不会产生额外的内存分配。
607///
608/// # 参数
609///
610/// * `rec` - 单条日志记录的文本（通常由 `RecordSplitter` 或相关函数产生）
611///
612/// # 返回值
613///
614/// 返回解析后的 `ParsedRecord`，所有字段都是对输入文本的引用。
615///
616/// # 示例
617///
618/// ```rust
619/// use dm_database_parser_sqllog::parse_record;
620///
621/// let record_text = "2025-08-12 10:57:09.562 (EP[0] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp) SELECT 1";
622/// let parsed = parse_record(record_text);
623/// println!("用户: {}, 事务ID: {}", parsed.user, parsed.trxid);
624/// ```
625pub fn parse_record(rec: &'_ str) -> ParsedRecord<'_> {
626    // 1) 将记录分割为 ts / meta_raw / body
627    let (ts, meta_raw, body) = split_ts_meta_body(rec);
628
629    // 2) 解析 meta 字段
630    let meta = parse_meta(meta_raw);
631
632    // 3) 从 body 解析数值指标
633    let (execute_time_ms, row_count, execute_id) = parse_body_metrics(body);
634
635    ParsedRecord {
636        ts,
637        meta_raw,
638        ep: meta.ep,
639        sess: meta.sess,
640        thrd: meta.thrd,
641        user: meta.user,
642        trxid: meta.trxid,
643        stmt: meta.stmt,
644        appname: meta.appname,
645        ip: meta.ip,
646        body,
647        execute_time_ms,
648        row_count,
649        execute_id,
650    }
651}
652
653#[cfg(test)]
654mod tests {
655    use super::*;
656
657    #[test]
658    fn test_split_by_ts_records() {
659        let log_text = "2023-10-05 14:23:45.123 (EP[12345] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp)\nSELECT * FROM users
6602023-10-05 14:24:00.456 (EP[12346] sess:2 thrd:2 user:guest trxid:0 stmt:2 appname:MyApp)\nINSERT INTO orders VALUES (1, 'item');\n";
661        let (records, errors) = split_by_ts_records_with_errors(log_text);
662
663        assert_eq!(records.len(), 2);
664        assert_eq!(errors.len(), 0);
665    }
666
667    #[test]
668    fn test_split_with_leading_errors() {
669        let log_text = "garbage line 1\ngarbage line 2\n2023-10-05 14:23:45.123 (EP[12345] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp)\nSELECT 1\n";
670        let (records, errors) = split_by_ts_records_with_errors(log_text);
671
672        assert_eq!(records.len(), 1);
673        assert_eq!(errors.len(), 2);
674        assert!(records[0].contains("SELECT 1"));
675    }
676
677    #[test]
678    fn test_record_splitter_iterator() {
679        let log_text = "garbage\n2023-10-05 14:23:45.123 (EP[1] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp) foo\n2023-10-05 14:23:46.456 (EP[2] sess:2 thrd:2 user:guest trxid:0 stmt:2 appname:MyApp) bar\n";
680        let it = RecordSplitter::new(log_text);
681        assert_eq!(it.leading_errors_slice().unwrap().trim(), "garbage");
682        let v: Vec<&str> = it.collect();
683        assert_eq!(v.len(), 2);
684    }
685
686    #[test]
687    fn test_parse_simple_log_sample() {
688        let log_text = "2025-08-12 10:57:09.562 (EP[0] sess:0x7fb24f392a30 thrd:757794 user:HBTCOMS_V3_PROD trxid:688489653 stmt:0x7fb236077b70 appname: ip:::ffff:10.3.100.68) EXECTIME: 0ms ROWCOUNT: 1 EXEC_ID: 289655185\n2025-08-12 10:57:09.562 (EP[0] sess:0x7fb24f392a30 thrd:757794 user:HBTCOMS_V3_PROD trxid:0 stmt:NULL appname:) TRX: START\n";
689
690        let (records, errors) = split_by_ts_records_with_errors(log_text);
691        assert_eq!(errors.len(), 0);
692        assert_eq!(records.len(), 2);
693
694        let r0 = parse_record(records[0]);
695        assert_eq!(r0.execute_time_ms, Some(0));
696        assert_eq!(r0.row_count, Some(1));
697        assert_eq!(r0.execute_id, Some(289655185));
698        assert_eq!(r0.ip, Some("10.3.100.68"));
699        assert_eq!(r0.appname, "");
700
701        let r1 = parse_record(records[1]);
702        assert!(r1.body.contains("TRX: START"));
703    }
704
705    #[test]
706    fn test_missing_sess_field_should_be_error() {
707        // 缺少 sess: 字段 - 找不到有效记录，所有内容都是前导错误
708        let log_text = "garbage1\n2023-10-05 14:23:45.123 (EP[12345] thrd:1 user:admin trxid:0 stmt:1 appname:MyApp)\nSELECT 1\n";
709        let (records, errors) = split_by_ts_records_with_errors(log_text);
710        assert_eq!(records.len(), 0);
711        // 如果找不到第一个有效记录，leading_errors_slice 返回 None，所以 errors 为空
712        // 但整个文本都没有有效记录，所以 records 也为空
713        assert_eq!(errors.len(), 0);
714    }
715
716    #[test]
717    fn test_missing_thrd_field_should_be_error() {
718        // 缺少 thrd: 字段 - 找不到有效记录
719        let log_text = "garbage2\n2023-10-05 14:23:45.123 (EP[12345] sess:1 user:admin trxid:0 stmt:1 appname:MyApp)\nSELECT 1\n";
720        let (records, errors) = split_by_ts_records_with_errors(log_text);
721        assert_eq!(records.len(), 0);
722        assert_eq!(errors.len(), 0); // 找不到第一个有效记录时，leading_errors_slice 返回 None
723    }
724
725    #[test]
726    fn test_missing_user_field_should_be_error() {
727        // 缺少 user: 字段 - 找不到有效记录
728        let log_text = "garbage3\n2023-10-05 14:23:45.123 (EP[12345] sess:1 thrd:1 trxid:0 stmt:1 appname:MyApp)\nSELECT 1\n";
729        let (records, errors) = split_by_ts_records_with_errors(log_text);
730        assert_eq!(records.len(), 0);
731        assert_eq!(errors.len(), 0);
732    }
733
734    #[test]
735    fn test_missing_trxid_field_should_be_error() {
736        // 缺少 trxid: 字段 - 找不到有效记录
737        let log_text = "garbage4\n2023-10-05 14:23:45.123 (EP[12345] sess:1 thrd:1 user:admin stmt:1 appname:MyApp)\nSELECT 1\n";
738        let (records, errors) = split_by_ts_records_with_errors(log_text);
739        assert_eq!(records.len(), 0);
740        assert_eq!(errors.len(), 0);
741    }
742
743    #[test]
744    fn test_missing_stmt_field_should_be_error() {
745        // 缺少 stmt: 字段 - 找不到有效记录
746        let log_text = "garbage5\n2023-10-05 14:23:45.123 (EP[12345] sess:1 thrd:1 user:admin trxid:0 appname:MyApp)\nSELECT 1\n";
747        let (records, errors) = split_by_ts_records_with_errors(log_text);
748        assert_eq!(records.len(), 0);
749        assert_eq!(errors.len(), 0);
750    }
751
752    #[test]
753    fn test_missing_appname_field_should_be_error() {
754        // 缺少 appname: 字段 - 找不到有效记录
755        let log_text = "garbage6\n2023-10-05 14:23:45.123 (EP[12345] sess:1 thrd:1 user:admin trxid:0 stmt:1)\nSELECT 1\n";
756        let (records, errors) = split_by_ts_records_with_errors(log_text);
757        assert_eq!(records.len(), 0);
758        assert_eq!(errors.len(), 0);
759    }
760
761    #[test]
762    fn test_wrong_field_order_should_be_error() {
763        // 字段顺序错误：sess 和 thrd 交换 - 找不到有效记录
764        let log_text = "garbage\n2023-10-05 14:23:45.123 (EP[12345] thrd:1 sess:1 user:admin trxid:0 stmt:1 appname:MyApp)\nSELECT 1\n";
765        let (records, errors) = split_by_ts_records_with_errors(log_text);
766        assert_eq!(records.len(), 0);
767        assert_eq!(errors.len(), 0);
768    }
769
770    #[test]
771    fn test_invalid_timestamp_format_should_be_error() {
772        // 时间戳格式错误（缺少毫秒部分） - 找不到有效记录
773        let log_text = "garbage\n2023-10-05 14:23:45 (EP[12345] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp)\nSELECT 1\n";
774        let (records, errors) = split_by_ts_records_with_errors(log_text);
775        assert_eq!(records.len(), 0);
776        assert_eq!(errors.len(), 0);
777    }
778
779    #[test]
780    fn test_invalid_timestamp_length_should_be_error() {
781        // 时间戳长度不足 - 找不到有效记录
782        let log_text = "garbage\n2023-10-05 14:23:45.1 (EP[12345] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp)\nSELECT 1\n";
783        let (records, errors) = split_by_ts_records_with_errors(log_text);
784        assert_eq!(records.len(), 0);
785        assert_eq!(errors.len(), 0);
786    }
787
788    #[test]
789    fn test_missing_space_after_timestamp_should_be_error() {
790        // 时间戳后没有空格 - 找不到有效记录
791        let log_text = "garbage\n2023-10-05 14:23:45.123(EP[12345] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp)\nSELECT 1\n";
792        let (records, errors) = split_by_ts_records_with_errors(log_text);
793        assert_eq!(records.len(), 0);
794        assert_eq!(errors.len(), 0);
795    }
796
797    #[test]
798    fn test_missing_ep_bracket_should_be_error() {
799        // EP[ 后没有闭合括号 ] - 找不到有效记录
800        let log_text = "garbage\n2023-10-05 14:23:45.123 (EP[12345 sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp)\nSELECT 1\n";
801        let (records, errors) = split_by_ts_records_with_errors(log_text);
802        assert_eq!(records.len(), 0);
803        assert_eq!(errors.len(), 0);
804    }
805
806    #[test]
807    fn test_no_timestamp_line_should_be_error() {
808        // 没有时间戳的普通行
809        let log_text = "garbage line 1\ngarbage line 2\njust a normal line\n2023-10-05 14:23:45.123 (EP[12345] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp)\nSELECT 1\n";
810        let (records, errors) = split_by_ts_records_with_errors(log_text);
811        assert_eq!(records.len(), 1);
812        assert_eq!(errors.len(), 3); // 3 行错误内容
813    }
814
815    #[test]
816    fn test_mixed_valid_and_invalid_records() {
817        // 混合有效和无效记录
818        // 注意：next() 方法只检查时间戳，不验证字段，所以所有有时间戳的行都会被当作记录
819        let log_text = "2023-10-05 14:23:45.123 (EP[12345] sess:1 thrd:1 user:admin trxid:0 stmt:1 appname:MyApp)\nSELECT 1\ninvalid line\n2023-10-05 14:24:00.456 (EP[12346] sess:2 thrd:2 user:guest trxid:0 stmt:2 appname:MyApp)\nINSERT 1\n2023-10-05 14:24:01.789 (EP[12347] sess:3)\ninvalid record\n";
820        let (records, errors) = split_by_ts_records_with_errors(log_text);
821        // next() 方法只检查时间戳，所以所有有时间戳的行都会被当作记录（即使字段不完整）
822        assert_eq!(records.len(), 3);
823        // invalid line 会被包含在第一个和第二个记录之间，所以 errors 中只有前导错误（如果有）
824        assert_eq!(errors.len(), 0); // 第一个记录有效，所以没有前导错误
825    }
826}
dm_database_parser_sqllog/parser.rs

dm_database_parser_sqllog/
parser.rs
