[SPARK-47673][SS] Implementing TTL for ListState

[ericm-db]

What changes were proposed in this pull request?

This PR adds support for expiring state based on TTL for ListState. Using this functionality, Spark users can specify a TTL Mode for transformWithState operator, and provide a ttlDuration for each value in ListState. TTL support for Map State will be added in future PRs. Once the ttlDuration has expired, the value will not be returned as part of get() and would be cleaned up at the end of the micro-batch.

Why are the changes needed?

These changes are needed to support TTL for ListState. The PR supports specifying ttl for processing time.

Does this PR introduce any user-facing change?

Yes, modifies the ListState interface for specifying ttlDuration

How was this patch tested?

Added the TransformWithListStateTTLSuite, ListStateSuite, StatefulProcessorHandleSuite

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

No

[sahnib]

Thanks for making these changes. Reviewed the code logic and left some comments/questions. Still reviewing testcases.

[anishshri-db]

Should we update the PR description ?

[sahnib]

Thanks for making these changes. Left a few comments.

[sahnib]

LGTM, thanks for adding this functionality

[ericm-db]

@HeartSaVioR PTAL, thanks!

[HeartSaVioR]

Not fully reviewed TransformWithListStateTTLSuite as I found oddness of the testing, as well as the logic itself.

It's great to test the case where elements in the middle are expired first, but the test doesn't seem to catch the edge case. Please correct me if I'm missing on my explanation of edge case.

[HeartSaVioR]

Wait, if we allow the change of TTL during query restart, it is possible that earlier elements in the list has later expiration time.

How this will work with ListStateImplWithTTL.get()? dropWhile would not work.

E.g. some elements in the middle weren't expired at batch N but going to be expired at batch N + 1, and earlier elements are somehow having later expiration time and do not expire at batch N + 1. Calling get() in batch N + 1 will produce all elements including the elements which are going to be expired in batch N + 1, because earlier elements will terminate dropWhile.

[ericm-db]

I'm a little confused by this comment - I think using dropWhile should be okay since we are only returning one row at a time in the iterator, and getNext is called after every invocation of next()

Consider the following case (e for expired, v for valid):
[v1, v2, v3, e1, e2, e3, v4, v5, v6]
Calling dropWhile will drop the first elements (of which there are none in this example) and point the iterator to v1, and we set the currentRow to v1. After we return v1, we set the currentRow to v2, then v3 (as dropWhile doesn't drop these elements)
Now, we have returned v3, and unsafeRowValuesIterator is pointing at e4. We would call dropWhile on this, causing the following three elements to be dropped, and the iterator is consumed

[HeartSaVioR]

Ah you're right, it was confusing because you're creating a new iterator based on underlying iterator. It's a logical (view) iterator which also moves the underlying iterator.

[ericm-db]

No worries

[HeartSaVioR]

+1

[HeartSaVioR]

Thanks! Merging to master.