[proxima-direct-core] #356 cache StreamElement in TimeBoundedVersionedCache

Author: je-ik

direct/core/src/main/java/cz/o2/proxima/direct/core/view/LocalCachedPartitionedView.java

@@ -23,6 +23,7 @@
 import cz.o2.proxima.core.storage.StreamElement;
 import cz.o2.proxima.core.storage.commitlog.Position;
 import cz.o2.proxima.core.util.ExceptionUtils;
+import cz.o2.proxima.core.util.Optionals;
 import cz.o2.proxima.core.util.Pair;
 import cz.o2.proxima.direct.core.CommitCallback;
 import cz.o2.proxima.direct.core.OnlineAttributeWriter;
@@ -108,20 +109,15 @@ protected void onCache(StreamElement element, boolean overwrite) {
       final Pair<Long, Payload> oldVal;
       synchronized (cache) {
         oldVal = cache.get(element.getKey(), attrName, Long.MAX_VALUE);
-        long sequentialId = element.hasSequentialId() ? element.getSequentialId() : 0L;
-        updated =
-            cache.put(
-                element.getKey(),
-                attrName,
-                element.getStamp(),
-                sequentialId,
-                overwrite,
-                parsed.orElse(null));
+        updated = cache.put(element, overwrite);
       }
       if (updated) {
         updateCallback.accept(
             element,
-            oldVal != null ? Pair.of(oldVal.getFirst(), oldVal.getSecond().getData()) : null);
+            oldVal != null && !oldVal.getSecond().getData().isDelete()
+                ? Pair.of(
+                    oldVal.getFirst(), Optionals.get(oldVal.getSecond().getData().getParsed()))
+                : null);
       }
     }
   }
@@ -276,7 +272,7 @@ public <T> Optional<KeyValue<T>> get(
       if (desc.isWildcard()) {
         // check for wildcard delete
         Pair<Long, Payload> wildcard = cache.get(key, desc.toAttributePrefix(), stamp);
-        if (wildcard != null && wildcard.getSecond().getData() == null) {
+        if (wildcard != null && wildcard.getSecond().getData().isDelete()) {
           // this is delete
           // move the required stamp after the delete
           deleteStamp = wildcard.getFirst();
@@ -285,7 +281,7 @@ public <T> Optional<KeyValue<T>> get(
       final long filterStamp = deleteStamp;
       return Optional.ofNullable(cache.get(key, attribute, stamp))
           .filter(e -> e.getFirst() >= filterStamp)
-          .flatMap(e -> Optional.ofNullable(toKv(key, attribute, e)));
+          .flatMap(e -> Optional.ofNullable(toKv(e)));
     }
   }
 
@@ -339,7 +335,7 @@ private void scanWildcardPrefix(
           return null;
         },
         (attr, e) -> {
-          KeyValue<Object> kv = toKv(key, attr, e);
+          KeyValue<Object> kv = toKv(e);
           if (kv != null) {
             if (missing.getAndDecrement() != 0) {
               consumer.accept(kv);
@@ -368,45 +364,33 @@ public void close() {
     cache.clear();
   }
 
-  private @Nullable <T> KeyValue<T> toKv(
-      String key, String attribute, @Nullable Pair<Long, Payload> p) {
-
-    Optional<AttributeDescriptor<Object>> attrDesc = entity.findAttribute(attribute, true);
-    if (attrDesc.isPresent()) {
-      return toKv(key, attribute, attrDesc.get(), p);
-    }
-    log.warn("Missing attribute {} in entity {}", attribute, entity);
-    return null;
-  }
-
   @SuppressWarnings("unchecked")
-  private @Nullable <T> KeyValue<T> toKv(
-      String key, String attribute, AttributeDescriptor<?> attr, @Nullable Pair<Long, Payload> p) {
-
-    if (p == null || p.getSecond() == null || p.getSecond().getData() == null) {
+  private @Nullable <T> KeyValue<T> toKv(@Nullable Pair<Long, Payload> p) {
+    if (p == null || p.getSecond() == null || p.getSecond().getData().isDelete()) {
       return null;
     }
-    if (p.getSecond().getSeqId() > 0) {
+    StreamElement data = p.getSecond().getData();
+    if (data.getSequentialId() > 0) {
       return KeyValue.of(
-          entity,
-          (AttributeDescriptor<T>) attr,
-          p.getSecond().getSeqId(),
-          key,
-          attribute,
-          new RawOffset(attribute),
-          (T) p.getSecond().getData(),
-          null,
-          p.getFirst());
+          data.getEntityDescriptor(),
+          (AttributeDescriptor<T>) data.getAttributeDescriptor(),
+          data.getSequentialId(),
+          data.getKey(),
+          data.getAttribute(),
+          new RawOffset(data.getAttribute()),
+          (T) data.getParsed().orElse(null),
+          data.getValue(),
+          data.getStamp());
     }
     return KeyValue.of(
-        entity,
-        (AttributeDescriptor<T>) attr,
-        key,
-        attribute,
-        new RawOffset(attribute),
-        (T) p.getSecond().getData(),
-        null,
-        p.getFirst());
+        data.getEntityDescriptor(),
+        (AttributeDescriptor<T>) data.getAttributeDescriptor(),
+        data.getKey(),
+        data.getAttribute(),
+        new RawOffset(data.getAttribute()),
+        (T) data.getParsed().orElse(null),
+        data.getValue(),
+        data.getStamp());
   }
 
   @Override

direct/core/src/main/java/cz/o2/proxima/direct/core/view/TimeBoundedVersionedCache.java

@@ -29,6 +29,7 @@
 import java.util.Map;
 import java.util.NavigableMap;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -63,7 +64,6 @@ static class Payload {
 
   @Nullable
   synchronized Pair<Long, Payload> get(String key, String attribute, long stamp) {
-
     NavigableMap<String, NavigableMap<Long, Payload>> attrMap;
     attrMap = cache.get(key);
     if (attrMap != null) {
@@ -107,16 +107,15 @@ synchronized void scan(
       return;
     }
     String lastParent = null;
-    Pair<Long, Payload> parentEntry = null;
+    Pair<Long, Payload> parentEntry;
     long parentTombstoneStamp = stamp;
     for (Map.Entry<String, NavigableMap<Long, Payload>> e : attrMap.tailMap(offset).entrySet()) {
-
       if (e.getKey().startsWith(prefix)) {
         if (!e.getKey().equals(offset)) {
           if (lastParent == null || !e.getKey().startsWith(lastParent)) {
             lastParent = parentRecordExtractor.apply(e.getKey());
             parentEntry = lastParent == null ? null : get(key, lastParent, stamp);
-            boolean isDelete = parentEntry != null && parentEntry.getSecond().getData() == null;
+            boolean isDelete = parentEntry != null && parentEntry.getSecond().getData().isDelete();
             parentTombstoneStamp = isDelete ? parentEntry.getFirst() : -1;
           }
           Map.Entry<Long, Payload> floorEntry = e.getValue().floorEntry(stamp);
@@ -194,45 +193,56 @@ synchronized void keys(int offset, int limit, Consumer<String> keyConsumer) {
     }
   }
 
-  synchronized boolean put(
-      String key,
-      String attribute,
-      long stamp,
-      long sequentialId,
-      boolean overwrite,
-      @Nullable Object value) {
-
+  synchronized boolean put(StreamElement element, boolean overwrite) {
     AtomicBoolean updated = new AtomicBoolean();
     cache.compute(
-        key,
-        (k, attrMap) -> {
-          if (attrMap == null) {
-            attrMap = new TreeMap<>();
-          }
-          NavigableMap<Long, Payload> valueMap =
-              attrMap.computeIfAbsent(attribute, tmp -> new TreeMap<>());
-          if (valueMap.isEmpty() || valueMap.firstKey() - keepDuration < stamp) {
-            final Payload oldPayload = valueMap.get(stamp);
-            if (overwrite || oldPayload == null || oldPayload.overridable) {
-              logPayloadUpdateIfNecessary(key, attribute, stamp, value);
-              Payload newPayload = new Payload(value, sequentialId, !overwrite);
-              valueMap.put(stamp, newPayload);
-              updated.set(!newPayload.equals(oldPayload));
-            }
-          }
-          long first;
-          while (!valueMap.isEmpty() && (first = valueMap.firstKey()) + keepDuration < stamp) {
-            valueMap.remove(first);
-          }
-          return attrMap;
-        });
+        element.getKey(),
+        (k, attrMap) -> updateAttrMapByElement(element, overwrite, updated, attrMap));
     return updated.get();
   }
 
-  private void logPayloadUpdateIfNecessary(
-      String key, String attribute, long stamp, @Nullable Object value) {
+  private NavigableMap<String, NavigableMap<Long, Payload>> updateAttrMapByElement(
+      StreamElement element,
+      boolean overwrite,
+      AtomicBoolean updated,
+      @Nullable NavigableMap<String, NavigableMap<Long, Payload>> attrMap) {
+
+    if (attrMap == null) {
+      attrMap = new TreeMap<>();
+    }
+    String attribute =
+        element.isDeleteWildcard()
+            ? element.getAttributeDescriptor().toAttributePrefix()
+            : element.getAttribute();
+    long stamp = element.getStamp();
+
+    NavigableMap<Long, Payload> valueMap =
+        attrMap.computeIfAbsent(attribute, tmp -> new TreeMap<>());
+    if (valueMap.isEmpty() || valueMap.firstKey() - keepDuration < stamp) {
+      final Payload oldPayload = valueMap.get(stamp);
+      if (overwrite || oldPayload == null || oldPayload.overridable) {
+        logPayloadUpdateIfNecessary(attribute, stamp, element);
+        Payload newPayload = new Payload(element, !overwrite);
+        valueMap.put(stamp, newPayload);
+        Object oldParsed =
+            Optional.ofNullable(oldPayload == null ? null : oldPayload.getData())
+                .flatMap(StreamElement::getParsed)
+                .orElse(null);
+        updated.set(!element.getParsed().equals(oldParsed));
+      }
+    }
+    long first;
+    while (!valueMap.isEmpty() && (first = valueMap.firstKey()) + keepDuration < stamp) {
+      valueMap.remove(first);
+    }
+    return attrMap;
+  }
+
+  private void logPayloadUpdateIfNecessary(String attribute, long stamp, StreamElement element) {
 
     if (log.isDebugEnabled()) {
+      String key = element.getKey();
+      Object value = element.getParsed().orElse(null);
       AttributeDescriptor<Object> attrDesc = entity.findAttribute(attribute, true).orElse(null);
       if (attrDesc != null) {
         log.debug(