[FLINK-39783][table] Translate LATERAL SNAPSHOT to LateralSnapshotJoinOperator

[fhueske]

What is the purpose of the change

This PR is part of FLIP-579.
It adds the optimizer rules to translate the built-in SNAPSHOT function in a LATERAL context into a custom operator.
In this PR, the operator implementation is just a stub, see #28329 for the actual implementation.

Brief change log

• Adds optimizer nodes:
  • FlinkLogicalLateralSnapshotJoin
  • StreamPhysicalLateralSnapshotJoin
  • StreamExecLateralSnapshotJoin
• Adds optimizer rules
  • LogicalCorrelateToJoinFromLateralSnapshotRule translates a SNAPSHOT table function call under a regular Join into a FlinkLogicalLateralSnapshotJoin
  • StreamPhysicalLateralSnapshotJoinRule translates the FlinkLogicalLateralSnapshotJoin into a StreamPhysicalLateralSnapshotJoin. This is basically a 1:1 translation (adding hash-partitioning ).

Special handling:

• Calcs that were pushed between the table function call and the join are maintained and applied before the new LateralSnapshotJoin. The new join requires an event-time attribute on the build input. The rules have special handling for ensuring that the time attribute is forwarded into the join. If it was removed due to projection pushdown, it is re-added by tweaking the projection before the join and removed by adding a calc after the join.
• The rules ensure that the LateralSnapshotJoin materializes the event-time attribute, i.e., the type is changed to a regular timestamp.
• Build-side proc-time attributes are materialized before the new join operator.
• The translation logic ensures that the probe-side only consumes append-only changes, the build-side consumes full retraction changes (with update_before)
• The logical translation rule ensures that SNAPSHOT args are literals by expression constant-folding
• disableSystemArguments(true) is set on the SNAPSHOT function definition. Currently, this flag is not correctly handled (see FLINK-40079]. The current behavior is that system-args are not rejected but do also not affect the join.

Verifying this change

Added unit tests checking plans and validations.

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): no
• 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? no
• If yes, how is the feature documented? n/a

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Co-authored-by: Claude Code 2.1.148 (Opus 4.7)

[flinkbot]

CI report:

• 3255f79 Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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

[fhueske]

@flinkbot run azure