Repository: NetEase/arctic (java)

Issue:
[Bug]: Find duplicate records when enable rocksdb map
### What happened?

When user write to Iceberg table in flink upsert mode, and spillable map was enabled in Optimizer, it is possible that the equailty-delete ids do not take effect after self-optimization, meaning that duplicates may be generated

### Affects Versions

master/0.4.x

### What engines are you seeing the problem on?

Optimizer

### How to reproduce

1. enable spillable map
2. start self-optimizing if reach the threshold for disk spillable 

### Relevant log output

_No response_

### Anything else

_No response_

### Code of Conduct

- [X] I agree to follow this project's Code of Conduct

Pull request description:
[ARCTIC-1751] Update the memory map in spillable map when new value was existed for the key
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://arctic.netease.com/ch/contribute/
  2. If the PR is related to an issue in https://github.com/NetEase/arctic/issues, add '[ARCTIC-XXXX]' in your PR title, e.g., '[ARCTIC-XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][ARCTIC #XXXX] Your PR title ...'.
-->

## Why are the changes needed?
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about its use case.
  2. If you fix a bug, you can clarify why it is a bug.
-->

fix #1751 and #1194 

## Brief change log

get corrent eq-delete file and the sequence number to make sure duplicate records are filtered

## How was this patch tested?
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] Run test locally before making a pull request

## Documentation

  - Does this pull request introduce a new feature? (no)
  - If yes, how is the feature documented? (not documented)


Patch:
diff --git a/core/pom.xml b/core/pom.xml
index 42c07f9c0..80ba37dbb 100644
--- a/core/pom.xml
+++ b/core/pom.xml
@@ -33,7 +33,7 @@
     <url>https://arctic.netease.com</url>
 
     <properties>
-        <rocksdb.version>5.17.2</rocksdb.version>
+        <rocksdb.version>7.10.2</rocksdb.version>
         <kryo.version>2.24.0</kryo.version>
         <jol.version>0.16</jol.version>
     </properties>
diff --git a/core/src/main/java/com/netease/arctic/utils/map/SimpleSpillableMap.java b/core/src/main/java/com/netease/arctic/utils/map/SimpleSpillableMap.java
index 7a29c6e3c..7534e6257 100644
--- a/core/src/main/java/com/netease/arctic/utils/map/SimpleSpillableMap.java
+++ b/core/src/main/java/com/netease/arctic/utils/map/SimpleSpillableMap.java
@@ -111,6 +111,11 @@ public class SimpleSpillableMap<K, T> implements SimpleMap<K, T> {
       this.currentInMemoryMapSize = this.memoryMap.size() * this.estimatedPayloadSize;
     }
 
+    if (memoryMap.containsKey(key)) {
+      memoryMap.put(key, value);
+      return;
+    }
+
     if (this.currentInMemoryMapSize < maxInMemorySizeInBytes) {
       if (memoryMap.put(key, value) == null) {
         currentInMemoryMapSize += this.estimatedPayloadSize;
@@ -127,9 +132,8 @@ public class SimpleSpillableMap<K, T> implements SimpleMap<K, T> {
     if (memoryMap.containsKey(key)) {
       currentInMemoryMapSize -= estimatedPayloadSize;
       memoryMap.remove(key);
-    } else {
-      diskBasedMap.ifPresent(map -> map.delete(key));
     }
+    diskBasedMap.ifPresent(map -> map.delete(key));
   }
 
   public void close() {


Test patch:
diff --git a/core/src/test/java/com/netease/arctic/utils/map/TestSimpleSpillableMap.java b/core/src/test/java/com/netease/arctic/utils/map/TestSimpleSpillableMap.java
index 57d15d7f6..6cd4a5066 100644
--- a/core/src/test/java/com/netease/arctic/utils/map/TestSimpleSpillableMap.java
+++ b/core/src/test/java/com/netease/arctic/utils/map/TestSimpleSpillableMap.java
@@ -1,6 +1,7 @@
 package com.netease.arctic.utils.map;
 
-import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -46,6 +47,71 @@ public class TestSimpleSpillableMap {
     map.close();
   }
 
+  @Test
+  public void testSpillableMapConsistency() {
+    SimpleSpillableMap<Key, Value> actualMap = new SimpleSpillableMap<>(
+        5 * (keySize + valueSize),
+        null,
+        new DefaultSizeEstimator<>(),
+        new DefaultSizeEstimator<>());
+
+    Map<Key, Value> expectedMap = Maps.newHashMap();
+    for (int i = 0; i < 10; i++) {
+      Key key = new Key();
+      Value value = new Value();
+      expectedMap.put(key, value);
+      actualMap.put(key, value);
+    }
+    Assert.assertTrue(actualMap.getSizeOfFileOnDiskInBytes() > 0);
+    Assert.assertTrue(actualMap.getMemoryMapSize() < expectedMap.size());
+    assertSimpleMaps(actualMap, expectedMap);
+
+    // update new value
+    Sets.newHashSet(expectedMap.keySet())
+        .forEach(k -> {
+          Value newValue = new Value();
+          actualMap.put(k, newValue);
+          expectedMap.put(k, newValue);
+          assertSimpleMaps(actualMap, expectedMap);
+        });
+
+    Sets.newHashSet(expectedMap.keySet())
+        .forEach(k -> {
+          actualMap.delete(k);
+          expectedMap.remove(k);
+          assertSimpleMaps(actualMap, expectedMap);
+        });
+  }
+
+  @Test
+  public void testSpillableMapRePut() {
+    SimpleSpillableMap<Key, Value> actualMap = new SimpleSpillableMap<>(
+        (keySize + valueSize),
+        null,
+        new DefaultSizeEstimator<>(),
+        new DefaultSizeEstimator<>());
+
+    Key k1 = new Key();
+    Value v1 = new Value();
+    Key k2 = new Key();
+    Value v2 = new Value();
+
+    actualMap.put(k1, v1);
+    actualMap.put(k2, v2);
+
+    // remove k1 from memory
+    actualMap.delete(k1);
+    Assert.assertEquals(v2, actualMap.get(k2));
+    // put a new value for k2
+    Value v3 = new Value();
+    actualMap.put(k2, v3);
+    Assert.assertEquals(v3, actualMap.get(k2));
+
+    actualMap.delete(k2);
+    // should not exist in memory or on disk
+    Assert.assertNull(actualMap.get(k2));
+  }
+
   private SimpleSpillableMap<Key, Value> testMap(long expectMemorySize, int expectKeyCount) {
     SimpleSpillableMap<Key, Value> actualMap = new SimpleSpillableMap<>(expectMemorySize * (keySize + valueSize),
         null, new DefaultSizeEstimator<>(), new DefaultSizeEstimator<>());
@@ -57,9 +123,7 @@ public class TestSimpleSpillableMap {
       expectedMap.put(key, value);
       actualMap.put(key, value);
     }
-    for (Key key : expectedMap.keySet()) {
-      Assert.assertEquals(expectedMap.get(key), actualMap.get(key));
-    }
+    assertSimpleMaps(actualMap, expectedMap);
     Assert.assertEquals(expectMemorySize, actualMap.getMemoryMapSize());
     Assert.assertEquals(
         expectMemorySize * (keySize + valueSize),
@@ -67,11 +131,15 @@ public class TestSimpleSpillableMap {
     return actualMap;
   }
 
+  private <K, V> void assertSimpleMaps(SimpleMap<K, V> actualMap, Map<K, V> expectedMap) {
+    for (K key : expectedMap.keySet()) {
+      Assert.assertEquals(expectedMap.get(key), actualMap.get(key));
+    }
+  }
+
   private static class Key implements Serializable {
     String id = UUID.randomUUID().toString();
 
-    Long num = random.nextLong();
-
     @Override
     public int hashCode() {
       return id.hashCode();