Hi Soumitra, Thanks for the FLIP. This is little outside of my scope of expertise, but I wanted to reply to one thing that you said, being
> Flink should not dictate a particular way of writing an application. It is > up to the user to decide how to express their computation using different > primitives. Different primitives provide different runtime characteristics, > have different resource needs, and may need different tunings to achieve best > performance. I don't think this is a true statement at all. Users struggle with writing good, performant streaming applications, and we need to make it as easy for them to achieve this. I would argue that in the declarative landscape of APIs, it's actually Flink that under the hood decides how computation is being achieved, and not the user. In my opinion, Flink should actually be (a bit) more opinionated towards users and guide them to the right approach, and offer primitives specifically for users who need lower level interfaces for specialized use cases. Best regards, Martijn Op do 4 jun 2026 om 07:30 schreef Soumitra Kumar <[email protected]>: > > Hi Roman, > > Thanks for your comments! > > I agree that there are several ways to implement the event reordering, and > in fact I have 3 implementations with different runtime characteristics. I > don't know how MapState can help with generic ordering, but my knowledge on > that is limited. > > Here is my line of thinking and justification of the FLIP: > > 1. Flink is a wonderful distributed processing engine and it already > leverages RocksDB as one of the state backends. > 2. Flink should not dictate a particular way of writing an application. > It is up to the user to decide how to express their computation using > different primitives. Different primitives provide different runtime > characteristics, have different resource needs, and may need different > tunings to achieve best performance. > 3. RocksDB is a C/C++ engine and supports associative merge operators. > In fact, ListState is built using one of the merge operators defined in C++ > in RocksDB. > 4. How cool would it be to enable support for Java-based associative > merge operators in FRocksDB! This would enable Flink application developers > to build aggregations using associative binary operations. A few > interesting such aggregations are set union, set intersection, bloom > filters, HLLs, Count-Min Sketch, etc. > > Needless to say, the devil is in the details to ensure this works at > reasonable performance. We can discuss and address those concerns at the PR > stage. I am motivated and excited about this enhancement. I have a few > things to add to the FLIP based on prior comments, and I am available to > answer any questions or concerns to make progress on the FLIP. > > Thanks again, best, > -Soumitra. > > On Wed, Jun 3, 2026 at 1:30 AM Roman Khachatryan <[email protected]> wrote: > > > Thanks for the clarifications, > > > > I briefly took a look at the use case > > > THIS <https://github.com/soumitrak/flink_streaming_event_reordering> > > repo. > > > Essentially, the scenario is to reorder events based on event time, and > > > process the ordered events every minute. In this case, I am building a > > > sorted list using an associative operator that implements merging two > > > sorted lists. > > > > Isn't it possible to use MapState keyed by event time? > > The sorting will come for free on RocksDB and PUT will only add the new > > element without touching the existing ones. > > (there is an API to check whether it's sorted or not) > > > > >> 6. Alternative: ListState > > > ListState uses the "stringappendtest" merge operator, defined in the C++ > > > layer in FrocksDB, that concatenates strings using comma char. I can > > > implement the sorted list using this construct, > > > > I think we're talking about different things. My proposal was to use > > ListState without > > using the merge operator directly. Yes it will load all the records for > > aggregation - but > > that's the same as in the original proposal, isn't it? > > Anyways, if the requirement is to sort those records then using MapState > > keyed by > > the sort key looks like a better idea to me. > > > > > > Regards, > > Roman > > > > > > On Mon, Jun 1, 2026 at 6:44 AM Soumitra Kumar <[email protected]> > > wrote: > > > > > Hi Roman, > > > > > > Thanks for your thorough review, please find the comments below. > > > > > > On Sun, May 31, 2026 at 2:42 PM Roman Khachatryan <[email protected]> > > > wrote: > > > > > > > 1. Is it possible that reads will become more costly in some scenarios? > > > > > > > > Per my understanding, on compaction RocksDB would need to read and > > write > > > > back the same state (compared to the current implementation) > > > > So the benefit is deferring and "batching" those read/write operations. > > > > > > > > Current approach > > > > add: read + deserialize + aggregate + serialize + write (IO#1) > > > > get: read + deserialize (IO#2) > > > > > > > > Proposed approach > > > > add: serialize + write (IO#1) > > > > compact: read + deserialize + aggregate + serialize + write (IO#2) > > > > get: read + deserialize (IO#3) > > > > > > > > But if adds are mostly followed by gets then there's more IO because > > > > RocksDB needs to make an additional call via JNI before serving reads, > > > > right? > > > > If that's the case, we need to have two implementations and select one > > > > depending on the case? > > > > > > > > > > Your calculation of the cost of the proposed approach is correct. If > > there > > > is frequent read-after-write, then the merge operator does not add value. > > > Users should use the existing read-modify-write based state variables > > > instead. The proposal does not change the behavior of any of the existing > > > state variables. Users can decide when to use the new state variables > > that > > > support custom Reduce/Aggregate functions. > > > > > > > > > > 2. Speaking more generally, could you list the motivating use cases, > > e.g. > > > > existing Flink DataStream Operators, custom operators, SQL operators? > > > > If the impact depends on the operator, it would be helpful to > > understand > > > > how the existing operators would be affected. > > > > > > > > > > I got the idea of this enhancement based on a real use case at my work. I > > > have captured the scenarios in THIS > > > <https://github.com/soumitrak/flink_streaming_event_reordering> repo. > > > Essentially, the scenario is to reorder events based on event time, and > > > process the ordered events every minute. In this case, I am building a > > > sorted list using an associative operator that implements merging two > > > sorted lists. > > > > > > In general, the proposal should benefit write-heavy (less read) scenarios > > > where the computation can be a Reduce/Aggregate function. The existing > > > state variables do read-modify-write and become a bottleneck for > > ingestion > > > for write-heavy use cases. Check the performance using the prototype > > > implementation HERE > > > < > > > > > https://github.com/soumitrak/flink_streaming_event_reordering#performance-comparison > > > > > > > . > > > > > > > > > > 3. Could you clarify how the Reduce/Aggregate function calls will be > > > > dispatched? > > > > In theory, if an operator has two such states, their states will end up > > > in > > > > the same RocksDB instance and we need to distinguish them, right? > > > > > > > > > RocksDB allows one merge operator per ColumnFamily, and AFAIU, Flink > > > creates one ColumnFamily per state variable. Currently, Flink sets the > > > "stringappendtest" > > > < > > > > > https://github.com/apache/flink/blob/6c756a084d40f4716cdfdb4fdd71b0c919c448e6/flink-state-backends/flink-statebackend-rocksdb/src/main/java/org/apache/flink/state/rocksdb/RocksDBKeyedStateBackend.java#L133 > > > > > > > merge operator for all ColumnFamilies. In this proposal, Flink can set a > > > user provided Java based Reduce/Aggregate function as the merge operator. > > > In your example, if the operator has two state variables, then there will > > > be two ColumnFamilies, and Flink can set a custom ReduceFunction for one > > > variable and a custom AggregateFunction for the other variable. We don't > > > need to change anything at this layer in Flink/RocksDB. > > > > > > > > > > 4. Function dispatch during restore > > > > IIRC, aggregate functions might not be available during restore > > > operation - > > > > because state descriptors containing these functions are set later. > > > > Is this a concern? Do we need to disable compactions during restore? > > > > > > > > > > Great point! The Reduce/Aggregate functions in Flink are already > > > serializable and in the proposal they are serialized in the savepoint and > > > are restored when rocksdb is loaded. This allows rocksdb to call these > > > functions during compaction before state descriptors are called. This way > > > we don't need to disable compaction during restore. > > > > > > > > > > 5. Remote compactions > > > > In the light of Disaggregated State Backend, Flink might benefit from > > > > remote compactions. > > > > This may help with predictability (no compaction impact on processing) > > > and > > > > achieving better cost/utilization. > > > > AFAIK ForSt doesn't support it ATM, but it would be great to keep this > > > > possibility open. > > > > And I think the current proposal will make remote compactions > > > problematic. > > > > > > > > > > TBH, I don't understand ForSt in detail to comment on this item. Since > > the > > > proposal is exposing associative merge operators, it should not be an > > issue > > > to support in ForSt. In fact, if ForSt does not support associative merge > > > operators, then I will volunteer to add it, but let's get this proposal > > > first. > > > > > > > > > > 6. Alternative: ListState > > > > Have you considered using ListState to accumulate the records; and then > > > > reduce/aggregate on read (or incrementally on write)? > > > > Basically, doing the same thing as you proposed but purely in Java. > > > > > > > > I think it should address all the concerns I mentioned above; plus > > > > - removes the associative requirement (because we control the order) > > > > - removes RocksDB dependency > > > > - gives some flexibility when to aggregate (incrementally on write > > and/or > > > > on read) > > > > > > > > (I couldn't find this in the Rejected Alternatives, sorry if I missed > > > > that). > > > > > > > > > > ListState uses the "stringappendtest" merge operator, defined in the C++ > > > layer in FrocksDB, that concatenates strings using comma char. I can > > > implement the sorted list using this construct, but the read will be more > > > expensive than the proposal, since the sorting will happen during the > > read. > > > The proposal adds support for pure Java based associative merge operators > > > that will be invoked by RocksDB using JNI. There will be JNI overhead, > > but > > > it is not on the write thread. The write thread is as performant as the > > > ListState implementation. I have a prototype implementation that reduces > > > the JNI overhead. I think this can be a great addition to Flink and, just > > > like other advanced features, it requires understanding/tuning of > > rocksdb. > > > > > > Best, > > > -Soumitra. > > > > > > > > > > > > > > On Tue, May 26, 2026 at 3:50 PM Zakelly Lan <[email protected]> > > > wrote: > > > > > > > > > Hi Soumitra, > > > > > > > > > > I'm somehow getting confused. I was expecting there is no public API > > > > > change in Flink side, and users could use the original state > > > descriptors > > > > to > > > > > create ReducingState/AggregatingState. And in the implementation > > side, > > > > > taking RocksDBReducingState as example, a wrapper of user defined > > > > > ReducingFunction that extends the AbstractAssociativeMergeOperator is > > > > > provided to the frocksdb side. This may be a simple way of > > > optimization, > > > > > offering no additional new use cases. I may have missed some > > > information, > > > > > so please correct me if I'm wrong. > > > > > > > > > > > > > > > And BTW, CC to @roman <[email protected]> and @Yuan Mei > > > > > <[email protected]> as you may be interested in this topic. > > > > > > > > > > > > > > > Best, > > > > > Zakelly > > > > > > > > > > On Mon, May 25, 2026 at 12:05 PM Soumitra Kumar < > > > > [email protected]> > > > > > wrote: > > > > > > > > > >> Hi Zakelly, > > > > >> > > > > >> I have removed the ReducingMergeState.set method to minimize the > > > changes > > > > >> to > > > > >> public APIs. The modified FLIP is at the same - > > > > >> > > > > >> > > > > > > > > > https://docs.google.com/document/d/1HwEDRGoSZIUU1SYxTih4qp8FM6LjTdIrDs7CJHm4iB0/edit?usp=sharing > > > > >> > > > > >> Community members, > > > > >> This proposal is a great addition to FrocksDB/Flink to enable > > > > applications > > > > >> to harness the power of RocksDB. This addition will help write-heavy > > > > >> workloads and let users build associative data structures in a > > > streaming > > > > >> fashion. > > > > >> > > > > >> Please review, and I can answer any question. > > > > >> > > > > >> Best, > > > > >> -Soumitra. > > > > >> > > > > >> On Tue, May 19, 2026 at 8:56 PM Soumitra Kumar < > > > > [email protected]> > > > > >> wrote: > > > > >> > > > > >> > Hi Zakelly, > > > > >> > > > > > >> > Thanks for your review and +1! > > > > >> > > > > > >> > On Tue, May 19, 2026 at 8:19 AM Zakelly Lan < > > [email protected]> > > > > >> wrote: > > > > >> > > > > > >> >> Thanks for the FLIP! I'd +1 for the direction of enhancing the > > > > reducing > > > > >> >> and > > > > >> >> aggregating state. It's important that we could leverage > > Rocksdb's > > > > >> merge > > > > >> >> operators to eliminate unnecessary `get()`. However I have a few > > > > >> >> questions: > > > > >> >> > > > > >> >> 1. I see you will introduce `AbstractAssociativeMergeOperator` in > > > > >> >> frocksdb side, so how could user pass this instance to the > > RocksDB > > > > >> State > > > > >> >> backend and what is the relationship with the Flink's > > > > `ReduceFunction` > > > > >> or > > > > >> >> `AggregateFunction`. I would suggest we may 'translate' user's > > > > >> >> `ReduceFunction` into some frocksdb's merge operator, thus for > > > flink > > > > >> we > > > > >> >> still maintain the original experience. WDYT? > > > > >> >> > > > > >> > > > > > >> > Yes, that is the whole idea. Since ReduceFunction and > > > > AggregateFunction > > > > >> > are existing primitives and they are associative and > > serializable, I > > > > am > > > > >> > building on them. > > > > >> > > > > > >> > This is how they are wired. There are two new classes > > > > >> > RocksDBReducingMergeOperator, RocksDBAggregatingMergeOperator > > > > >> implementing > > > > >> > AbstractAssociativeMergeOperator in Flink. The ColumnFamily is > > > > >> configured > > > > >> > with one of them. RocksDBReducingMergeOperator for > > > ReducingMergeState, > > > > >> and > > > > >> > RocksDBAggregatingMergeOperator for AggregatingMergeState types. > > > These > > > > >> > classes get the callback from frocksdb, handle the serde, and call > > > the > > > > >> user > > > > >> > defined ReduceFunction and AggregateFunction. > > > > >> > > > > > >> > > > > > >> >> > > > > >> >> 2. I read you introduce a new `ReducingMergeState` with a new > > > `set()` > > > > >> >> method compared with `ReducingState`, is this necessary? I mean > > if > > > we > > > > >> >> intend to optimize performance solely through the use of a merge > > > > >> operator, > > > > >> >> this is not necessary, right? I do not recommend introducing too > > > many > > > > >> >> public APIs, as this would force us to consider their > > > > >> >> semantics. Specifically for example, in changelog stream > > > processing, > > > > if > > > > >> >> the > > > > >> >> `merge` operation were to permit state-setting operations, it > > would > > > > >> >> complicate potential future retraction (or reverse > > > merge/aggregating) > > > > >> >> operations. WDYT? > > > > >> >> > > > > >> > > > > > >> > ReducingMergeState.set is just a shortcut for clear-add. I agree > > > with > > > > >> your > > > > >> > point, I will remove it. > > > > >> > > > > > >> > Best, > > > > >> > -Soumitra. > > > > >> > > > > > >> > > > > > >> >> On Tue, May 5, 2026 at 11:50 AM Soumitra Kumar < > > > > >> [email protected]> > > > > >> >> wrote: > > > > >> >> > > > > >> >> > Dear Community Members, > > > > >> >> > > > > > >> >> > I want to start discussion on the two tickets I filed recently: > > > > >> >> > Add support for Java based AssociativeMergeOperator via JNI > > > > >> >> > <https://issues.apache.org/jira/browse/FLINK-39455> > > > > >> >> > Support ReducingMergeState and AggregatingMergeState backed by > > > Java > > > > >> >> based > > > > >> >> > associative merge operators > > > > >> >> > <https://issues.apache.org/jira/browse/FLINK-39456> > > > > >> >> > > > > > >> >> > Copying the motivation from the FLIP doc > > > > >> >> > < > > > > >> >> > > > > > >> >> > > > > >> > > > > > > > > > https://docs.google.com/document/d/1HwEDRGoSZIUU1SYxTih4qp8FM6LjTdIrDs7CJHm4iB0/edit?usp=sharing > > > > >> >> > > > > > > >> >> > : > > > > >> >> > > > > > >> >> > Flink supports RocksDBReducingState and RocksDBAggregatingState > > > > state > > > > >> >> > variables that do a synchronous read-modify-write on every add > > > > call. > > > > >> >> While > > > > >> >> > this works great in many scenarios, for write-heavy workloads > > > this > > > > >> can > > > > >> >> be > > > > >> >> > expensive and may become a bottleneck. > > > > >> >> > RocksDB's AssociativeMergeOperator is a storage-level primitive > > > > >> designed > > > > >> >> > for commutative and associative operations — integer counters, > > > set > > > > >> >> union, > > > > >> >> > list append, approximate sketches, top-K structures, Bloom > > > filter, > > > > >> and > > > > >> >> > similar patterns. However, frocksdb (the RocksDB fork used in > > > > Flink) > > > > >> >> does > > > > >> >> > not support Java based associative merge operators. > > > > >> >> > > > > > >> >> > This FLIP has two parts: > > > > >> >> > 1. Support for Java based AssociativeMergeOperator in frocksdb > > > via > > > > >> JNI > > > > >> >> > 2. Support ReducingMergeState and AggregatingMergeState backed > > by > > > > >> Java > > > > >> >> > based associative merge operators > > > > >> >> > > > > > >> >> > The first part proposes exposing the associative merge operator > > > as > > > > a > > > > >> >> Java > > > > >> >> > class in frocksdb with minimal JNI overhead. RocksDB can call > > > these > > > > >> >> > operators during flushing and compaction. > > > > >> >> > The second part leverages the frocksdb support developed in the > > > > first > > > > >> >> part > > > > >> >> > to support ReducingMergeState and AggregatingMergeState state > > > > >> variables > > > > >> >> > with user defined ReduceFunction and AggregateFunction using > > > > rocksdb > > > > >> >> > backend. > > > > >> >> > > > > > >> >> > This enhancement opens up a powerful feature of rocksdb to > > Java. > > > > >> Flink > > > > >> >> > users can use it to build interesting associative data > > structures > > > > >> >> > on streaming data. I have added benchmark details from a > > > prototype > > > > >> >> > implementation in the FLIP doc. > > > > >> >> > > > > > >> >> > Looking forward to feedback. > > > > >> >> > > > > > >> >> > FLIP in Google doc > > > > >> >> > < > > > > >> >> > > > > > >> >> > > > > >> > > > > > > > > > https://docs.google.com/document/d/1HwEDRGoSZIUU1SYxTih4qp8FM6LjTdIrDs7CJHm4iB0/edit?usp=sharing > > > > >> >> > > > > > > >> >> > > > > > >> >> > Best, > > > > >> >> > -Soumitra. > > > > >> >> > > > > > >> >> > > > > >> > > > > > >> > > > > > > > > > > > > > >
