[flink] Implement FLIP-314 LineageVertexProvider for source and sink connectors

[jsingh-yelp]

Purpose

• Implement LineageVertexProvider which helps flink (FLIP-314) to generate lineage events with additional details which flink don't have access to.
• It generates a proper namespace and passes down table config options with additional details such as primary_keys, partition_keys e.t.c
• Linked issue: [Feature] Support native Flink 2.0 job lineage (FLIP-314) for Paimon tables #7306

Tests

• Added unit tests for the change
• Also was able to do end to end manual testing for change. This is how the final lineage event will look like using openlineage: https://i.fluffy.cc/7KGHgtJ3x003hvs4lK2rrBDNqgmF9MLZ.html#L87-88,L95-120,L129-131

API and Format

• This change does not change any API
• Note: This PR changes how PaimonDataStreamScanProvider and PaimonDataStreamSinkProvider are constructed callers now use createProvider() factory methods instead of direct constructors. This is necessary because the FLIP-314 implementation requires the returned ScanRuntimeProvider/SinkRuntimeProvider to also implement LineageVertexProvider, an interface that only exists in Flink 1.20+. To maintain backward compatibility with older Flink versions, the factory methods delegate to a version-specific DataStreamProviderFactory: the Flink 2.x variant wraps the provider with lineage support, while the Flink 1.x stub returns it unchanged.

Documentation

• Yes I think this is worth adding documentation for, I am happy to add docs for it if someone can point me where to add the specific documentation for this.
• Yes this introduces a new feature of adding support of native lineage generation of Paimon tables for Flink.

Note: This change also depends on two other changes and one of them is already merged. Other two changes:

• Flink: [FLINK-39219] [table] Support lineage extraction for DataStreamscan/sink providers flink#27727
• Openlineage: [Already Merged] Add flink support for DatasetConfigFacet for flink native listener OpenLineage/OpenLineage#4368

[jsingh-yelp]

This is required because since now we have to import DataStreamScanProvider in flink 2-common. Code ref: https://github.com/apache/flink/blob/c0479c74f2b736aaed00b4d9fb1b14ad296e562d/flink-table/flink-table-api-java-bridge/src/main/java/org/apache/flink/table/connector/source/DataStreamScanProvider.java#L40

[jsingh-yelp]

@JingsongLi can I please have a review on this

[JingsongLi]

Cool, @yunfengzhou-hub can you take a look?

[yunfengzhou-hub]

Thanks for the PR. Generally LGTM. Left one comment as below.

[yunfengzhou-hub]

Compared with delegate pattern, would it be better to directly have PaimonDataStreamScanProvider implement LineageVertexProvider?

Same for LineageAwarePaimonDataStreamSinkProvider.

[jsingh-yelp]

Thanks for the review @yunfengzhou-hub. The reason for the delegate pattern is backward compatibility. PaimonDataStreamScanProvider lives in paimon-flink-common, which is built against multiple Flink versions, including ones older than 1.20. Since LineageVertexProvider only exists in Flink 1.20+, having PaimonDataStreamScanProvider directly implement it would cause a ClassNotFoundException at build/load time for older Flink versions.

By using the delegate pattern with a version-specific DataStreamProviderFactory, the lineage-aware wrapper is only created in the paimon-flink2-common module (which targets Flink 2.x), and the older Flink modules never reference LineageVertexProvider at all.