[Umbrella] Support Aggregation Merge Engine

[platinumhamburg]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

When using Aggregation Merge Engine with multiple concurrent Flink jobs writing to the same primary key table, the following issues occur:

1. Concurrent Conflicts: Multiple writers updating the same aggregation columns simultaneously cause data inconsistency
2. No Exactly-Once Guarantee: Job failover cannot guarantee aggregation accuracy
3. Duplicate Calculations: Data replay after job restart leads to duplicate aggregation
4. State Loss: Lack of state management to track committed data

These issues make Aggregation Merge Engine unsuitable for production scenarios requiring strict data accuracy.

Solution

The complete solution has been proposed in the FIP document FIP-21: Aggregation Merge Engine.

1. State Management

• WriterState: Tracks maximum committed offset per TableBucket
• BucketOffsetTracker: Client-side offset tracking
• Integration: Uses Flink's Operator State for persistence across checkpoints

2. Undo Recovery

• Mechanism: On failover, undoes uncommitted data written after last checkpoint
• Implementation: Reads old values at committed offset and overwrites new values
• Guarantee: Ensures data consistency after job restart

Anything else?

No response

Willingness to contribute

• I'm willing to submit a PR!

[Jackeyzhe]

@platinumhamburg Hi, I've read the FIP document, it looks like a good design. And I have a little question: Does the UpdateWriter need to append changelog into LogTablet during the undo flow? Maybe there're some clients which consume the LogTablet also need to rollback data.

[platinumhamburg]

Hi, I've read the FIP document, it looks like a good design. And I have a little question: Does the UpdateWriter need to append changelog into LogTablet during the undo flow? Maybe there're some clients which consume the LogTablet also need to rollback data.

Yes. The undo operation is performed as a normal update to the KvTablet. So it not only updates the KV state, but also generates the corresponding undo changelog and appends it to the LogTablet. Therefore, all downstream clients that consume the log stream can consume the rollback logs normally.

[zuston]

Nice feature. Looking forward to see this in the 0.9 version.