Develop lineage Parser plugins

Parser plugins convert job files to standard SQLink lineage. REST handles uploading, parsing requests, persistence, and queries; plugins implement the SPI.

1. Runtime flow

Job JSON
  -> Parser.parse(content, extension)
  -> LineageContainer (Job / Step / Line)
  -> SQLink persistence
  -> lineageGraph/query, getTransform, getImpact
  -> Lineage on DAM metadata detail pages

DAM display requires parsed tables to match harvested objects through connection identity, namespace, schema, and name. A successful SQLink graph does not establish DAM association.

2. Package contract

Include plugin.json and an SPI implementation JAR:

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

Implement parse(String content, String extension), stable getType(), and supported extensions. WHALETUNNEL and SEATUNNEL have distinct engine types and entry classes.

3. Standard outputs

Step table nodes

Field	Requirement
id	Unique and stable within parsing; referenced by columns
name	Actual source/sink table
schema	Actual or agreed default; log default use
properties.$connectionIdentifier	Match SQLink mapping and DAM source
properties.$input_table	true for source
properties.$output_table	true for sink
properties.$with_table	Only CTE or temporary nodes
properties.SCHEMA	Same as schema
properties.TABLE_NAME	Table name, separate from connection ID

Column nodes

Field	Meaning
id	Stable, typically Step ID plus field name
stepId	Existing owning Step
name	Field name
schema, table	Table identity
order	SQL SELECT and sink column order

Line relationships

Table edges use sourceStepId and targetStepId. Field edges use source and target Column IDs and both Step IDs. When fields are unavailable, emit table-level lineage only and explain the limitation.

4. SeaTunnel Parser

Read seatunnel.job_config, including:

• plugin_name, plugin_input, plugin_output, connector settings.
• table_path, table_name, database, database_name, datasource_name, schema.
• columns, fields, SQL/query, aliases, CTEs, derived tables, joins, FROM/INTO.
• FieldRemover, FieldRenamer, TableRenamer, convert_case.
• Sink SQL generation and column order.

SQL parsing explains flows. Filter conditions usually do not create field lineage themselves. Delegate complex expressions to the SQL parser. Check field lists, order, and aliases before changing edge algorithms.

5. WhaleTunnel Parser

Read whaletunnel.job_config and exported exportData for sources, transforms, and sinks. Minimal table-level input:

{
  "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 must match connection mapping; schema, table, and fields must match DAM. Every jobLines plugin ID must exist. Normalize source/table/result fields into Steps before reusing graph builders.

6. Upload, parse, and query

1. POST /sqlink/lineage/getLineageParseEngines: confirm registered type and extensions.
2. POST /sqlink/lineage/uploadLineageFile: submit lineageEngine, objectType, file, folderId, defaultSchema, parse.
3. POST /sqlink/lineage/parseLineage: submit lineageIds; use confirmed onlyFailed or incrementalParsing for retries.
4. POST /sqlink/task/getTaskParseDetail: inspect parseStatus, parseRate, notFoundTables, errorType, errorMessage.
5. POST /sqlink/lineage/getLineageTableByPage: verify source and sink tables.
6. POST /sqlink/lineageGraph/query: verify TABLE/COLUMN nodes and edges.
7. Read transformations and impact through getTransform and getImpact.
8. Search DAM metadata and verify lineage on corresponding detail pages.

Complete table queries use POST /sqlink/llm/getLineageOfTables:

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

A table name alone is ambiguous. Use supported direction values FORWARD, BACKWARD, or ALL.

7. Tests and acceptance

Cover one source/sink, multiple schemas, same-name tables, aliases, CTEs, renames, removed fields, SQL queries, default schema, unsupported tasks, and parsing failures. Targeted Java tests:

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

Verify correct getType(), source and sink Steps, $output_table, $connectionIdentifier, schema, TABLE_NAME, SCHEMA_ID, table edge count and direction, field names and order, readable unsupported-task errors, and agreement across graph, table pagination, and DAM details.

8. Troubleshooting

Symptom	Check
SQLink graph but no DAM lineage	Connection, namespace, schema/table, reharvesting
Changed JSON has no effect	New lineage ID versus queries using the old file
Engine missing	Manifest, plugin directory, factory registration, extensions
Table-only lineage	Fields, order, aliases, SQL parser
Quoted lowercase Oracle names fail	Exact case and quoting for schema, table, fields

Create a concrete Parser for each new file format, update normalization for field candidates, and update shared builders for common graph rules. Avoid copying an entire Parser or placing connection IDs in namespaces.