[SPARK-46913][SS] Add support for processing/event time based timers with transformWithState operator

[anishshri-db]

What changes were proposed in this pull request?

Add support for processing/event time based timers with transformWithState operator

Why are the changes needed?

Changes are required to add event-driven timer based support for stateful streaming applications based on arbitrary state API with the transformWithState operator

As part of this change - we introduce a bunch of functions that users can use within the StatefulProcessor logic. Using the StatefulProcessorHandle, users can do the following:

• register timer at a given timestamp
• delete timer at a given timestamp
• list timers

Note that all the above operations are tied to the implicit grouping key.

In terms of the implementation, we make use of additional column families to support the operations mentioned above. For registered timers, we maintain a primary index (as a col family) that keeps the mapping between the grouping key and expiry timestamp. This col family is used to add and delete timers with direct access to the key and also for listing registered timers for a given grouping key using prefix scan. We also maintain a secondary index that inverts the ordering of the timestamp and grouping key. We will incorporate the use of the range scan encoder for this col family in a separate PR.

Few additional constraints:

• only registered timers are tracked and occupy storage (locally and remotely)
• col families starting with _ are reserved and cannot be used as state variables
• timers are checkpointed as before
• users have to provide a timeoutMode to the operator. Currently, they can choose to not register timeouts or register timeouts that are processing-time based or event-time based. However, this mode has to be declared upfront within the operator arguments.

Does this PR introduce any user-facing change?

Yes

How was this patch tested?

Added unit tests as well as pseudo-integration tests

StatefulProcessorHandleSuite

13:58:42.463 WARN org.apache.spark.sql.execution.streaming.state.StatefulProcessorHandleSuite:

===== POSSIBLE THREAD LEAK IN SUITE o.a.s.sql.execution.streaming.state.StatefulProcessorHandleSuite, threads: rpc-boss-3-1 (daemon=true), shuffle-boss-6-1 (daemon=true) =====
[info] Run completed in 4 seconds, 559 milliseconds.
[info] Total number of tests run: 8
[info] Suites: completed 1, aborted 0
[info] Tests: succeeded 8, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.

TransformWithStateSuite

13:48:41.858 WARN org.apache.spark.sql.streaming.TransformWithStateSuite:

===== POSSIBLE THREAD LEAK IN SUITE o.a.s.sql.streaming.TransformWithStateSuite, threads: QueryStageCreator-0 (daemon=true), state-store-maintenance-thread-0 (daemon=true), ForkJoinPool.commonPool-worker-4 (daemon=true), state-store-maintenance-thread-1 (daemon=true), QueryStageCreator-1 (daemon=true), rpc-boss-3-1 (daemon=true), F
orkJoinPool.commonPool-worker-3 (daemon=true), QueryStageCreator-2 (daemon=true), QueryStageCreator-3 (daemon=true), state-store-maintenance-task (daemon=true), ForkJoinPool.com...
[info] Run completed in 1 minute, 32 seconds.
[info] Total number of tests run: 20
[info] Suites: completed 1, aborted 0
[info] Tests: succeeded 20, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.

Was this patch authored or co-authored using generative AI tooling?

No

[anishshri-db]

@HeartSaVioR - PTAL, thx !

[sahnib]

Still reviewing, posting initial set of comments.

[sahnib]

Thanks for making these changes.

[sahnib]

LGTM. One minor comment, thanks.

[sahnib]

LGTM.

[HeartSaVioR]

Went through implementation part, will continue with tests.

[HeartSaVioR]

Done reviewing for the first round.

[HeartSaVioR]

+1 pending CI

[HeartSaVioR]

Thanks! Merging to master.