fix(sync): 定向名单分片查询 —— 7 万 id 撑爆 ClickHouse max_query_size

测试服定向补数(PAC_COHORT_ONLY_PATIENT=@file,70,283 个 id)实跑 fatal:
  Syntax error: failed at position 262142
262142 = 256 KiB,ClickHouse max_query_size 默认上限。7 万 id 拼 IN (...) 约 630KB,
超 2.4 倍;重试 3 次全败,patients upserted: 0(没写坏任何数据)。

⚠️ 这与 resolveOnlyPatientIds 的 `@file` 是**两个不同的上限**,之前混为一谈了:
  · @file 解的是环境变量 128KB(E2BIG)—— 传参侧
  · 本次是 SQL 文本长度 —— 服务端解析侧
文件读进来了,SQL 照样超。注释里"大名单走文件"的承诺此前并不成立。

改为按 10k 分片跑(≈90KB/片,离上限有充足余量),其余条件(cursor / clinics / union 分支)
每片原样带上,结果用 Map 按 (key, tenant) 去重合并 —— 与单条 SQL 的结果集等价。
名单未超阈值时仍走单条 SQL,行为不变。

分片时不套 orderTail 的 LIMIT:PAC_COHORT_LIMIT 采样与分片叠加会"每片各取 N"而超量,
且定向重摄本就是显式点名,不该再被采样截断。

补 3 项回归(含"7 万单条必超上限"的反证),共 18 项。

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

apps/pac-service/src/modules/sync/cold-import/clickhouse-source.service.ts

@@ -550,12 +550,58 @@ export class ClickHouseSourceService {
       const selectCols =
         (tenant_key_column ? `${patient_key_column}, ${tenant_key_column}` : patient_key_column) + extraSelect;
       const listFrom = unionMode ? `(${incBranches.join(' UNION ALL ')})` : patient_list_from;
+      // ⭐ 定向名单分片跑 —— ClickHouse `max_query_size` 默认 **256 KiB**,一条 SQL 装不下大名单。
+      //   2026-08-01 测试服实测:7 万个 id 拼成 `IN (...)` 约 630KB →
+      //     `Syntax error: failed at position 262142`(= 256 KiB 边界),重试 3 次全败、整跑 fatal。
+      //   ⚠️ 这跟 resolveOnlyPatientIds 的 `@file` 是**两个不同的上限**:那个解的是环境变量
+      //     128KB(E2BIG,传参侧),这个是 SQL 文本长度(服务端解析侧)。文件读进来了,SQL 还是超。
+      //   分片只切 ONLY_PATIENT 这一维:其余条件(cursor / clinics / union 分支)每片原样带上,
+      //   DISTINCT 在合并后用 Map 去重 —— 与单条 SQL 的结果集完全等价。
+      //   注:PAC_COHORT_LIMIT 采样与分片叠加时按"每片各取 LIMIT"处理会超量,故分片时不走
+      //   orderTail 的 LIMIT(定向重摄本就是显式点名,不该再被采样截断)。
+      const ID_CHUNK = 10_000; // 10k × ~9B ≈ 90KB,离 256KiB 有足够余量(id 变长也不会顶到)
+      const idChunks: string[][] =
+        ids.length > ID_CHUNK
+          ? Array.from({ length: Math.ceil(ids.length / ID_CHUNK) }, (_, i) =>
+              ids.slice(i * ID_CHUNK, (i + 1) * ID_CHUNK),
+            )
+          : [];
+      const started = Date.now();
+      let rows: Array<Record<string, unknown>>;
+      if (idChunks.length > 0) {
+        this.logger.log(
+          `[clickhouse·cohort] 定向名单 ${ids.length} 个 > ${ID_CHUNK} → 分 ${idChunks.length} 片查询(CH max_query_size 256KiB)`,
+        );
+        const merged = new Map<string, Record<string, unknown>>();
+        for (const [i, chunk] of idChunks.entries()) {
+          const quotedChunk = chunk.map((id) => `'${id.replace(/'/g, "''")}'`).join(', ');
+          // 用本片的 IN 替换掉整名单那一条(其余 whereParts 保持不变)
+          const chunkWhere = whereParts.map((w) =>
+            w.startsWith(`${patient_key_column} IN (`) || w.startsWith(`${patient_key_column} = `)
+              ? `${patient_key_column} IN (${quotedChunk})`
+              : w,
+          );
+          const chunkSql = `SELECT DISTINCT ${selectCols} FROM ${listFrom} WHERE ${chunkWhere.join(' AND ')}`;
+          const part = (await this.queryJsonWithRetry(
+            client,
+            chunkSql,
+            `list-patient-keys[${i + 1}/${idChunks.length}]`,
+          )) as Array<Record<string, unknown>>;
+          for (const r of part) {
+            merged.set(
+              `${String(r[patient_key_column] ?? '')}|||${tenant_key_column ? String(r[tenant_key_column] ?? '') : ''}`,
+              r,
+            );
+          }
+        }
+        rows = [...merged.values()];
+      } else {
         const sql = `SELECT DISTINCT ${selectCols} FROM ${listFrom}${whereSql}${orderTail}`;
         this.logger.log(
           `[clickhouse·cohort] list patient keys${unionMode ? `(union of ${incBranches.length} cursored tables)` : ''} — ${sql.slice(0, 300)}`,
         );
-      const started = Date.now();
-      const rows = (await this.queryJsonWithRetry(client, sql, 'list-patient-keys')) as Array<Record<string, unknown>>;
+        rows = (await this.queryJsonWithRetry(client, sql, 'list-patient-keys')) as Array<Record<string, unknown>>;
+      }
       const keys: CohortKey[] = rows.map((r) => ({
         key: String(r[patient_key_column] ?? ''),
         tenant: tenant_key_column ? String(r[tenant_key_column] ?? '') : undefined,

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

@@ -156,6 +156,37 @@ describe('增量列患者的 UNION 分支 — 别名列 + 顶层表名', () => {
   });
 });
 
+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));