管道构建（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 秒；管道运行是同步执行（建 run → 执行完才返回），大管道可能接近或超过 30 秒——超时请用「运行历史」确认 run 状态

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. 响应立即返回 run 记录（status 通常为 running），前端 alert 管道运行已启动 run_id=<id> status=<status>；
3. 终态判定：展开该管道「运行历史」，刷新后查看 run 的 status 是否变为 success/failed；finished_at 非空即执行完毕；error 列展示失败原因（SQL 执行错误、连接失败、质量 error 违规、write_mode 不支持等）。
4. 成功路径：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 是同步执行；若管道步骤耗时超过 client.js 30 秒超时，前端请求中断但后端 run 仍在跑，返回的响应丢失；另外运行历史列表是手动刷新（点「运行」只在已展开该管道历史时自动 loadRuns），不会轮询。
• 排查步骤：① 重新点「运行历史」/「收起历史」再展开，强制重新 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 not implemented」

• 现象：选择 changelog 写模式运行管道，run 标 failed，错误信息包含 not implemented。
• 原因：changelog（按主键 upsert + 变更历史）是 P2 预留功能，syncToTargetObject 对 changelog 直接返回错误。
• 排查步骤：① 编辑管道把写入模式改为 append / append_only_new / snapshot_replace / snapshot_replace_and_remove；② 若必须变更历史，改用多条 insert/update 步骤手动实现，或等待后续版本。

9. 已知缺陷与边界