[FLINK-33672] Use MapState.entries() instead of keys() and get() in over window

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.19.0
• Component/s: Table SQL / Runtime
• Labels: pull-request-available
• Assignee: Zakelly Lan
• Reporter: Zakelly Lan
• Created: 28/Nov/23 08:34
• Updated: 12/Dec/23 03:30
• Resolved: 12/Dec/23 02:15

Description

In code logic related with over windows, such as org.apache.flink.table.runtime.operators.over.ProcTimeRangeBoundedPrecedingFunction

private transient MapState<Long, List<RowData>> inputState;

public void onTimer(
        long timestamp,
        KeyedProcessFunction<K, RowData, RowData>.OnTimerContext ctx,
        Collector<RowData> out)
        throws Exception {
    //...
    Iterator<Long> iter = inputState.keys().iterator();
    //...
    while (iter.hasNext()) {
        Long elementKey = iter.next();
        if (elementKey < limit) {
            // element key outside of window. Retract values
            List<RowData> elementsRemove = inputState.get(elementKey);
            // ...
        }
    }
    //...
} 

As we can see, there is a combination of key iteration and get the value for iterated key from inputState. However for RocksDB, the key iteration calls entry iteration, which means actually we could replace it by entry iteration without introducing any extra overhead. And as a result, we could save a function call of get() by using getValue() of iterated entry at very low cost.

Issue Links

• links to: GitHub Pull Request #23855