minxhe opened a new pull request, #29795: URL: https://github.com/apache/flink/pull/29795
## What is the purpose of the change FLINK-41355 added async state versions of the SQL interval join but kept the synchronous functions for a join with the `EARLY_FIRE` hint (FLIP-497), because only they implement early fire. This pull request ports the interval-join early fire of FLINK-40171 and FLINK-40172 (designed by Weiqing Yang) to the async functions, so `table.exec.async-state.enabled` selects them with or without the hint. Jira: https://issues.apache.org/jira/browse/FLINK-41356 (subtask of FLINK-36157, related to FLIP-497 / FLINK-36953) ### Depends on - FLINK-41355 (#29794): this pull request is stacked on it; only the last commit is new. - FLINK-41091 (#29571) and FLINK-41095 (#29575), like FLINK-41355: expired buckets and their fired bits are deleted with `MapState#asyncPutAll` of null values. ## Brief change log - `AsyncStateTimeIntervalJoin` keeps the early-fire state of `TimeIntervalJoin`, with the same names, types and registration conditions: `IntervalJoin{Left,Right}Fired` when early fire is enabled for an outer join, and `IntervalJoin{Left,Right}EarlyFireSchedule` for processing-time early fire on a row-time join. A savepoint of the synchronous early-fire join restores into the async functions. - Same rules as the synchronous version: an unmatched outer row is padded at its early-fire time, the pad is retracted when a later match arrives, and a fired row is not padded again at expiry; inner joins ignore the hint. Processing-time early fire on a row-time join uses processing-time timers and a schedule, and `onTimer` tells them apart by `ctx.timeDomain()`. Every pad is stamped `INSERT`, since a retraction changes the row kind of the reused pad row. - Before scanning an outer cache with early fire enabled, the function reads that cache's fired bits in one scan, so the cache callbacks stay synchronous and keep their output order. The fired bits of expired buckets are deleted with the buckets, in one `asyncPutAll` of null values; on full expiry they are deleted like the buckets (bulk delete, or `asyncClear()` above 65,536). As in the synchronous version, a schedule entry is removed by its own timer only. - `AsyncStateRowTimeIntervalJoin` and `AsyncStateProcTimeIntervalJoin` take the early-fire delay (and the cross-domain flag for row time) like their synchronous versions. `EmitAwareCollector#armRetraction` and `#disarm` become public for the async package. - `StreamExecIntervalJoin` no longer keeps the synchronous functions for the `EARLY_FIRE` hint. No new option, JSON plan change or ExecNode version change. - Docs: the supported-operator entry of the disaggregated state page (en and zh) drops "without the `EARLY_FIRE` hint". ## Verifying this change This change added tests and can be verified as follows: - The 13 early-fire cases of `RowTimeIntervalJoinTest` and the 3 of `ProcTimeIntervalJoinTest` now run against both the synchronous and the async functions, through a shared harness factory that replaces 13 copies of the constructor call. New case `testRowTimeLeftOuterEarlyFireThenExpiryOnJoin`: a fired row that expires on the record path is not padded again. Three cases also assert that no state is left after full expiry. - `StreamExecIntervalJoinAsyncStateTest#testEarlyFireJoin` expects the async operator for an `EARLY_FIRE` join with async state on. - `IntervalJoinAsyncStateRestoreTest` also restores the committed `interval-join-early-fire` savepoint, and the new `IntervalJoinProcTimeEarlyFireAsyncStateRestoreTest` restores `interval-join-proc-time-early-fire`, both taken with the synchronous functions. Unlike the plain interval-join programs, both use restored state after restore: the first retracts a pad emitted before the savepoint, the second emits a pad from the restored schedule and processing-time timer. No new savepoint files. - Mutations confirmed that the tests catch: the planner fallback kept, the retraction disabled, the restored early-fire state ignored, fired bits ignored on record-path expiry, and fired bits left after full expiry. - As for FLINK-41355, async state requests complete synchronously on the HashMap and RocksDB adaptor, and no test runs on ForSt. 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,IntervalJoinProcTimeEarlyFireAsyncStateRestoreTest,IntervalJoinITCase,IntervalJoinTest,EarlyFireJoinHintTest' test # BUILD SUCCESS: flink-table-runtime 66 run, 0 failures; flink-table-planner 128 run, 0 failures, 4 skipped (the @Disabled # restore-file generators). Per class: RowTimeIntervalJoinTest 47, ProcTimeIntervalJoinTest 12, AsyncStateRowTimeIntervalJoinTest 7, # StreamExecIntervalJoinAsyncStateTest 6, IntervalJoinITCase 63, IntervalJoinTest 28, EarlyFireJoinHintTest 18, IntervalJoinRestoreTest 5, # IntervalJoinAsyncStateRestoreTest 4, IntervalJoinProcTimeEarlyFireRestoreTest 2, IntervalJoinProcTimeEarlyFireAsyncStateRestoreTest 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 early-fire state names and serializers of the synchronous functions - The runtime per-record code paths (performance sensitive): yes, for interval joins with the `EARLY_FIRE` hint 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. A job with async state enabled switches its early-fire interval joins to async state on its first restore with this change; the state layout is unchanged (tested from the committed synchronous early-fire savepoints on RocksDB). As noted in FLINK-41355, switching a timer-using operator between synchronous and async state on ForSt is not covered. - 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]
