[Fix #1691] Moving from int index to string generated index

Signed-off-by: Francisco Javier Tirado Sarti <ftirados@ibm.com>

Author: fjtirado

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/PersistenceUtils.java

@@ -15,11 +15,13 @@
  */
 package io.serverlessworkflow.impl.persistence.hashing;
 
+import com.github.f4b6a3.ulid.UlidFactory;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Consumer;
@@ -28,8 +30,10 @@
 
 public class HashMappingCoordinator {
 
+  private static final UlidFactory idFactory = UlidFactory.newMonotonicInstance();
+
   public static final HashMappingCoordinator build(
-      Function<String, Map<String, List<byte[]>>> retriever,
+      Function<String, Map<String, Map<String, byte[]>>> retriever,
       Consumer<Map<String, List<HashMappingInfo>>> writer) {
     return new HashMappingCoordinator(retriever, writer);
   }
@@ -57,19 +61,19 @@ public boolean isTransient() {
     }
   }
 
-  private record PendingWrite(String key, int index, BytesWithFlag bytes) {}
+  private record PendingWrite(String key, String index, BytesWithFlag bytes) {}
 
-  private final Function<String, Map<String, List<byte[]>>> retriever;
+  private final Function<String, Map<String, Map<String, byte[]>>> retriever;
   private final Consumer<Map<String, List<HashMappingInfo>>> writer;
 
   private HashMappingCoordinator(
-      Function<String, Map<String, List<byte[]>>> retriever,
+      Function<String, Map<String, Map<String, byte[]>>> retriever,
       Consumer<Map<String, List<HashMappingInfo>>> writer) {
     this.retriever = retriever;
     this.writer = writer;
   }
 
-  private static Map<String, Map<String, List<BytesWithFlag>>> mappingInfo =
+  private static Map<String, Map<String, Map<String, BytesWithFlag>>> mappingInfo =
       new ConcurrentHashMap<>();
 
   private Map<String, List<PendingWrite>> pending = new HashMap<>();
@@ -78,52 +82,45 @@ private HashMappingInfo from(PendingWrite pending) {
     return new HashMappingInfo(pending.key, pending.index, pending.bytes.bytes);
   }
 
-  public int calculateIndex(String instanceId, HashItem item, byte[] bytes) {
-    Map<String, List<BytesWithFlag>> instanceMap = getInstanceMap(instanceId);
+  public String calculateIndex(String instanceId, HashItem item, byte[] bytes) {
+    Map<String, Map<String, BytesWithFlag>> instanceMap = getInstanceMap(instanceId);
     String key = item.key();
     synchronized (instanceMap) {
-      List<BytesWithFlag> list = instanceMap.computeIfAbsent(key, __ -> new ArrayList<>());
-      if (list.isEmpty()) {
-        int index = 0;
-        BytesWithFlag bytesWithFlag = new BytesWithFlag(bytes);
-        list.add(bytesWithFlag);
-        addWrite(instanceId, key, index, bytesWithFlag);
-        return index;
-      }
-      for (int i = 0; i < list.size(); i++) {
-        BytesWithFlag bytesWithFlag = list.get(i);
-        if (Arrays.equals(bytesWithFlag.bytes, bytes)) {
-          if (bytesWithFlag.isTransient()) {
-            addWrite(instanceId, key, i, bytesWithFlag);
+      Map<String, BytesWithFlag> duplicateMap =
+          instanceMap.computeIfAbsent(key, __ -> new HashMap<>());
+
+      for (Entry<String, BytesWithFlag> entry : duplicateMap.entrySet()) {
+        if (Arrays.equals(entry.getValue().bytes, bytes)) {
+          if (entry.getValue().isTransient()) {
+            addWrite(instanceId, key, entry.getKey(), entry.getValue());
           }
-          return i;
+          return entry.getKey();
         }
       }
+      String index = idFactory.create().toString();
       BytesWithFlag bytesWithFlag = new BytesWithFlag(bytes);
-      list.add(bytesWithFlag);
-      int index = list.size() - 1;
+      duplicateMap.put(index, bytesWithFlag);
       addWrite(instanceId, key, index, bytesWithFlag);
       return index;
     }
   }
 
-  private void addWrite(String instanceId, String key, int index, BytesWithFlag bytes) {
+  private void addWrite(String instanceId, String key, String index, BytesWithFlag bytes) {
     pending
         .computeIfAbsent(instanceId, __ -> new ArrayList<>())
         .add(new PendingWrite(key, index, bytes));
   }
 
-  public Optional<byte[]> readBytes(String instanceId, HashItem item, int index) {
-    Map<String, List<BytesWithFlag>> instanceMap = getInstanceMap(instanceId);
+  public Optional<byte[]> readBytes(String instanceId, HashItem item, String index) {
+    Map<String, Map<String, BytesWithFlag>> instanceMap = getInstanceMap(instanceId);
     synchronized (instanceMap) {
       return Optional.ofNullable(instanceMap.get(item.key()))
-          .filter(list -> list.size() > index)
-          .map(list -> list.get(index))
+          .map(m -> m.get(index))
           .map(bytes -> bytes.bytes);
     }
   }
 
-  private Map<String, List<BytesWithFlag>> getInstanceMap(String instanceId) {
+  private Map<String, Map<String, BytesWithFlag>> getInstanceMap(String instanceId) {
     return mappingInfo.computeIfAbsent(
         instanceId,
         k ->
@@ -132,9 +129,11 @@ private Map<String, List<BytesWithFlag>> getInstanceMap(String instanceId) {
                     Collectors.toMap(
                         Map.Entry::getKey,
                         e ->
-                            e.getValue().stream()
-                                .map(b -> new BytesWithFlag(b, true))
-                                .collect(Collectors.toCollection(ArrayList::new)))));
+                            e.getValue().entrySet().stream()
+                                .collect(
+                                    Collectors.toMap(
+                                        Entry::getKey,
+                                        v -> new BytesWithFlag(v.getValue(), true))))));
   }
 
   public void persist() {
@@ -143,7 +142,7 @@ public void persist() {
       String instanceId = item.getKey();
       List<HashMappingInfo> list = new ArrayList<>();
       result.put(instanceId, list);
-      Map<String, List<BytesWithFlag>> instanceMap = mappingInfo.get(instanceId);
+      Map<String, Map<String, BytesWithFlag>> instanceMap = mappingInfo.get(instanceId);
       if (instanceMap != null) {
         synchronized (instanceMap) {
           item.getValue()
@@ -162,7 +161,7 @@ public void persist() {
   public void afterCommit() {
     for (Map.Entry<String, List<PendingWrite>> item : pending.entrySet()) {
       String instanceId = item.getKey();
-      Map<String, List<BytesWithFlag>> instanceMap = mappingInfo.get(instanceId);
+      Map<String, Map<String, BytesWithFlag>> instanceMap = mappingInfo.get(instanceId);
       if (instanceMap != null) {
         synchronized (instanceMap) {
           item.getValue().forEach(v -> v.bytes.persist());
@@ -176,7 +175,7 @@ public void afterRollback() {
     pending.clear();
   }
 
-  public void afterRemove(String instanceId) {
+  public static void afterRemove(String instanceId) {
     mappingInfo.remove(instanceId);
   }
 }

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingCoordinator.java

@@ -15,4 +15,4 @@
  */
 package io.serverlessworkflow.impl.persistence.hashing;
 
-public record HashMappingInfo(String key, int index, byte[] bytes) {}
+public record HashMappingInfo(String key, String index, byte[] bytes) {}

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/HashMappingInfo.java

@@ -22,7 +22,7 @@
 public class IntegerHashItem implements HashItem {
 
   public static final byte ID = 1;
-  public static final int SIZE_THRESHOLD = 4 * 4;
+  public static final int SIZE_THRESHOLD = 32;
 
   private final int hashCode;
 

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/IntegerHashItem.java

@@ -23,7 +23,7 @@
 public class MD5HashItem implements HashItem {
 
   public static final byte ID = 2;
-  public static final int SIZE_THRESHOLD = 128 * 2;
+  public static final int SIZE_THRESHOLD = 150;
 
   @Override
   public byte id() {

impl/persistence/api/src/main/java/io/serverlessworkflow/impl/persistence/hashing/MD5HashItem.java

@@ -33,7 +33,6 @@
 import io.serverlessworkflow.impl.persistence.CompletedTaskInfo;
 import io.serverlessworkflow.impl.persistence.PersistenceInstanceInfo;
 import io.serverlessworkflow.impl.persistence.PersistenceTaskInfo;
-import io.serverlessworkflow.impl.persistence.PersistenceUtils;
 import io.serverlessworkflow.impl.persistence.RetriedTaskInfo;
 import io.serverlessworkflow.impl.persistence.hashing.HashFactory;
 import io.serverlessworkflow.impl.persistence.hashing.HashItem;
@@ -70,15 +69,13 @@ protected BytesMapInstanceTransaction(WorkflowBufferFactory factory, HashFactory
         HashMappingCoordinator.build(this::retrieveBlobData, this::writeBlobData);
   }
 
-  private Map<String, List<byte[]>> retrieveBlobData(String instanceId) {
-    Map<String, Map<Integer, byte[]>> result = new HashMap<>();
+  private Map<String, Map<String, byte[]>> retrieveBlobData(String instanceId) {
+    Map<String, Map<String, byte[]>> result = new HashMap<>();
     for (Map.Entry<String, byte[]> entry : blobData(instanceId).entrySet()) {
       String[] splitted = entry.getKey().split(SEPARATOR);
-      result
-          .computeIfAbsent(splitted[0], __ -> new HashMap<>())
-          .put(Integer.parseInt(splitted[1]), entry.getValue());
+      result.computeIfAbsent(splitted[0], __ -> new HashMap<>()).put(splitted[1], entry.getValue());
     }
-    return PersistenceUtils.mapMapToMapList(result);
+    return result;
   }
 
   private void writeBlobData(Map<String, List<HashMappingInfo>> writeInfo) {
@@ -99,7 +96,7 @@ public void removeProcessInstance(WorkflowContextData workflowContext) {
     super.removeProcessInstance(workflowContext);
     String instanceId = workflowContext.instanceData().id();
     removeBlobData(instanceId);
-    hashCoordinator.afterRemove(instanceId);
+    HashMappingCoordinator.afterRemove(instanceId);
   }
 
   @Override
@@ -280,10 +277,10 @@ private PersistenceTaskInfo readVersion0(WorkflowInputBuffer buffer) {
 
   private void writeLargeObject(
       HashItem item, WorkflowInstanceData instanceData, WorkflowOutputBuffer writer, byte[] bytes) {
-    int index = hashCoordinator.calculateIndex(instanceData.id(), item, bytes);
+    String index = hashCoordinator.calculateIndex(instanceData.id(), item, bytes);
     writer.writeByte(item.id());
     item.writeKey(writer);
-    writer.writeShort((short) index);
+    writer.writeString(index);
   }
 
   private Object readLargeObject(String instanceId, WorkflowInputBuffer buffer) {
@@ -295,7 +292,7 @@ private Object readLargeObject(String instanceId, WorkflowInputBuffer buffer) {
                   bufferFactory.input(
                       new ByteArrayInputStream(
                           hashCoordinator
-                              .readBytes(instanceId, item, buffer.readShort())
+                              .readBytes(instanceId, item, buffer.readString())
                               .orElseThrow()))) {
                 return input.readObject();
               }

impl/persistence/bigmap/src/main/java/io/serverlessworkflow/impl/persistence/bigmap/BytesMapInstanceTransaction.java

@@ -37,6 +37,7 @@
 import io.serverlessworkflow.impl.executors.TransitionInfo;
 import io.serverlessworkflow.impl.persistence.PersistenceInstanceHandlers;
 import io.serverlessworkflow.impl.persistence.WorkflowPersistenceInstance;
+import io.serverlessworkflow.impl.persistence.hashing.HashMappingCoordinator;
 import java.io.IOException;
 import java.time.Duration;
 import java.time.Instant;
@@ -163,9 +164,12 @@ void testWorkflowInstance() throws InterruptedException {
       assertThat(stream.count()).isEqualTo(1);
     }
     definition.close();
+
     instance =
         (WorkflowPersistenceInstance)
             handlers.reader().find(definition, workflowInstance.id()).orElseThrow();
+
+    HashMappingCoordinator.afterRemove(instance.id());
     updateWContext = mock(WorkflowContext.class);
     updateTContext = mock(TaskContext.class);
     when(updateTContext.position()).thenReturn(position2);