[FLINK-40336][table] Add TemporalRowTimeJoinOperatorV2 to improve join on sorted state backends

[mateczagany]

What is the purpose of the change

On every watermark, emitResultAndCleanUpState will scan the full leftState of every key, which includes a full deserialization, including all the elements which have have higher timestamps than the current watermark. This work is completely unneeded, and for some specific Flink jobs, the speedup this new operator brings can be very much noticable.

With RocksDB and ForSt, when iterating the state, the keys returned will always be ordered in a serialized-key byte order, so it's easy to take an advantage of that, and exit early from the iteration of the state.

Brief change log

• Add TemporalRowTimeJoinOperatorV2 (flink-table-runtime):
  • probe side keyed by nested LeftTimeIndexKey
  • on RocksDB/ForSt (detected via getBackendTypeIdentifier(), same idiom as InputSortBuffer / LateralSnapshotJoinOperator), both states are iterated in ascending time order with early exit at the watermark
  • probe rows are re-sorted by arrival index before emission, preserving the exact output order of the old operator
  • unordered backends (heap) fall back to full scans equivalent to the old operator
• Add @ExecNodeMetadata(version = 2) to StreamExecTemporalJoin. createJoinOperator picks the operator by getVersion(). Version 1 plans keep instantiating TemporalRowTimeJoinOperator
• Add restore test fixtures (compiled plan + savepoint) for stream-exec-temporal-join_2 for all four temporal join test programs.
• Register LeftTimeIndexKeySerializer in TypeSerializerTestCoverageTest.

Verifying this change

• TemporalRowTimeJoinOperatorV2Test: all scenarios of TemporalRowTimeJoinOperatorTest mirrored with identical inputs/expected outputs
  • each run on both heap and RocksDB backends
  • asserting the sorted fast path is actually taken on RocksDB
  • plus new scenarios for arrival-order emission with out-of-order probe rows
• LeftTimeIndexKeySerializerTest: SerializerTestBase battery plus a property test asserting unsigned lexicographic byte order equals numeric (timestamp, index) order.
• TemporalJoinRestoreTest: version 1 savepoints restore into the old operator, version 2 savepoints into the new one (both on RocksDB)
• TemporalJoinITCase passes unmodified; fresh plans now compile to stream-exec-temporal-join_2 and run the new operator end-to-end.
• The existing TemporalRowTimeJoinOperatorTest is untouched and still passes

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): yes
• 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? not applicable

---

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

• Yes (please specify the tool below)

Generated-by: [Tool Name and Version]

[flinkbot]

CI report:

• 8b64241 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

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