管道构建（Pipelines）

1. 页面概览

1.1 是什么

管道构建（PipelinesPage.vue，路由 /foundry/pipelines）是 LightFoundry 的「SQL 步骤声明式管道」管理界面，对标 Palantir Foundry 的 Pipeline Builder（Transforms）。与可视化 DAG 画布不同，本实现采用单数据源内有序 SQL 步骤：一条管道 = 一个源连接器（source_connector_id）+ 一串顺序执行的 SQL 步骤（sql_steps，类型 select/insert/update/create_table_as）+ 可选的写入模式（write_mode）+ 可选的触发方式（manual/cron）+ 可选的启用开关。管道可真实运行：后端建运行记录 → 连接真实数据源 → 顺序执行 SQL 步骤 → 质量规则检查 → 按 write_mode 同步到目标对象 → 落库运行状态与行数。

页面同时提供运行历史查看：选中管道可展开该管道的运行记录表（状态/rows_in/rows_out/错误），用于验收每次运行的结果。V5 Stage 0 起，cron 触发的管道在保存时同步注册到平台统一调度器（platform/scheduler），实现真正的定时调度。

1.2 核心价值

1.3 一句话总结

2. 访问入口

2.1 路由与菜单

• 路由：/foundry/pipelines
• 路由名称：FoundryPipelines
• 菜单位置：Foundry 左侧边栏「管道构建」（FoundryLayout 菜单「数据集成」区块）
• 源码文件：action/web/src/views/PipelinesPage.vue（约 486 行）

2.2 认证与权限

• 需要登录：是（路由 meta.requiresAuth = true）
• 登录方式：AIP 统一登录，client.js 请求拦截器自动附带 Authorization: Bearer <aip_token>；401 时响应拦截器清理凭证并跳转 /login
• 页面无角色限制；管道执行读取的真实数据源连接配置由后端连接器工厂白名单控制（默认仅放行 MYSQL/SQLSERVER/POSTGRESQL，SQLITE 需显式授权）
• 404 排错：访问 /foundry/pipelines 404，先确认 Foundry 后端（18081）已启动且路由已注册（router/index.js 第 233-237 行）

2.3 端口与 API 前缀

• 后端端口：18081（Foundry）
• API 前缀：/api/v1（action/web/src/api/client.js）
• 请求超时：30 秒；管道运行已任务化/异步化——POST /pipelines/:id/run 在 15s 快路径内同步完成则 200 返回 run 信息，超时返回 202 {task_id,status:"running",type:"pipeline_run"}，本页随即退避轮询 GET /system/tasks/:id（1.5s 起 ×1.5 封顶 10s，上限 5 分钟，超时后露出「继续轮询」按钮）直至终态，大管道不再触发前端 30s 超时

3. 界面布局

┌──────────────────────────────────────────────────────────────┐
│ ① 页头：「管道构建」标题 + 「新建管道」/「收起表单」按钮         │
│ ② 操作结果提示条（alert 成功/失败，可「关闭」）                 │
│ ③ 创建/编辑表单卡片（点「新建管道」或「编辑」才展开）           │
│    ├─ 名称/源连接器 ID/目标对象类型 ID/写入模式/触发类型/cron  │
│    ├─ 描述 + 启用勾选                                          │
│    └─ SQL 步骤动态列表（+ 步骤 / 每步名称+类型+SQL+删除）       │
│       └─ [保存] [取消]                                         │
│ ④ 管道列表卡片（表格）                                         │
│    └─ ID/名称/描述/写入模式/触发/启用/操作（运行/继续轮询*/    │
│       停用/编辑/删除/运行历史）  *轮询超时后条件出现            │
│ ⑤ 运行历史卡片（选中管道展开）                                 │
│    └─ ID/状态/触发/rows_in/rows_out/开始/结束/错误             │
└──────────────────────────────────────────────────────────────┘

各板块职责：

• ① 页头：标题 + 新建/收起按钮（同一按钮切换）。
• ② 提示条：全局唯一反馈通道，showAlert(msg, type) 驱动。
• ③ 创建/编辑表单：showForm 控制显隐；editingPipeline 非空时标题为「编辑管道」，否则「新建管道」。编辑时名称（api_name）输入框 disabled（不可改）。
• ④ 管道列表：fetchPipelines 加载 GET /pipelines，空列表提示「暂无管道，请点击"新建管道"添加。」；每行操作按钮齐全。
• ⑤ 运行历史：点某管道「运行历史」展开该管道 run 列表（showRunsId 记录当前展开管道），再点变「收起历史」；runs 为空显示「暂无运行记录」。

4. 交互元素详解

4.1 页头与表单

4.2 管道列表操作

4.3 运行历史表

5. 后端关联

5.1 API 客户端

action/web/src/api/client.js：baseURL /api/v1、timeout 30000、Bearer aip_token、401 跳登录（同指标页，复用同一实例，PipelinesPage 直接 import apiClient from '../api/client.js'）。

5.2 端点表

5.3 响应结构

统一成功包装 {code: 0, data: ...}（okData）；失败 {code, error}（errorResponse）。

创建成功：

{ "code": 0, "data": { "id": 3 } }

管道列表（GET /pipelines）：

{
  "code": 0,
  "data": [
    {
      "id": 3,
      "name": "orders_sync",
      "description": "",
      "source_connector_id": 1,
      "target_object_type_id": 2,
      "sql_steps": [ { "name": "extract", "sql": "SELECT * FROM orders", "transform_type": "select" } ],
      "write_mode": "append",
      "trigger_type": "cron",
      "cron_expr": "0 2 * * *",
      "enabled": true
    }
  ]
}

运行触发（POST /pipelines/:id/run）：

{
  "code": 0,
  "data": {
    "id": 17,
    "pipeline_id": 3,
    "status": "running",
    "started_at": "2026-08-30T02:00:00Z",
    "rows_in": 0,
    "rows_out": 0,
    "triggered_by": "web"
  }
}

运行历史（GET /pipelines/:id/runs）：

{
  "code": 0,
  "data": [
    { "id": 17, "pipeline_id": 3, "status": "success", "started_at": "...", "finished_at": "...", "rows_in": 1000, "rows_out": 1000, "error": "", "triggered_by": "web" }
  ]
}

5.4 关联模块表

5.5 关键机制

• RunPipeline 七步执行：① 校验管道存在且 enabled（disabled 报「pipeline "xx" is disabled」）；② 建 run 记录（status=running）→ ③ 连真实数据源（connector 工厂创建后需显式 Connect）→ ④ 顺序执行 SQL 步骤：select 走 ExecuteSQL（读入行数 += 返回行数），create_table_as 走 ExecWrite 后 count 该表，insert 计入写入行数、update 不计（DDL RowsAffected 不可靠）→ ⑤ 质量规则执行（scope=pipeline 且 enabled），违规记 issue，severity=error 违规计数 → ⑥ 目标对象同步（write_mode 落盘）→ ⑦ 更新 run 状态与 rows_in/rows_out；success 后自动打血缘。
• write_mode 语义（目标对象同步，G17）：
  • append：INSERT INTO 目标表 SELECT * FROM 结果表；
  • append_only_new：按对象主键 WHERE pk NOT IN (SELECT pk FROM 目标表) 仅插新行；对象无主键时降级为 append；
  • snapshot_replace / snapshot_replace_and_remove：DELETE FROM 目标表 后全量 INSERT（「移除已删除行」由全量替换天然覆盖）；
  • changelog：按主键 upsert + 变更历史，P2 未实现，运行时返回提示性错误使 run 标 failed。
• 质量阻断：error 级违规使 run 标 failed 并阻断目标对象同步（问题隔离）；无目标对象的管道其 SQL 步骤结果保留（仅告警 + 问题隔离）。
• rows_out 语义：有结果表时重算（COUNT(*)），select-only 管道（无结果表）保持 rows_out=rows_in。
• 调度器同步：创建/更新/启停/删除管道时 syncSchedule 幂等注册/反注册调度条目（pipeline:<id>）；triggeredBy="cron" 复用 RunPipeline；cron 解析失败的管道记日志跳过，不阻断启动。
• 血缘打点：run 成功后 RecordLineage 打点源表→管道（单源表携带字段级映射）与管道→对象；血缘失败/未注入不阻断 run。

6. 核心流程详解

6.1 创建一条管道（主流程）

1. 点「新建管道」展开表单；
2. 填名称（api_name）、源连接器 ID（数据源 id，必填）、目标对象类型 ID（可选）；
3. 选写入模式（默认 append）与触发类型（默认 manual；选 cron 则补 cron 表达式）；
4. 填描述、保持「启用」勾选；
5. 「SQL 步骤（至少 1 个）」区点「+ 步骤」，填写步骤名、选变换类型（select/insert/update/create_table_as）、写 SQL；
6. 点「保存」：前端校验——过滤出「名称与 SQL 均非空」的步骤，无有效步骤则 alert「至少需要 1 个有效 SQL 步骤（名称与 SQL 均需填写）」；cron 类型无表达式则 alert「cron 触发类型需要填写 cron 表达式」；
7. 创建走 POST /pipelines（附 name），成功 alert 管道创建成功 id=<id> 并刷新列表；后端同步注册 cron 调度。

6.2 编辑管道

• 列表「编辑」→ 表单回填（含 sql_steps 逐项映射 transform_type 缺省 select）→ 修改后可编辑字段 → 保存走 PUT /pipelines/:id（不含 name，name 前端 disabled）→ 成功 alert「管道已更新」并刷新。
• 后端 UpdatePipeline 重新校验 sql_steps/source_connector_id/cron_expr/write_mode/target 存在性，并 syncSchedule 同步调度条目。

6.3 运行与终态判定（后端任务化 + 前端退避轮询）

1. 点「运行」→ confirm → POST /pipelines/:id/run（body {triggered_by:'web'}）；
2. 后端任务化双响应：
   • 快路径 200：15s 内执行完成，data 为 run 记录（含 task_id），alert 管道运行已启动 run_id=<id> status=<status>；
   • 超时受理 202：执行超过 15s 转入后台，data={task_id,status:"running",type:"pipeline_run"}，alert 管道已受理，正在执行（task_id=…）并启动轮询。
3. 轮询终态（退避 + 封顶 + 超时后继续轮询）：轮询 GET /system/tasks/:id，延迟 1.5s 起、每次 ×1.5、封顶 10s，总上限 5 分钟，直至：
   • success：alert 管道运行成功 run_id=…（run 记录在 task.result），已展开该管道历史则自动刷新；
   • failed/cancelled：alert-error 展示 task.error/task.message，并自动刷新运行历史；
   • 单次轮询请求失败不中断任务，按退避节奏继续重试；
   • 到达 5 分钟上限仍未终态：alert 提示「轮询已达 5 分钟上限；可点击「继续轮询」获取终态」，并在该管道行露出「继续轮询」按钮，由用户一键续轮询直至终态。
4. 轮询期该管道「运行」按钮显示「运行中」并禁用；组件卸载自动停止轮询（onBeforeUnmount(stopRunPolling) 清定时器并置取消标记）。该模式与「实体消解」（FusionPage.vue）的运行轮询一致。
5. 成功路径：rows_in/rows_out 反映实际读入/写出行数，且自动打血缘（可在「数据血缘」页按管道名查上游/下游）。

6.4 启用/停用与删除

• 启停：POST /pipelines/:id/enable|disable，成功提示并刷新；后端同步注册/反注册调度条目；停用后 RunPipeline 直接拒绝执行（报 disabled）。
• 删除：confirm 后 DELETE /pipelines/:id，运行历史保留；若该管道历史已展开则前端自动收起。后端删除定义并反注册调度条目（scheduler_runs 历史保留）。

6.5 cron 定时调度流

• 触发类型选 cron 并填表达式 + 勾选启用 → 保存后 syncSchedule 注册到平台统一调度器；
• foundry 启动时 RegisterScheduled 遍历存量 enabled+cron 管道批量注册；
• 每次触发 RunPipeline(ctx, id, "cron")，失败信息经返回 error 落 scheduler_runs 供排障；
• 停用/删除/改为 manual/清空表达式都会反注册调度条目。

7. 权限与安全

• 认证：AIP 统一登录（aip_token）；路由 requiresAuth；401 跳登录。
• 数据源访问控制：管道执行使用 source_connector_id 对应的真实数据源连接配置，后端经 connector.CreateConnector 工厂创建——工厂 allow-list 默认仅放行 MYSQL/SQLSERVER/POSTGRESQL，SQLITE 需显式授权；SQL 步骤直接执行用户书写的 SQL（白名单/参数化由连接器保证），因此管道定义能力等同于授予该数据源的高权限执行，创建/编辑管道应视为敏感写操作。
• 写操作防护：运行、删除均有 confirm 二次确认；保存/运行/启停/删除均 busy 置位防连点；表单对 cron 表达式与步骤做前端必填校验。
• 质量红线：error 级质量违规会阻断目标对象同步写入（问题隔离），防止脏数据落目标对象。
• 注入与标识符：目标对象同步 SQL 中表名经 normalizeTable 去包裹符号后拼接；对象主键列取自本体定义（非用户自由输入）。

8. 常见问题与排错

8.1 点「运行」后 alert「管道保存失败」或「运行失败」且无明确错误

• 现象：运行管道报错或保存报错，提示文本只有通用「运行失败：...」。
• 原因：最常见是后端 18081 未启动（网络层 error）、source_connector_id 指向的数据源不存在/类型未授权（如 SQLITE 未在连接器工厂 allow-list 放行）、或 SQL 步骤语法错误。错误详情在后端返回的 err.response?.data?.error，前端只把它拼进 alert。
• 排查步骤：① 确认 Foundry 后端进程在跑（18081）；② 用浏览器 DevTools Network 查看 POST /pipelines/:id/run 的响应体 error 字段具体文案；③ 若为「data source x not found」→ 到「数据源」页确认 id 正确；若为「pipeline "xx" is disabled」→ 先「启用」再运行；④ 若为 SQL 报错 → 编辑管道修正步骤 SQL 后重试。

8.2 运行历史一直停在「running」或状态不更新

• 现象：展开「运行历史」后某条 run 始终显示 running，或点「运行」后长时间无反馈。
• 原因：POST /run 已任务化（15s 快路径 200 / 超时 202 task_id），页面正常情况下会自动退避轮询至终态并提示结果；若仍停在 running，多为后端 SQL 步骤执行缓慢/卡死（连接器挂起），或已超过页面轮询上限（5 分钟）而暂停自动轮询（管道仍在后台执行）。
• 排查步骤：① 点「运行」后若看到「管道已受理，正在执行（task_id=…）」属正常受理提示，等待终态会 alert 管道运行成功/失败；② 若看到「轮询已达 5 分钟上限」提示，点该管道行的「继续轮询」按钮续轮询（无需重新触发运行）；③ 也可重新点「运行历史」/「收起历史」再展开，强制重新 GET /pipelines/:id/runs；④ 若仍是 running 且时间很久，检查后端日志中该 run 的 SQL 步骤是否卡死（连接器挂起）；⑤ 正常终态应显示 success/failed 且 finished_at 非空；⑥ 不要重复点「运行」避免产生多份 run。

8.3 保存管道报「pipeline "xx" already exists」或重复键错误

• 现象：创建管道保存失败，提示重复键。
• 原因：name 字段有唯一索引，后端 CreatePipeline 先 GetByName 校验；若同名管道已存在（含刚删除、历史保留的）则拒绝。
• 排查步骤：① 列表确认是否已存在同名管道；② 删除是物理删除定义（运行历史保留），删除后同名可重建；③ 名称必须唯一，编辑时 name 不可改（前端 disabled）。

8.4 cron 管道不按时触发

• 现象：管道触发类型为 cron、启用中，但到点没跑。
• 原因：cron 调度依赖 platform/scheduler 注入（V5 Stage 0）；若该进程未注入调度器（nil）则 cron_expr 不激活；或表达式非法导致注册失败（启动时记日志跳过）；或管道被停用（停用即反注册调度条目）。
• 排查步骤：① 确认管道 enabled=true 且 trigger_type=cron、cron_expr 非空；② 检查后端启动日志是否有「管道 cron 注册失败」warn；③ 表达式须为合法 cron（如 0 2 * * *）；④ 改保存一次管道（触发 syncSchedule 重注册）后再观察；⑤ 定时触发的 run 在运行历史中 triggered_by=cron，可据此确认。

8.5 保存/运行管道报「write_mode=changelog 暂未实现」

• 现象：选择 changelog 写模式保存（创建/更新）管道，保存失败并提示 write_mode=changelog 暂未实现，请使用 append/append_only_new/snapshot_replace/snapshot_replace_and_remove；存量 changelog 管道运行时 run 标 failed。
• 原因：changelog（按主键 upsert + 变更历史）是 P2 预留功能，创建/更新即被后端显式 400 拒绝（pipeline/service.go:415-416 显式返回 InvalidRequestError，属设计如此、非缺陷；W2-01 起改为显式拒绝以免静默按普通写入误导用户），不会进入可运行态；表单下拉中 changelog 亦禁用并标注「暂未实现」；运行时兜底错误仅对存量 changelog 管道可见。
• 排查步骤：① 编辑管道把写入模式改为 append / append_only_new / snapshot_replace / snapshot_replace_and_remove；② 若必须变更历史，改用多条 insert/update 步骤手动实现，或等待后续版本。

9. 已知缺陷与边界