Pipeline Spec

表达 Pipeline 的定义语义。name 为自然键；dagConfig 为 DAG JSON，运行时被反序列化为 PipelineDagDTO；INPUT 节点中的 datasetId 支持按数据集名称解析。

资源类型与自然键

资源类型	自然键	说明
`Pipeline`	`name`	Pipeline 名称，全局唯一

PipelineSpecDTO

字段	类型	说明
`name`	string	名称，作为自然键
`displayName`	string	显示名称占位字段
`description`	string	描述
`runOnInstall`	boolean	是否随包安装自动运行；缺省视为 `true`，显式 `false` 时不进入 `installPlan` 运行节点
`dagConfig`	Map<String, Object>	DAG 配置 JSON，运行时被反序列化为 `PipelineDagDTO`

DAG 结构（PipelineDagDTO）

字段	类型	说明
`nodes`	List<PipelineNodeDTO>	节点列表
`edges`	List<PipelineEdgeDTO>	边列表，描述节点间的数据流向

节点（PipelineNodeDTO）

字段	类型	说明
`id`	string	节点唯一标识
`type`	string	算子类型：`INPUT` / `FILTER` / `JOIN` / `AGGREGATE` / `OUTPUT` / `PROJECTION` / `UNION`
`name`	string	节点显示名称
`description`	string	节点业务描述（可选）
`config`	NodeConfig	节点多态配置，根据 `type` 反序列化为对应子类型

边（PipelineEdgeDTO）— Source / Target

字段	类型	说明
`id`	string	连线唯一标识
`source`	string	起始节点（上游）的 `id`
`target`	string	目标节点（下游）的 `id`
`sourceHandle`	string	起始节点的输出端口标识
`targetHandle`	string	目标节点的输入端口标识；适用于多输入节点，如 JOIN 的 left/right

节点配置

INPUT（数据源输入）

字段	类型	说明
`datasetId`	string	引用的数据集 ID/名称，按名称解析
`columns`	List<string>	需要读取的列名列表

OUTPUT（数据输出）

字段	类型	说明
`writeMode`	string	未使用。写入模式占位字段：`OVERWRITE` / `APPEND`。
`targetTableName`	string	底层数据源中生成的物理表名
`registerDataset`	boolean	是否自动注册为系统新资产，默认 `false`
`targetCatalogId`	string	未使用。新注册资产所属分类 ID 占位字段。
`materializationType`	string	物化类型：`MATERIALIZED` / `LOGICAL`，默认 `LOGICAL`

FILTER（行过滤）

字段	类型	说明
`root`	FilterGroup	过滤表达式根节点

FilterRule

字段	类型	说明
`column`	string	参与过滤的列名
`operator`	string	操作符，如 `=`、`>`、`<`、`>=`、`<=` 等
`value`	any	比较值，可为 string、number、boolean 或 array

FilterGroup

字段	类型	说明
`condition`	string	逻辑组合：`AND` / `OR`
`children`	List<FilterExpression>	子表达式列表，元素为 FilterRule 或 FilterGroup

JOIN（表关联）

字段	类型	说明
`joinType`	string	关联类型：`LEFT` / `INNER` / `RIGHT` / `FULL`
`leftKey`	string	左表关联字段
`rightKey`	string	右表关联字段
`leftKeys`	List<string>	左表关联字段列表（复合键），与 `rightKeys` 按位置配对；配置后优先于 `leftKey`/`rightKey`
`rightKeys`	List<string>	右表关联字段列表（复合键），与 `leftKeys` 按位置配对

复合键条件以 AND 连接，如 l.A = r.A AND l.B = r.B。leftKeys/rightKeys 长度相等且非空时生效，否则回退单键。

AGGREGATE（聚合）

字段	类型	说明
`groupBy`	List<string>	分组字段列表（GROUP BY 维度）
`aggregations`	List<AggregateRule>	聚合函数配置列表

AggregateRule

字段	类型	说明
`column`	string	参与计算的列名
`func`	string	聚合函数：`SUM` / `COUNT` / `AVG` / `MAX` / `MIN`
`alias`	string	结果重命名字段

PROJECTION（列投影）

字段	类型	说明
`columns`	string	投影的列名字符串，通常以逗号分隔

转换算子（Transformations）

Pipeline 通过不同类型的节点表达数据转换：

算子类型	作用
`INPUT`	从数据集读取源数据
`FILTER`	按条件过滤数据行
`JOIN`	多路数据关联
`AGGREGATE`	分组聚合计算
`PROJECTION`	选择并透传指定列
`UNION`	将两个及以上上游分支按列名纵向合并
`OUTPUT`	将结果写入目标数据集或物理表

UNION（纵向合并）

字段	类型	说明
`unionType`	string	合并类型：`UNION_ALL`（不去重，默认）/ `UNION`（跨分支去重）

调度（Schedule）

当前 PipelineSpecDTO 未包含 schedule / cron 字段；runOnInstall 仅控制是否随包安装自动运行。Pipeline 的定时调度触发由运行时执行接口负责，不在本 Spec 中声明。

示例

多表关联并过滤后输出

apiVersion: foundry.loambase.ai/v1alpha1
kind: Pipeline
metadata:
  name: example-sales-pipeline
spec:
  name: example-sales-pipeline
  description: 关联订单与客户数据并过滤大额订单
  dagConfig:
    nodes:
      - id: in-orders
        type: INPUT
        name: 订单输入
        config:
          datasetId: sales-orders
          columns:
            - order_id
            - customer_id
            - amount
      - id: in-customers
        type: INPUT
        name: 客户输入
        config:
          datasetId: sales-customers
          columns:
            - customer_id
            - customer_name
            - region
      - id: join-customer
        type: JOIN
        name: 关联客户
        config:
          joinType: INNER
          leftKey: customer_id
          rightKey: customer_id
      - id: filter-amount
        type: FILTER
        name: 过滤大额订单
        config:
          root:
            condition: AND
            children:
              - column: amount
                operator: '>'
                value: 1000
      - id: proj-out
        type: PROJECTION
        name: 输出投影
        config:
          columns: order_id, customer_id, customer_name, region, amount
      - id: out-1
        type: OUTPUT
        name: 结果输出
        config:
          registerDataset: true
          materializationType: MATERIALIZED
    edges:
      - source: in-orders
        target: join-customer
      - source: in-customers
        target: join-customer
      - source: join-customer
        target: filter-amount
      - source: filter-amount
        target: proj-out
      - source: proj-out
        target: out-1

消费位置

foundry-pipeline-biz: PipelineDomainProjectionAdapter.project 将 name、description 投影到 fd_pipeline；dagConfig 会反序列化为 PipelineDagDTO 并保存为 DAG。