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    /// # 为什么非要有这个
166    ///
167    /// 抢占是「先 SELECT 出最该跑的那笔，再 UPDATE 占坑」。不加锁的话，
168    /// N 个推进 worker 的 SELECT 会**全部选中同一行**，然后挤在 UPDATE
169    /// 上排队，最后只有一个成功、其余白跑一轮。实测 Postgres 上
170    /// 1 个 worker 71 笔/秒、8 个也才 127 笔/秒 —— 并行度基本没了。
171    ///
172    /// `FOR UPDATE SKIP LOCKED` 让每个 worker 自动跳过别人正在抢的行，
173    /// 各拿各的，这才是真并行。
174    ///
175    /// - Postgres 9.5+ / MySQL 8.0+ 都支持
176    /// - **sqlite 返回空串**：它没有行锁，写操作本来就是全库串行的，
177    ///   加了也没用（语法上还不认）
178    pub fn skip_locked(&self) -> &'static str {
179        match self {
180            Self::Sqlite => "",
181            _ => " FOR UPDATE SKIP LOCKED",
182        }
183    }
184}
185
186/// 值超过列宽
187#[derive(Debug, Clone, PartialEq, Eq)]
188pub struct TooLong {
189    pub col: &'static str,
190    pub len: usize,
191    pub max: usize,
192}
193
194impl std::fmt::Display for TooLong {
195    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
196        write!(
197            f,
198            "{} 超长：{} 个字符，上限 {}（MySQL 上这一列是 VARCHAR({})）",
199            self.col, self.len, self.max, self.max
200        )
201    }
202}
203
204impl std::error::Error for TooLong {}
205
206/// 写库前挡住超长的值。**必须在 Rust 侧挡，不能指望数据库报错。**
207///
208/// MySQL 的 `INSERT IGNORE` 会把 strict mode 的 1406 降级成 1265 警告，
209/// 然后**静默截断**（实测：`VARCHAR(32)` 塞 100 个字符 →
210/// `Warning 1265 Data truncated`，行插进去了，存的是前 32 个字符）。
211///
212/// 而幂等插入全都得用 `INSERT IGNORE`（见 [`Backend::q`]），所以在 MySQL 上
213/// 超长值不是"插入失败"，是**"插入成功但内容被改了"**：
214///
215/// - `payload` 被截断 → 存进去的是坏 JSON，事务再也推不动
216/// - `gid` 被截断 → 两笔不相关的长 gid 事务在屏障表里**撞成同一行**，
217///   一笔的执行会被另一笔当成"已处理过"跳过
218///
219/// 三家的原生行为还完全不一致：postgres 直接报错、sqlite 根本没有长度限制。
220/// 统一在这里挡住，三种后端拿到同一个错误。
221///
222/// 按**字符**数算而不是字节 —— MySQL 的 `VARCHAR(n)` 数的是字符。
223pub fn check_len(col: &'static str, val: &str, max: usize) -> Result<(), TooLong> {
224    let len = val.chars().count();
225    if len > max {
226        return Err(TooLong { col, len, max });
227    }
228    Ok(())
229}
230
231#[cfg(test)]
232mod tests {
233    use super::*;
234
235    #[test]
236    fn 从url认后端() {
237        assert_eq!(Backend::from_url("sqlite::memory:"), Backend::Sqlite);
238        assert_eq!(Backend::from_url("sqlite:/tmp/a.db"), Backend::Sqlite);
239        assert_eq!(Backend::from_url("postgres://u:p@h/db"), Backend::Postgres);
240        assert_eq!(
241            Backend::from_url("postgresql://u:p@h/db"),
242            Backend::Postgres
243        );
244        assert_eq!(
245            Backend::from_url("mysql://root:x@h:3306/db"),
246            Backend::MySql
247        );
248        assert_eq!(Backend::from_url("MySQL://ROOT@H/DB"), Backend::MySql);
249        assert_eq!(Backend::from_url("mariadb://root@h/db"), Backend::MySql);
250    }
251
252    #[test]
253    fn 占位符按方言转换() {
254        let t = "INSERT INTO t (a,b,c) VALUES (?,?,?)";
255        assert_eq!(
256            Backend::Postgres.q(t),
257            "INSERT INTO t (a,b,c) VALUES ($1,$2,$3)"
258        );
259        assert_eq!(
260            Backend::Sqlite.q(t),
261            "INSERT INTO t (a,b,c) VALUES ($1,$2,$3)"
262        );
263        // MySQL 原样保留
264        assert_eq!(Backend::MySql.q(t), t);
265    }
266
267    #[test]
268    fn 冲突忽略按方言展开() {
269        let t = "{INS} t (k) VALUES (?) {NOCONFLICT}";
270        assert_eq!(
271            Backend::Postgres.q(t),
272            "INSERT INTO t (k) VALUES ($1) ON CONFLICT DO NOTHING"
273        );
274        assert_eq!(Backend::MySql.q(t), "INSERT IGNORE INTO t (k) VALUES (?) ");
275    }
276
277    #[test]
278    fn 自由文本列在mysql上必须是varchar() {
279        // sqlx::Any 读 MySQL 的 TEXT 会当成 BLOB，解不成 String
280        assert_eq!(Backend::MySql.text(8192), "VARCHAR(8192)");
281        assert_eq!(Backend::Postgres.text(8192), "TEXT");
282        assert_eq!(Backend::Sqlite.text(8192), "TEXT");
283    }
284
285    #[test]
286    fn 主键列类型按方言() {
287        // MySQL 不能对 TEXT 建索引，必须定长
288        assert_eq!(Backend::MySql.id_text(), "VARCHAR(128)");
289        assert_eq!(Backend::Postgres.id_text(), "TEXT");
290        assert_eq!(Backend::Sqlite.id_text(), "TEXT");
291    }
292
293    #[test]
294    fn 索引方式按方言二选一() {
295        // 非 MySQL：独立 CREATE INDEX，且不要内联
296        assert!(Backend::Postgres.create_index("i", "t", "a,b").is_some());
297        assert_eq!(Backend::Postgres.inline_index("i", "a,b"), "");
298        // MySQL：反过来
299        assert!(Backend::MySql.create_index("i", "t", "a,b").is_none());
300        assert_eq!(Backend::MySql.inline_index("i", "a,b"), ", KEY i (a,b)");
301    }
302
303    #[test]
304    fn 编号从1开始且连续() {
305        let t = "UPDATE t SET a=?, b=? WHERE c=? AND d=?";
306        assert_eq!(
307            Backend::Postgres.q(t),
308            "UPDATE t SET a=$1, b=$2 WHERE c=$3 AND d=$4"
309        );
310    }
311}
dtmrs_core/dialect.rs

dtmrs_core/
dialect.rs
