SQLink 血缘 Parser 插件开发

本页给出从输入文件到 DAM 血缘展示的完整开发路径。Parser 是 SQLink 的 SPI 插件，不是把一组 REST 接口打包成 JAR。插件负责识别一种作业文件、构造标准血缘对象，SQLink 负责上传、调度解析、保存结果和提供查询接口。

1. 运行链路

作业 JSON
  -> Parser.parse(content, extension)
  -> LineageContainer(Job / Step / Line)
  -> SQLink 保存解析结果
  -> lineageGraph/query、getTransform、getImpact
  -> DAM 元数据详情页的血缘展示

要在 DAM 详情页显示完整血缘，解析出的表必须能和 DAM 已采集对象匹配。匹配依赖连接标识、命名空间、Schema 和表名；SQLink 图显示成功不代表 DAM 元数据已经关联成功。

2. 插件包契约

插件包至少包含 plugin.json 和实现 Parser SPI 的 JAR：

{
  "type": "SEATUNNEL",
  "displayName": "Apache SeaTunnel Lineage Parser",
  "loadClass": "com.datablau.sqlink.plugin.seatunnel.SeatunnelLineageParser",
  "build": "1.0.0",
  "supportFileExtensions": ["json"]
}

Parser 入口契约为 parse(String content, String extension)，并应提供稳定的 getType() 和支持扩展名声明。WHALETUNNEL 与 SEATUNNEL 是两个不同引擎类型，不能只替换显示名称后复用错误的入口类。

3. 标准输出对象

Step 表节点

字段	含义	实现要求
id	表节点唯一 ID	在同一次解析中稳定，字段节点引用该 ID
name	展示表名	与 source/sink 实际表名一致
schema	Schema	缺失时使用约定的默认 Schema，并在日志中说明
properties.$connectionIdentifier	数据源连接标识	必须能匹配 SQLink 目录映射和 DAM 数据源
properties.$input_table	是否输入表	source 节点设为 true
properties.$output_table	是否输出表	sink 节点设为 true
properties.$with_table	是否 CTE/临时表	仅对中间过程节点设置
properties.SCHEMA	图展示 Schema	与 schema 保持一致
properties.TABLE_NAME	图展示表名	不要把表名放进连接标识

Column 字段节点

字段	含义
id	通常由 step ID 和字段名组成的稳定 ID
stepId	所属表节点 ID，必须指向真实 Step
name	字段名
schema、table	所属表定位信息
order	字段顺序，SQL select 和 sink 列表需要它

Line 关系

表级关系使用 sourceStepId 和 targetStepId。字段级关系使用 source 和 target 指向 Column ID，同时保留两端 Step ID。没有输入或输出字段时可以只生成表级关系，并在解析详情中说明原因。

4. SeaTunnel Parser

SeaTunnel 文件通常以 seatunnel.job_config 为根配置，Parser 需要处理：

• plugin_name、plugin_input、plugin_output 和 connector 参数；
• table_path、table_name、database、database_name、datasource_name、schema；
• columns、fields、SQL/query、别名、CTE、derived table、join、from/into；
• FieldRemover、FieldRenamer、TableRenamer 和 convert_case；
• sink SQL 生成和字段顺序保持。

SQL 解析只负责数据流向。过滤条件本身通常不会产生字段血缘，复杂表达式应交由 SQL parser 处理。字段级血缘不准确时，先检查输入字段列表、输出字段列表、字段顺序和别名，再修改连线算法。

5. WhaleTunnel Parser

WhaleTunnel Parser 读取 whaletunnel.job_config 和作业导出的 exportData，处理 source、transform、sink。最小表级输入：

{
  "syncJobDefinitionDtoList": [{
    "jobTasks": [
      {
        "pluginId": "source_1",
        "type": "SOURCE",
        "connectorType": "Jdbc",
        "dataSourceName": "source_connection",
        "dataSourceOption": "{\"schema\":\"sales\",\"table\":\"customer\"}",
        "selectTableFields": "{\"columns\":[\"customer_id\"]}"
      },
      {
        "pluginId": "sink_1",
        "type": "SINK",
        "connectorType": "Jdbc",
        "dataSourceName": "target_connection",
        "dataSourceOption": "{\"schema\":\"mart\",\"table\":\"customer\"}",
        "selectTableFields": "{\"columns\":[\"customer_id\"]}"
      }
    ],
    "jobLines": [{"inputPluginId":"source_1","targetPluginId":"sink_1"}]
  }]
}

dataSourceName 必须能匹配 SQLink connection mapping，Schema、table 和 column 必须与 DAM 采集结果一致。jobLines 的 plugin ID 必须存在，否则图会出现孤立节点。WhaleTunnel 的 source/table/result 字段应先归一化为统一 Step，再复用公共建图逻辑。

6. 上传、解析和查询

1. POST /sqlink/lineage/getLineageParseEngines，确认插件的 typeName 和扩展名已经注册。
2. POST /sqlink/lineage/uploadLineageFile，表单提交 lineageEngine、objectType、file、folderId、defaultSchema、parse。
3. POST /sqlink/lineage/parseLineage，提交 lineageIds；失败重试时可使用 onlyFailed 或 incrementalParsing。
4. POST /sqlink/task/getTaskParseDetail，检查 parseStatus、parseRate、notFoundTables、errorType、errorMessage。
5. POST /sqlink/lineage/getLineageTableByPage，确认源表和目标表已经生成。
6. POST /sqlink/lineageGraph/query，检查 TABLE、COLUMN 节点和表/字段边。
7. POST /sqlink/lineageGraph/getTransform 和 getImpact，核对加工逻辑及影响范围。
8. POST /metadata/entities/searchMetadata，确认 DAM 中对应表已采集，并在元数据详情页验证血缘。

完整表血缘查询使用 POST /sqlink/llm/getLineageOfTables，必须提交完整路径：

{
  "namespace": "<source-or-catalog>",
  "schema": "sales",
  "table": "customer",
  "direction": "ALL"
}

只传表名会在多数据源或多 Schema 中产生歧义。direction 按目标版本使用 FORWARD、BACKWARD 或 ALL。

7. 测试与验收

测试至少覆盖：单 source/sink、多 Schema、同名表、别名、CTE、字段重命名、字段删除、SQL query、默认 Schema、未支持任务类型和解析失败。Java 项目可用以下方式运行定向测试：

$env:JAVA_HOME = 'C:\Program Files\Java\jdk-17'
mvn test "-Dtest=SeatunnelLineageParserTest,WhaleTunnelLineageParserTest"

验收标准：

• getType() 返回正确的 SEATUNNEL 或 WHALETUNNEL；
• source 和 sink Step 存在，sink 具有 $output_table；
• $connectionIdentifier、Schema、TABLE_NAME、SCHEMA_ID 正确；
• 表级 Line 数量和方向正确；
• 输入字段、输出字段和字段级 Line 能按字段名及顺序对应；
• 不支持任务有可读错误，不能静默生成半条血缘；
• SQLink 图、解析表分页、DAM 元数据详情页三处结果一致。

8. 常见问题定位

现象	优先检查
SQLink 有图，DAM 没有血缘	connection mapping、namespace、Schema/table 一致性和 DAM 是否重新采集
修改 JSON 后结果没变化	是否创建了新的 lineageId，查询是否仍使用旧文件
引擎列表没有插件	manifest、插件目录、Parser factory 注册和文件扩展名
只有表级没有字段级	columns/fields 是否存在、字段顺序、别名、SQL parser 是否介入
Oracle 小写带引号表失败	保持 Schema、表名和字段的大小写及引号与实际对象一致

开发新 Parser 时，新增文件类型就新建具体 Parser；输入字段候选变化修改 Table 归一化层；公共建图规则变化修改 Builder；不要复制整套已有 Parser，也不要把连接标识塞进 namespace。