feat(infra): 数据库方言接缝 SqlDialect（PG 实装 + 11g MERGE 编译级交付）(ACM2-29 P3-B)

- SqlDialect 接口：快照 upsert / 主行存在性保障两类方言敏感 SQL； PostgreSqlDialect 承接现有 ON CONFLICT 路径，JdbcFlightSchdRepository 已接入 - Oracle11gDialect：MERGE INTO 模板（编译级交付，未经目标库验证）——激活门控写入 KDoc：11.2 补丁 × JDK25 × ojdbc × Flyway(oracle11g location) × 连接池组合现场实测通过才允许切换，MERGE 绑定顺序适配随激活完成 - db/migration/oracle11g/README.md：11g 迁移位置占位与前置条件清单（CLOB 映射、空串=NULL 语义回归、display 视图 11g 版）验证：MSGX_PG_PORT=5433 真实 PG 全量 108 用例 0 失败 0 跳过
2026-09-08 17:10:40 +08:00
parent 35c91c67b9
commit 107be5d74a
3 changed files with 104 additions and 15 deletions
@@ -0,0 +1,72 @@
+package com.gzzn.omms.msgexchange.infra.persistence.dialect
+
+/**
+ * v2 §6（ACM2-29 P3-B）：数据库方言接缝。
+ *
+ * 原则（v2 §1/§6）：同一业务模型支持 PostgreSQL 与 Oracle 11g，方言只存在于基础设施适配层，
+ * 业务层与仓储语义不按数据库类型分支。
+ *
+ * 接缝范围（随仓储 SQL 演进扩充）：
+ * - 航班主行快照 upsert：PG `INSERT .. ON CONFLICT` vs 11g `MERGE INTO`（11g 无 ON CONFLICT）
+ * - 增量路径主行存在性保障：同上
+ *
+ * 激活门控（本接缝为编译级交付，**未经目标库验证**）：
+ * Oracle 11.2 补丁级别 × JDK 25 × ojdbc 驱动 × Flyway（db/migration/oracle11g location）
+ * × 连接池组合必须在现场 11g 实测通过后，才允许把生产方言切到 [Oracle11gDialect]；
+ * MERGE 绑定顺序适配、CLOB（SRVT/VIPF/MAFL_TEXT）与空串=NULL 语义回归一并纳入激活清单。
+ * 验收证据要求见 docs/flight-state-design-v2.md §6/§8。
+ */
+interface SqlDialect {
+    /** 航班主行快照 upsert：按 SCALAR_WRITE_COLUMNS 生成整体替换 SQL。 */
+    fun flightSnapshotUpsertSql(columns: List<String>): String
+
+    /** 增量路径主行存在性保障（新插 FDAY=NULL，已有行仅推进 updated_at）。 */
+    fun flightRowEnsureSql(): String
+}
+
+/** PostgreSQL 方言：当前生产路径。 */
+object PostgreSqlDialect : SqlDialect {
+
+    override fun flightSnapshotUpsertSql(columns: List<String>): String = buildString {
+        append("INSERT INTO flight_schd (flid, fday, ")
+        append(columns.joinToString(", "))
+        append(", last_message_id, state_version, created_at, updated_at) VALUES (?, ?, ")
+        append(columns.joinToString(", ") { "?" })
+        append(", ?, ?, ?, ?) ON CONFLICT (flid) DO UPDATE SET fday = EXCLUDED.fday, ")
+        append(columns.joinToString(", ") { "$it = EXCLUDED.$it" })
+        append(", last_message_id = EXCLUDED.last_message_id, state_version = EXCLUDED.state_version, updated_at = EXCLUDED.updated_at")
+    }
+
+    override fun flightRowEnsureSql(): String =
+        "INSERT INTO flight_schd (flid, created_at, updated_at) VALUES (?, ?, ?) " +
+            "ON CONFLICT (flid) DO UPDATE SET updated_at = EXCLUDED.updated_at"
+}
+
+/**
+ * Oracle 11g 方言：`MERGE INTO` 单语句 upsert（11g 无 `ON CONFLICT`）。
+ *
+ * **编译级交付，未在目标 11g 上验收**（见 [SqlDialect] 激活门控）。
+ * 绑定顺序与 PG 不同（USING 子句的 flid 绑定 + MATCHED/NOT MATCHED 两份列绑定），
+ * 激活时需同步实现 11g 绑定适配，禁止直接复用 PG 绑定序列。
+ */
+object Oracle11gDialect : SqlDialect {
+
+    override fun flightSnapshotUpsertSql(columns: List<String>): String = buildString {
+        append("MERGE INTO flight_schd t ")
+        append("USING (SELECT ? AS FLID FROM dual) s ON (t.flid = s.FLID) ")
+        append("WHEN MATCHED THEN UPDATE SET t.fday = ?, ")
+        append(columns.joinToString(", ") { "t.$it = ?" })
+        append(", t.last_message_id = ?, t.state_version = ?, t.updated_at = ? ")
+        append("WHEN NOT MATCHED THEN INSERT (flid, fday, ")
+        append(columns.joinToString(", "))
+        append(", last_message_id, state_version, created_at, updated_at) VALUES (")
+        append((listOf("?", "?") + columns.map { "?" } + listOf("?", "?", "?")).joinToString(", "))
+        append(")")
+    }
+
+    override fun flightRowEnsureSql(): String =
+        "MERGE INTO flight_schd t " +
+            "USING (SELECT ? AS FLID FROM dual) s ON (t.flid = s.FLID) " +
+            "WHEN MATCHED THEN UPDATE SET t.updated_at = ? " +
+            "WHEN NOT MATCHED THEN INSERT (flid, created_at, updated_at) VALUES (?, ?, ?)"
+}
@@ -391,6 +391,10 @@ class JdbcFlightSchdRepository(
    private val ds: DataSource,
 ) : FlightSchdRepository {

+    /** 方言接缝（P3-B）：当前生产 PG；切 11g 需现场验收通过后显式替换（见 SqlDialect 激活门控）。 */
+    private val dialect: com.gzzn.omms.msgexchange.infra.persistence.dialect.SqlDialect =
+        com.gzzn.omms.msgexchange.infra.persistence.dialect.PostgreSqlDialect
+
    companion object {
        /** SCHD.FLTR 标量列 + legacy 派生列（与 V1.1.0 迁移一致，全部可空 VARCHAR）。 */
        private val SCALAR_COLUMNS: List<String> = listOf(
@@ -893,33 +897,22 @@ class JdbcFlightSchdRepository(
    }


-    /** v2 主表行存在性保障（增量路径新插 FDAY=NULL）。 */
+    /** v2 主表行存在性保障（增量路径新插 FDAY=NULL）；SQL 由方言接缝生成（P3-B）。 */
    private fun ensureFlightRow(flid: String, now: Instant) {
        val ts = now.toSqlTimestamp()
-        ds.update(
-            "INSERT INTO flight_schd (flid, created_at, updated_at) VALUES (?, ?, ?) " +
-                "ON CONFLICT (flid) DO UPDATE SET updated_at = EXCLUDED.updated_at",
-        ) { ps ->
+        ds.update(dialect.flightRowEnsureSql()) { ps ->
            ps.setString(1, flid)
            ps.setTimestamp(2, ts)
            ps.setTimestamp(3, ts)
        }
    }

-    /** v2 快照：标量列全量替换（集合不写槽位/里程碑列）。 */
+    /** v2 快照：标量列全量替换（集合不写槽位/里程碑列）；SQL 由方言接缝生成（P3-B）。 */
    private fun upsertSnapshotScalars(day: String, state: com.gzzn.omms.msgexchange.domain.flight.FlightNextState, now: Instant) {
        val stored = flattenScalarsOnly(state.scalars)
        val sqlDate = toSqlDate(day)
        val sqlTimestamp = now.toSqlTimestamp()
-        val sql = buildString {
-            append("INSERT INTO flight_schd (flid, fday, ")
-            append(SCALAR_WRITE_COLUMNS.joinToString(", "))
-            append(", last_message_id, state_version, created_at, updated_at) VALUES (?, ?, ")
-            append(SCALAR_WRITE_COLUMNS.joinToString(", ") { "?" })
-            append(", ?, ?, ?, ?) ON CONFLICT (flid) DO UPDATE SET fday = EXCLUDED.fday, ")
-            append(SCALAR_WRITE_COLUMNS.joinToString(", ") { "$it = EXCLUDED.$it" })
-            append(", last_message_id = EXCLUDED.last_message_id, state_version = EXCLUDED.state_version, updated_at = EXCLUDED.updated_at")
-        }
+        val sql = dialect.flightSnapshotUpsertSql(SCALAR_WRITE_COLUMNS)
        ds.update(sql) { ps ->
            var i = 1
            ps.setString(i++, state.flid)
@@ -0,0 +1,24 @@
+# Oracle 11g 专用迁移位置（ACM2-29 P3-B，门控未激活）
+
+本目录是现场 11g 部署形态的 Flyway migration location 占位。当前 **不接入任何环境的
+Flyway 配置**；PG 路径使用 `classpath:db/migration`，两者互不混用。
+
+## 激活前置条件（全部满足才允许接入）
+
+1. 现场 11.2 补丁级别、数据库字符集、DBA 权限清单拿到，且可提供可测试的目标库。
+2. JDK 25 × ojdbc 驱动（具体版本）× Flyway Oracle 支持 × 连接池组合在目标库实测通过
+   ——不能以"PG 通过"代替 Oracle 验收（docs/flight-state-design-v2.md §6）。
+3. `SqlDialect` 切到 `Oracle11gDialect` 前，MERGE 绑定顺序适配完成并通过
+   `FlightSchdJdbcPgTest` 同等粒度的 11g 集成测试。
+
+## 计划内容
+
+- V1.0.0 起：管道表（PROC_STATE/MSG_EVENT/PUMP_JOB/REQ_TRACK/REF_MASTER/PIPELINE_LOCK）
+  的 11g DDL——`NUMBER`/序列、`TIMESTAMP WITH TIME ZONE`、`VARCHAR2` BYTE/CHAR 语义钉死。
+- FLIGHT_SCHD 宽表 + V1.3.0 等价明细表 DDL（无 `ON CONFLICT`/`LATERAL` 依赖）。
+- `SRVT_TEXT/VIPF_TEXT/MAFL_TEXT` 映射 `CLOB`（PG 为 TEXT）。
+- `flight_schd_display` 11g 版视图（无 LATERAL，用标量子查询）。
+- 空串按 NULL 的语义回归：命令层 presence 信息不得被 11g 空串语义吞掉
+  （docs/flight-state-semantics.md §5）。
+
+版本号与 PG location 各自独立推进，禁止复用版本号语义。