[KV] Add retract support for Aggregation Merge Engine

[platinumhamburg]

Search before asking

• I searched in the 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

• I'm willing to submit a PR!