tillrohrmann commented on a change in pull request #9501: [FLINK-12697] [State
Backends] Support on-disk state storage for spill-able heap backend
URL: https://github.com/apache/flink/pull/9501#discussion_r334021225
##########
File path:
flink-state-backends/flink-statebackend-heap-spillable/src/main/java/org/apache/flink/runtime/state/heap/CopyOnWriteSkipListStateMap.java
##########
@@ -310,57 +299,21 @@ public S removeAndGetOld(K key, N namespace) {
* @param keyByteBuffer byte buffer storing the key.
* @param keyOffset offset of the key.
* @param keyLen length of the key.
- * @return id of the node. NIL_NODE will be returned if key does no
exist.
+ * @return the state. Null will be returned if key does not exist.
*/
@VisibleForTesting
- long getNode(ByteBuffer keyByteBuffer, int keyOffset, int keyLen) {
- int deleteCount = 0;
- long prevNode = findPredecessor(keyByteBuffer, keyOffset, 1);
- long currentNode = helpGetNextNode(prevNode, 0);
- long nextNode;
-
- int c;
- while (currentNode != NIL_NODE) {
- nextNode = helpGetNextNode(currentNode, 0);
-
- boolean isRemoved = isNodeRemoved(currentNode);
- if (isRemoved) {
- // remove the node physically when there is no
snapshot running
- if (highestRequiredSnapshotVersion == 0 &&
deleteCount < numKeysToDeleteOneTime) {
- doPhysicalRemove(currentNode, prevNode,
nextNode);
-
logicallyRemovedNodes.remove(currentNode);
- totalSize--;
- deleteCount++;
- } else {
- prevNode = currentNode;
- }
- currentNode = nextNode;
- continue;
- }
-
- c = compareByteBufferAndNode(keyByteBuffer, keyOffset,
keyLen, currentNode);
-
- // find the key
- if (c == 0) {
- return currentNode;
- }
-
- // the key is less than the current node, and nodes
after current
- // node can not be equal to the key.
- if (c < 0) {
- break;
- }
-
- prevNode = currentNode;
- currentNode = helpGetNextNode(currentNode, 0);
- }
-
- return NIL_NODE;
+ S getNode(ByteBuffer keyByteBuffer, int keyOffset, int keyLen) {
+ Tuple4<Long, Long, Boolean, S> result =
iterateAndProcess(keyByteBuffer, keyOffset, keyLen,
+ (tuple3, isRemoved) -> {
+ long currentNode = tuple3.f1;
+ return isRemoved ? null :
getNodeStateHelper(currentNode);
+ });
+ return result.f2 ? result.f3 : null;
Review comment:
I have no idea what this expresses. I think there is a way to make it more
expressive.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
