From 20aafb9ab03cf09362a94cf1ca4a04cd14a29a3a Mon Sep 17 00:00:00 2001 From: windyboy Date: Tue, 8 Sep 2026 15:18:13 +0800 Subject: [PATCH] =?UTF-8?q?feat(processing):=20=E5=BF=AB=E7=85=A7=E6=B5=81?= =?UTF-8?q?=E7=A8=8B=E9=94=81=E5=86=85=E5=A4=8D=E6=A0=B8=E6=94=B6=E6=95=9B?= =?UTF-8?q?=20+=20=E5=BC=82=E5=B8=B8=E6=98=BE=E5=BC=8F=E6=B8=85=E9=99=A4?= =?UTF-8?q?=E8=AF=AD=E4=B9=89=20+=20=E8=AF=AD=E4=B9=89=E7=9F=A9=E9=98=B5?= =?UTF-8?q?=E5=AE=9A=E7=A8=BF=20(ACM2-29=20P2-5/M3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - SnapshotFlow 收敛到 v2 §5 事务顺序：锁外仅解析校验；取得 PIPELINE_LOCK 后 锁内复核快照身份（last_message_id 重放短路）→ 锁内读代与当前航班态 → 计算 nextState → 写入 → CAS → 事件 → SUCCEEDED。锁内 CAS 失败属数据异常， 一律回滚 FAILED(INFRA)，删除「版本号相同即视为重放」的推断路径（评审 F7/F8 收口） - M3/F4：异常对象（FDIV/FRET/FLAB）显式清除语义——null 字面量/空对象/空串 翻译为 Clear 命令；FlightNextState.clearedKeys 携带本次清除键；增量路径按 映射把前缀标量列置 NULL；库内清除后读视图无该键（线格式保持键缺失，显式 清除表达留待阶段 2 按下游契约定） - StubFlightSchd 增量分支改为 nextState 权威整体替换（引擎已合并当前态）， 天然承载 Clear；flattenScalarsOnly 去除冗余循环（L2） - docs/flight-state-semantics.md：16 类结构语义矩阵定稿（P0-A 交付物落库） - 新测试：引擎 0 标记/异常清除/混排 fail-fast 单测；JDBC 异常清除回环用例 验证：MSGX_PG_PORT=5433 真实 PG ./gradlew test --rerun-tasks 105 用例 0 失败 0 跳过 --- docs/flight-state-semantics.md | 60 +++++++++++++++++ .../domain/flight/FlightNextState.kt | 2 + .../domain/flight/FlightStateEngine.kt | 24 ++++++- .../persistence/jdbc/JdbcPgRepositories.kt | 28 ++++++-- .../infra/stub/StubRepositories.kt | 10 +-- .../msgexchange/processing/SnapshotFlow.kt | 64 ++++++++++--------- .../domain/flight/FlightStateEngineTest.kt | 36 +++++++++++ .../persistence/jdbc/FlightSchdJdbcPgTest.kt | 45 +++++++++++++ 8 files changed, 223 insertions(+), 46 deletions(-) create mode 100644 docs/flight-state-semantics.md diff --git a/docs/flight-state-semantics.md b/docs/flight-state-semantics.md new file mode 100644 index 0000000..b442e13 --- /dev/null +++ b/docs/flight-state-semantics.md @@ -0,0 +1,60 @@ +# 运营航班字段语义矩阵（v2 权威口径） + +状态：定稿（ACM2-29 P2-5 收口，原 P0-A 交付物）。 +依据：[flight-state-design-v2.md](flight-state-design-v2.md) §3/§4/§9、[SIS_AODB_RMS-V0.1.md](legacy/SIS_AODB_RMS-V0.1.md)、`FlightStateEngine` 实现。 +本文是「当前有效规则」的唯一索引；decision-flight-state.md 中历史口径与此冲突时以本文为准。 + +## 1. 命令模型 + +解码/Handler 层产生显式命令，仓储不做字符串猜测： + +| 命令 | 标量 | 集合 | +|---|---|---| +| Unchanged（未出现） | 不修改 | 不修改 | +| Set(value) | 覆盖 | — | +| Clear | 删除该键，物理列置 NULL | 删除全部明细行 | +| Replace(list) | — | 同事务 DELETE + 批量 INSERT 整集合替换 | +| Apply(item, sourceSeq) | — | 按 SOURCE_SEQ 定位更新，缺失则追加 | + +- 序号属性为 `"0"` 的条目是协议显式清除标记：**仅当集合内全部条目均为 0** 时翻译为 Clear；与常规条目混排属非法结构，fail fast 拒绝。 +- 空数组 `[]` = Replace(空集)：明细行清空（区别于 Unchanged 的键缺失）。 +- 非法 JSON、未知形状、超容输入不得截断后成功。 + +## 2. 16 类结构映射 + +| 集合键 | 存储载体 | 序号属性 | 出现语义 | 显式清除 | Apply | +|---|---|---|---|---|---| +| GTDT 登机口 | `flight_gate` | GTNO | Replace | GTNO=0 全清 | 支持（GTNO 定位） | +| CKDT 值机柜台 | `flight_checkin` | CKNO | Replace | CKNO=0 全清 | 支持；同 CHKC 多舱位分配按 CKNO 区分，不按资源号去重 | +| CLDT 行李转盘 | `flight_belt` | CLNO | Replace | CLNO=0 全清 | 支持 | +| PSDT 计划机位 | `flight_stand_plan` | PSNO | Replace | PSNO=0 全清 | 支持 | +| CHDT 行李滑槽 | `flight_chute` | CHNO | Replace | CHNO=0 全清 | 支持 | +| DELY 延误 | `flight_delay` | 无（DLNO 非协议属性） | Replace（当前单有效延误 + 历史保留） | `[]` 清空 | **不支持**（协议无定位键；Apply 对 DELY 拒绝） | +| ABTM 靠撤桥 | `flight_bridge_op` | ASNO | Replace（操作全集） | ASNO=0 全清 | 支持（ASNO 定位；多次靠/撤桥各占一行） | +| CHOT 轮挡 | `flight_chock_op` | CSNO | Replace | CSNO=0 全清 | 支持（CHID=ON/OFF 为业务属性，非序号） | +| ROUT 计划航路 | `flight_route_point(route_kind=ROUT)` | RTNO | Replace | RTNO=0 全清 | 支持 | +| ERUT 扩展航路 | `flight_route_point(route_kind=ERUT)` | RTNO | Replace | RTNO=0 全清 | 支持 | +| FDIV 备降 | 主表前缀列 `FDIV_*` | — | Set（对象） | `"null"`/`{}`/空串 → Clear | — | +| FRET 返航 | 主表前缀列 `FRET_*` | — | Set | 同上 | — | +| FLAB 中止 | 主表前缀列 `FLAB_*` | — | Set | 同上 | — | +| SRVT 服务 | 主表 `SRVT_TEXT`（无损 JSON 文本） | — | Set | 同异常（Clear 置 NULL） | — | +| VIPF 贵宾 | 主表 `VIPF_TEXT` | — | Set | 同上 | — | +| MAFL 共享航班 | 主表 `MAFL_TEXT` | — | Set | 同上 | 一旦出现关联查询需求，升级为显式关系表（v2 §8.2） | + +公共明细列：`FLID` FK、`ORDINAL`（本集合内输入顺序，1 起）、`SOURCE_SEQ`（源序号字符串，不强转整数、不设唯一）、`RECORD_VERSION`、`CREATED_AT/UPDATED_AT`；PK `(FLID, ORDINAL)`。相同资源号不代表同一条分配记录，**禁止按资源号去重**。 + +## 3. 清除的库内与线上表达 + +- 库内：标量/异常/文本键 Clear → 对应列 `NULL`；集合键 Clear/Replace(空) → 明细行删除。 +- 线上（KAFKA_SCHD）：被清除的键**缺失**（不输出 `"[]"`/`"null"`）。下游契约如需显式清除表达（如 `"GTDT":[]`），由阶段 2 Handler 按下游契约决定，不在存储层擅自改变线格式（v2 §9.1）。 + +## 4. 快照（DNLD）与增量（FLOP/ADFT） + +- DNLD：外层标量字段级合并；集合为整包快照 Replace；集合键缺失 = Unchanged（不自动清空旧资源）。删除航班走 `SCHD_GEN` 差集 + `deleteDiffByDay`（FDAY 域化，ADFT/跨代迁移行受保护）。 +- FLOP：字段级合并；Handler 产生的集合键出现即全量替换。 +- 重放判定凭 `schd_gen.last_message_id`（快照）与 `identity_key`（消息去重）；版本号只是顺序令牌。锁内 CAS 失败一律回滚 `FAILED(INFRA)`。 + +## 5. 时区与空值 + +- SIS 时间串（`DDMONYYHHMM` 机场当地时）原值保存；类型化提升按查询需要另行定案（ACM2-29 遗留 3）。 +- PG 保存空字符串；Oracle 11g 空串按 NULL——命令层保留 presence 信息，清除一律走显式 Clear 命令，不把「缺失、空串、NULL」混同（v2 §6）。 diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightNextState.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightNextState.kt index 2e6a903..43faa7c 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightNextState.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightNextState.kt @@ -9,6 +9,8 @@ data class FlightNextState( val collections: Map>>, val stateVersion: Long, val lastMessageId: String, + /** 本次迁移中被显式清除的标量/异常/文本键（仓储据此写 NULL 列；快照全量替换下无意义）。 */ + val clearedKeys: Set = emptySet(), ) { fun toFlightFields(mapper: ObjectMapper = ObjectMapper()): FlightFields { val out = linkedMapOf() diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngine.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngine.kt index 04ac662..90f01bf 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngine.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngine.kt @@ -14,6 +14,17 @@ object FlightStateEngine { "GTDT", "CKDT", "CLDT", "PSDT", "CHDT", "DELY", "ABTM", "CHOT", "ROUT", "ERUT", ) + /** 1:0..1 单值异常结构：主表前缀标量列承载；显式 null 载荷 = 清除（v2 §3.2/§9.1）。 */ + val EXCEPTION_KEYS = setOf("FDIV", "FRET", "FLAB") + + /** 异常/单值载荷的显式清除形态：空串、null 字面量、空对象（Oracle 空串即 NULL 语义的显式来源）。 */ + private fun isClearPayload(value: String): Boolean { + val trimmed = value.trim() + if (trimmed.isEmpty() || trimmed == "null") return true + val node = runCatching { mapper.readTree(trimmed) }.getOrNull() ?: return false + return node.isNull || (node.isObject && node.size() == 0) + } + private val SEQ_ATTR = mapOf( "GTDT" to "GTNO", "CKDT" to "CKNO", @@ -46,6 +57,7 @@ object FlightStateEngine { ) } } + key in EXCEPTION_KEYS && isClearPayload(value) -> scalars[key] = ScalarCommand.Clear else -> scalars[key] = ScalarCommand.Set(value) } } @@ -62,12 +74,19 @@ object FlightStateEngine { val baseCollections = current?.collections?.mapValues { it.value.toMutableList() } ?.toMutableMap() ?: mutableMapOf() val baseVersion = current?.stateVersion ?: 0L + val cleared = mutableSetOf() commands.scalars.forEach { (key, cmd) -> when (cmd) { ScalarCommand.Unchanged -> Unit - is ScalarCommand.Set -> baseScalars[key] = cmd.value - ScalarCommand.Clear -> baseScalars.remove(key) + is ScalarCommand.Set -> { + baseScalars[key] = cmd.value + cleared.remove(key) + } + ScalarCommand.Clear -> { + baseScalars.remove(key) + cleared += key + } } } @@ -96,6 +115,7 @@ object FlightStateEngine { collections = baseCollections.mapValues { it.value.toList() }, stateVersion = nextVersion, lastMessageId = messageId, + clearedKeys = cleared.toSet(), ) } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt index 2e2b9a1..be93fa9 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/JdbcPgRepositories.kt @@ -518,8 +518,6 @@ class JdbcFlightSchdRepository( /** v2：仅标量/异常/文本列 flatten；集合键由明细表读写，不经过槽位列。 */ private fun flattenScalarsOnly(scalars: Map): Map { - val fields = linkedMapOf() - scalars.forEach { (key, value) -> fields[key] = value } val out = mutableMapOf() scalars.forEach { (key, value) -> when { @@ -527,7 +525,7 @@ class JdbcFlightSchdRepository( key in setOf("SRVT", "VIPF", "MAFL") -> out["${key}_TEXT"] = value } } - flattenExceptions(fields, out) + flattenExceptions(scalars, out) return out } @@ -936,15 +934,31 @@ class JdbcFlightSchdRepository( } } - /** v2 增量：仅更新出现的标量列 + 追踪字段。 */ + /** 显式清除键 → 需要置 NULL 的物理列（异常前缀列 / 文本列 / 普通标量列）。 */ + private fun clearColumnsFor(keys: Set): List = keys.flatMap { key -> + when { + key == "FDIV" -> listOf("FDIV_DDES", "FDIV_DDIR", "FDIV_REMC") + key == "FRET" -> listOf("FRET_REID", "FRET_RSN") + key == "FLAB" -> listOf("FLAB_ARES", "FLAB_RSN") + key == "SRVT" -> listOf("SRVT_TEXT") + key == "VIPF" -> listOf("VIPF_TEXT") + key == "MAFL" -> listOf("MAFL_TEXT") + key in SCALAR_KEY_SET -> listOf(key) + else -> emptyList() + } + } + + /** v2 增量：仅更新出现的标量列 + 显式清除列置 NULL + 追踪字段。 */ private fun upsertIncrementalScalars(state: com.gzzn.omms.msgexchange.domain.flight.FlightNextState, now: Instant) { val stored = flattenScalarsOnly(state.scalars) - if (stored.isEmpty()) { + val clearedColumns = clearColumnsFor(state.clearedKeys) + if (stored.isEmpty() && clearedColumns.isEmpty()) { updateTrackingFields(state, now) return } - val assignments = stored.keys.map { "$it = ?" }.toMutableList() - assignments += listOf("last_message_id = ?", "state_version = ?", "updated_at = ?") + val assignments = stored.keys.map { "$it = ?" } + + clearedColumns.map { "$it = NULL" } + + listOf("last_message_id = ?", "state_version = ?", "updated_at = ?") val sql = "UPDATE flight_schd SET ${assignments.joinToString(", ")} WHERE flid = ?" val sqlTimestamp = now.toSqlTimestamp() ds.update(sql) { ps -> diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt index 1c74464..4e49165 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/infra/stub/StubRepositories.kt @@ -335,21 +335,17 @@ class StubFlightSchd : FlightSchdRepository { ) detailCollections[state.flid] = state.collections } else { - val mergedScalars = (existing?.fields?.filterKeys { - it == "FLID" || it !in com.gzzn.omms.msgexchange.infra.persistence.jdbc.FlightDetailTables.DETAIL_COLLECTION_KEYS - } ?: emptyMap()) + scalarFields + // nextState 由引擎在当前态上合并而来，是完整权威态：直接整体替换（含清除语义） records[state.flid] = Record( flid = state.flid, fday = existing?.fday, - fields = mergedScalars, + fields = scalarFields, stateVersion = state.stateVersion, lastMessageId = state.lastMessageId, createdAt = existing?.createdAt ?: now, updatedAt = now, ) - val mergedCollections = (detailCollections[state.flid] ?: emptyMap()).toMutableMap() - state.collections.forEach { (key, items) -> mergedCollections[key] = items } - detailCollections[state.flid] = mergedCollections + detailCollections[state.flid] = state.collections } } } diff --git a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt index 0f6eadb..20325d4 100644 --- a/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt +++ b/src/main/kotlin/com/gzzn/omms/msgexchange/processing/SnapshotFlow.kt @@ -57,35 +57,39 @@ class SnapshotFlow( return } - val gen = flightSchd.getGen(day) - if (gen?.lastMessageId == messageId) { - txManager.inTransaction { - procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED) - } - safeBackfill(head, msg) - log.info("snapshot replay no-op id={} day={}", head.cminmsgsId, day) - return - } - - val expected = gen?.version ?: 0L - val newFlids = normalized.map { it.first }.toSet() - val delFields = gen?.flids?.minus(newFlids) ?: emptySet() - val newVersion = expected + 1L - - val nextStates = normalized.map { (flid, fields) -> - val current = flightSchd.findNextStateByFlid(flid) - val commands = FlightStateEngine.commandsFromFields(flid, fields, snapshotReplace = true) - FlightStateEngine.apply(current, commands, messageId, bumpVersion = true) - } - + // v2 §5 事务流程：锁外只做解析与整包校验；取得 PIPELINE_LOCK 后在锁内 + // 复核快照身份 → 读取当前代与当前航班态 → 计算 nextState → 写入 → 提交。 + var replayNoOp = false try { txManager.inTransaction { + // 锁内复核快照身份：同一消息已成功提交 → 重放短路（不加版本、不重复发事件） + val gen = flightSchd.getGen(day) + if (gen?.lastMessageId == messageId) { + procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED) + replayNoOp = true + return@inTransaction + } + + val expected = gen?.version ?: 0L + val newFlids = normalized.map { it.first }.toSet() + val delFields = gen?.flids?.minus(newFlids) ?: emptySet() + val newVersion = expected + 1L + + // 锁内读取当前航班态，计算 nextState（单写者互斥下读到的一定是已提交最新态） + val nextStates = normalized.map { (flid, fields) -> + val current = flightSchd.findNextStateByFlid(flid) + val commands = FlightStateEngine.commandsFromFields(flid, fields, snapshotReplace = true) + FlightStateEngine.apply(current, commands, messageId, bumpVersion = true) + } + flightSchd.persistNextStates(day, nextStates, snapshotReplace = true) if (delFields.isNotEmpty()) { flightSchd.deleteDiffByDay(day, delFields) } + // 锁内 CAS：expected 即锁内刚读到的最新版本，仍失败属数据异常—— + // 一律回滚进入可重试 FAILED(INFRA)，绝不凭版本号推断“已经是我的提交”（v2 §5） val casSuccess = flightSchd.putGenIfVersion( day = day, expected = expected, @@ -97,14 +101,9 @@ class SnapshotFlow( ), ) if (!casSuccess) { - val again = flightSchd.getGen(day) - val replayIdentity = again?.lastMessageId == messageId && again.version == newVersion - if (!replayIdentity) { - throw CasConflictException( - "gen-cas-conflict: expected=$expected current=${again?.version} msg=${again?.lastMessageId}", - ) - } - log.info("gen idempotent replay within tx id={} day={}", head.cminmsgsId, day) + throw CasConflictException( + "gen-cas-conflict: expected=$expected msg=$messageId (lock-held CAS must not fail)", + ) } val schdKind = msg.kind as? MsgKind.Schd @@ -128,8 +127,13 @@ class SnapshotFlow( procState.update(head.cminmsgsId, ProcStatus.SUCCEEDED) } + // 提交后补偿：重放短路同样补做回填（其待办可能仍在重试中） safeBackfill(head, msg) - log.info("snapshot SUCCEEDED id={} day={} flights={}", head.cminmsgsId, day, normalized.size) + if (replayNoOp) { + log.info("snapshot replay no-op id={} day={}", head.cminmsgsId, day) + } else { + log.info("snapshot SUCCEEDED id={} day={} flights={}", head.cminmsgsId, day, normalized.size) + } } catch (e: CasConflictException) { log.warn("gen CAS conflict -> FAILED(INFRA) id={} msg={}", head.cminmsgsId, e.message) procFailure.fail(head, ErrorClass.INFRA, e.message ?: "gen-cas-conflict") diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngineTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngineTest.kt index f0743da..66a36d3 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngineTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/domain/flight/FlightStateEngineTest.kt @@ -142,4 +142,40 @@ class FlightStateEngineTest { val next = FlightStateEngine.apply(current, commands, "m2", bumpVersion = true) assertEquals(emptyList>(), next.collections["GTDT"]) } + + @Test + fun `exception null payload translates to explicit clear and apply records cleared key`() { + val current = FlightNextState( + "F1", + mapOf("FLNO" to "CA100", "FRET" to """{"REID":"R1","RSN":"diverted"}"""), + emptyMap(), + 3L, + "m1", + ) + val commands = FlightStateEngine.commandsFromFields("F1", mapOf("FRET" to "null"), snapshotReplace = false) + assertEquals(ScalarCommand.Clear, commands.scalars["FRET"]) + + val next = FlightStateEngine.apply(current, commands, "m2", bumpVersion = true) + assertFalse(next.scalars.containsKey("FRET")) + assertTrue(next.clearedKeys.contains("FRET")) + assertEquals("CA100", next.scalars["FLNO"]) + + // 非空载荷仍为 Set，且清空后的下一跳 Set 恢复键（clearedKeys 不跨消息携带） + val restored = FlightStateEngine.apply( + next, + FlightStateEngine.commandsFromFields("F1", mapOf("FRET" to """{"REID":"R2"}"""), snapshotReplace = false), + "m3", + bumpVersion = true, + ) + assertEquals("""{"REID":"R2"}""", restored.scalars["FRET"]) + assertFalse(restored.clearedKeys.contains("FRET")) + } + + @Test + fun `exception clear accepts null literal empty object and empty string`() { + for (payload in listOf("null", "{}", "")) { + val commands = FlightStateEngine.commandsFromFields("F1", mapOf("FDIV" to payload), snapshotReplace = false) + assertEquals(ScalarCommand.Clear, commands.scalars["FDIV"], "payload=[$payload] must be Clear") + } + } } diff --git a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightSchdJdbcPgTest.kt b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightSchdJdbcPgTest.kt index 4bdaf45..b07ec2d 100644 --- a/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightSchdJdbcPgTest.kt +++ b/src/test/kotlin/com/gzzn/omms/msgexchange/infra/persistence/jdbc/FlightSchdJdbcPgTest.kt @@ -693,6 +693,51 @@ class FlightSchdJdbcPgTest { assertEquals(Triple("Z1", null, 1), after) } + @Test + fun `PG dialect - exception object explicit clear writes null columns and omits wire key`() { + val day = "2026-09-07" + val flid = "TEST_EXC_CLEAR" + val engine = com.gzzn.omms.msgexchange.domain.flight.FlightStateEngine + + // 1. 快照写入返航异常对象 → 主表前缀标量列 + repo.persistNextStates( + day, + listOf( + engine.apply( + null, + engine.commandsFromFields(flid, mapOf("FRET" to """{"REID":"CA108","RSN":"diverted"}"""), snapshotReplace = true), + "msg-exc-1", bumpVersion = true, + ), + ), + snapshotReplace = true, + ) + val stored1 = ds.queryOne( + "SELECT fret_reid, fret_rsn FROM flight_schd WHERE flid = ?", + { ps -> ps.setString(1, flid) }, + ) { rs -> listOf(rs.getString("fret_reid"), rs.getString("fret_rsn")) } + assertEquals(listOf("CA108", "diverted"), stored1) + + // 2. 增量显式清除（FRET="null"）→ 列置 NULL，读视图无 FRET 键（评审 F4 修复闭环） + repo.persistNextStates( + null, + listOf( + engine.apply( + repo.findNextStateByFlid(flid), + engine.commandsFromFields(flid, mapOf("FRET" to "null"), snapshotReplace = false), + "msg-exc-2", bumpVersion = true, + ), + ), + snapshotReplace = false, + ) + val stored2 = ds.queryOne( + "SELECT fret_reid, fret_rsn FROM flight_schd WHERE flid = ?", + { ps -> ps.setString(1, flid) }, + ) { rs -> listOf(rs.getString("fret_reid"), rs.getString("fret_rsn")) } + assertEquals(listOf(null, null), stored2) + assertNull(repo.findByFlid(flid)!!["FRET"]) + assertNull(repo.findNextStateByFlid(flid)!!.scalars["FRET"]) + } + private fun assertLegacyCollectionColumnsAllNull(flid: String) { val columnList = com.gzzn.omms.msgexchange.support.LEGACY_COLLECTION_STORAGE_COLUMNS.joinToString(", ") ds.queryOne(