> > With that in mind, we could only offer a couple of selected > temporal/sorted > state implementations that are handled internally, but not really a > generic > one - even if you let the user explicitly handle binary keys...
If we want to have a generic interface that is portable between different state backends and allows for all the use-cases described above, lexicographical binary sort sounds reasonable, because you need to be able to push sorting out of the JVM boundary. Only trade off I can think of is that as long as you stay within the JVM (heap state backend), you need to pay a slight key serialization cost, which is IMO ok-ish. Do you have any future state backend ideas in mind, that might not work with this assumption? - I'm really starting to like the idea of having a BinarySortedMapState + higher level / composite states. D. On Fri, Apr 22, 2022 at 1:58 PM David Morávek <david.mora...@gmail.com> wrote: > Isn't allowing a TemporalValueState just a special case of b.III? So if a >> user >> of the state wants that, then they can leverage a simple API vs. if you >> want >> fancier duplicate handling, you'd just go with TemporalListState and >> implement >> the logic you want? > > > Yes it is. But it IMO doesn't justify adding a new state primitive. My > take would be that as long as we can build TVS using other existing state > primitives (TLS) we should treat it as a "composite state". We currently > don't have a good user facing API to do that, but it could be added in > separate FLIP. > > eg. something along the lines of > > TemporalValueState<String> state = getRuntimeContext().getCompositeState( > new CompositeStateDescriptor<>( > "composite", new TemporalValueState(type))); > > On Fri, Apr 22, 2022 at 1:44 PM Nico Kruber <n...@apache.org> wrote: > >> David, >> >> 1) Good points on the possibility to make the TemporalListState generic >> -> actually, if you think about it more, we are currently assuming that >> all >> state backends use the same comparison on the binary level because we add >> an >> appropriate serializer at an earlier abstraction level. This may not >> actually >> hold for all (future) state backends and can limit further >> implementations (if >> you think this is something to keep in mind!). >> >> So we may have to push this serializer implementation further down the >> stack, >> i.e. our current implementation is one that fits RocksDB and that alone... >> >> With that in mind, we could only offer a couple of selected >> temporal/sorted >> state implementations that are handled internally, but not really a >> generic >> one - even if you let the user explicitly handle binary keys... >> >> >> 2) Duplicates >> >> Isn't allowing a TemporalValueState just a special case of b.III? So if a >> user >> of the state wants that, then they can leverage a simple API vs. if you >> want >> fancier duplicate handling, you'd just go with TemporalListState and >> implement >> the logic you want? >> >> >> >> Nico >> >> On Friday, 22 April 2022 10:43:48 CEST David Morávek wrote: >> > Hi Yun & Nico, >> > >> > few thoughts on the discussion >> > >> > 1) Making the TemporalListState generic >> > >> > This is just not possible with the current infrastructure w.r.t type >> > serializers as the sorting key *needs to be comparable on the binary >> level* >> > (serialized form). >> > >> > What I could imagine, is introducing some kind of >> `Sorted(List|Map)State` >> > with explicit binary keys. User would either have to work directly with >> > `byte[]` keys or provide a function for transforming keys into the >> binary >> > representation that could be sorted (this would have to be different >> from >> > `TypeSerializer` which could get more fancy with the binary >> representation, >> > eg. to save up space -> varints). >> > >> > This kind of interface might be really hard to grasp by the pipeline >> > authors. There needs to be a deeper understanding how the byte >> comparison >> > works (eg. it needs to be different from the java byte comparison which >> > compares bytes as `signed`). This could be maybe partially mitigated by >> > providing predefined `to binary sorting key` functions for the common >> > primitives / types. >> > >> > 2) Duplicates >> > >> > I guess, this all boils down to dealing with duplicates / values for the >> > >> > > same timestamp. >> > >> > We should never have duplicates. Let's try to elaborate on what having >> the >> > duplicates really means in this context. >> > >> > a) Temporal Table point of view >> > >> > There could be only a single snapshot of the table at any given point in >> > time (~ physics). If we allow for duplicates we violate this, as it's >> not >> > certain what the actual state of the table is at that point in time. In >> > case of the temporal join, what should the elements from the other side >> > join against? >> > >> > If we happen to have a duplicate, it actually brings us to b) causality >> > (which could actually answer the previous question). >> > >> > b) Causality >> > >> > When building any kind of state machine, it's important to think about >> > causality (if we switch the order of events, state transitions no longer >> > result in the same state). Temporal table is a specific type of the >> state >> > machine. >> > >> > There are several approaches to mitigate this: >> > I) nano-second precision -> the chance that two events affecting the >> same >> > thing happen at the exactly same nanosecond is negligible (from the >> > physical standpoint) >> > II) the sorting key is tuple of (timestamp, sequential id) -> an example >> > could be early firings (you get speculative results from a windowed >> > aggregation with timestamp = EOW, but you can easily assign the order in >> > which these have happened) >> > III) last write with the same timestamp wins -> this is a special case >> of >> > II) when we're sure that elements with the duplicate timestamp come in >> order >> > >> > c) Secondary Sort >> > >> > Handling the actual duplicates requires secondary sort key, which might >> > complicate the `to binary sorting key` interface discussed in 1) - >> > basically some kind of user provided duplicate handler. >> > >> > d) Temporal Value State >> > >> > The above point apply to the temporal value state as well, it just >> pushes >> > the responsibility away from the state interface. I'm still not >> convinced >> > that this is a right direction. >> > >> > - >> > >> > For most use cases I've seen, the millisecond precision is more than >> enough >> > (+ the last write wins as a fallback). Supporting the use cases where >> it's >> > not actually enough (I've seen that as well in the past, especially with >> > the early firings), might be actually a good case for a more generic >> form >> > of the state, that we've discussed in 1). >> > >> > 3) Map vs List >> > >> > I think this also boils down to the discussion of how to handle >> duplicates. >> > From the commonly accepted contracts, list implies that there could be >> > duplicates and map implies otherwise. One concern about `Map` is that it >> > also implies that you should be able to do a point query. >> > >> > Best, >> > D. >> > >> > . >> > >> > On Fri, Apr 22, 2022 at 9:21 AM Yun Tang <myas...@live.com> wrote: >> > > Hi Nico, >> > > >> > > I did not mean that we need to support all APIs in NavigableMap, and >> it is >> > > indeed too heavily to implement them all. >> > > Moreover, I also prefer iterator-like API instead of the original >> #tailMap >> > > like API. I just use NavigableMap's API to show examples. >> > > >> > > I think we can introduce new APIs like: >> > > >> > > SortedMapState<UK, UV> extends State >> > > >> > > Map.Entry<UK, UV> firstEntry() throws Exception; >> > > Map.Entry<UK, UV> lastEntry() throws Exception; >> > > Iterator<Map.Entry<UK, UV>> headIterator(UK endUserKey) throws >> > > Exception; >> > > Iterator<Map.Entry<UK, UV>> tailIterator(UK startUserKey) throws >> > > >> > > Exception; >> > > >> > > Iterator<Map.Entry<UK, UV>> subIterator(UK startUserKey, UK >> endUserKey) >> > > >> > > throws Exception; >> > > >> > > Since SortedMapState has several new APIs, I prefer to introduce new >> state >> > > descriptor to distinguish with the original map state. >> > > >> > > For the API of SortedMapOfListsState, I would not be strong against, >> and I >> > > just want to know the actual benefits if we really want to introduce >> API. >> > > >> > > When talking about the part of changelog state backend, my concern is >> > > about how to group keys together in the changelog logger. >> > > I can share a problem, and before this I need to spend some time on >> how to >> > > implement serializer to keep the order of serialized bytes same as >> > > original >> > > java objects first. >> > > For the fixed-length serializer, such as LongSerializer, we just need >> to >> > > ensure all numbers are positive or inverting the sign bit. >> > > However, for non-fixed-length serializer, such as StringSerializer, it >> > > will write the length of the bytes first, which will break the natural >> > > order if comparing the bytes. Thus, we might need to avoid to write >> the >> > > length in the serialized bytes. >> > > On the other hand, changelog logger would record operation per key >> one by >> > > one in the logs. We need to consider how to distinguish each key in >> the >> > > combined serialized byte arrays. >> > > >> > > Best >> > > Yun Tang >> > > >> > > ------------------------------ >> > > *From:* Nico Kruber <n...@apache.org> >> > > *Sent:* Thursday, April 21, 2022 23:50 >> > > *To:* dev <dev@flink.apache.org> >> > > *Cc:* David Morávek <david.mora...@gmail.com>; Yun Tang < >> myas...@live.com> >> > > *Subject:* Re: [DISCUSS] FLIP-220: Temporal State >> > > >> > > Thanks Yun Tang for your clarifications. >> > > Let me keep my original structure and reply in these points... >> > > >> > > 3. Should we generalise the Temporal***State to offer arbitrary key >> types >> > > and >> > > not just Long timestamps? >> > > >> > > The use cases you detailed do indeed look similar to the ones we were >> > > optimising in our TemporalState PoC... >> > > >> > > I don't think, I'd like to offer a full implementation of NavigableMap >> > > though >> > > because that seems quite some overhead to implement while we can >> cover the >> > > mentioned examples with the proposed APIs already when using >> iterators as >> > > well >> > > as single-value retrievals. >> > > So far, when we were iterating from the smallest key, we could just >> use >> > > Long.MIN_VALUE and start from there. That would be difficult to >> generalise >> > > for >> > > arbitrary data types because you may not always know the smallest >> possible >> > > value for a certain serialized type (unless we put this into the >> > > appropriate >> > > serializer interface). >> > > >> > > I see two options here: >> > > a) a slim API but using NULL as an indicator for smallest/largest >> > > depending on >> > > the context, e.g. >> > > >> > > - `readRange(null, key)` means from beginning to key >> > > - `readRange(key, null)` means from key to end >> > > - `readRange(null, null)` means from beginning to end >> > > - `value[AtOr]Before(null)` means largest available key >> > > - `value[AtOr]After(null)` means smallest available key >> > > >> > > b) a larger API with special methods for each of these use cases >> similar >> > > to >> > > what NavigableMap has but based on iterators and single-value >> functions >> > > only >> > > >> > > > BTW, I prefer to introduce another state descriptor instead of >> current >> > > >> > > map >> > > >> > > > state descriptor. >> > > >> > > Can you elaborate on this? We currently don't need extra >> functionality, so >> > > this would be a plain copy of the MapStateDescriptor... >> > > >> > > > For the API of SortedMapOfListsState, I think this is a bit bounded >> to >> > > > current implementation of RocksDB state-backend. >> > > >> > > Actually, I don't think this is special to RocksDB but generic to all >> > > state >> > > backends that do not hold values in memory and allow fast append-like >> > > operations. >> > > Additionally, since this is a very common use case and RocksDB is also >> > > widely >> > > used, I wouldn't want to continue without this specialization. For a >> > > similar >> > > reason, we offer ListState and not just ValueState<List>... >> > > >> > > >> > > 4. ChangelogStateBackend >> > > >> > > > For the discussion of ChangelogStateBackend, you can think of >> changelog >> > > > state-backend as a write-ahead-log service. And we need to record >> the >> > > > changes to any state, thus this should be included in the design >> doc as >> > > >> > > we >> > > >> > > > need to introduce another kind of state, especially you might need >> to >> > > > consider how to store key bytes serialized by the new serializer >> (as we >> > > > might not be able to write the length in the beginning of serialized >> > > >> > > bytes >> > > >> > > > to make the order of bytes same as natural order). >> > > >> > > Since the ChangelogStateBackend "holds the working state in the >> underlying >> > > delegatedStateBackend, and forwards state changes to State >> Changelog", I >> > > honestly still don't see how this needs special handling. As long as >> the >> > > delegated state backend suppors sorted state, ChangelogStateBackend >> > > doesn't >> > > have to do anything special except for recording changes to state. >> Our PoC >> > > simply uses the namespace for these keys and that's the same thing the >> > > Window >> > > API is already using - so there's nothing special here. The order in >> the >> > > log >> > > doesn't have to follow the natural order because this is only required >> > > inside >> > > the delegatedStateBackend, isn't it? >> > > >> > > >> > > Nico >> > > >> > > On Wednesday, 20 April 2022 17:03:11 CEST Yun Tang wrote: >> > > > Hi Nico, >> > > > >> > > > Thanks for your clarification. >> > > > For the discussion about generalizing Temporal state to sorted map >> > > >> > > state. I >> > > >> > > > could give some examples of how to use sorted map state in min/max >> with >> > > > retract functions. >> > > > >> > > > As you know, NavigableMap in java has several APIs like: >> > > > Map.Entry<K,V> firstEntry(); >> > > > Map.Entry<K,V> lastEntry(); >> > > > NavigableMap<K,V> tailMap(K fromKey, boolean inclusive) >> > > > >> > > > The #firstEntry API could be used in >> > > > MinWithRetractAggFunction#updateMin, >> > > > #lastEntry could be used in MaxWithRetractAggFunction#updateMax, and >> > > > #tailMap could be used in FirstValueWithRetractAggFunction#retract. >> > > >> > > If we >> > > >> > > > can introduce SortedMap-like state, these functions could be >> benefited. >> > > > BTW, I prefer to introduce another state descriptor instead of >> current >> > > >> > > map >> > > >> > > > state descriptor. >> > > > For the API of SortedMapOfListsState, I think this is a bit bounded >> to >> > > > current implementation of RocksDB state-backend. >> > > > >> > > > For the discussion of ChangelogStateBackend, you can think of >> changelog >> > > > state-backend as a write-ahead-log service. And we need to record >> the >> > > > changes to any state, thus this should be included in the design >> doc as >> > > >> > > we >> > > >> > > > need to introduce another kind of state, especially you might need >> to >> > > > consider how to store key bytes serialized by the new serializer >> (as we >> > > > might not be able to write the length in the beginning of serialized >> > > >> > > bytes >> > > >> > > > to make the order of bytes same as natural order). >> > > > >> > > > Best >> > > > Yun Tang. >> > > > ________________________________ >> > > > From: Nico Kruber <n...@apache.org> >> > > > Sent: Wednesday, April 20, 2022 0:38 >> > > > To: dev <dev@flink.apache.org> >> > > > Cc: Yun Tang <myas...@live.com>; David Morávek < >> david.mora...@gmail.com> >> > > > Subject: Re: [DISCUSS] FLIP-220: Temporal State >> > > > >> > > > Hi all, >> > > > I have read the discussion points from the last emails and would >> like to >> > > > add >> > > >> > > my two cents on what I believe are the remaining points to solve: >> > > > 1. Do we need a TemporalValueState? >> > > > >> > > > I guess, this all boils down to dealing with duplicates / values >> for the >> > > > same >> > > >> > > timestamp. Either you always have to account for them and thus always >> > > >> > > > have to store a list anyway, or you only need to join with "the >> latest" >> > > >> > > (or >> > > >> > > > the only) value for a given timestamp and get a nicer API and lower >> > > > overhead for that use case. >> > > > At the easiest, you can make an assumption that there is only a >> single >> > > > value >> > > >> > > for each timestamp by contract, e.g. by increasing the timestamp >> > > >> > > > precision and interpreting them as nanoseconds, or maybe >> milliseconds >> > > > are >> > > > already good enough. If that contract breaks, however, you will get >> into >> > > > undefined behaviour. >> > > > The TemporalRowTimeJoinOperator, for example, currently just assumes >> > > > that >> > > > there is only a single value on the right side of the join >> (rightState) >> > > >> > > and >> > > >> > > > I >> > > >> > > believe many use cases can make that assumption or otherwise you'd >> have >> > > >> > > > to define the expected behaviour for multiple values at the same >> > > >> > > timestamp, >> > > >> > > > e.g. "join with the most recent value at the time of the left side >> and >> > > > if >> > > > there are multiple values, choose X". >> > > > >> > > > I lean towards having a ValueState implementation as well (in >> addition >> > > > to >> > > > lists). >> > > > >> > > > >> > > > 2. User-facing API (Iterators vs. valueAtOr[Before|After]) >> > > > >> > > > I like the iterable-based APIs that David M was proposing, i.e. >> > > > - Iterable<TimestampedValue<T>> readRange(long minTimestamp, long >> > > > limitTimestamp); >> > > > - void clearRange(long minTimestamp, long limitTimestamp); >> > > > >> > > > However, I find Iterables rather cumbersome to work with if you >> actually >> > > > only >> > > >> > > need a single value, e.g. the most recent one. >> > > >> > > > For iterating over a range of values, however, they feel more >> natural to >> > > >> > > me >> > > >> > > > than our proposal. >> > > > >> > > > Actually, if we generalise the key type (see below), we may also >> need to >> > > > offer >> > > >> > > additional value[Before|After] functions to cover "+1" iterations >> > > >> > > > where we cannot simply add 1 as we do now. >> > > > >> > > > (a) How about offering both Iterables and >> > > > value[AtOrBefore|AtOrAfter|Before| >> > > >> > > After]? >> > > >> > > > This would be similar to what NavigableMap [2] is offering but with >> a >> > > >> > > more >> > > >> > > > explicit API than "ceiling", "floor",... >> > > > >> > > > (b) Our API proposal currently also allows iterating backwards >> which is >> > > >> > > not >> > > >> > > > covered by the readRange proposal - we could, however, just do that >> if >> > > > minTimestamp > limitTimestamp). What do you think? >> > > > >> > > > (c) When implementing the iterators, I actually also see two >> different >> > > > modes >> > > >> > > which may differ in performance: I call them iteration with eager >> > > >> > > > vs. lazy value retrieval. Eager retrieval may retrieve all values >> in a >> > > > range at once and make them available in memory, e.g. for smaller >> data >> > > >> > > sets >> > > >> > > > similar to what TemporalRowTimeJoinOperator is doing for the right >> side >> > > >> > > of >> > > >> > > > the join. This can be spare a lot of Java<->JNI calls and let >> RocksDB >> > > > iterate only once (as long at things fit into memory). Lazy >> retrieval >> > > >> > > would >> > > >> > > > fetch results one-by-one. -> We could set one as default and allow >> the >> > > >> > > user >> > > >> > > > to override that behaviour. >> > > > >> > > > 3. Should we generalise the Temporal***State to offer arbitrary key >> > > > types >> > > > and >> > > >> > > not just Long timestamps? >> > > >> > > > @Yun Tang: can you describe in more detail where you think this >> would be >> > > > needed for SQL users? I don't quite get how this would be >> beneficial. >> > > > The >> > > > example you linked doesn't quite show the same behaviour. >> > > > >> > > > Other than this, I could see that you can leverage such a >> generalisation >> > > > for >> > > >> > > arbitrary joins between, for example, IDs and ID ranges which don't >> > > >> > > > have a time component attached to it. Given that this shouldn't be >> too >> > > > difficult to expose (the functionality has to exist anyway, but >> > > > otherwise >> > > > buried into Flink's internals). We'd just have to find suitable >> names. >> > > > >> > > > (a) I don't think TemporalListState<T> is actually >> SortedMapState<Long, >> > > > List<T>> because we need efficient "add to list" primitives which >> cannot >> > > > easily be made available with a single generic SortedMapState... >> > > > >> > > > (b) So the most expressive (yet kind-of ugly) names could be >> > > > - SortedMapState<Long, ValueType> >> > > > - SortedMapOfListsState<Long, List<ValueType>> >> > > > >> > > > (c) For both of these, we could then re-use the existing >> > > >> > > MapStateDescriptor >> > > >> > > > to >> > > >> > > define key and value/list-element types and require that the key >> type / >> > > >> > > > serializer implements a certain RetainingSortOrderSerializer >> interface >> > > >> > > (and >> > > >> > > > think about a better name for this) which defines the contract that >> the >> > > > binary sort order is the same as the Java Object one. >> > > > -> that can also be verified at runtime to fail early. >> > > > >> > > > >> > > > 4. ChangelogStateBackend: we don't think this needs special >> attention - >> > > >> > > it >> > > >> > > > is >> > > >> > > just delegating to the other backends anyway and these methods are >> > > >> > > > already adapted in our POC code >> > > > >> > > > >> > > > @David M, Yun Tang: let me/us know what you think about these >> proposals >> > > > >> > > > >> > > > >> > > > Nico >> > > > >> > > > >> > > > [2] >> > > >> > > https://docs.oracle.com/javase/8/docs/api/java/util/NavigableMap.html >> > > >> > > > On Thursday, 14 April 2022 14:15:53 CEST Yun Tang wrote: >> > > > > Hi David Anderson, >> > > > > >> > > > > >> > > > > >> > > > > I feel doubted that no motivating use case for this >> generalization to >> > > > > SortedMapState. From our internal stats, SQL user would use much >> more >> > > > > cases >> > > >> > > of min/max with retract functions [1] compared with interval join. >> > > >> > > > > From my understanding, the TemporalListState<T> is actually >> > > > > SortedMapState<Long, List<T>>, while TemporalValueState<T> is >> > > > > SortedMapState<Long, T>. As you can see, if we just restrict >> > > > > SortedMapState >> > > >> > > with the key type as Long, all current two new interfaces >> > > >> > > > > could be replaced. >> > > > > >> > > > > >> > > > > >> > > > > Moreover, once we introduce temporal state, the extension would be >> > > > > limited. >> > > >> > > Apart from Long, many other types could be comparable, e.g. >> > > >> > > > > TimeStamp, Int, Float and so on. How could we handle these feature >> > > > > request after >> > > > > TemporalState merged? I don't think introducing too many state >> types >> > > >> > > is a >> > > >> > > > > good idea. We can only support Long type for the 1st version when >> > > > > introducing SortedMapState, and then extends it to many other more >> > > >> > > types >> > > >> > > > > in >> > > >> > > the future. This could balance the feature requests with clean >> > > >> > > > > interfaces design. And thus, we can also use sorted map state in >> the >> > > > > popular min/max functions. >> > > > > >> > > > > >> > > > > >> > > > > >> > > > > >> > > > > By the way, current FLIP lacks of consideration of the work on >> > > >> > > changelog >> > > >> > > > > state-backend once two new state types are introduced. >> > > > > >> > > > > >> > > > > >> > > > > >> > > > > >> > > > > [1] >> > > >> > > >> https://github.com/apache/flink/blob/master/flink-table/flink-table-runtim >> > > >> > > > > e >> > > >> > > >> /src/main/java/org/apache/flink/table/runtime/functions/aggregate/MinWith >> > > >> > > > > Ret ractAggFunction.java >> > > > > >> > > > > >> > > > > >> > > > > Best, >> > > > > Yun Tang >> > > > > ________________________________ >> > > > > From: David Morávek <david.mora...@gmail.com> >> > > > > Sent: Wednesday, April 13, 2022 19:50 >> > > > > To: dev <dev@flink.apache.org> >> > > > > Subject: Re: [DISCUSS] FLIP-220: Temporal State >> > > > > >> > > > > >> > > > > >> > > > > Here is a very naive implementation [1] from a prototype I did few >> > > >> > > months >> > > >> > > > > back that uses list and insertion sort. Since the list is sorted >> we >> > > > > can >> > > > > use >> > > >> > > binary search to create sub-list, that could leverage the same thing >> > > >> > > > > I've described above. >> > > > > >> > > > > >> > > > > >> > > > > I think back then I didn't go for the SortedMap as it would be >> hard to >> > > > > implement with the current heap state backend internals and would >> have >> > > > > bigger memory overhead. >> > > > > >> > > > > >> > > > > >> > > > > The ideal solution would probably use skip list [2] to lower the >> > > >> > > overhead >> > > >> > > > > of the binary search, while maintaining a reasonable memory >> footprint. >> > > > > Other than that it could be pretty much the same as the prototype >> > > > > implementation [1]. >> > > > > >> > > > > >> > > > > >> > > > > [1] >> > > >> > > >> https://github.com/dmvk/flink/blob/ecdbc774b13b515e8c0943b2c143fb1e34eca6f >> > > >> > > > > 0/ >> > > >> > > >> flink-runtime/src/main/java/org/apache/flink/runtime/state/heap/HeapTempo >> > > >> > > > > ral ListState.java >> > > > >> > > > [2] https://en.wikipedia.org/wiki/Skip_list >> > > > >> > > > > Best, >> > > > > D. >> > > > > >> > > > > >> > > > > >> > > > > On Wed, Apr 13, 2022 at 1:27 PM David Morávek < >> david.mora...@gmail.com >> > > > > >> > > > > wrote: >> > > > > > Hi David, >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > It seems to me that at least with the heap-based state backend, >> > > > > > readRange >> > > > > > >> > > > > >> is going to have to do a lot of unnecessary work to implement >> this >> > > > > >> isEmpty() operation, since it have will to consider the entire >> > > > > >> range >> > > > > >> from >> > > > > >> MIN_VALUE to MAX_VALUE. (Maybe we should add an explicit >> isEmpty >> > > > > >> method? >> > > > > >> I'm not convinced we need it, but it would be cheaper to >> implement. >> > > >> > > Or >> > > >> > > > > >> perhaps this join can be rewritten to not need this operation; >> I >> > > > > >> haven't >> > > > > >> thought enough about that alternative.) >> > > > > > >> > > > > > I think this really boils down to how the returned iterable is >> going >> > > >> > > to >> > > >> > > > > > be >> > > > > > implemented. Basically for checking whether state is empty, you >> need >> > > >> > > to >> > > >> > > > > > do >> > > > > > something along the lines of: >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > Iterables.isEmpty(state.readRange(Long.MIN_VALUE, MAX_VALUE)); >> // >> > > > > > basically checking `hasNext() == false` or `isEmpty()` in case >> of >> > > > > > `Collection` >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > Few notes: >> > > > > > 1) It could be lazy (the underlying collection doesn't have to >> be >> > > > > > materialized - eg. in case of RocksDB); >> > > > > > 2) For HeapStateBackend it depends on the underlying >> implementation. >> > > > > > I'd >> > > > > > probably do something along the lines of sorted tree (eg. >> SortedMap >> > > > > > / >> > > > > > NavigableMap), that allows effective range scans / range >> deletes. >> > > >> > > Then >> > > >> > > > > > you >> > > > > > could simply do something like (from top of the head): >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > @Value >> > > > > > class TimestampedKey<K> { >> > > > > > >> > > > > > K key; >> > > > > > long timestamap; >> > > > > > >> > > > > > } >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > SortedMap<TimestampedKey<K>, V> internalState; >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > Iterable<TimestampedValue<V>> readRange(long min, long max) { >> > > > > > >> > > > > > return toIterable(internalState.subMap(new >> > > > > > TimestampedKey(currentKey(), >> > > > > > >> > > > > > min), new TimestampedKey(currentKey(), max))); >> > > > > > } >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > This should be fairly cheap. The important bit is that the >> returned >> > > > > > iterator is always non-null, but could be empty. >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > Does that answer your question? >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > D. >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > >> > > > > > On Wed, Apr 13, 2022 at 12:21 PM David Anderson < >> > > >> > > da...@alpinegizmo.com> >> > > >> > > > > > wrote: >> > > > > >> Yun Tang and Jingsong, >> > > > > >> >> > > > > >> >> > > > > >> >> > > > > >> >> > > > > >> >> > > > > >> Some flavor of OrderedMapState is certainly feasible, and I do >> see >> > > > > >> some >> > > > > >> appeal in supporting Binary**State. >> > > > > >> >> > > > > >> >> > > > > >> >> > > > > >> >> > > > > >> >> > > > > >> However, I haven't seen a motivating use case for this >> > > >> > > generalization, >> > > >> > > > > >> and >> > > > > >> would rather keep this as simple as possible. By handling >> Longs we >> > > >> > > can >> > > >> > > > > >> already optimize a wide range of use cases. >> > > > > >> >> > > > > >> >> > > > > >> >> > > > > >> >> > > > > >> >> > > > > >> David >> > > > > >> >> > > > > >> On Tue, Apr 12, 2022 at 9:21 AM Yun Tang <myas...@live.com> >> wrote: >> > > > > >> > Hi David, >> > > > > >> > >> > > > > >> > Could you share some explanations why SortedMapState cannot >> work >> > > >> > > in >> > > >> > > > > >> > details? I just cannot catch up what the statement below >> means: >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > This was rejected as being overly difficult to implement in >> a way >> > > > > >> > that >> > > > > >> > would cleanly leverage RocksDB’s iterators. >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > Best >> > > > > >> > Yun Tang >> > > > > >> > ________________________________ >> > > > > >> > From: Aitozi <gjying1...@gmail.com> >> > > > > >> > Sent: Tuesday, April 12, 2022 15:00 >> > > > > >> > To: dev@flink.apache.org <dev@flink.apache.org> >> > > > > >> > Subject: Re: [DISCUSS] FLIP-220: Temporal State >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > Hi David >> > > > > >> > >> > > > > >> > I have look through the doc, I think it will be a good >> > > > > >> > improvement >> > > > > >> >> > > > > >> to >> > > > > >> >> > > > > >> > this pattern usage, I'm interested in it. Do you have some >> POC >> > > >> > > work >> > > >> > > > > >> > to >> > > > > >> > share for a closer look. >> > > > > >> > Besides, I have one question that can we support expose the >> > > > > >> > namespace >> > > > > >> > in >> > > > > >> > the different state type not limited to `TemporalState`. By >> this, >> > > > > >> > user >> > > > > >> >> > > > > >> can >> > > > > >> >> > > > > >> > specify the namespace >> > > > > >> > and the TemporalState is one of the special case that it use >> > > > > >> > timestamp >> > > > > >> >> > > > > >> as >> > > > > >> >> > > > > >> > the namespace. I think it will be more extendable. >> > > > > >> > >> > > > > >> > What do you think about this ? >> > > > > >> > >> > > > > >> > Best, >> > > > > >> > Aitozi. >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > >> > > > > >> > David Anderson <dander...@apache.org> 于2022年4月11日周一 20:54写道： >> > > > > >> > >> > > > > >> > > Greetings, Flink developers. >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > I would like to open up a discussion of a proposal [1] to >> add a >> > > > > >> > > new >> > > > > >> >> > > > > >> kind >> > > > > >> >> > > > > >> > of >> > > > > >> > >> > > > > >> > > state to Flink. >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > The goal here is to optimize a fairly common pattern, >> which is >> > > > > >> > > using >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > MapState<Long, List<Event>> >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > to store lists of events associated with timestamps. This >> > > >> > > pattern >> > > >> > > > > >> > > is >> > > > > >> >> > > > > >> used >> > > > > >> >> > > > > >> > > internally in quite a few operators that implement sorting >> and >> > > > > >> > > joins, >> > > > > >> >> > > > > >> and >> > > > > >> >> > > > > >> > > it also shows up in user code, for example, when >> implementing >> > > > > >> > > custom >> > > > > >> > > windowing in a KeyedProcessFunction. >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > Nico Kruber, Seth Wiesman, and I have implemented a POC >> that >> > > > > >> > > achieves >> > > > > >> >> > > > > >> a >> > > > > >> >> > > > > >> > > more than 2x improvement in throughput when performing >> these >> > > > > >> >> > > > > >> operations >> > > > > >> >> > > > > >> > on >> > > > > >> > >> > > > > >> > > RocksDB by better leveraging the capabilities of the >> RocksDB >> > > >> > > state >> > > >> > > > > >> > backend. >> > > > > >> > >> > > > > >> > > See FLIP-220 [1] for details. >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > Best, >> > > > > >> > > David >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > >> > > > > >> > > [1] https://cwiki.apache.org/confluence/x/Xo_FD >> > > > >> > > > Dr. Nico Kruber | Solutions Architect >> > > > >> > > > Follow us @VervericaData Ververica >> > > > -- >> > > > Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany >> > > > -- >> > > > Ververica GmbH >> > > > Registered at Amtsgericht Charlottenburg: HRB 158244 B >> > > > Managing Directors: Yip Park Tung Jason, Jinwei (Kevin) Zhang, Karl >> > > > Anton >> > > > Wehner >> > > >> > > -- >> > > Dr. Nico Kruber | Solutions Architect >> > > >> > > Follow us @VervericaData Ververica >> > > -- >> > > Join Flink Forward - The Apache Flink Conference >> > > Stream Processing | Event Driven | Real Time >> > > -- >> > > Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany >> > > -- >> > > Ververica GmbH >> > > Registered at Amtsgericht Charlottenburg: HRB 158244 B >> > > Managing Directors: Yip Park Tung Jason, Jinwei (Kevin) Zhang, Karl >> Anton >> > > Wehner >> >> >> -- >> Dr. Nico Kruber | Solutions Architect >> >> Follow us @VervericaData Ververica >> -- >> Join Flink Forward - The Apache Flink Conference >> Stream Processing | Event Driven | Real Time >> -- >> Ververica GmbH | Invalidenstrasse 115, 10115 Berlin, Germany >> -- >> Ververica GmbH >> Registered at Amtsgericht Charlottenburg: HRB 158244 B >> Managing Directors: Yip Park Tung Jason, Jinwei (Kevin) Zhang, Karl Anton >> Wehner >> >> >>
