[FLINK-31275] extend lineage listener with QueryOperationEvent

[pawel-big-lebowski]

What is the purpose of the change

This change should enable extracting column level lineage for Flink SQL jobs.
This can be achieved by JobStatusChangedListener if Flink exposes QueryOperation.
More details on the approach can be found in docs: https://docs.google.com/document/d/1XmbHy6XqBrMoH9rkSyOG0wbwQZgf0epz-07lr_NfikI/edit?usp=sharing

Brief change log

• *Introduce new QueryOperationEvent which implements QueryOperationEvent *
• notify query operations from TableEnvironmentImpl

Does this pull request potentially affect one of the following parts:

• Dependencies (does it add or upgrade a dependency): no
• The public API, i.e., is any changed class annotated with @Public(Evolving): yes
• The serializers: no
• The runtime per-record code paths (performance sensitive): no
• Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: no
• The S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? yes
• If yes, how is the feature documented? Added extra info on job listeners' page.

[flinkbot]

CI report:

• fd6e351 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[davidradl]

and for Table API jobs?

[davidradl]

I would like to see some tests showing the column lineage is produced as expected for different topologies.

[davidradl]

some comments:

• it would be great to get this very useful capability into Flink. Thanks for the contribution
• I see that the document pointed to, has some side effects - are these still the case, if so do we need to document them?
• I see that spark implements CLL at the optimization level. Are we doing CCL on the optimized logical plan? If not what is the thinking around this?

[davidradl]

Reviewed by Chi on 30/01/2025 Go back to the submitter with review comments.

[pawel-big-lebowski]

Thank you @davidradl for your attention and comments provided.

I see that the document pointed to, has some side effects - are these still the case, if so do we need to document them?

The linked google doc describes in details motivation behind this PR and proves column level lineage metadata can be obtained with this change. It also contains my concern as this change creates another instance of JobStatusChangedListener. Within TableEnvironmentImpl I was not able to reach the listeners from executors (#24754 -> this PR introduced the listeners). I included a comment within the code.

I would like to see some tests showing the column lineage is produced as expected for different topologies.

This PR provides an ability to enrich existing lineage metadata collected. The enrichment can be done only through jobId. That allows correlating column level lineage extracted from Calcite RelNode's with lineage from DefaultJobCreatedEvent, as both events share same jobId. Within this change, job status listeners are notified on all the execEnv.executeAsync calls, when job client is created.

I don't think it's possible to test all the possible TableEnvironmentImpl method calls or all the SQLs possible. However, this change notifies the listener always when having jobId + queryOperation.

I see that spark implements CLL at the optimization level. Are we doing CCL on the optimized logical plan? If not what is the thinking around this?

Although it may be useful to extract column level lineage from CompiledPlan, as it is fully optimized and one can directly execute the plan, I was not able to find the proper abstractions within CompiledPlan to achieve this. I found Calcite's RelNode enables CLL extraction and the same approach has been chosen for another project - flink-sql-lineage - https://github.com/HamaWhiteGG/flink-sql-lineage.

I think there should be in future another CompiledPlanEvent that notifies the job listener as well. In this case, it will be the listener to decide whether to extract lineage metadata from RelNodes, CompiledPlan, etc. However, at the moment I didn't find the way to connect CompiledPlan with the JobId.