rustcdc 0.1.5

Embeddable Rust CDC library focused on correctness-first capture primitives
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

//! SQL Server DDL extraction from CDC events.

use super::{
    extract_captured_ddl, parse_create_table_schema_for_dialect, parse_schema_table_for_dialect,
    CapturedDdl, DdlDialect, DdlExtractor,
};
use crate::schema_history::TableSchema;

/// SQL Server-specific DDL statement extractor.
///
/// Handles CREATE/ALTER/DROP TABLE statements from SQL Server CDC.
#[derive(Default)]
pub struct SqlServerDdlExtractor;

impl SqlServerDdlExtractor {
    /// Create a new SQL Server DDL extractor.
    pub fn new() -> Self {
        Self
    }
}

impl DdlExtractor for SqlServerDdlExtractor {
    fn extract_ddl(&self, message: &str) -> Option<CapturedDdl> {
        extract_captured_ddl(DdlDialect::SqlServer, message)
    }

    fn parse_schema_table(&self, statement: &str) -> Option<(String, String)> {
        parse_schema_table_for_dialect(DdlDialect::SqlServer, statement)
    }

    fn parse_create_table_schema(&self, statement: &str) -> Option<TableSchema> {
        parse_create_table_schema_for_dialect(DdlDialect::SqlServer, statement)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn sqlserver_ddl_extractor_parses_create_table() {
        let extractor = SqlServerDdlExtractor::new();
        let sql = "CREATE TABLE [dbo].[users] ([id] INT PRIMARY KEY, [name] VARCHAR(255) NOT NULL)";

        let ddl = extractor.extract_ddl(sql);
        assert!(ddl.is_some());

        let ddl = ddl.unwrap();
        assert_eq!(ddl.ddl_type, "CREATE_TABLE");
        assert_eq!(ddl.table, "users");
    }

    #[test]
    fn sqlserver_ddl_extractor_parses_drop_table() {
        let extractor = SqlServerDdlExtractor::new();
        let sql = "DROP TABLE [dbo].[users]";

        let ddl = extractor.extract_ddl(sql);
        assert!(ddl.is_some());

        let ddl = ddl.unwrap();
        assert_eq!(ddl.ddl_type, "DROP_TABLE");
        assert_eq!(ddl.table, "users");
    }

    #[test]
    fn sqlserver_ddl_extractor_ignores_non_ddl() {
        let extractor = SqlServerDdlExtractor::new();
        let sql = "INSERT INTO users VALUES (1, 'Alice')";

        let ddl = extractor.extract_ddl(sql);
        assert!(ddl.is_none());
    }

    #[test]
    fn sqlserver_primary_key_extraction() {
        let ddl = SqlServerDdlExtractor::new()
            .extract_ddl("CREATE TABLE [users] ([id] INT PRIMARY KEY, [name] VARCHAR(255))")
            .unwrap();
        let schema = ddl.result_schema.unwrap();
        assert_eq!(schema.primary_keys, vec!["id"]);
    }
}