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?
> >> > > > >>>
> >> > > > >>>
> >> > > >
> >> > >
> >> >
> >>
> >
>
