THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
This proposal is an alternative solution to KIP-450, which instead of proposing a new sliding window construct, we would just try to optimize the tumbling window where window length is >> advance step.
Motivation
With a tumbling window length of N, and an advance step of M << N, each window aggregation update involves the following: for each window of the M / N windows this record falls into, issue a update (get, aggregate, and then put).
The total cost of it:
Read: M / N Write: M / N
Instead, we can aggregate per overlapping advance step (let's call it sub-window), and then return the aggregated value across all the overlapping period that this window covers. More specifically, we would only do an update on one sub-window, and then we would return the value by further aggregating the values of all sub-windows. This is a very common window aggregation techniques (see references below). The update is a single read plus a single write, but the further aggregation involves M / N - 1 reads
The total cost of it:
Read: M / N - 1 + 1 = M / N Write: 1
In addition, if the distribution of records falling into the sub-windows is sparse (i.e. a given window would only have records in a very of its sub-windows), then the underlying store's get calls could be more efficient to return empty results (e.g. RocksDB's bloom-filter).
We assume the major cost is on state store IO, and the further aggregation CPU is neglectable.
Proposal
We can refactor the implementation of WindowAggregateProcessor / SessionAggregateProcessor such that aggregate values are organized per sub-window, and then we apply aggregations on-the-fly for each process function call across all the sub-windows of a given window. But the aggregation of sub-windows cannot rely on the existing user-provided "Aggregator" API since the aggregate result type and the aggregating value type may not be the same.
In addition, we can further optimize our implementation by buffering the partial sub-window aggregations to reduce repeating fetches on the latest sub-window: this is similar to the window-slicing / pre-aggregation techniques.
API Changes
As pointed out above, we need the further "combine" function in addition to the existing aggregate result
public interface CombineAggregator<K, V, VA> extends Aggregator<K, V, VA> {
VA combine(final VA agg1, final VA agg2);
}
References:
https://openproceedings.org/2019/conf/edbt/EDBT19_paper_171.pdf