[flink] Implement FLIP-314 LineageVertexProvider for non-table source & sink APIs

[jsingh-yelp]

Purpose

• Linked Issue: [Feature] Support native Flink 2.0 job lineage (FLIP-314) for Paimon tables #7306
• Add FLIP-314 lineage support for Paimon's non-Table (DataStream) source & sink APIs. This PR adds LineageVertexProvider support for jobs that use FlinkSourceBuilder and FlinkSink directly.
• This pull request extends upon the work done previously here: [flink] Implement FLIP-314 LineageVertexProvider for source and sink connectors #7311 where we enhanced existing Paimon Flink Table API lineage by adding dataset namespace handling and enriching datasets with Paimon catalog & table config metadata.

What is covered?

Component	Approach	Covers
DataStream Sources	PaimonDataStreamSource wraps any Paimon Source with LineageVertexProvider	StaticFileStoreSource, ContinuousFileStoreSource, AlignedContinuousFileStoreSource, MonitorSource (via FlinkSourceBuilder)
DataStream Sinks	PaimonDiscardingSink extends DiscardingSink with lineage	All FlinkSink subclasses (append, upsert, CDC fixed/dynamic bucket)
Format Table Sink	FormatTableSink implements LineageVertexProvider directly	FlinkFormatTableDataStreamSink (Parquet/ORC direct writes)

What is still not covered?

Not covered	Reason
Multi-table CDC sinks (FlinkCdcMultiTableSink)	Tables are created dynamically at runtime, while getLineageVertex() is called during graph construction before the output tables are known. This can be revisited in a future improvement.
CompactorSourceBuilder / SystemTableSource	Maintenance or metadata-oriented paths rather than regular source-to-sink user data pipelines.

Tests

• Added tests for the changes done in this PR.
• Did manual testing for various use cases I had readily available for both Paimon as a source and sink:
  • Kafka to Paimon:
    • Sinking data from kafka to paimon, in this test case Paimon uses KafkaSyncTableAction → FlinkSink.
    • Tested with fixed bucket CdcFixedBucketSink (bucket=1 or 4) and CdcDynamicBucketSink (bucket = -1).
    • Example lineage event: https://i.fluffy.cc/XlDmn38MqjPKFJ99TTpPLXZrsTd7SXgJ.html#L54-96
  • Paimon to Cassandra:
    • Sinking data from paimon to cassandra, in this test case Paimon uses FlinkSourceBuilder via Datastream APIs.
    • Example lineage event: https://i.fluffy.cc/3HMH7j6sqD3JQDtvP4x6zPK0JC1MW2gp.html#L36-81

Example paimon table looks like this inside the lineage event:

{
      "namespace": "file:///tmp/paimon-warehouse",
      "name": "jdbc.jaskaran_test.default_value",
      "facets": {
        "schema": {
          "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.47.1/integration/flink",
          "_schemaURL": "https://openlineage.io/spec/facets/1-2-0/SchemaDatasetFacet.json#/$defs/SchemaDatasetFacet",
          "fields": [
            {
              "name": "id",
              "type": "STRING NOT NULL"
            },
            {
              "name": "age",
              "type": "INT"
            },
            {
              "name": "source",
              "type": "STRING"
            }
          ]
        },
        "symlinks": {
          "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.47.1/integration/flink",
          "_schemaURL": "https://openlineage.io/spec/facets/1-0-1/SymlinksDatasetFacet.json#/$defs/SymlinksDatasetFacet",
          "identifiers": []
        },
        "config": {
          "_producer": "https://github.com/OpenLineage/OpenLineage/tree/1.47.1/integration/flink",
          "_schemaURL": "https://openlineage.io/spec/2-0-2/OpenLineage.json#/$defs/DatasetFacet",
          "bucket": "-1",
          "catalog.warehouse": "file:///tmp/paimon-warehouse",
          "path": "file:/tmp/paimon-warehouse/jaskaran_test.db/default_value",
          "write-only": "true",
          "changelog-producer": "input",
          "partition-keys": "",
          "type": "paimon",
          "primary-keys": "id"
        }
      }
    }

[jsingh-yelp]

cc: @JingsongLi and @yunfengzhou-hub can I please a review on this PR, this is extension to work done previously in:
#7311

[leaves12138]

Thanks for the contribution. I am holding off on approval because the current CI status still has a failing build_test job. Please fix or rerun it, then I can take another pass.

[jsingh-yelp]

@leaves12138 I re-triggered the CI and all tests are passing now, can you please give another pass. Thanks a lot.

[jsingh-yelp]

@leaves12138 Sorry to re-tag you can I please have review on this PR?

[jsingh-yelp]

@JingsongLi / @leaves12138 gentle reminder can I please have a review on this PR?

@JingsongLi once this PR is merged, I can add docs on how paimon support native lineage for flink.

[jsingh-yelp]

@JingsongLi Can I please get a review on this PR?

[jsingh-yelp]

@JingsongLi can you please give another pass to this PR?

[JingsongLi]

This also changes the existing table/SQL lineage path. PaimonDataStreamScanProvider and PaimonDataStreamSinkProvider pass the Flink tableIdentifier.asSummaryString() as an explicit name, but for JDBC catalogs this branch ignores it and emits catalog-key + "." + table.fullName() instead, while non-JDBC catalogs keep the Flink identifier. That makes table/SQL dataset names metastore-dependent even though the new derivation is only needed when the non-table DataStream APIs do not have a name. Please keep defaultName when it is non-null, and only derive the catalog-key name when no explicit name was provided; a JDBC-metastore regression for the explicit-name overload would catch this.

[jsingh-yelp]

@JingsongLi This is by design, the catalog-key override for JDBC metastore on the Table API path is intentional. Catalog names which is used in tableIdentifier.asSummaryString() are arbitrary per-job the same physical table can appear under different lineage dataset names depending on how the job defines its catalog name.

catalog-key is a more stable cross-job identifier that users can explicitly configure. This is particularly valuable when the same database and table exist across different catalogs the catalog-key disambiguates them.

For context, this is the follow up ticket to support this in Flink: https://issues.apache.org/jira/browse/FLINK-39935

[JingsongLi]

Thanks for the context. I still think the explicit defaultName should win in this overload. The Table/SQL providers pass the resolved Flink table identifier here, so overriding it only for JDBC catalogs changes the existing table/SQL lineage name even though the catalog-key derivation is only needed for the new DataStream APIs where no name is available. That makes the same SQL/table job report a different dataset name after switching the catalog metastore/config, which is a surprising behavior change for this PR. Please keep defaultName when it is non-null and use catalog-key + "." + table.fullName() only for the null-name DataStream overload (or split the Table/SQL naming change into a separate documented change with compatibility discussion and tests).

[jsingh-yelp]

@JingsongLi Makes sense, i have added the fix as suggested. I can create a separate PR with reasoning for the table/sql flow.

[jsingh-yelp]

@JingsongLi Can you please share your thoughts for this comment: #8001 (comment) I think catalog-key provided by jdbc is more stable then catalog-name in case it is available, but please let me know your thoughts.

[jsingh-yelp]

@JingsongLi I have added the suggested fix here: #8001 (comment) please let me know your thoughts thanks.