[SPARK-31014][CORE][3.0] InMemoryStore: remove key from parentToChild…

…renMap when removing key from CountingRemoveIfForEach

### What changes were proposed in this pull request?

This patch addresses missed spot on SPARK-30964 (#27716) - SPARK-30964 added secondary index which defines the relationship between parent - children and able to operate all children for given parent faster.

While SPARK-30964 handled the addition and deletion of secondary index in InstanceList properly, it missed to add code to handle deletion of secondary index in CountingRemoveIfForEach, resulting to the leak of indices. This patch adds the deletion of secondary index in CountingRemoveIfForEach.

### Why are the changes needed?

Described above.

### Does this PR introduce any user-facing change?

No.

### How was this patch tested?

N/A, as relevant field and class are marked as private, and it cannot be checked in higher level. I'm not sure we want to adjust scope to add a test.

Closes #27825 from HeartSaVioR/SPARK-31014-branch-3.0.

Authored-by: Jungtaek Lim (HeartSaVioR) <kabhwan.opensource@gmail.com>
Signed-off-by: Dongjoon Hyun <dongjoon@apache.org>

common/kvstore/src/main/java/org/apache/spark/util/kvstore/InMemoryStore.java

@@ -177,7 +177,7 @@ private static class InstanceList<T> {
      * iterators.  https://bugs.openjdk.java.net/browse/JDK-8078645
      */
     private static class CountingRemoveIfForEach<T> implements BiConsumer<Comparable<Object>, T> {
-      private final ConcurrentMap<Comparable<Object>, T> data;
+      private final InstanceList<T> instanceList;
       private final Predicate<? super T> filter;
 
       /**
@@ -189,17 +189,15 @@ private static class CountingRemoveIfForEach<T> implements BiConsumer<Comparable
        */
       private int count = 0;
 
-      CountingRemoveIfForEach(
-          ConcurrentMap<Comparable<Object>, T> data,
-          Predicate<? super T> filter) {
-        this.data = data;
+      CountingRemoveIfForEach(InstanceList<T> instanceList, Predicate<? super T> filter) {
+        this.instanceList = instanceList;
         this.filter = filter;
       }
 
       @Override
       public void accept(Comparable<Object> key, T value) {
         if (filter.test(value)) {
-          if (data.remove(key, value)) {
+          if (instanceList.delete(key, value)) {
             count++;
           }
         }
@@ -253,7 +251,7 @@ int countingRemoveAllByIndexValues(String index, Collection<?> indexValues) {
         return count;
       } else {
         Predicate<? super T> filter = getPredicate(ti.getAccessor(index), indexValues);
-        CountingRemoveIfForEach<T> callback = new CountingRemoveIfForEach<>(data, filter);
+        CountingRemoveIfForEach<T> callback = new CountingRemoveIfForEach<>(this, filter);
 
         // Go through all the values in `data` and delete objects that meets the predicate `filter`.
         // This can be slow when there is a large number of entries in `data`.
@@ -278,7 +276,22 @@ public void put(T value) throws Exception {
 
     public boolean delete(Object key) {
       boolean entryExists = data.remove(asKey(key)) != null;
-      if (entryExists && hasNaturalParentIndex) {
+      if (entryExists) {
+        deleteParentIndex(key);
+      }
+      return entryExists;
+    }
+
+    public boolean delete(Object key, T value) {
+      boolean entryExists = data.remove(asKey(key), value);
+      if (entryExists) {
+        deleteParentIndex(key);
+      }
+      return entryExists;
+    }
+
+    private void deleteParentIndex(Object key) {
+      if (hasNaturalParentIndex) {
         for (NaturalKeys v : parentToChildrenMap.values()) {
           if (v.remove(asKey(key))) {
             // `v` can be empty after removing the natural key and we can remove it from
@@ -289,7 +302,6 @@ public boolean delete(Object key) {
           }
         }
       }
-      return entryExists;
     }
 
     public int size() {