[KV] Add retract support for Aggregation Merge Engine
- Dominant language
- Java
- Stars
- 2.1k
- Forks
- 625
- Avg merge
- 3d 14h
- Merged PRs (30d)
- 97
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.
### Motivation
The Aggregation Merge Engine currently supports upsert and deleteoperations. However, there is no way to **reverse a previously applied aggregation contribution** without deleting the entire row.
Consider an e-commerce scenario: a table aggregates total revenue per shop using `SUM`. When a refund occurs, the system needs to subtract the refunded amount from the running total. Today, this requires the Flink compute layer to maintain full aggregation state — defeating the purpose of pushing aggregation down to the storage engine.
### Solution
Introduce a **retract** operation that computes the inverse of a previous aggregation at the storage layer:
- `SUM`: subtracts the retracted value
- `PRODUCT`: divides by the retracted value
- `LAST_VALUE` / `LAST_VALUE_IGNORE_NULLS`: clears to null Functions that cannot be mathematically reversed (`MAX`, `MIN`, `FIRST_VALUE`, `LISTAGG`, etc.) do not support retract.
### Anything else?
_No response_
### Willingness to contribute
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Start by locating the Aggregation Merge Engine and reviewing its existing upsert and delete paths. Trace how SUM, PRODUCT, LAST_VALUE, and LAST_VALUE_IGNORE_NULLS are implemented, along with the unsupported MAX, MIN, FIRST_VALUE, and LISTAGG functions. Done means retract behavior is covered for supported functions and unsupported functions remain excluded.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 52/100