[FLINK-24597] RocksdbStateBackend getKeysAndNamespaces would return duplicate data when using MapState

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.14.0, 1.12.4, 1.13.3
• Fix Version/s: 1.13.6, 1.14.3, 1.15.0
• Component/s: API / State Processor, Runtime / State Backends
• Labels: pull-request-available
• Assignee: Yue Ma
• Reporter: Yue Ma
• Created: 20/Oct/21 06:43
• Updated: 15/Dec/21 01:44
• Resolved: 05/Nov/21 03:03

Description

For example, in RocksdbStateBackend , if we worked in VoidNamespace , and And use the ValueState like below .

            // insert record
            for (int i = 0; i < 3; ++i) {
                keyedStateBackend.setCurrentKey(i);
                testValueState.update(String.valueOf(i));
            }

Then we get all the keysAndNamespace according the method RocksDBKeyedStateBackend#getKeysAndNamespaces().The result of the traversal is
<1,VoidNamespace>,<2,VoidNamespace>,<3,VoidNamespace> ,which is as expected.

Thus，if we use MapState , and update the MapState with different user key, the getKeysAndNamespaces would return duplicate data with same keyAndNamespace.

            // insert record
            for (int i = 0; i < 3; ++i) {
                keyedStateBackend.setCurrentKey(i);
                mapState.put("userKeyA_" + i, "userValue");
                mapState.put("userKeyB_" + i, "userValue");
            }

The result of the traversal is
<1,VoidNamespace>,<1,VoidNamespace>,<2,VoidNamespace>,<2,VoidNamespace>,<3,VoidNamespace>,<3,VoidNamespace>.

By reading the code, I found that the main reason for this problem is in the implementation of RocksStateKeysAndNamespaceIterator.
In the hasNext method, when a new keyAndNamespace is created, there is no comparison with the previousKeyAndNamespace. So we can refer to RocksStateKeysIterator to implement the same logic should solve this problem.
 
 

Issue Links

• links to: GitHub Pull Request #17525