minxhe opened a new pull request, #29794: URL: https://github.com/apache/flink/pull/29794
## What is the purpose of the change The SQL interval join always uses synchronous state, even with `table.exec.async-state.enabled`, so a job with an interval join falls back to synchronous state on a disaggregated state backend such as ForSt. `StreamExecIntervalJoin` still has two `TODO: add async version ... to use AsyncKeyedCoProcessOperator`. This pull request adds the async state versions of the row-time and processing-time interval join, as a step of FLIP-473. The DataStream interval join already has one (FLINK-37003). Jira: https://issues.apache.org/jira/browse/FLINK-41355 (subtask of FLINK-36157) ### Depends on FLINK-41091 / FLINK-41095 (must merge first) Expired rows are deleted with one `MapState#asyncPutAll` that maps their timestamps to `null`, which the `MapState` contract defines as a removal. Two backends do not honor that yet: - FLINK-41091 (#29571): ForSt `asyncPutAll` with null values. - FLINK-41095 (#29575): `MapStateAdaptor`, used for async state on HashMap and RocksDB. This pull request does not include them. Without FLINK-41095, the async test cases of `RowTimeIntervalJoinTest` and `ProcTimeIntervalJoinTest` fail (expired rows stay in state), and the synchronous cases pass. I will update this branch from `master` once both are merged. ## Brief change log - New `AsyncStateRowTimeIntervalJoin` and `AsyncStateProcTimeIntervalJoin` on a shared base in `org.apache.flink.table.runtime.operators.join.interval.asyncprocessing`, run by the existing `AsyncKeyedCoProcessOperatorWithWatermarkDelay` / `AsyncKeyedCoProcessOperator`. They use the state names, types and serializers of `TimeIntervalJoin`, and its join, outer-padding and cleanup-timer rules, including FLINK-41067 (rows whose join key can never match are not cached). - Records of several keys are in flight at a time, so each record and timer captures its join key and operator time when it starts. `IntervalJoinFunction` gets a `join` overload and a `keyCannotMatch` check that take the join key explicitly. `EmitAwareCollector` and `TimeIntervalStreamJoinTestBase` become public so the async package can use them. - Cleanup: the changed buckets of a scanned cache are written after the scan, then all expired buckets are removed with one `asyncPutAll` of null values; nothing is sent when nothing expired. When no row of a key survives, the timer state is cleared and the cache is bulk-deleted, or cleared with `asyncClear()` above 65,536 row times (on HashMap and RocksDB the adaptor turns a bulk delete into one remove per entry). The cleanup timer of a side only scans keys when that side needs no outer padding. - Two differences from the synchronous functions: rows at time zero or earlier are kept and joined like later rows (the synchronous cleanup can drop a surviving row at time zero or earlier), and cleanup times saturate at `Long.MAX_VALUE` instead of overflowing. As in the synchronous version, the event-time operator time is never below zero. - `useInterruptibleTimers` (FLINK-38810) is not enabled for the async functions: `AbstractAsyncKeyOrderedStreamOperator#processWatermark` is final and advances the timers directly, so async operators do not use interruptible timers. - `StreamExecIntervalJoin` selects the async functions when `table.exec.async-state.enabled` is true and the join has no `EARLY_FIRE` hint (FLIP-497); with the hint, the synchronous functions stay. Async early fire can follow as a separate subtask. Both TODOs are removed. - No new option, JSON plan change or ExecNode version change. As for the other async-capable ExecNodes, the async flag is not a consumed option, so a compiled plan follows the current setting. - Docs: the interval join is added to the supported operators of the disaggregated state page (en and zh), without the `EARLY_FIRE` hint. ## Verifying this change This change added tests and can be verified as follows: - `RowTimeIntervalJoinTest` and `ProcTimeIntervalJoinTest` run their shared cases against both the synchronous and the async functions (9 and 3 cases), including the FLINK-41067 cases. Early-fire and interruptible-timer cases stay synchronous. - New `AsyncStateRowTimeIntervalJoinTest` covers what only the async functions do: rows at time zero and earlier, cleanup-time saturation with a `Long.MAX_VALUE` cleanup interval, partial expiry padding each outer row exactly once on both the record and the timer path, full expiry at and above the 65,536 bulk-delete limit, restoring a snapshot of the synchronous function (joining and padding restored rows and firing restored timers), and an exception in a state callback failing the operator. - New `StreamExecIntervalJoinAsyncStateTest`: operator selection for processing and row time with async state on and off, the `EARLY_FIRE` hint keeping the synchronous operator, and a compiled plan that is identical with async state on and off and selects the async operator when loaded with async state on. - New `IntervalJoinAsyncStateRestoreTest` restores the committed `IntervalJoinRestoreTest` plans and savepoints, taken with the synchronous operator, with async state on. These programs check that the async operator accepts the synchronous state, but their output after restore does not use restored rows, so `AsyncStateRowTimeIntervalJoinTest#testRestoreSyncSnapshot` (above) covers that restored rows and timers are joined, padded and fired. No new savepoint files are added. - `IntervalJoinITCase` is parameterized like `RankITCase`: HashMap with and without async state, and RocksDB. - Mutations confirmed that the new tests catch: the frontier starting at 0, the synchronous survivor rule, unsaturated cleanup times, skipping the expired-row deletion on the record path or on the timer path, the planner ignoring the flag or the `EARLY_FIRE` hint, the async function not reading restored state, and a different timer-state serializer. - On the HashMap and RocksDB adaptor, async state requests complete synchronously, so these tests never hold two records of the same key in flight, and no test here runs on ForSt. A ForSt test needs a `flink-statebackend-forst` test dependency in `flink-table-runtime` (see FLINK-36354). Question for reviewers: should it come here or with FLINK-36354? Commands run with JDK 17 and the Maven wrapper (3.9.16), on this branch plus the FLINK-41091 and FLINK-41095 commits: ```sh ./mvnw -B -ntp -Dfast -Dflink.markBundledAsOptional=false -Dflink.forkCountUnitTest=1 \ -Dsurefire.failIfNoSpecifiedTests=false -pl flink-table/flink-table-planner -am \ -Dtest='RowTimeIntervalJoinTest,ProcTimeIntervalJoinTest,AsyncStateRowTimeIntervalJoinTest,StreamExecIntervalJoinAsyncStateTest,IntervalJoinAsyncStateRestoreTest,IntervalJoinRestoreTest,IntervalJoinProcTimeEarlyFireRestoreTest,IntervalJoinITCase,IntervalJoinTest,EarlyFireJoinHintTest' test # BUILD SUCCESS: flink-table-runtime 48 run, 0 failures; flink-table-planner 125 run, 0 failures, 3 skipped (the @Disabled # restore-file generators). Per class: RowTimeIntervalJoinTest 32, ProcTimeIntervalJoinTest 9, # AsyncStateRowTimeIntervalJoinTest 7, StreamExecIntervalJoinAsyncStateTest 6, IntervalJoinITCase 63, IntervalJoinTest 28, # EarlyFireJoinHintTest 18, IntervalJoinRestoreTest 5, # IntervalJoinAsyncStateRestoreTest 3, IntervalJoinProcTimeEarlyFireRestoreTest 2 (one skipped each) ./mvnw -B -ntp -pl flink-table/flink-table-runtime,flink-table/flink-table-planner spotless:check checkstyle:check # BUILD SUCCESS, 0 Checkstyle violations ``` Module and full-reactor `clean verify` and the ArchUnit tests were not run; CI provides the full run. ## 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 async functions use the state names and serializers of the synchronous functions - The runtime per-record code paths (performance sensitive): yes, the interval join uses async state when `table.exec.async-state.enabled` is true; nothing changes when it is false - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: yes. The operator keeps its UID and state layout, so a job can restore across the switch where the backend restores the same descriptors (tested from synchronous savepoints on RocksDB, and synchronous snapshots on HashMap). A job that already runs with `table.exec.async-state.enabled` switches its interval joins to async state on its first restore with this change. On ForSt, the synchronous backend keeps timers in ForSt by default and the async backend keeps them in a heap queue, so switching a timer-using operator between them is not covered, as for the other async operators with timers. - The S3 file system connector: no ## Documentation - Does this pull request introduce a new feature? yes - If yes, how is the feature documented? docs (disaggregated state page, supported operators) --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes (please specify the tool below) Generated-by: GitHub Copilot CLI 1.0.102-0 (Claude Opus 5.5) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
