Type Members

1.  class ListStateImplWithTTL[S] extends OneToManyTTLState with ListState[S]

   Class that provides a concrete implementation for a list state state associated with state variables (with ttl expiration support) used in the streaming transformWithState operator.

   Class that provides a concrete implementation for a list state state associated with state variables (with ttl expiration support) used in the streaming transformWithState operator.

   S

   - data type of object that will be stored

2.  class MapStateImplWithTTL[K, V] extends OneToOneTTLState with MapState[K, V] with Logging

   Class that provides a concrete implementation for map state associated with state variables (with ttl expiration support) used in the streaming transformWithState operator.

   Class that provides a concrete implementation for map state associated with state variables (with ttl expiration support) used in the streaming transformWithState operator.

   K

   - type of key for map state variable

   V

   - type of value for map state variable

   returns

   - instance of MapState of type [K,V] that can be used to store state persistently

3.  abstract class OneToManyTTLState extends TTLState

   OneToManyTTLState is an implementation of TTLState for stateful variables that associate a single key with multiple values; every value has its own expiration timestamp.

   OneToManyTTLState is an implementation of TTLState for stateful variables that associate a single key with multiple values; every value has its own expiration timestamp.

   We need an efficient way to find all the values that have expired, but we cannot issue point-wise deletes to the elements, since they are merged together using the RocksDB StringAppendOperator for merging. As such, we cannot keep a secondary index on the key (expirationMs, groupingKey, indexInList), since we have no way to delete a specific indexInList from the RocksDB value. (In the future, we could write a custom merge operator that can handle tombstones for deleted indexes, but RocksDB doesn't support custom merge operators written in Java/Scala.)

   Instead, we manage expiration per grouping key instead. Our secondary index will look like (expirationMs, groupingKey) -> EMPTY_ROW. This way, we can quickly find all the grouping keys that contain at least one element that has expired.

   To make sure that we aren't "late" in cleaning up expired values, this secondary index maps from the minimum expiration in a list and a grouping key to the EMPTY_VALUE. This index is called the "TTL index" in the code (to be consistent with OneToOneTTLState), though it behaves more like a work queue of lists that need to be cleaned up.

   Since a grouping key may have a large list and we need to quickly know what the minimum expiration is, we need to reverse this work queue index. This reversed index maps from key to the minimum expiration in the list, and it is called the "min-expiry" index.

   Note: currently, this is only used by ListState with TTL.

4.  abstract class OneToOneTTLState extends TTLState

   OneToOneTTLState is an implementation of TTLState that is used to manage TTL for state variables that need a single secondary index to efficiently manage records with an expiration.

   OneToOneTTLState is an implementation of TTLState that is used to manage TTL for state variables that need a single secondary index to efficiently manage records with an expiration.

   The primary index for state variables that can use a OneToOneTTLState have the form of: [elementKey -> (value, elementExpiration)]. You'll notice that, given a timestamp, it would take linear time to probe the primary index for all of its expired values.

   As a result, this class uses helper methods from TTLState to maintain the secondary index from [(elementExpiration, elementKey) -> EMPTY_ROW].

   For an explanation of why this structure is not always sufficient (e.g. why the class OneToManyTTLState is needed), please visit its class-doc comment.

5.  trait TTLState extends AnyRef

   Any state variable that wants to support TTL must implement this trait, which they can do by extending OneToOneTTLState or OneToManyTTLState.

   Any state variable that wants to support TTL must implement this trait, which they can do by extending OneToOneTTLState or OneToManyTTLState.

   The only required methods here are ones relating to evicting expired and all state, via clearExpiredStateForAllKeys and clearAllStateForElementKey, respectively. How classes do this is implementation detail, but the general pattern is to use secondary indexes to make sure cleanup scans theta(records to evict), not theta(all records).

   There are two broad patterns of implementing stateful variables, and thus there are two broad patterns for implementing TTL. The first is when there is a one-to-one mapping between an element key [1] and a value; the primary and secondary index management for this case is implemented by OneToOneTTLState. When a single element key can have multiple values, all of which can expire at their own, unique times, then OneToManyTTLState should be used.

   In either case, implementations need to use some sort of secondary index that orders element keys by expiration time. This base functionality is provided by methods in this trait that read/write/delete to the so-called "TTL index". It is a secondary index with the layout of (expirationMs, elementKey) -> EMPTY_ROW. The expirationMs is big-endian encoded to allow for efficient range scans to find all expired keys.

   TTLState (or any abstract sub-classes) should never deal with encoding or decoding UnsafeRows to and from their user-facing types. The stateful variable themselves should be doing this; all other TTLState sub-classes should be concerned only with writing, reading, and deleting UnsafeRows and their associated expirations from the primary and secondary indexes. [2]

   [1]. You might ask, why call it "element key" instead of "grouping key"? This is because a single grouping key might have multiple elements, as in the case of a map, which has composite keys of the form (groupingKey, mapKey). In the case of ValueState, though, the element key is the grouping key. To generalize to both cases, this class should always use the term elementKey.)

   [2]. You might also ask, why design it this way? We want the TTLState abstract sub-classes to write to both the primary and secondary indexes, since they both need to stay in sync; co-locating the logic is cleanest.

6.  class ValueStateImplWithTTL[S] extends OneToOneTTLState with ValueState[S]

   Class that provides a concrete implementation for a single value state associated with state variables (with ttl expiration support) used in the streaming transformWithState operator.

   Class that provides a concrete implementation for a single value state associated with state variables (with ttl expiration support) used in the streaming transformWithState operator.

   S

   - data type of object that will be stored