Hi Roman, I appreciate you taking the time to review the prototype, and thanks for the feedback. I will work on concrete suggestions and generate data in the next few days.
I want to highlight that this FLIP exposes the merge operator already used by ListState variables to end users, rather than introducing a new way to parallelize associative operations in Flink. ListState uses the "stringappendtest" merge operator defined in the C++ layer of FRocksDB, and this is why ListState is performant — RocksDB performs aggregation async. This FLIP has two parts: 1. Modify FRocksDB to enable support for associative merge operators in Java 2. Expose custom associative merge operators using the existing ReducingState and AggregatingState primitives Best, -Soumitra. On Mon, Jul 6, 2026 at 4:44 PM Roman Khachatryan <[email protected]> wrote: > Hi Soumitra, > > Thanks for your reply. I took a closer look at the prototype repo [1], and > I'd like to make one general point first, because I think it's the crux of > the discussion. > > 1. Associativity is better exploited in the dataflow than in the storage > layer > > The FLIP's premise is: the operation is associative, so aggregation can be > deferred and done asynchronously by RocksDB. But if an operation is > associative, Flink can already exploit that property at the dataflow level > — with partial (local/global, two-phase) aggregation. That distributes both > the writes and the aggregation CPU across the cluster; a storage-level > merge operator can only shift aggregation CPU to compaction threads on the > same TaskManager, and does nothing for the actual scaling problem of > write-heavy aggregation, which is a hot key on a single subtask. > > Your own price-stats example shows this: PriceStatsWindowFunction is keyed > by a constant ("global"), so it runs on one subtask regardless of > parallelism — the bottleneck is the topology, not state access. > Restructured as two-phase aggregation (per-subtask partial > min/max/sum/count, then combining a few dozen partial records per window — > what Flink SQL already does automatically for such queries), each > accumulator is a few dozen bytes of hot state, and read-modify-write on it > costs nothing. Meanwhile the merge-state version accumulates an unbounded > operand chain that get() must fold via JNI callbacks at window end. > > So for genuinely associative aggregations, the framework-level solution > dominates: it scales horizontally, uses stable APIs, keeps state small — > and has none of the costs discussed below. I think the FLIP needs to show a > workload where merge operators beat that, not where they beat > read-modify-write ValueState. > > 2. The prototype itself falls back to task-thread aggregation > > In ClickStreamReorderUsingMergeState.onTimer(), every timer fire calls > bufferState.get() (folding all operands), drains ready events, and writes > the entire remaining buffer back via setAcc() — commented "Setting it > irrespective of change to save the cost of merge in next get call". I.e., > compaction doesn't keep up at 30s granularity, and the prototype > compensates by squashing the operand chain on the task thread on every > timer. That is exactly the ListState + periodic-aggregation alternative I > proposed, with extra JNI machinery underneath. The async-compaction benefit > does not materialize in the FLIP's own motivating use case. > > Also note the prototype depends on setAcc() (in processElement and onTimer) > — the set-style API removed from the FLIP after Zakelly's feedback. The > benchmarked design is not the proposed design. > > 3. The benchmark doesn't measure the proposal's delta > > The ValueState baseline serializes the buffer via Kryo > (TypeHint<ArrayList<...>>) on every event, while the merge variant got a > hand-written TypeSerializer — so the 2x gap partly measures the serializer. > More importantly, the repo has no ListState or MapState variant, so the > alternatives under discussion were never measured. ListState.add() is the > same blind rocksdb.merge() call as the proposed state (you confirmed this), > so it removes the same write bottleneck. And for reordering specifically, > the buffer must be drained up to the watermark and the rest kept — with a > merged sorted list that's a full read + write-back per drain, while > MapState keyed by timestamp deletes only the released entries (the > CepOperator pattern). > > 4. TTL is unresolved > > In the prototype, TTL on the merge state is commented out ("// SK: Fix it") > while the auxiliary ValueStates (needed for size/max-timestamp bookkeeping > and read/written per event — so the write path isn't read-free either) do > have TTL, allowing buffer and metadata to expire independently. TTL > semantics over unmerged operand chains isn't a PR-stage detail. > > 5. The design-level blockers remain > > Serializing user functions into savepoints bypasses serializer migration > and re-introduces Java-serialized user code with its upgrade fragility; > requiring job-specific Java code during compaction rules out > remote/external compaction for those column families. "Users handle schema > changes in their functions" and "ForSt just won't use it" don't resolve > these concerns — they confirm them. A public state type that one backend > cannot implement is a one-way door. > > Concrete suggestion: add ListState and MapState variants to the same > harness (same serializer, same drain logic) and re-run — and, per point 1, > compare the price-stats pipeline against a standard two-phase aggregation. > Those are falsifiable experiments; if merge state shows a meaningful > advantage there, that's real evidence for the FLIP. My expectation, given > the identical write paths, is that it won't — but I'm happy to be proven > wrong by data. > > [1] https://github.com/soumitrak/flink_streaming_event_reordering > > Regards, > Roman > > > On Sun, Jul 5, 2026 at 11:59 PM Soumitra Kumar <[email protected]> > wrote: > > > Hi Roman, > > > > Please let me know your thoughts. > > > > Thanks for your review! > > > > Best, > > -Soumitra. > > > > On Thu, Jun 25, 2026, 9:52 PM Soumitra Kumar <[email protected]> > > wrote: > > > > > Hi Roman, > > > > > > Sorry for the late reply, I was travelling. Please find my comments > > inline. > > > > > > On Fri, Jun 19, 2026 at 8:36 AM Roman Khachatryan <[email protected]> > > > wrote: > > > > > >> I might be missing something, but I don't see why the existing > > mechanisms > > >> can't solve the problem; especially simple MapState (2) or ListState > > (6). > > >> At the same time, I have concerns about the proposal itself (1), (4), > > (5): > > >> > > > > > > Existing solutions can solve the problem, but not with the same > run-time > > > characteristics. Two things: > > > 1. I have not tried MapState and ListState, but (most likely) they will > > > increase the size of the state. > > > 2. I can construct an example where the read time would be more than > the > > > proposed solution, since the aggregation will happen on read vs async. > > > > > > > > >> > > 1. Is it possible that reads will become more costly in some > > >> scenarios? > > >> > 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. > > >> > > >> My point is that it not only doesn't improve; it degrades the > > performance > > >> in these read-after-write cases. > > >> So the proposal adds one more (expert-level) way to tune the runtime. > > >> Which I believe we should avoid. > > >> > > > > > > The proposal adds support for associative operators, and the example > (the > > > event reordering) is just one such operation. Currently, AFAIK, there > are > > > two solutions in Flink: > > > 1. Use ValueState and implement read-modify-write > > > 2. Use ListState and perform the aggregation during read > > > > > > Users can pick one of them based on their needs. The proposal exposes > the > > > already supported associative merge operators in RocksDB to Java in > > > FrocksDB layer, and leverages that to add new types of state variables > in > > > Flink. I agree that this pattern is not helpful for read-after-write > > > scenarios. In fact, read-modify-write is best for this situation. The > > > proposed solution will help in write-heavy applications. There are > > several > > > pointers on the web where merge operators in rocksdb can be helpful. > > > Quoting https://artem.krylysov.com/blog/2023/04/19/how-rocksdb-works/ > > > "Merge is a good fit for write-heavy streaming applications ..." > > > > > > The proposal adds one more way to implement associative operation, and > I > > > agree that to get the best performance this requires rocksdb tuning. > > IMHO, > > > the async merging done by RocksDB distributes and parallelizes nicely, > > and > > > fits within the processing style supported by Flink. So, we should add > > > support for that. > > > > > > > > >> > > > 2. Speaking more generally, could you list the motivating use > > cases > > >> > > 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) > > >> > don't know how MapState can help with generic ordering, but my > > knowledge > > >> on that is limited. > > >> > > >> A common pattern is to use MapState<Long, ...>, where keys are > > timestamps. > > >> Such a MapState, when backed by RocksDB, is automatically sorted by > > >> timestamps. > > >> > > > > > > As I said, I have not used MapState. But, the proposal is adding > support > > > for generic associative merge operations by exposing existing > > functionality > > > in RocksDB. > > > > > > > > >> > > 4. Function dispatch 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. > > >> > > >> That means that in case of schema change, compactions will be using > old > > >> schema, right? That way, I'm afraid it can bypass state migration. > > >> > > > > > > User needs to handle the schema changes in the Reduce/Aggregate > > functions. > > > As long as it is done properly, I don't see any additional issue from > > this > > > proposal. > > > > > > > > >> > > 5. Remote compactions > > >> > 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. > > >> > > >> My concern is about an external compaction component: > > >> this proposal forces it to have job-specific java code > > >> instead of having only C++ code only (or whatever is used in state > > >> backend). > > >> > > > > > > I have not used ForSt backed, and don't know the details. I will invite > > > comments from experts, but my 2 cents is that we don't need to use the > > > merge operators in the ForSt backend, so there won't be any side > effects > > of > > > this enhancement to the ForSt backend. > > > > > > > > >> > > 6. Alternative: ListState > > >> > 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. > > >> > > >> In the current proposal, read will trigger sorting as well - if the > data > > >> is > > >> not > > >> compacted/sorted yet. And it will add more latency than with ListState > > >> because of the extra read/write pass. > > >> > > > > > > Yes, if the flush/compaction has not happened before read, then read > will > > > invoke associative operator callback. > > > > > > > > >> ListState solution provides flexibility to choose when this work > > happens: > > >> 1. Periodically (using processing time timers) - similar to > compactions > > >> 2. On reads > > >> 3. On writes incrementally > > >> 4. Some combination > > >> > > > > > > 1. The compaction happens in different threads than writes in RocksDB. > > > However, in the case of KeyedProcessFunction, the processElement() and > > > onTimer() are single-threaded in Flink. Both methods are executed > > > sequentially by the exact same task thread. The proposal will perform > > > better because of the async aggregation. > > > 2. If the flush/compaction has happened before read, then read will be > > way > > > faster than ListState, else not. > > > 3. Write will be similar, since both use rocksdb.merge. > > > > > > Thanks for your comments! Best, > > > -Soumitra. > > > > > > > > >> > > >> On Mon, Jun 15, 2026 at 9:13 PM Soumitra Kumar < > > [email protected]> > > >> wrote: > > >> > > >> > ---------- Forwarded message --------- > > >> > From: Soumitra Kumar <[email protected]> > > >> > Date: Mon, Jun 15, 2026, 12:12 PM > > >> > Subject: Re: [DISCUSS] FLIP-XXX Support ReducingMergeState and > > >> > AggregatingMergeState backed by Java based associative merge > operators > > >> > To: <[email protected]> > > >> > > > >> > > > >> > Hi Roman, > > >> > > > >> > I replied to your questions a while back. Let me forward the thread > to > > >> > [email protected] . > > >> > > > >> > Best, > > >> > -Soumitra. > > >> > > > >> > On Mon, Jun 15, 2026, 12:48 AM Roman Khachatryan <[email protected]> > > >> wrote: > > >> > > > >> > > Hello Soumitra Kumar, > > >> > > > > >> > > It would be great to get the answers to the questions above I > > posted - > > >> > > unless the problem is solved and the FLIP isn't necessary. > > >> > > > > >> > > Regards, > > >> > > Roman > > >> > > > > >> > > > > >> > > On Sun, Jun 14, 2026 at 9:41 PM Soumitra Kumar < > > >> [email protected] > > >> > > > > >> > > wrote: > > >> > > > > >> > > > Hello Members, > > >> > > > > > >> > > > Thank you for your review so far. I don't have any open issues > at > > >> this > > >> > > > moment. Please let me know if there is any issue for me to > > >> > > clarify/address. > > >> > > > > > >> > > > Best, > > >> > > > -Soumitra. > > >> > > > > > >> > > > On Mon, Jun 8, 2026 at 10:09 PM Soumitra Kumar < > > >> > [email protected] > > >> > > > > > >> > > > wrote: > > >> > > > > > >> > > > > Hi Han, > > >> > > > > > > >> > > > > I have added a section on TTL of ReducingMergeState and > > >> > > > > AggregatingMergeState - HERE > > >> > > > > < > > >> > > > > > >> > > > > >> > > > >> > > > https://docs.google.com/document/d/1HwEDRGoSZIUU1SYxTih4qp8FM6LjTdIrDs7CJHm4iB0/edit?tab=t.0#heading=h.mqp1qeixcg45 > > >> > > > > > >> > > > , > > >> > > > > please review. > > >> > > > > > > >> > > > > Best, > > >> > > > > -Soumitra. > > >> > > > > > > >> > > > > On Mon, Jun 1, 2026 at 11:02 PM Soumitra Kumar < > > >> > > [email protected] > > >> > > > > > > >> > > > > wrote: > > >> > > > > > > >> > > > >> Hi Han, > > >> > > > >> > > >> > > > >> Thanks for your review and encouragement! > > >> > > > >> > > >> > > > >> #1 - Users can migrate from ReducingState to > > ReducingMergeState, > > >> but > > >> > > it > > >> > > > >> has to be a conscious decision knowing the rocksdb > implication. > > >> We > > >> > > > should > > >> > > > >> plan to create a few howto docs monitoring and tuning rocksdb > > to > > >> get > > >> > > the > > >> > > > >> best out of the merge operators. Theoretically, it is > possible > > to > > >> > > build > > >> > > > an > > >> > > > >> automatic migration path, but I would not favor that because > of > > >> the > > >> > > > >> different runtime characteristics of ReducingState and > > >> > > > ReducingMergeState. > > >> > > > >> The checkpoints/savepoints for > > >> > > ReducingMergeState/AggregatingMergeState > > >> > > > >> state variables will serialize the Reduce/Aggregate function > as > > >> > well. > > >> > > > >> > > >> > > > >> #2 - "Will this introduce different semantics when State TTL > is > > >> > > enabled" > > >> > > > >> - Can you elaborate on this? TBH, I have not planned the > > details > > >> of > > >> > > the > > >> > > > TTL > > >> > > > >> of ReducingMergeState/AggregatingMergeState variables yet, > but > > >> the > > >> > TTL > > >> > > > >> should be applied on the variable, not on individual > operands. > > I > > >> > will > > >> > > > add a > > >> > > > >> section on TTL of these variables in the FLIP. > > >> > > > >> > > >> > > > >> Best, > > >> > > > >> -Soumitra. > > >> > > > >> > > >> > > > >> On Mon, Jun 1, 2026 at 3:03 AM Han Yin <[email protected] > > > > >> > wrote: > > >> > > > >> > > >> > > > >>> Hi Sumatra, > > >> > > > >>> Thanks for the FLIP. The ability to leverage RocksDB merge > > >> > operators > > >> > > in > > >> > > > >>> Reducing/Aggregating state is a really meaningful > improvement. > > >> > > > >>> > > >> > > > >>> I share similar concerns about the user interface with the > > >> previous > > >> > > > >>> comments: > > >> > > > >>> • If new state types are introduced, can users migrate > > their > > >> > > > >>> existing jobs from ReducingState to ReducingMergeState? > Since > > >> the > > >> > > core > > >> > > > >>> logic of the ReduceFunction remains the same, one would > > expect a > > >> > > > >>> straightforward migration path. If yes, will > > >> checkpoints/savepoints > > >> > > > remain > > >> > > > >>> compatible across this switch (and back)? > > >> > > > >>> • Will this introduce different semantics when State TTL > > is > > >> > > > enabled? > > >> > > > >>> > > >> > > > >>> > > >> > > > > > >> > > > > >> > > > >> > > > > > >
