[FLINK-13034] Improve the performance when checking whether mapstate is empty for RocksDBStateBackend

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.6.3, 1.7.2, 1.8.1
• Fix Version/s: 1.10.0
• Component/s: Runtime / State Backends
• Labels: pull-request-available
• Release Note: We have added a new method MapState#isEmpty() which enables users to check whether a map state is empty. The new method is 40% faster than mapState.keys().iterator().hasNext() when using the RocksDB state backend.
• Assignee: Yun Tang
• Reporter: Yun Tang
• Created: 30/Jun/19 17:50
• Updated: 03/Feb/20 12:17
• Resolved: 06/Nov/19 20:43

Description

Currently, there existed several scenarios to check whether map state is empty in Flink source code, e.g.TemporalRowTimeJoinOperator, AbstractRowTimeUnboundedPrecedingOver.
Developers would use below command to check whether the map state is empty:

boolean noRecordsToProcess = !inputState.keys().iterator().hasNext();

However, if we use RocksDBStateBackend, inputState.keys().iterator().hasNext() would actually call 1 seek and 128 next actions in RocksDBMapState, in which the redundant next actions are not what we want.

I have two options to improve this:

• Modify RocksDBMapState back to previous design which would first load one element and then load more elements in the follow-up queries. However, this would effect the performance of other map state methods.
• Add a isEmpty() method in the public evolving interface MapState, so that we could use it to check whether the map state is empty without any redundant RocksDB actions.

I prefer to the 2nd option.

 

Issue Links

• is caused by: FLINK-8639 Fix always need to seek multiple times when iterator RocksDBMapState
• is related to: FLINK-13846 Implement benchmark case on MapState#isEmpty
• links to: GitHub Pull Request #9255