dtmrs_core/
dialect.rs

1//! 方言层 —— 一套 SQL 模板跑 sqlite / postgres / mysql。
2//!
3//! 加 MySQL 之前只有两种后端，`sqlx::Any` 配 `$N` 占位符就够了。MySQL 一进来
4//! **同时打破三条**，所以必须有这一层。全部实测（sqlx 0.8 / pg16 / mysql8.0.44）：
5//!
6//! | | sqlite | postgres | mysql |
7//! |---|---|---|---|
8//! | `$1` 占位符 | ✅ | ✅ | ❌ `Unknown column '$1'` |
9//! | `?` 占位符 | ✅ | ❌ 语法错误 | ✅ |
10//! | `ON CONFLICT DO NOTHING` | ✅ | ✅ | ❌ 1064 语法错误 |
11//! | `INSERT IGNORE` | ❌ | ❌ | ✅ |
12//! | `TEXT PRIMARY KEY` | ✅ | ✅ | ❌ 1170 要 key length |
13//! | `CREATE INDEX IF NOT EXISTS` | ✅ | ✅ | ❌ 1064 语法错误 |
14//!
15//! 还有一条**踩过就忘不了**的：MySQL 的 `ON DUPLICATE KEY UPDATE` 在重复时
16//! `rows_affected` 返回 **1**（不是 0），拿它做幂等判断会把"已存在"误判成
17//! "刚插入"。所以 MySQL 必须用 `INSERT IGNORE`。
18//!
19//! # 写 SQL 的规矩
20//!
21//! 模板里统一用 `?` 当占位符（跟 MySQL 一致），非 MySQL 后端由 [`Backend::q`]
22//! 自动换成 `$1..$n`。
23//!
24//! ⚠ 所以**模板的字符串字面量里不能出现 `?`** —— 会被当成占位符。
25//! 目前所有 SQL 都满足（字面量只有 `''` 和 `'prepared'` 这类）。
26
27/// 支持的后端
28#[derive(Debug, Clone, Copy, PartialEq, Eq)]
29pub enum Backend {
30    Sqlite,
31    Postgres,
32    MySql,
33}
34
35impl std::fmt::Display for Backend {
36    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
37        f.write_str(match self {
38            Self::Sqlite => "sqlite",
39            Self::Postgres => "postgres",
40            Self::MySql => "mysql",
41        })
42    }
43}
44
45impl Backend {
46    pub fn from_url(url: &str) -> Self {
47        let u = url.trim().to_ascii_lowercase();
48        if u.starts_with("mysql") || u.starts_with("mariadb") {
49            Self::MySql
50        } else if u.starts_with("postgres") {
51            Self::Postgres
52        } else {
53            Self::Sqlite
54        }
55    }
56
57    /// 主键/索引列的字符串类型。
58    ///
59    /// MySQL 不能对 `TEXT` 建索引（1170：要 key length），必须用定长 VARCHAR。
60    /// 128 跟 DTM 的 `varchar(128)` 对齐；复合主键三列合计仍在 InnoDB 限内。
61    pub fn id_text(&self) -> &'static str {
62        match self {
63            Self::MySql => "VARCHAR(128)",
64            _ => "TEXT",
65        }
66    }
67
68    /// 短标识列（op 之类），MySQL 下不必给到 128
69    pub fn id_short(&self) -> &'static str {
70        match self {
71            Self::MySql => "VARCHAR(45)",
72            _ => "TEXT",
73        }
74    }
75
76    /// 自由文本列（payload / url / 原因之类）。
77    ///
78    /// ⚠ **MySQL 上只能用 `VARCHAR`**：经 `sqlx::Any` 读 MySQL 的 `TEXT`
79    /// （含 `LONGTEXT`、`MEDIUMTEXT`、显式 `CHARACTER SET utf8mb4`）一律报
80    /// `mismatched types; Rust type String is not compatible with SQL type BLOB`。
81    /// 五种写法实测下来只有 `VARCHAR` 能解成 String。
82    ///
83    /// 代价是有长度上限：`n` 要够装最长的内容。MySQL 单行还有 65535 字节的
84    /// 总限制（utf8mb4 每字符 4 字节），所以别把每列都开得很大。
85    pub fn text(&self, n: usize) -> String {
86        match self {
87            Self::MySql => format!("VARCHAR({n})"),
88            _ => "TEXT".to_string(),
89        }
90    }
91
92    /// [`Self::id_text`] 列能装多少个字符
93    pub const ID_MAX: usize = 128;
94
95    /// [`Self::id_short`] 列能装多少个字符
96    pub const ID_SHORT_MAX: usize = 45;
97
98    /// 把 SQL 模板渲染成这个后端能吃的语句。
99    ///
100    /// 做三件事：
101    /// 1. `?` → `$1..$n`（MySQL 保持 `?`）
102    /// 2. `{INS}` → `INSERT IGNORE INTO`（MySQL）/ `INSERT INTO`（其它）
103    /// 3. `{NOCONFLICT}` → 空（MySQL）/ `ON CONFLICT DO NOTHING`（其它）
104    pub fn q(&self, template: &str) -> String {
105        let s = template
106            .replace("{INS}", self.insert_ignore())
107            .replace("{NOCONFLICT}", self.no_conflict());
108        self.placeholders(&s)
109    }
110
111    fn insert_ignore(&self) -> &'static str {
112        match self {
113            Self::MySql => "INSERT IGNORE INTO",
114            _ => "INSERT INTO",
115        }
116    }
117
118    fn no_conflict(&self) -> &'static str {
119        match self {
120            // MySQL 靠 INSERT IGNORE 达到同样效果，句尾不需要东西
121            Self::MySql => "",
122            _ => "ON CONFLICT DO NOTHING",
123        }
124    }
125
126    fn placeholders(&self, sql: &str) -> String {
127        if *self == Self::MySql {
128            return sql.to_string();
129        }
130        let mut out = String::with_capacity(sql.len() + 8);
131        let mut n = 0;
132        for c in sql.chars() {
133            if c == '?' {
134                n += 1;
135                out.push('$');
136                out.push_str(&n.to_string());
137            } else {
138                out.push(c);
139            }
140        }
141        out
142    }
143
144    /// 建索引。MySQL 不支持 `CREATE INDEX IF NOT EXISTS`，只能靠
145    /// 建表时内联 `KEY`，所以这里对 MySQL 返回 `None`。
146    pub fn create_index(&self, name: &str, table: &str, cols: &str) -> Option<String> {
147        match self {
148            Self::MySql => None,
149            _ => Some(format!(
150                "CREATE INDEX IF NOT EXISTS {name} ON {table}({cols})"
151            )),
152        }
153    }
154
155    /// 建表时内联的索引定义。只有 MySQL 用得上（见 [`Self::create_index`]）。
156    pub fn inline_index(&self, name: &str, cols: &str) -> String {
157        match self {
158            Self::MySql => format!(", KEY {name} ({cols})"),
159            _ => String::new(),
160        }
161    }
162}
163
164/// 值超过列宽
165#[derive(Debug, Clone, PartialEq, Eq)]
166pub struct TooLong {
167    pub col: &'static str,
168    pub len: usize,
169    pub max: usize,
170}
171
172impl std::fmt::Display for TooLong {
173    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
174        write!(
175            f,
176            "{} 超长：{} 个字符，上限 {}（MySQL 上这一列是 VARCHAR({})）",
177            self.col, self.len, self.max, self.max
178        )
179    }
180}
181
182impl std::error::Error for TooLong {}
183
184/// 写库前挡住超长的值。**必须在 Rust 侧挡，不能指望数据库报错。**
185///
186/// MySQL 的 `INSERT IGNORE` 会把 strict mode 的 1406 降级成 1265 警告，
187/// 然后**静默截断**（实测：`VARCHAR(32)` 塞 100 个字符 →
188/// `Warning 1265 Data truncated`，行插进去了，存的是前 32 个字符）。
189///
190/// 而幂等插入全都得用 `INSERT IGNORE`（见 [`Backend::q`]），所以在 MySQL 上
191/// 超长值不是"插入失败"，是**"插入成功但内容被改了"**：
192///
193/// - `payload` 被截断 → 存进去的是坏 JSON，事务再也推不动
194/// - `gid` 被截断 → 两笔不相关的长 gid 事务在屏障表里**撞成同一行**，
195///   一笔的执行会被另一笔当成"已处理过"跳过
196///
197/// 三家的原生行为还完全不一致：postgres 直接报错、sqlite 根本没有长度限制。
198/// 统一在这里挡住，三种后端拿到同一个错误。
199///
200/// 按**字符**数算而不是字节 —— MySQL 的 `VARCHAR(n)` 数的是字符。
201pub fn check_len(col: &'static str, val: &str, max: usize) -> Result<(), TooLong> {
202    let len = val.chars().count();
203    if len > max {
204        return Err(TooLong { col, len, max });
205    }
206    Ok(())
207}
208
209#[cfg(test)]
210mod tests {
211    use super::*;
212
213    #[test]
214    fn 从url认后端() {
215        assert_eq!(Backend::from_url("sqlite::memory:"), Backend::Sqlite);
216        assert_eq!(Backend::from_url("sqlite:/tmp/a.db"), Backend::Sqlite);
217        assert_eq!(Backend::from_url("postgres://u:p@h/db"), Backend::Postgres);
218        assert_eq!(
219            Backend::from_url("postgresql://u:p@h/db"),
220            Backend::Postgres
221        );
222        assert_eq!(
223            Backend::from_url("mysql://root:x@h:3306/db"),
224            Backend::MySql
225        );
226        assert_eq!(Backend::from_url("MySQL://ROOT@H/DB"), Backend::MySql);
227        assert_eq!(Backend::from_url("mariadb://root@h/db"), Backend::MySql);
228    }
229
230    #[test]
231    fn 占位符按方言转换() {
232        let t = "INSERT INTO t (a,b,c) VALUES (?,?,?)";
233        assert_eq!(
234            Backend::Postgres.q(t),
235            "INSERT INTO t (a,b,c) VALUES ($1,$2,$3)"
236        );
237        assert_eq!(
238            Backend::Sqlite.q(t),
239            "INSERT INTO t (a,b,c) VALUES ($1,$2,$3)"
240        );
241        // MySQL 原样保留
242        assert_eq!(Backend::MySql.q(t), t);
243    }
244
245    #[test]
246    fn 冲突忽略按方言展开() {
247        let t = "{INS} t (k) VALUES (?) {NOCONFLICT}";
248        assert_eq!(
249            Backend::Postgres.q(t),
250            "INSERT INTO t (k) VALUES ($1) ON CONFLICT DO NOTHING"
251        );
252        assert_eq!(Backend::MySql.q(t), "INSERT IGNORE INTO t (k) VALUES (?) ");
253    }
254
255    #[test]
256    fn 自由文本列在mysql上必须是varchar() {
257        // sqlx::Any 读 MySQL 的 TEXT 会当成 BLOB，解不成 String
258        assert_eq!(Backend::MySql.text(8192), "VARCHAR(8192)");
259        assert_eq!(Backend::Postgres.text(8192), "TEXT");
260        assert_eq!(Backend::Sqlite.text(8192), "TEXT");
261    }
262
263    #[test]
264    fn 主键列类型按方言() {
265        // MySQL 不能对 TEXT 建索引，必须定长
266        assert_eq!(Backend::MySql.id_text(), "VARCHAR(128)");
267        assert_eq!(Backend::Postgres.id_text(), "TEXT");
268        assert_eq!(Backend::Sqlite.id_text(), "TEXT");
269    }
270
271    #[test]
272    fn 索引方式按方言二选一() {
273        // 非 MySQL：独立 CREATE INDEX，且不要内联
274        assert!(Backend::Postgres.create_index("i", "t", "a,b").is_some());
275        assert_eq!(Backend::Postgres.inline_index("i", "a,b"), "");
276        // MySQL：反过来
277        assert!(Backend::MySql.create_index("i", "t", "a,b").is_none());
278        assert_eq!(Backend::MySql.inline_index("i", "a,b"), ", KEY i (a,b)");
279    }
280
281    #[test]
282    fn 编号从1开始且连续() {
283        let t = "UPDATE t SET a=?, b=? WHERE c=? AND d=?";
284        assert_eq!(
285            Backend::Postgres.q(t),
286            "UPDATE t SET a=$1, b=$2 WHERE c=$3 AND d=$4"
287        );
288    }
289}
dtmrs_core/dialect.rs

dtmrs_core/
dialect.rs
