merge: feat/jvs-dw-complex-case-gate → main(jvs-dw 潜在治疗池不进召回 + 增量链路四处修复)

apps/pac-service/data/jvs-dw/manifest.yaml

@@ -115,9 +115,12 @@ sql_source:
     clinic_scope:
       from_table:  dw_group.fact_emr_treatment_out
       org_column:  organization_id
-    # 增量「反向拉主档」的来源表 —— 这些表有自己的 cursor,它们变了就要把对应患者的主档补拉回来
-    # (主档 cursor 是 last_visit_time,人不来诊拉不到)。前四张是历史默认(EMR 编辑等场景);
-    # fact_complex_cases_out 是本次新增:宿主开/关复杂病例时把人带进 cohort,跟进闸才跟得上。
+    # 增量「反向拉主档」的来源表 —— 这些表变了就把对应患者的主档补拉回来
+    # (主档 cursor 是 last_visit_time,人不来诊拉不到)。
+    # ⚠️ **仅 single-shot(非 cohort)模式走这条**。jvs-dw 日常增量是 cohort 模式,
+    #   靠 incremental.per_query 的 cursor 拼 UNION 分支列患者(见上面 fact_complex_cases_out
+    #   的注释),不经过反向拉。这里保持声明是为了两点:① 该列表原先四张表名硬编码在通用
+    #   service 里,挪进 yaml 才符合"宿主差异只在 yaml";② single-shot 路径仍然会用到。
     reverse_pull_from:
       - fact_appointment_out
       - fact_emr_treatment_out
@@ -151,8 +154,15 @@ sql_source:
 
     # ── 复杂病例(潜在治疗池)──
     # ⚠️ 本表**不产 fact、无 assembler**,拉它只为一件事:增量时把"病例变了但人没来诊"的患者
-    #   带进 cohort(见 cohort.reverse_pull_from)。真正的判定在上面主档 SQL 的标量子查询里。
+    #   带进 cohort。真正的判定在上面主档 SQL 的标量子查询里。
     #   不拉它的话:患者不来诊 → 主档 cursor(last_visit_time)拉不到 → 病例开/关 PAC 永远不知道。
+    #   ⭐ 生效的是下面 incremental.per_query 给它配的 cursor:listPatientPairs 会把每张
+    #     **配了 cursor 且有水位**的表拼成 UNION 分支来列患者(cohort 模式走这条,不走反向拉)。
+    #     2026-08-01 测试服实测:接入后 UNION 从 6 张变 7 张,窗口内 19 个"末次就诊在几个月前、
+    #     但病例刚变"的患者被正确带进 cohort 并写上闸。
+    #   ⚠️ **新表首轮不生效**:无历史水位 → cursorValue 为空 → 该分支被跳过(见
+    #     clickhouse-source.service.ts 的 `if (!cfg.cursorValue) continue`)。第一轮只建水位,
+    #     第二轮起才开始感知变化。上线时别把首轮的"没反应"当 bug。
     # changed_at:updated_gmt_at 有 44% 为 NULL(建后没改过,恰是刚入池的新病例),
     #   直接拿它当 cursor 会 `NULL > x` 恒 UNKNOWN 静默漏掉这批 → 必须 coalesce 到 created_gmt_at。
     fact_complex_cases_out: |

apps/pac-service/src/modules/sync/cold-import/manifest.schema.ts

@@ -126,6 +126,11 @@ export const ClickHouseSourceSchema = z.object({
       ///   患者不来诊则主档拉不到他 —— 但他的**事实**可能变了(EMR 被编辑、复杂病例被关闭)。
       ///   增量跑完后从这些表里收集患者键,反向补拉一次主档,保证"事实变了主档也在场"。
       ///
+      /// ⚠️ **只作用于 single-shot(非 cohort)模式** —— reversePullPatientMaster 仅在
+      ///   loadAllTables 里调用。cohort 模式下同一诉求由 listPatientPairs 的 UNION 分支满足
+      ///   (每张"配了 cursor 且有水位"的表各出一条分支来列患者),不经过反向拉;
+      ///   且主档在 cohort 模式下不再注入 cursor,被带进来的患者主档一定拉得到。
+      ///
       /// 【为什么要可配】此前是四张 jvs-dw 表名硬编码在 service 里,新增一张要改通用代码;
       ///   而"哪些表的变化该带出主档"本就是**宿主口径**(取决于该宿主哪些表有独立 cursor)。
       ///   不配 → 保持历史默认(见 DEFAULT_REVERSE_PULL_FROM),行为不变。

apps/pac-service/tests/cohort-only-patient.spec.ts

 import * as fs from 'node:fs';
 import * as os from 'node:os';
 import * as path from 'node:path';
-import { resolveOnlyPatientIds } from '../src/modules/sync/cold-import/clickhouse-source.service';
+import {
+  resolveOnlyPatientIds,
+  resolveOnlyPatientKeys,
+} from '../src/modules/sync/cold-import/clickhouse-source.service';
 
 /**
  * PAC_COHORT_ONLY_PATIENT(定向重摄名单)解析。
@@ -64,4 +67,42 @@ describe('PAC_COHORT_ONLY_PATIENT 解析', () => {
     process.env.PAC_COHORT_ONLY_PATIENT = '@/no/such/pids.txt';
     expect(() => resolveOnlyPatientIds()).toThrow(/不存在/);
   });
+
+  /**
+   * 命名空间维(manifest cohort.tenant_key_column 声明的那一维)。
+   *
+   * 患者主键只在命名空间内唯一 —— jvs-dw 的 patient_id 261067 在两个品牌下是两个人。
+   * 2026-08-01 测试服实测:7 万个纯 id 的定向名单列出 **140,566** 个 cohort key(翻倍),
+   * 一半是另一命名空间下的同号患者,白摄一倍数据。
+   * ⚠️ 维名不写死("brand" 只是 jvs-dw 填进 tenant_key_column 的值),单命名空间宿主写纯 id。
+   */
+  describe('命名空间维', () => {
+    test('纯 id → 无 tenant(单命名空间宿主,行为与改动前一致)', () => {
+      process.env.PAC_COHORT_ONLY_PATIENT = '1855960,1855959';
+      expect(resolveOnlyPatientKeys()).toEqual([{ key: '1855960' }, { key: '1855959' }]);
+    });
+
+    test('`id|命名空间` → 复合键', () => {
+      process.env.PAC_COHORT_ONLY_PATIENT = '261067|瑞尔,261067|瑞泰';
+      expect(resolveOnlyPatientKeys()).toEqual([
+        { key: '261067', tenant: '瑞尔' },
+        { key: '261067', tenant: '瑞泰' },
+      ]);
+    });
+
+    test('TAB 分隔 → 同义(SQL dump TSV 可直接喂)', () => {
+      tmpFile = path.join(os.tmpdir(), `pids-tsv-${Date.now()}.txt`);
+      fs.writeFileSync(tmpFile, '261067\t瑞尔\n261068\t瑞泰\n');
+      process.env.PAC_COHORT_ONLY_PATIENT = `@${tmpFile}`;
+      expect(resolveOnlyPatientKeys()).toEqual([
+        { key: '261067', tenant: '瑞尔' },
+        { key: '261068', tenant: '瑞泰' },
+      ]);
+    });
+
+    test('resolveOnlyPatientIds 仍只返回 key —— 旧调用点(是否定向模式的判定)不受影响', () => {
+      process.env.PAC_COHORT_ONLY_PATIENT = '261067|瑞尔,261068|瑞泰';
+      expect(resolveOnlyPatientIds()).toEqual(['261067', '261068']);
+    });
+  });
 });

apps/pac-service/tests/complex-case-gate-sql.spec.ts

@@ -127,6 +127,66 @@ describe('injectIncrementalCursor — 带派生列的主档 SQL', () => {
   });
 });
 
+describe('增量列患者的 UNION 分支 — 别名列 + 顶层表名', () => {
+  /// 复刻 listPatientPairs 里 incBranches 的构造(见 clickhouse-source.service.ts)
+  const branchOf = (q: string, cursorCol: string, cursorVal: string) =>
+    `SELECT patient_id, brand FROM (${priv.injectIncrementalCursor(q, cursorCol, cursorVal)})`;
+
+  test('⭐ 主档分支查的是主档表,不是 SELECT 列里子查询的那张', () => {
+    const b = branchOf(MASTER_SQL, 'last_visit_time', '2026-07-30 08:15:00');
+    // 2026-08-01 测试服 dry-run 真实翻车过:分支被拼成
+    //   SELECT patient_id, brand FROM dw_group.fact_complex_cases_out WHERE last_visit_time > …
+    // → 表名取自子查询、cursor 列取自主档,CH 报 Missing columns: 'last_visit_time' 'patient_id'
+    expect(b).not.toMatch(/FROM\s+dw_group\.fact_complex_cases_out\s+WHERE\s+last_visit_time/);
+    expect(b).toContain('FROM dw_group.fact_client_out');
+  });
+
+  test('⭐ 复杂病例分支能读到别名列 —— 物理表只有 customer_id,没有 patient_id', () => {
+    const q = `
+      SELECT customer_id AS patient_id, brand,
+             coalesce(updated_gmt_at, toDateTime(created_gmt_at)) AS changed_at
+      FROM dw_group.fact_complex_cases_out
+      WHERE (customer_id, brand) IN (
+        SELECT patient_id, brand FROM dw_group.fact_client_out WHERE last_visit_time IS NOT NULL
+      )`;
+    const b = branchOf(q, 'changed_at', '2026-07-30 08:15:00');
+    // 外层 SELECT patient_id 必须落在**子查询之上**(别名在那里才存在)
+    expect(b).toMatch(/^SELECT patient_id, brand FROM \(/);
+    expect(b).toContain('customer_id AS patient_id');
+  });
+});
+
+describe('定向名单分片 — ClickHouse max_query_size 256KiB', () => {
+  /// CH 服务端默认上限;超过即 `Syntax error: failed at position 262142`
+  const MAX_QUERY_SIZE = 256 * 1024;
+  const ID_CHUNK = 10_000;
+  const mkIds = (n: number) => Array.from({ length: n }, (_, i) => String(1_000_000 + i));
+
+  test('⭐ 7 万 id 单条 SQL 会超 256KiB —— 这正是 2026-08-01 定向补数 fatal 的原因', () => {
+    const oneShot = `${mkIds(70_283).map((id) => `'${id}'`).join(', ')}`;
+    expect(oneShot.length).toBeGreaterThan(MAX_QUERY_SIZE);
+  });
+
+  test('⭐ 按 10k 分片后每片都远低于上限', () => {
+    const ids = mkIds(70_283);
+    const chunks = Array.from({ length: Math.ceil(ids.length / ID_CHUNK) }, (_, i) =>
+      ids.slice(i * ID_CHUNK, (i + 1) * ID_CHUNK),
+    );
+    expect(chunks).toHaveLength(8);
+    for (const c of chunks) {
+      const clause = c.map((id) => `'${id}'`).join(', ');
+      expect(clause.length).toBeLessThan(MAX_QUERY_SIZE / 2); // 留一半余量给 SQL 其余部分
+    }
+    // 分片是**无损**切分:并集 = 原名单,不重不漏
+    expect(chunks.flat()).toEqual(ids);
+  });
+
+  test('名单不超过阈值时不分片(保持单条 SQL,行为不变)', () => {
+    const ids = mkIds(500);
+    expect(ids.length > ID_CHUNK).toBe(false);
+  });
+});
+
 describe('manifest 契约', () => {
   const raw = readFileSync(join(__dirname, '../data/jvs-dw/manifest.yaml'), 'utf-8');
   const manifest = ColdImportManifestSchema.parse(yaml.load(raw));