Merge pull request #22508 from gosubpl/wip/22350-delta-crdt-orset-friends

ORMap and friends are delta-CRDTs (#22350)

akka-distributed-data/src/main/protobuf/ReplicatedDataMessages.proto

@@ -75,6 +75,31 @@ message ORMap {
   repeated Entry entries = 2;
 }
 
+message ORMapDeltaGroup {
+  message MapEntry {
+    optional string stringKey = 1;
+    required OtherMessage value = 2;
+    optional sint32 intKey = 3;
+    optional sint64 longKey = 4;
+    optional OtherMessage otherKey = 5;
+  }
+  message Entry {
+    required ORMapDeltaOp operation = 1;
+    required ORSet underlying = 2;
+    required sint32 zeroTag = 3;
+    optional MapEntry entryData = 4;
+  }
+
+  repeated Entry entries = 1;
+}
+
+enum ORMapDeltaOp {
+  ORMapPut = 0;
+  ORMapRemove = 1;
+  ORMapRemoveKey = 2;
+  ORMapUpdate = 3;
+}
+
 message LWWMap {
   message Entry {
     optional string stringKey = 1;
@@ -112,7 +137,8 @@ message ORMultiMap {
 
    required ORSet keys = 1;
    repeated Entry entries = 2;
-}
+   optional bool withValueDeltas = 3;
+ }
 
 
 

akka-distributed-data/src/main/scala/akka/cluster/ddata/LWWMap.scala

@@ -6,9 +6,10 @@ package akka.cluster.ddata
 import akka.cluster.Cluster
 import akka.cluster.UniqueAddress
 import akka.annotation.InternalApi
+import akka.cluster.ddata.ORMap.LWWMapTag
 
 object LWWMap {
-  private val _empty: LWWMap[Any, Any] = new LWWMap(ORMap.empty)
+  private val _empty: LWWMap[Any, Any] = new LWWMap(ORMap.emptyWithLWWMapTag)
   def empty[A, B]: LWWMap[A, B] = _empty.asInstanceOf[LWWMap[A, B]]
   def apply(): LWWMap[Any, Any] = _empty
   /**
@@ -47,10 +48,11 @@ object LWWMap {
 @SerialVersionUID(1L)
 final class LWWMap[A, B] private[akka] (
   private[akka] val underlying: ORMap[A, LWWRegister[B]])
-  extends ReplicatedData with ReplicatedDataSerialization with RemovedNodePruning {
+  extends DeltaReplicatedData with ReplicatedDataSerialization with RemovedNodePruning {
   import LWWRegister.{ Clock, defaultClock }
 
   type T = LWWMap[A, B]
+  type D = ORMap.DeltaOp
 
   /**
    * Scala API: All entries of the map.
@@ -141,6 +143,14 @@ final class LWWMap[A, B] private[akka] (
   @InternalApi private[akka] def remove(node: UniqueAddress, key: A): LWWMap[A, B] =
     new LWWMap(underlying.remove(node, key))
 
+  override def resetDelta: LWWMap[A, B] =
+    new LWWMap(underlying.resetDelta)
+
+  override def delta: Option[D] = underlying.delta
+
+  override def mergeDelta(thatDelta: D): LWWMap[A, B] =
+    new LWWMap(underlying.mergeDelta(thatDelta))
+
   override def merge(that: LWWMap[A, B]): LWWMap[A, B] =
     new LWWMap(underlying.merge(that.underlying))
 

akka-distributed-data/src/main/scala/akka/cluster/ddata/ORMap.scala

@@ -7,6 +7,9 @@ import akka.cluster.Cluster
 import akka.cluster.UniqueAddress
 import akka.util.HashCode
 import akka.annotation.InternalApi
+import akka.cluster.ddata.ORMap.{ AtomicDeltaOp, ZeroTag }
+
+import scala.collection.immutable
 
 object ORMap {
   private val _empty: ORMap[Any, ReplicatedData] = new ORMap(ORSet.empty, Map.empty)
@@ -22,6 +25,125 @@ object ORMap {
    */
   def unapply[A, B <: ReplicatedData](m: ORMap[A, B]): Option[Map[A, B]] = Some(m.entries)
 
+  sealed trait DeltaOp extends ReplicatedDelta with RequiresCausalDeliveryOfDeltas {
+    type T = DeltaOp
+    override def zero: DeltaReplicatedData
+  }
+
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[akka] def emptyWithPNCounterMapTag[A, B <: ReplicatedData]: ORMap[A, B] = new ORMap(ORSet.empty, Map.empty, zeroTag = PNCounterMapTag)
+
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[akka] def emptyWithORMultiMapTag[A, B <: ReplicatedData]: ORMap[A, B] = new ORMap(ORSet.empty, Map.empty, zeroTag = ORMultiMapTag)
+
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[akka] def emptyWithORMultiMapWithValueDeltasTag[A, B <: ReplicatedData]: ORMap[A, B] = new ORMap(ORSet.empty, Map.empty, zeroTag = ORMultiMapWithValueDeltasTag)
+
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[akka] def emptyWithLWWMapTag[A, B <: ReplicatedData]: ORMap[A, B] = new ORMap(ORSet.empty, Map.empty, zeroTag = LWWMapTag)
+
+  /**
+   * INTERNAL API
+   * Tags for ORMap.DeltaOp's, so that when the Replicator needs to re-create full value from delta,
+   * the right map type will be used
+   */
+  @InternalApi private[akka] trait ZeroTag {
+    def zero: DeltaReplicatedData
+    def value: Int
+  }
+
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[akka] case object VanillaORMapTag extends ZeroTag {
+    override def zero: DeltaReplicatedData = ORMap.empty
+    override final val value: Int = 0
+  }
+
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[akka] case object PNCounterMapTag extends ZeroTag {
+    override def zero: DeltaReplicatedData = PNCounterMap.empty
+    override final val value: Int = 1
+  }
+
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[akka] case object ORMultiMapTag extends ZeroTag {
+    override def zero: DeltaReplicatedData = ORMultiMap.empty
+    override final val value: Int = 2
+  }
+
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[akka] case object ORMultiMapWithValueDeltasTag extends ZeroTag {
+    override def zero: DeltaReplicatedData = ORMultiMap.emptyWithValueDeltas
+    override final val value: Int = 3
+  }
+
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[akka] case object LWWMapTag extends ZeroTag {
+    override def zero: DeltaReplicatedData = LWWMap.empty
+    override final val value: Int = 4
+  }
+
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[akka] sealed abstract class AtomicDeltaOp[A, B <: ReplicatedData] extends DeltaOp {
+    def underlying: ORSet.DeltaOp
+    def zeroTag: ZeroTag
+    override def zero: DeltaReplicatedData = zeroTag.zero
+
+    override def merge(that: DeltaOp): DeltaOp = that match {
+      case other: AtomicDeltaOp[A, B] ⇒ DeltaGroup(Vector(this, other))
+      case DeltaGroup(ops)            ⇒ DeltaGroup(this +: ops)
+    }
+  }
+
+  // PutDeltaOp contains ORSet delta and full value
+  /** INTERNAL API */
+  @InternalApi private[akka] final case class PutDeltaOp[A, B <: ReplicatedData](underlying: ORSet.DeltaOp, value: (A, B), zeroTag: ZeroTag = VanillaORMapTag) extends AtomicDeltaOp[A, B] {
+  }
+
+  // UpdateDeltaOp contains ORSet delta and either delta of value (in case where underlying type supports deltas) or full value
+  /** INTERNAL API */
+  @InternalApi private[akka] final case class UpdateDeltaOp[A, X <: ReplicatedDelta](underlying: ORSet.DeltaOp, values: Map[A, X], zeroTag: ZeroTag = VanillaORMapTag) extends AtomicDeltaOp[A, X] {
+  }
+
+  // RemoveDeltaOp does not contain any value at all - the propagated 'value' map is empty
+  /** INTERNAL API */
+  @InternalApi private[akka] final case class RemoveDeltaOp[A, B <: ReplicatedData](underlying: ORSet.DeltaOp, zeroTag: ZeroTag = VanillaORMapTag) extends AtomicDeltaOp[A, B] {
+  }
+
+  // RemoveKeyDeltaOp contains a single value - to provide the recipient with the removed key for value map
+  /** INTERNAL API */
+  @InternalApi private[akka] final case class RemoveKeyDeltaOp[A, B <: ReplicatedData](underlying: ORSet.DeltaOp, removedKey: A, zeroTag: ZeroTag = VanillaORMapTag) extends AtomicDeltaOp[A, B] {
+  }
+
+  // DeltaGroup is effectively a causally ordered list of individual deltas
+  /** INTERNAL API */
+  @InternalApi private[akka] final case class DeltaGroup[A, B <: ReplicatedData](ops: immutable.IndexedSeq[DeltaOp]) extends DeltaOp {
+    override def merge(that: DeltaOp): DeltaOp = that match {
+      case DeltaGroup(thatOps)       ⇒ DeltaGroup(ops ++ thatOps)
+      case that: AtomicDeltaOp[A, B] ⇒ DeltaGroup(ops :+ that)
+    }
+
+    override def zero: DeltaReplicatedData = ops.headOption.fold(ORMap.empty[A, B].asInstanceOf[DeltaReplicatedData])(_.zero)
+  }
 }
 
 /**
@@ -35,10 +157,15 @@ object ORMap {
 @SerialVersionUID(1L)
 final class ORMap[A, B <: ReplicatedData] private[akka] (
   private[akka] val keys:    ORSet[A],
-  private[akka] val values: Map[A, B])
+  private[akka] val values:  Map[A, B],
-  extends ReplicatedData with ReplicatedDataSerialization with RemovedNodePruning {
+  private[akka] val zeroTag: ZeroTag               = ORMap.VanillaORMapTag,
+  override val delta:        Option[ORMap.DeltaOp] = None)
+  extends DeltaReplicatedData with ReplicatedDataSerialization with RemovedNodePruning {
+
+  import ORMap.{ PutDeltaOp, UpdateDeltaOp, RemoveDeltaOp, RemoveKeyDeltaOp }
 
   type T = ORMap[A, B]
+  type D = ORMap.DeltaOp
 
   /**
    * Scala API: All entries of the map.
@@ -100,8 +227,11 @@ final class ORMap[A, B <: ReplicatedData] private[akka] (
         "`ORMap.put` must not be used to replace an existing `ORSet` " +
           "value, because important history can be lost when replacing the `ORSet` and " +
           "undesired effects of merging will occur. Use `ORMultiMap` or `ORMap.updated` instead.")
-    else
+    else {
-      new ORMap(keys.add(node, key), values.updated(key, value))
+      val putDeltaOp = PutDeltaOp(keys.resetDelta.add(node, key).delta.get, key → value, zeroTag)
+      // put forcibly damages history, so we propagate full value that will overwrite previous values
+      new ORMap(keys.add(node, key), values.updated(key, value), zeroTag, Some(newDelta(putDeltaOp)))
+    }
 
   /**
    * Scala API: Replace a value by applying the `modify` function on the existing value.
@@ -124,12 +254,30 @@ final class ORMap[A, B <: ReplicatedData] private[akka] (
   /**
    * INTERNAL API
    */
-  @InternalApi private[akka] def updated(node: UniqueAddress, key: A, initial: B)(modify: B ⇒ B): ORMap[A, B] = {
+  @InternalApi private[akka] def updated(node: UniqueAddress, key: A, initial: B, valueDeltas: Boolean = false)(modify: B ⇒ B): ORMap[A, B] = {
-    val newValue = values.get(key) match {
+    val (oldValue, hasOldValue) = values.get(key) match {
-      case Some(old) ⇒ modify(old)
+      case Some(old) ⇒ (old, true)
-      case _         ⇒ modify(initial)
+      case _         ⇒ (initial, false)
+    }
+    // Optimization: for some types - like GSet, GCounter, PNCounter and ORSet  - that are delta based
+    // we can emit (and later merge) their deltas instead of full updates.
+    // However to avoid necessity of tombstones, the derived map type needs to support this
+    // with clearing the value (e.g. removing all elements if value is a set)
+    // before removing the key - like e.g. ORMultiMap does
+    oldValue match {
+      case _: DeltaReplicatedData if valueDeltas ⇒
+        val newValue = modify(oldValue.asInstanceOf[DeltaReplicatedData].resetDelta.asInstanceOf[B])
+        val newValueDelta = newValue.asInstanceOf[DeltaReplicatedData].delta
+        val deltaOp = newValueDelta match {
+          case Some(d) if hasOldValue ⇒ UpdateDeltaOp(keys.resetDelta.add(node, key).delta.get, Map(key → d), zeroTag)
+          case _                      ⇒ PutDeltaOp(keys.resetDelta.add(node, key).delta.get, key → newValue, zeroTag)
+        }
+        new ORMap(keys.add(node, key), values.updated(key, newValue), zeroTag, Some(newDelta(deltaOp)))
+      case _ ⇒
+        val newValue = modify(oldValue)
+        val deltaOp = PutDeltaOp(keys.resetDelta.add(node, key).delta.get, key → newValue, zeroTag)
+        new ORMap(keys.add(node, key), values.updated(key, newValue), zeroTag, Some(newDelta(deltaOp)))
     }
-    new ORMap(keys.add(node, key), values.updated(key, newValue))
   }
 
   /**
@@ -150,13 +298,24 @@ final class ORMap[A, B <: ReplicatedData] private[akka] (
    * INTERNAL API
    */
   @InternalApi private[akka] def remove(node: UniqueAddress, key: A): ORMap[A, B] = {
-    new ORMap(keys.remove(node, key), values - key)
+    // for removals the delta values map emitted will be empty
+    val removeDeltaOp = RemoveDeltaOp(keys.resetDelta.remove(node, key).delta.get, zeroTag)
+    new ORMap(keys.remove(node, key), values - key, zeroTag, Some(newDelta(removeDeltaOp)))
   }
 
-  override def merge(that: ORMap[A, B]): ORMap[A, B] = {
+  /**
-    val mergedKeys = keys.merge(that.keys)
+   * INTERNAL API
+   * This function is only to be used by derived maps that avoid remove anomalies
+   * by keeping the vvector (in form of key -> value pair) for deleted keys
+   */
+  @InternalApi private[akka] def removeKey(node: UniqueAddress, key: A): ORMap[A, B] = {
+    val removeKeyDeltaOp = RemoveKeyDeltaOp(keys.resetDelta.remove(node, key).delta.get, key, zeroTag)
+    new ORMap(keys.remove(node, key), values, zeroTag, Some(newDelta(removeKeyDeltaOp)))
+  }
+
+  private def dryMerge(that: ORMap[A, B], mergedKeys: ORSet[A], valueKeysIterator: Iterator[A]): ORMap[A, B] = {
     var mergedValues = Map.empty[A, B]
-    mergedKeys.elementsMap.keysIterator.foreach { key ⇒
+    valueKeysIterator.foreach { key ⇒
       (this.values.get(key), that.values.get(key)) match {
         case (Some(thisValue), Some(thatValue)) ⇒
           if (thisValue.getClass != thatValue.getClass) {
@@ -174,8 +333,110 @@ final class ORMap[A, B <: ReplicatedData] private[akka] (
         case (None, None) ⇒ throw new IllegalStateException(s"missing value for $key")
       }
     }
+    new ORMap(mergedKeys, mergedValues, zeroTag = zeroTag)
+  }
 
-    new ORMap(mergedKeys, mergedValues)
+  override def merge(that: ORMap[A, B]): ORMap[A, B] = {
+    val mergedKeys = keys.merge(that.keys)
+    dryMerge(that, mergedKeys, mergedKeys.elementsMap.keysIterator)
+  }
+
+  /**
+   * INTERNAL API
+   * This function is only to be used by derived maps that avoid remove anomalies
+   * by keeping the vvector (in form of key -> value pair) for deleted keys
+   */
+  @InternalApi private[akka] def mergeRetainingDeletedValues(that: ORMap[A, B]): ORMap[A, B] = {
+    val mergedKeys = keys.merge(that.keys)
+    dryMerge(that, mergedKeys, (this.values.keySet ++ that.values.keySet).iterator)
+  }
+
+  override def resetDelta: ORMap[A, B] =
+    if (delta.isEmpty) this
+    else new ORMap[A, B](keys, values, zeroTag = zeroTag)
+
+  override def mergeDelta(thatDelta: ORMap.DeltaOp): ORMap[A, B] = {
+    // helper function to simplify folds below
+    def foldValues(values: List[(A, ReplicatedData)], initial: B) =
+      values.foldLeft(initial) {
+        case (acc: DeltaReplicatedData, (_, value: ReplicatedDelta)) ⇒
+          acc.mergeDelta(value.asInstanceOf[acc.D]).asInstanceOf[B]
+        case (acc, (_, value)) ⇒
+          acc.merge(value.asInstanceOf[acc.T]).asInstanceOf[B]
+      }
+
+    val mergedKeys: ORSet[A] = thatDelta match {
+      case d: AtomicDeltaOp[A, B] ⇒ keys.mergeDelta(d.underlying)
+      case ORMap.DeltaGroup(ops) ⇒
+        ops.foldLeft(keys)((acc, op) ⇒ acc.mergeDelta(op.asInstanceOf[AtomicDeltaOp[A, B]].underlying))
+    }
+
+    var mergedValues = Map.empty[A, B]
+    var tombstonedVals = Set.empty[A]
+    var thatValueDeltas: Map[A, List[(A, ReplicatedData)]] = Map.empty
+
+    val processDelta: PartialFunction[ORMap.DeltaOp, Unit] = {
+      case putOp: PutDeltaOp[A, B] ⇒
+        val key = putOp.value._1
+        thatValueDeltas += (key → (putOp.value :: Nil)) // put is destructive!
+      case _: RemoveDeltaOp[A, B] ⇒
+      // remove delta is only for the side effect of key being removed
+      // please note that if it is not preceded by update clearing the value
+      // anomalies will result
+      case removeKeyOp: RemoveKeyDeltaOp[A, B] ⇒
+        tombstonedVals = tombstonedVals + removeKeyOp.removedKey
+      case updateOp: UpdateDeltaOp[A, _] ⇒
+        val key = updateOp.values.head._1
+        val value = (key, updateOp.values.head._2)
+        if (thatValueDeltas.contains(key))
+          thatValueDeltas = thatValueDeltas + (key → (thatValueDeltas(key) :+ value))
+        else
+          thatValueDeltas += (key → (value :: Nil))
+    }
+
+    val processNestedDelta: PartialFunction[ORMap.DeltaOp, Unit] = {
+      case ORMap.DeltaGroup(ops) ⇒
+        ops.foreach {
+          processDelta.orElse {
+            case ORMap.DeltaGroup(args) ⇒
+              throw new IllegalStateException("Cannot nest DeltaGroups")
+          }
+        }
+    }
+
+    (processDelta orElse processNestedDelta)(thatDelta)
+
+    val aggregateValuesForKey: (A ⇒ Unit) = { key ⇒
+      (this.values.get(key), thatValueDeltas.get(key)) match {
+        case (Some(thisValue), Some(thatValues)) ⇒
+          val mergedValue = foldValues(thatValues, thisValue)
+          mergedValues = mergedValues.updated(key, mergedValue)
+        case (Some(thisValue), None) ⇒
+          mergedValues = mergedValues.updated(key, thisValue)
+        case (None, Some(thatValues)) ⇒
+          val (_, initialValue) = thatValues.head
+          val mergedValue = initialValue match {
+            case _: ReplicatedDelta ⇒
+              foldValues(thatValues, initialValue.asInstanceOf[ReplicatedDelta].zero.asInstanceOf[B])
+            case _ ⇒
+              foldValues(thatValues.tail, initialValue.asInstanceOf[B])
+          }
+          mergedValues = mergedValues.updated(key, mergedValue)
+        case (None, None) ⇒ throw new IllegalStateException(s"missing value for $key")
+      }
+    }
+
+    mergedKeys.elementsMap.keysIterator.foreach { aggregateValuesForKey }
+    tombstonedVals.foreach { aggregateValuesForKey }
+
+    new ORMap[A, B](mergedKeys, mergedValues, zeroTag = zeroTag)
+  }
+
+  private def newDelta(deltaOp: ORMap.DeltaOp) = delta match {
+    case Some(d) ⇒
+      d.merge(deltaOp)
+    case None ⇒
+      deltaOp
   }
 
   override def modifiedByNodes: Set[UniqueAddress] = {
@@ -199,7 +460,7 @@ final class ORMap[A, B <: ReplicatedData] private[akka] (
         acc.updated(key, data.prune(removedNode, collapseInto).asInstanceOf[B])
       case (acc, _) ⇒ acc
     }
-    new ORMap(prunedKeys, prunedValues)
+    new ORMap(prunedKeys, prunedValues, zeroTag = zeroTag)
   }
 
   override def pruningCleanup(removedNode: UniqueAddress): ORMap[A, B] = {
@@ -209,7 +470,7 @@ final class ORMap[A, B <: ReplicatedData] private[akka] (
         acc.updated(key, data.pruningCleanup(removedNode).asInstanceOf[B])
       case (acc, _) ⇒ acc
     }
-    new ORMap(pruningCleanupedKeys, pruningCleanupedValues)
+    new ORMap(pruningCleanupedKeys, pruningCleanupedValues, zeroTag = zeroTag)
   }
 
   // this class cannot be a `case class` because we need different `unapply`

akka-distributed-data/src/main/scala/akka/cluster/ddata/ORMultiMap.scala

@@ -3,22 +3,26 @@
  */
 package akka.cluster.ddata
 
-import akka.cluster.{ UniqueAddress, Cluster }
+import akka.cluster.{ Cluster, UniqueAddress }
 import akka.annotation.InternalApi
+import akka.cluster.ddata.ORMap._
 
 object ORMultiMap {
 
-  val _empty: ORMultiMap[Any, Any] = new ORMultiMap(ORMap.empty)
+  val _empty: ORMultiMap[Any, Any] = new ORMultiMap(ORMap.emptyWithORMultiMapTag, false)
+  val _emptyWithValueDeltas: ORMultiMap[Any, Any] = new ORMultiMap(ORMap.emptyWithORMultiMapTag, true)
   /**
    * Provides an empty multimap.
    */
   def empty[A, B]: ORMultiMap[A, B] = _empty.asInstanceOf[ORMultiMap[A, B]]
+  def emptyWithValueDeltas[A, B]: ORMultiMap[A, B] = _emptyWithValueDeltas.asInstanceOf[ORMultiMap[A, B]]
   def apply(): ORMultiMap[Any, Any] = _empty
 
   /**
    * Java API
    */
   def create[A, B](): ORMultiMap[A, B] = empty[A, B]
+  def createWithDeltaDelta[A, B](): ORMultiMap[A, B] = emptyWithValueDeltas[A, B]
 
   /**
    * Extract the [[ORMultiMap#entries]].
@@ -41,18 +45,28 @@ object ORMultiMap {
  * This class is immutable, i.e. "modifying" methods return a new instance.
  */
 @SerialVersionUID(1L)
-final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[A, ORSet[B]])
+final class ORMultiMap[A, B] private[akka] (
-  extends ReplicatedData with ReplicatedDataSerialization with RemovedNodePruning {
+  private[akka] val underlying:      ORMap[A, ORSet[B]],
+  private[akka] val withValueDeltas: Boolean)
+  extends DeltaReplicatedData with ReplicatedDataSerialization with RemovedNodePruning {
 
   override type T = ORMultiMap[A, B]
+  override type D = ORMap.DeltaOp
 
   override def merge(that: T): T =
-    new ORMultiMap(underlying.merge(that.underlying))
+    if (withValueDeltas == that.withValueDeltas) {
+      if (withValueDeltas)
+        new ORMultiMap(underlying.mergeRetainingDeletedValues(that.underlying), withValueDeltas)
+      else
+        new ORMultiMap(underlying.merge(that.underlying), withValueDeltas)
+    } else throw new IllegalArgumentException("Trying to merge two ORMultiMaps of different map sub-type")
 
   /**
    * Scala API: All entries of a multimap where keys are strings and values are sets.
    */
-  def entries: Map[A, Set[B]] =
+  def entries: Map[A, Set[B]] = if (withValueDeltas)
+    underlying.entries.collect { case (k, v) if underlying.keys.elements.contains(k) ⇒ k → v.elements }
+  else
     underlying.entries.map { case (k, v) ⇒ k → v.elements }
 
   /**
@@ -61,9 +75,10 @@ final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[
   def getEntries(): java.util.Map[A, java.util.Set[B]] = {
     import scala.collection.JavaConverters._
     val result = new java.util.HashMap[A, java.util.Set[B]]
-    underlying.entries.foreach {
+    if (withValueDeltas)
-      case (k, v) ⇒ result.put(k, v.elements.asJava)
+      underlying.entries.foreach { case (k, v) ⇒ if (underlying.keys.elements.contains(k)) result.put(k, v.elements.asJava) }
-    }
+    else
+      underlying.entries.foreach { case (k, v) ⇒ result.put(k, v.elements.asJava) }
     result
   }
 
@@ -71,6 +86,9 @@ final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[
    * Get the set associated with the key if there is one.
    */
   def get(key: A): Option[Set[B]] =
+    if (withValueDeltas && !underlying.keys.elements.contains(key))
+      None
+    else
       underlying.get(key).map(_.elements)
 
   /**
@@ -80,11 +98,11 @@ final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[
   def getOrElse(key: A, default: ⇒ Set[B]): Set[B] =
     get(key).getOrElse(default)
 
-  def contains(key: A): Boolean = underlying.contains(key)
+  def contains(key: A): Boolean = underlying.keys.elements.contains(key)
 
-  def isEmpty: Boolean = underlying.isEmpty
+  def isEmpty: Boolean = underlying.keys.elements.isEmpty
 
-  def size: Int = underlying.size
+  def size: Int = underlying.keys.elements.size
 
   /**
    * Convenience for put. Requires an implicit Cluster.
@@ -115,10 +133,10 @@ final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[
    * INTERNAL API
    */
   @InternalApi private[akka] def put(node: UniqueAddress, key: A, value: Set[B]): ORMultiMap[A, B] = {
-    val newUnderlying = underlying.updated(node, key, ORSet.empty[B]) { existing ⇒
+    val newUnderlying = underlying.updated(node, key, ORSet.empty[B], valueDeltas = withValueDeltas) { existing ⇒
       value.foldLeft(existing.clear(node)) { (s, element) ⇒ s.add(node, element) }
     }
-    new ORMultiMap(newUnderlying)
+    new ORMultiMap(newUnderlying, withValueDeltas)
   }
 
   /**
@@ -137,8 +155,14 @@ final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[
   /**
    * INTERNAL API
    */
-  @InternalApi private[akka] def remove(node: UniqueAddress, key: A): ORMultiMap[A, B] =
+  @InternalApi private[akka] def remove(node: UniqueAddress, key: A): ORMultiMap[A, B] = {
-    new ORMultiMap(underlying.remove(node, key))
+    if (withValueDeltas) {
+      val u = underlying.updated(node, key, ORSet.empty[B], valueDeltas = true) { existing ⇒ existing.clear(node) }
+      new ORMultiMap(u.removeKey(node, key), withValueDeltas)
+    } else {
+      new ORMultiMap(underlying.remove(node, key), withValueDeltas)
+    }
+  }
 
   /**
    * Scala API: Add an element to a set associated with a key. If there is no existing set then one will be initialised.
@@ -156,8 +180,8 @@ final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[
    * INTERNAL API
    */
   @InternalApi private[akka] def addBinding(node: UniqueAddress, key: A, element: B): ORMultiMap[A, B] = {
-    val newUnderlying = underlying.updated(node, key, ORSet.empty[B])(_.add(node, element))
+    val newUnderlying = underlying.updated(node, key, ORSet.empty[B], valueDeltas = withValueDeltas)(_.add(node, element))
-    new ORMultiMap(newUnderlying)
+    new ORMultiMap(newUnderlying, withValueDeltas)
   }
 
   /**
@@ -179,13 +203,17 @@ final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[
    */
   @InternalApi private[akka] def removeBinding(node: UniqueAddress, key: A, element: B): ORMultiMap[A, B] = {
     val newUnderlying = {
-      val u = underlying.updated(node, key, ORSet.empty[B])(_.remove(node, element))
+      val u = underlying.updated(node, key, ORSet.empty[B], valueDeltas = withValueDeltas)(_.remove(node, element))
       u.get(key) match {
-        case Some(s) if s.isEmpty ⇒ u.remove(node, key)
+        case Some(s) if s.isEmpty ⇒
+          if (withValueDeltas)
+            u.removeKey(node, key)
+          else
+            u.remove(node, key)
         case _ ⇒ u
       }
     }
-    new ORMultiMap(newUnderlying)
+    new ORMultiMap(newUnderlying, withValueDeltas)
   }
 
   /**
@@ -205,6 +233,14 @@ final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[
     else
       this
 
+  override def resetDelta: ORMultiMap[A, B] =
+    new ORMultiMap(underlying.resetDelta, withValueDeltas)
+
+  override def delta: Option[D] = underlying.delta
+
+  override def mergeDelta(thatDelta: D): ORMultiMap[A, B] =
+    new ORMultiMap(underlying.mergeDelta(thatDelta), withValueDeltas)
+
   override def modifiedByNodes: Set[UniqueAddress] =
     underlying.modifiedByNodes
 
@@ -212,10 +248,10 @@ final class ORMultiMap[A, B] private[akka] (private[akka] val underlying: ORMap[
     underlying.needPruningFrom(removedNode)
 
   override def pruningCleanup(removedNode: UniqueAddress): T =
-    new ORMultiMap(underlying.pruningCleanup(removedNode))
+    new ORMultiMap(underlying.pruningCleanup(removedNode), withValueDeltas)
 
   override def prune(removedNode: UniqueAddress, collapseInto: UniqueAddress): T =
-    new ORMultiMap(underlying.prune(removedNode, collapseInto))
+    new ORMultiMap(underlying.prune(removedNode, collapseInto), withValueDeltas)
 
   // this class cannot be a `case class` because we need different `unapply`
 

akka-distributed-data/src/main/scala/akka/cluster/ddata/ORSet.scala

@@ -91,7 +91,10 @@ object ORSet {
     }
   }
 
-  final case class DeltaGroup[A](ops: immutable.IndexedSeq[DeltaOp]) extends DeltaOp {
+  /**
+   * INTERNAL API
+   */
+  @InternalApi private[akka] final case class DeltaGroup[A](ops: immutable.IndexedSeq[DeltaOp]) extends DeltaOp {
     override def merge(that: DeltaOp): DeltaOp = that match {
       case thatAdd: AddDeltaOp[A] ⇒
         // merge AddDeltaOp into last AddDeltaOp in the group, if possible

akka-distributed-data/src/main/scala/akka/cluster/ddata/PNCounterMap.scala

@@ -6,10 +6,12 @@ package akka.cluster.ddata
 import akka.cluster.Cluster
 import akka.cluster.UniqueAddress
 import java.math.BigInteger
+
 import akka.annotation.InternalApi
+import akka.cluster.ddata.ORMap._
 
 object PNCounterMap {
-  def empty[A]: PNCounterMap[A] = new PNCounterMap(ORMap.empty)
+  def empty[A]: PNCounterMap[A] = new PNCounterMap(ORMap.emptyWithPNCounterMapTag)
   def apply[A](): PNCounterMap[A] = empty
   /**
    * Java API
@@ -30,9 +32,10 @@ object PNCounterMap {
 @SerialVersionUID(1L)
 final class PNCounterMap[A] private[akka] (
   private[akka] val underlying: ORMap[A, PNCounter])
-  extends ReplicatedData with ReplicatedDataSerialization with RemovedNodePruning {
+  extends DeltaReplicatedData with ReplicatedDataSerialization with RemovedNodePruning {
 
   type T = PNCounterMap[A]
+  type D = ORMap.DeltaOp
 
   /** Scala API */
   def entries: Map[A, BigInt] = underlying.entries.map { case (k, c) ⇒ k → c.value }
@@ -124,6 +127,14 @@ final class PNCounterMap[A] private[akka] (
   override def merge(that: PNCounterMap[A]): PNCounterMap[A] =
     new PNCounterMap(underlying.merge(that.underlying))
 
+  override def resetDelta: PNCounterMap[A] =
+    new PNCounterMap(underlying.resetDelta)
+
+  override def delta: Option[D] = underlying.delta
+
+  override def mergeDelta(thatDelta: D): PNCounterMap[A] =
+    new PNCounterMap(underlying.mergeDelta(thatDelta))
+
   override def modifiedByNodes: Set[UniqueAddress] =
     underlying.modifiedByNodes
 

akka-distributed-data/src/main/scala/akka/cluster/ddata/protobuf/ReplicatedDataSerializer.scala

@@ -149,6 +149,22 @@ private object ReplicatedDataSerializer {
     override def getValue(entry: rd.ORMultiMap.Entry): rd.ORSet = entry.getValue
   }
 
+  implicit object ORMapDeltaGroupEntry extends ProtoMapEntryWriter[rd.ORMapDeltaGroup.MapEntry, rd.ORMapDeltaGroup.MapEntry.Builder, dm.OtherMessage] with ProtoMapEntryReader[rd.ORMapDeltaGroup.MapEntry, dm.OtherMessage] {
+    override def setStringKey(builder: rd.ORMapDeltaGroup.MapEntry.Builder, key: String, value: dm.OtherMessage): rd.ORMapDeltaGroup.MapEntry = builder.setStringKey(key).setValue(value).build()
+    override def setLongKey(builder: rd.ORMapDeltaGroup.MapEntry.Builder, key: Long, value: dm.OtherMessage): rd.ORMapDeltaGroup.MapEntry = builder.setLongKey(key).setValue(value).build()
+    override def setIntKey(builder: rd.ORMapDeltaGroup.MapEntry.Builder, key: Int, value: dm.OtherMessage): rd.ORMapDeltaGroup.MapEntry = builder.setIntKey(key).setValue(value).build()
+    override def setOtherKey(builder: rd.ORMapDeltaGroup.MapEntry.Builder, key: dm.OtherMessage, value: dm.OtherMessage): rd.ORMapDeltaGroup.MapEntry = builder.setOtherKey(key).setValue(value).build()
+    override def hasStringKey(entry: rd.ORMapDeltaGroup.MapEntry): Boolean = entry.hasStringKey
+    override def getStringKey(entry: rd.ORMapDeltaGroup.MapEntry): String = entry.getStringKey
+    override def hasIntKey(entry: rd.ORMapDeltaGroup.MapEntry): Boolean = entry.hasIntKey
+    override def getIntKey(entry: rd.ORMapDeltaGroup.MapEntry): Int = entry.getIntKey
+    override def hasLongKey(entry: rd.ORMapDeltaGroup.MapEntry): Boolean = entry.hasLongKey
+    override def getLongKey(entry: rd.ORMapDeltaGroup.MapEntry): Long = entry.getLongKey
+    override def hasOtherKey(entry: rd.ORMapDeltaGroup.MapEntry): Boolean = entry.hasOtherKey
+    override def getOtherKey(entry: rd.ORMapDeltaGroup.MapEntry): OtherMessage = entry.getOtherKey
+    override def getValue(entry: rd.ORMapDeltaGroup.MapEntry): dm.OtherMessage = entry.getValue
+  }
+
 }
 
 /**
@@ -178,6 +194,11 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
   private val PNCounterKeyManifest = "g"
   private val ORMapManifest = "H"
   private val ORMapKeyManifest = "h"
+  private val ORMapPutManifest = "Ha"
+  private val ORMapRemoveManifest = "Hr"
+  private val ORMapRemoveKeyManifest = "Hk"
+  private val ORMapUpdateManifest = "Hu"
+  private val ORMapDeltaGroupManifest = "Hg"
   private val LWWMapManifest = "I"
   private val LWWMapKeyManifest = "i"
   private val PNCounterMapManifest = "J"
@@ -198,6 +219,11 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
     GCounterManifest → gcounterFromBinary,
     PNCounterManifest → pncounterFromBinary,
     ORMapManifest → ormapFromBinary,
+    ORMapPutManifest → ormapPutFromBinary,
+    ORMapRemoveManifest → ormapRemoveFromBinary,
+    ORMapRemoveKeyManifest → ormapRemoveKeyFromBinary,
+    ORMapUpdateManifest → ormapUpdateFromBinary,
+    ORMapDeltaGroupManifest → ormapDeltaGroupFromBinary,
     LWWMapManifest → lwwmapFromBinary,
     PNCounterMapManifest → pncountermapFromBinary,
     ORMultiMapManifest → multimapFromBinary,
@@ -225,6 +251,10 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
     case _: Flag                         ⇒ FlagManifest
     case _: LWWRegister[_]               ⇒ LWWRegisterManifest
     case _: ORMap[_, _]                  ⇒ ORMapManifest
+    case _: ORMap.PutDeltaOp[_, _]       ⇒ ORMapPutManifest
+    case _: ORMap.RemoveDeltaOp[_, _]    ⇒ ORMapRemoveManifest
+    case _: ORMap.RemoveKeyDeltaOp[_, _] ⇒ ORMapRemoveKeyManifest
+    case _: ORMap.UpdateDeltaOp[_, _]    ⇒ ORMapUpdateManifest
     case _: LWWMap[_, _]                 ⇒ LWWMapManifest
     case _: PNCounterMap[_]              ⇒ PNCounterMapManifest
     case _: ORMultiMap[_, _]             ⇒ ORMultiMapManifest
@@ -243,6 +273,7 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
     case _: ORMultiMapKey[_, _]          ⇒ ORMultiMapKeyManifest
 
     case _: ORSet.DeltaGroup[_]          ⇒ ORSetDeltaGroupManifest
+    case _: ORMap.DeltaGroup[_, _]       ⇒ ORMapDeltaGroupManifest
     case _: ORSet.FullStateDeltaOp[_]    ⇒ ORSetFullManifest
 
     case _ ⇒
@@ -259,6 +290,10 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
     case m: Flag                         ⇒ flagToProto(m).toByteArray
     case m: LWWRegister[_]               ⇒ lwwRegisterToProto(m).toByteArray
     case m: ORMap[_, _]                  ⇒ compress(ormapToProto(m))
+    case m: ORMap.PutDeltaOp[_, _]       ⇒ ormapPutToProto(m).toByteArray
+    case m: ORMap.RemoveDeltaOp[_, _]    ⇒ ormapRemoveToProto(m).toByteArray
+    case m: ORMap.RemoveKeyDeltaOp[_, _] ⇒ ormapRemoveKeyToProto(m).toByteArray
+    case m: ORMap.UpdateDeltaOp[_, _]    ⇒ ormapUpdateToProto(m).toByteArray
     case m: LWWMap[_, _]                 ⇒ compress(lwwmapToProto(m))
     case m: PNCounterMap[_]              ⇒ compress(pncountermapToProto(m))
     case m: ORMultiMap[_, _]             ⇒ compress(multimapToProto(m))
@@ -266,6 +301,7 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
     case m: VersionVector                ⇒ versionVectorToProto(m).toByteArray
     case Key(id)                         ⇒ keyIdToBinary(id)
     case m: ORSet.DeltaGroup[_]          ⇒ orsetDeltaGroupToProto(m).toByteArray
+    case m: ORMap.DeltaGroup[_, _]       ⇒ ormapDeltaGroupToProto(m).toByteArray
     case m: ORSet.FullStateDeltaOp[_]    ⇒ orsetToProto(m.underlying).toByteArray
     case _ ⇒
       throw new IllegalArgumentException(s"Can't serialize object of type ${obj.getClass} in [${getClass.getName}]")
@@ -512,8 +548,9 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
   }
 
   def ormapToProto(ormap: ORMap[_, _]): rd.ORMap = {
+    val ormapBuilder = rd.ORMap.newBuilder()
     val entries: jl.Iterable[rd.ORMap.Entry] = getEntries(ormap.values, rd.ORMap.Entry.newBuilder, otherMessageToProto)
-    rd.ORMap.newBuilder().setKeys(orsetToProto(ormap.keys)).addAllEntries(entries).build()
+    ormapBuilder.setKeys(orsetToProto(ormap.keys)).addAllEntries(entries).build()
   }
 
   def ormapFromBinary(bytes: Array[Byte]): ORMap[Any, ReplicatedData] =
@@ -536,9 +573,144 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
       entries)
   }
 
+  def singleMapEntryFromProto[PEntry <: GeneratedMessage, A <: GeneratedMessage, B <: ReplicatedData](entry: PEntry, valueCreator: A ⇒ B)(implicit eh: ProtoMapEntryReader[PEntry, A]): Map[Any, B] = {
+    val elem = if (eh.hasStringKey(entry)) Some(eh.getStringKey(entry) → valueCreator(eh.getValue(entry)))
+    else if (eh.hasIntKey(entry)) Some(eh.getIntKey(entry) → valueCreator(eh.getValue(entry)))
+    else if (eh.hasLongKey(entry)) Some(eh.getLongKey(entry) → valueCreator(eh.getValue(entry)))
+    else if (eh.hasOtherKey(entry)) Some(otherMessageFromProto(eh.getOtherKey(entry)) → valueCreator(eh.getValue(entry)))
+    else None
+    elem match {
+      case Some(e) ⇒ Map(e)
+      case _       ⇒ Map.empty[Any, B]
+    }
+  }
+
+  // wire protocol is always DeltaGroup
+  private def ormapPutFromBinary(bytes: Array[Byte]): ORMap.PutDeltaOp[Any, ReplicatedData] = {
+    val group = ormapDeltaGroupFromBinary(bytes)
+    if (group.ops.size == 1 && group.ops.head.isInstanceOf[ORMap.PutDeltaOp[_, _]])
+      group.ops.head.asInstanceOf[ORMap.PutDeltaOp[Any, ReplicatedData]]
+    else
+      throw new NotSerializableException("Improper ORMap delta put operation size or kind")
+  }
+
+  // wire protocol is always delta group
+  private def ormapRemoveFromBinary(bytes: Array[Byte]): ORMap.RemoveDeltaOp[Any, ReplicatedData] = {
+    val group = ormapDeltaGroupFromBinary(bytes)
+    if (group.ops.size == 1 && group.ops.head.isInstanceOf[ORMap.RemoveDeltaOp[_, _]])
+      group.ops.head.asInstanceOf[ORMap.RemoveDeltaOp[Any, ReplicatedData]]
+    else
+      throw new NotSerializableException("Improper ORMap delta remove operation size or kind")
+  }
+
+  // wire protocol is always delta group
+  private def ormapRemoveKeyFromBinary(bytes: Array[Byte]): ORMap.RemoveKeyDeltaOp[Any, ReplicatedData] = {
+    val group = ormapDeltaGroupFromBinary(bytes)
+    if (group.ops.size == 1 && group.ops.head.isInstanceOf[ORMap.RemoveKeyDeltaOp[_, _]])
+      group.ops.head.asInstanceOf[ORMap.RemoveKeyDeltaOp[Any, ReplicatedData]]
+    else
+      throw new NotSerializableException("Improper ORMap delta remove key operation size or kind")
+  }
+
+  // wire protocol is always delta group
+  private def ormapUpdateFromBinary(bytes: Array[Byte]): ORMap.UpdateDeltaOp[Any, ReplicatedDelta] = {
+    val group = ormapDeltaGroupFromBinary(bytes)
+    if (group.ops.size == 1 && group.ops.head.isInstanceOf[ORMap.UpdateDeltaOp[_, _]])
+      group.ops.head.asInstanceOf[ORMap.UpdateDeltaOp[Any, ReplicatedDelta]]
+    else
+      throw new NotSerializableException("Improper ORMap delta update operation size or kind")
+  }
+
+  // this can be made client-extendable in the same way as Http codes in Spray are
+  private def zeroTagFromCode(code: Int) = code match {
+    case ORMap.VanillaORMapTag.value              ⇒ ORMap.VanillaORMapTag
+    case ORMap.PNCounterMapTag.value              ⇒ ORMap.PNCounterMapTag
+    case ORMap.ORMultiMapTag.value                ⇒ ORMap.ORMultiMapTag
+    case ORMap.ORMultiMapWithValueDeltasTag.value ⇒ ORMap.ORMultiMapWithValueDeltasTag
+    case ORMap.LWWMapTag.value                    ⇒ ORMap.LWWMapTag
+    case _                                        ⇒ throw new IllegalArgumentException("Invalid ZeroTag code")
+  }
+
+  private def ormapDeltaGroupFromBinary(bytes: Array[Byte]): ORMap.DeltaGroup[Any, ReplicatedData] = {
+    val deltaGroup = rd.ORMapDeltaGroup.parseFrom(bytes)
+    val ops: Vector[ORMap.DeltaOp] =
+      deltaGroup.getEntriesList.asScala.map { entry ⇒
+        if (entry.getOperation == rd.ORMapDeltaOp.ORMapPut) {
+          val map = singleMapEntryFromProto(entry.getEntryData, (v: dm.OtherMessage) ⇒ otherMessageFromProto(v).asInstanceOf[ReplicatedData])
+          ORMap.PutDeltaOp(ORSet.AddDeltaOp(orsetFromProto(entry.getUnderlying)), map.head, zeroTagFromCode(entry.getZeroTag))
+        } else if (entry.getOperation == rd.ORMapDeltaOp.ORMapRemove) {
+          val map = singleMapEntryFromProto(entry.getEntryData, (v: dm.OtherMessage) ⇒ otherMessageFromProto(v).asInstanceOf[ReplicatedData])
+          ORMap.RemoveDeltaOp(ORSet.RemoveDeltaOp(orsetFromProto(entry.getUnderlying)), zeroTagFromCode(entry.getZeroTag))
+        } else if (entry.getOperation == rd.ORMapDeltaOp.ORMapRemoveKey) {
+          val map = singleMapEntryFromProto(entry.getEntryData, (v: dm.OtherMessage) ⇒ otherMessageFromProto(v).asInstanceOf[ReplicatedData])
+          ORMap.RemoveKeyDeltaOp(ORSet.RemoveDeltaOp(orsetFromProto(entry.getUnderlying)), map.keySet.head, zeroTagFromCode(entry.getZeroTag))
+        } else if (entry.getOperation == rd.ORMapDeltaOp.ORMapUpdate) {
+          val map = singleMapEntryFromProto(entry.getEntryData, (v: dm.OtherMessage) ⇒ otherMessageFromProto(v).asInstanceOf[ReplicatedDelta])
+          ORMap.UpdateDeltaOp(ORSet.AddDeltaOp(orsetFromProto(entry.getUnderlying)), map, zeroTagFromCode(entry.getZeroTag))
+        } else
+          throw new NotSerializableException(s"Unknown ORMap delta operation ${entry.getOperation}")
+      }(collection.breakOut)
+    ORMap.DeltaGroup(ops)
+  }
+
+  private def ormapPutToProto(addDelta: ORMap.PutDeltaOp[_, _]): rd.ORMapDeltaGroup = {
+    ormapDeltaGroupToProto(ORMap.DeltaGroup(scala.collection.immutable.IndexedSeq(addDelta.asInstanceOf[ORMap.DeltaOp])))
+  }
+
+  private def ormapRemoveToProto(addDelta: ORMap.RemoveDeltaOp[_, _]): rd.ORMapDeltaGroup = {
+    ormapDeltaGroupToProto(ORMap.DeltaGroup(scala.collection.immutable.IndexedSeq(addDelta.asInstanceOf[ORMap.DeltaOp])))
+  }
+
+  private def ormapRemoveKeyToProto(addDelta: ORMap.RemoveKeyDeltaOp[_, _]): rd.ORMapDeltaGroup = {
+    ormapDeltaGroupToProto(ORMap.DeltaGroup(scala.collection.immutable.IndexedSeq(addDelta.asInstanceOf[ORMap.DeltaOp])))
+  }
+
+  private def ormapUpdateToProto(addDelta: ORMap.UpdateDeltaOp[_, _]): rd.ORMapDeltaGroup = {
+    ormapDeltaGroupToProto(ORMap.DeltaGroup(scala.collection.immutable.IndexedSeq(addDelta.asInstanceOf[ORMap.DeltaOp])))
+  }
+
+  private def ormapDeltaGroupToProto(deltaGroup: ORMap.DeltaGroup[_, _]): rd.ORMapDeltaGroup = {
+    def createEntry(opType: rd.ORMapDeltaOp, u: ORSet[_], m: Map[_, _], zt: Int) = {
+      if (m.size > 1)
+        throw new IllegalArgumentException("Invalid size of ORMap delta map")
+      else {
+        val entryDataBuilder = rd.ORMapDeltaGroup.MapEntry.newBuilder()
+        m.headOption.map {
+          case (key: String, value) ⇒ entryDataBuilder.setStringKey(key).setValue(otherMessageToProto(value))
+          case (key: Int, value)    ⇒ entryDataBuilder.setIntKey(key).setValue(otherMessageToProto(value))
+          case (key: Long, value)   ⇒ entryDataBuilder.setLongKey(key).setValue(otherMessageToProto(value))
+          case (key, value)         ⇒ entryDataBuilder.setOtherKey(otherMessageToProto(key)).setValue(otherMessageToProto(value))
+        }
+        val builder = rd.ORMapDeltaGroup.Entry.newBuilder()
+          .setOperation(opType)
+          .setUnderlying(orsetToProto(u))
+          .setZeroTag(zt)
+        if (m.size > 0)
+          builder.setEntryData(entryDataBuilder.build())
+        builder
+      }
+    }
+
+    val b = rd.ORMapDeltaGroup.newBuilder()
+    deltaGroup.ops.foreach {
+      case ORMap.PutDeltaOp(op, pair, zt) ⇒
+        b.addEntries(createEntry(rd.ORMapDeltaOp.ORMapPut, op.asInstanceOf[ORSet.AddDeltaOp[_]].underlying, Map(pair), zt.value))
+      case ORMap.RemoveDeltaOp(op, zt) ⇒
+        b.addEntries(createEntry(rd.ORMapDeltaOp.ORMapRemove, op.asInstanceOf[ORSet.RemoveDeltaOp[_]].underlying, Map.empty, zt.value))
+      case ORMap.RemoveKeyDeltaOp(op, k, zt) ⇒
+        b.addEntries(createEntry(rd.ORMapDeltaOp.ORMapRemove, op.asInstanceOf[ORSet.RemoveDeltaOp[_]].underlying, Map(k → k), zt.value))
+      case ORMap.UpdateDeltaOp(op, m, zt) ⇒
+        b.addEntries(createEntry(rd.ORMapDeltaOp.ORMapUpdate, op.asInstanceOf[ORSet.AddDeltaOp[_]].underlying, m, zt.value))
+      case ORMap.DeltaGroup(u) ⇒
+        throw new IllegalArgumentException("ORMap.DeltaGroup should not be nested")
+    }
+    b.build()
+  }
+
   def lwwmapToProto(lwwmap: LWWMap[_, _]): rd.LWWMap = {
+    val lwwmapBuilder = rd.LWWMap.newBuilder()
     val entries: jl.Iterable[rd.LWWMap.Entry] = getEntries(lwwmap.underlying.entries, rd.LWWMap.Entry.newBuilder, lwwRegisterToProto)
-    rd.LWWMap.newBuilder().setKeys(orsetToProto(lwwmap.underlying.keys)).addAllEntries(entries).build()
+    lwwmapBuilder.setKeys(orsetToProto(lwwmap.underlying.keys)).addAllEntries(entries).build()
   }
 
   def lwwmapFromBinary(bytes: Array[Byte]): LWWMap[Any, Any] =
@@ -548,12 +720,13 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
     val entries = mapTypeFromProto(lwwmap.getEntriesList, lwwRegisterFromProto)
     new LWWMap(new ORMap(
       keys = orsetFromProto(lwwmap.getKeys),
-      entries))
+      entries, ORMap.LWWMapTag))
   }
 
   def pncountermapToProto(pncountermap: PNCounterMap[_]): rd.PNCounterMap = {
+    val pncountermapBuilder = rd.PNCounterMap.newBuilder()
     val entries: jl.Iterable[rd.PNCounterMap.Entry] = getEntries(pncountermap.underlying.entries, rd.PNCounterMap.Entry.newBuilder, pncounterToProto)
-    rd.PNCounterMap.newBuilder().setKeys(orsetToProto(pncountermap.underlying.keys)).addAllEntries(entries).build()
+    pncountermapBuilder.setKeys(orsetToProto(pncountermap.underlying.keys)).addAllEntries(entries).build()
   }
 
   def pncountermapFromBinary(bytes: Array[Byte]): PNCounterMap[_] =
@@ -563,12 +736,16 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
     val entries = mapTypeFromProto(pncountermap.getEntriesList, pncounterFromProto)
     new PNCounterMap(new ORMap(
       keys = orsetFromProto(pncountermap.getKeys),
-      entries))
+      entries, ORMap.PNCounterMapTag))
   }
 
   def multimapToProto(multimap: ORMultiMap[_, _]): rd.ORMultiMap = {
+    val ormultimapBuilder = rd.ORMultiMap.newBuilder()
     val entries: jl.Iterable[rd.ORMultiMap.Entry] = getEntries(multimap.underlying.entries, rd.ORMultiMap.Entry.newBuilder, orsetToProto)
-    rd.ORMultiMap.newBuilder().setKeys(orsetToProto(multimap.underlying.keys)).addAllEntries(entries).build()
+    ormultimapBuilder.setKeys(orsetToProto(multimap.underlying.keys)).addAllEntries(entries)
+    if (multimap.withValueDeltas)
+      ormultimapBuilder.setWithValueDeltas(true)
+    ormultimapBuilder.build()
   }
 
   def multimapFromBinary(bytes: Array[Byte]): ORMultiMap[Any, Any] =
@@ -576,9 +753,18 @@ class ReplicatedDataSerializer(val system: ExtendedActorSystem)
 
   def multimapFromProto(multimap: rd.ORMultiMap): ORMultiMap[Any, Any] = {
     val entries = mapTypeFromProto(multimap.getEntriesList, orsetFromProto)
-    new ORMultiMap(new ORMap(
+    val withValueDeltas = if (multimap.hasWithValueDeltas)
+      multimap.getWithValueDeltas
+    else false
+    new ORMultiMap(
+      new ORMap(
         keys = orsetFromProto(multimap.getKeys),
-      entries))
+        entries,
+        if (withValueDeltas)
+          ORMap.ORMultiMapWithValueDeltasTag
+        else
+          ORMap.ORMultiMapTag),
+      withValueDeltas)
   }
 
   def keyIdToBinary(id: String): Array[Byte] =

akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/ReplicatorMapDeltaSpec.scala

@@ -0,0 +1,281 @@
+/**
+ * Copyright (C) 2009-2016 Lightbend Inc. <http://www.lightbend.com>
+ */
+package akka.cluster.ddata
+
+import java.util.concurrent.ThreadLocalRandom
+
+import scala.concurrent.duration._
+
+import akka.cluster.Cluster
+import akka.cluster.ddata.Replicator._
+import akka.remote.testconductor.RoleName
+import akka.remote.testkit.MultiNodeConfig
+import akka.remote.testkit.MultiNodeSpec
+import akka.testkit._
+import com.typesafe.config.ConfigFactory
+
+object ReplicatorMapDeltaSpec extends MultiNodeConfig {
+  val first = role("first")
+  val second = role("second")
+  val third = role("third")
+  val fourth = role("fourth")
+
+  commonConfig(ConfigFactory.parseString("""
+    akka.loglevel = DEBUG
+    akka.actor.provider = "cluster"
+    akka.log-dead-letters-during-shutdown = off
+    """))
+
+  testTransport(on = true)
+
+  sealed trait Op
+  final case class Delay(n: Int) extends Op
+  final case class Incr(ki: (PNCounterMapKey[String], String), n: Int, consistency: WriteConsistency) extends Op
+  final case class Decr(ki: (PNCounterMapKey[String], String), n: Int, consistency: WriteConsistency) extends Op
+  final case class Add(ki: (ORMultiMapKey[String, String], String), elem: String, consistency: WriteConsistency) extends Op
+  final case class Remove(ki: (ORMultiMapKey[String, String], String), elem: String, consistency: WriteConsistency) extends Op
+
+  val timeout = 5.seconds
+  val writeTwo = WriteTo(2, timeout)
+  val writeMajority = WriteMajority(timeout)
+
+  val KeyPN = PNCounterMapKey[String]("A")
+  val KeyMM = ORMultiMapKey[String, String]("D")
+  val KeyA: (PNCounterMapKey[String], String) = (KeyPN, "a")
+  val KeyB: (PNCounterMapKey[String], String) = (KeyPN, "b")
+  val KeyC: (PNCounterMapKey[String], String) = (KeyPN, "c")
+  val KeyD: (ORMultiMapKey[String, String], String) = (KeyMM, "d")
+  val KeyE: (ORMultiMapKey[String, String], String) = (KeyMM, "e")
+  val KeyF: (ORMultiMapKey[String, String], String) = (KeyMM, "f")
+
+  def generateOperations(onNode: RoleName): Vector[Op] = {
+    val rnd = ThreadLocalRandom.current()
+
+    def consistency(): WriteConsistency = {
+      rnd.nextInt(100) match {
+        case n if n < 90  ⇒ WriteLocal
+        case n if n < 95  ⇒ writeTwo
+        case n if n < 100 ⇒ writeMajority
+      }
+    }
+
+    def rndPnCounterkey(): (PNCounterMapKey[String], String) = {
+      rnd.nextInt(3) match {
+        case 0 ⇒ KeyA
+        case 1 ⇒ KeyB
+        case 2 ⇒ KeyC
+      }
+    }
+
+    def rndOrSetkey(): (ORMultiMapKey[String, String], String) = {
+      rnd.nextInt(3) match {
+        case 0 ⇒ KeyD
+        case 1 ⇒ KeyE
+        case 2 ⇒ KeyF
+      }
+    }
+
+    var availableForRemove = Set.empty[String]
+
+    def rndAddElement(): String = {
+      // lower case a - j
+      val s = (97 + rnd.nextInt(10)).toChar.toString
+      availableForRemove += s
+      s
+    }
+
+    def rndRemoveElement(): String = {
+      if (availableForRemove.isEmpty)
+        "a"
+      else
+        availableForRemove.toVector(rnd.nextInt(availableForRemove.size))
+    }
+
+    (0 to (30 + rnd.nextInt(10))).map { _ ⇒
+      rnd.nextInt(4) match {
+        case 0 ⇒ Delay(rnd.nextInt(500))
+        case 1 ⇒ Incr(rndPnCounterkey(), rnd.nextInt(100), consistency())
+        case 2 ⇒ Decr(rndPnCounterkey(), rnd.nextInt(10), consistency())
+        case 3 ⇒
+          // ORSet
+          val key = rndOrSetkey()
+          // only removals for KeyF on node first
+          if (key == KeyF && onNode == first && rnd.nextBoolean())
+            Remove(key, rndRemoveElement(), consistency())
+          else
+            Add(key, rndAddElement(), consistency())
+      }
+    }.toVector
+  }
+
+}
+
+class ReplicatorMapDeltaSpecMultiJvmNode1 extends ReplicatorMapDeltaSpec
+class ReplicatorMapDeltaSpecMultiJvmNode2 extends ReplicatorMapDeltaSpec
+class ReplicatorMapDeltaSpecMultiJvmNode3 extends ReplicatorMapDeltaSpec
+class ReplicatorMapDeltaSpecMultiJvmNode4 extends ReplicatorMapDeltaSpec
+
+class ReplicatorMapDeltaSpec extends MultiNodeSpec(ReplicatorMapDeltaSpec) with STMultiNodeSpec with ImplicitSender {
+  import Replicator._
+  import ReplicatorMapDeltaSpec._
+
+  override def initialParticipants = roles.size
+
+  implicit val cluster = Cluster(system)
+  val fullStateReplicator = system.actorOf(Replicator.props(
+    ReplicatorSettings(system).withGossipInterval(1.second).withDeltaCrdtEnabled(false)), "fullStateReplicator")
+  val deltaReplicator = {
+    val r = system.actorOf(Replicator.props(ReplicatorSettings(system)), "deltaReplicator")
+    r ! Replicator.Internal.TestFullStateGossip(enabled = false)
+    r
+  }
+
+  var afterCounter = 0
+  def enterBarrierAfterTestStep(): Unit = {
+    afterCounter += 1
+    enterBarrier("after-" + afterCounter)
+  }
+
+  def join(from: RoleName, to: RoleName): Unit = {
+    runOn(from) {
+      cluster join node(to).address
+    }
+    enterBarrier(from.name + "-joined")
+  }
+
+  "delta-CRDT" must {
+    "join cluster" in {
+      join(first, first)
+      join(second, first)
+      join(third, first)
+      join(fourth, first)
+
+      within(15.seconds) {
+        awaitAssert {
+          fullStateReplicator ! GetReplicaCount
+          expectMsg(ReplicaCount(4))
+        }
+      }
+
+      enterBarrierAfterTestStep()
+    }
+
+    "propagate delta" in {
+      join(first, first)
+      join(second, first)
+      join(third, first)
+      join(fourth, first)
+
+      within(15.seconds) {
+        awaitAssert {
+          fullStateReplicator ! GetReplicaCount
+          expectMsg(ReplicaCount(4))
+        }
+      }
+      enterBarrier("ready")
+
+      runOn(first) {
+        // by setting something for each key we don't have to worry about NotFound
+        List(KeyA, KeyB, KeyC).foreach { key ⇒
+          fullStateReplicator ! Update(key._1, PNCounterMap.empty[String], WriteLocal)(_ increment key._2)
+          deltaReplicator ! Update(key._1, PNCounterMap.empty[String], WriteLocal)(_ increment key._2)
+        }
+        List(KeyD, KeyE, KeyF).foreach { key ⇒
+          fullStateReplicator ! Update(key._1, ORMultiMap.emptyWithValueDeltas[String, String], WriteLocal)(_ + (key._2, Set("a")))
+          deltaReplicator ! Update(key._1, ORMultiMap.emptyWithValueDeltas[String, String], WriteLocal)(_ + (key._2, Set("a")))
+        }
+      }
+      enterBarrier("updated-1")
+
+      within(5.seconds) {
+        awaitAssert {
+          val p = TestProbe()
+          List(KeyA, KeyB, KeyC).foreach { key ⇒
+            fullStateReplicator.tell(Get(key._1, ReadLocal), p.ref)
+            p.expectMsgType[GetSuccess[PNCounterMap[String]]].dataValue.get(key._2).get.intValue should be(1)
+          }
+        }
+        awaitAssert {
+          val p = TestProbe()
+          List(KeyD, KeyE, KeyF).foreach { key ⇒
+            fullStateReplicator.tell(Get(key._1, ReadLocal), p.ref)
+            p.expectMsgType[GetSuccess[ORMultiMap[String, String]]].dataValue.get(key._2) should ===(Some(Set("a")))
+          }
+        }
+      }
+
+      enterBarrierAfterTestStep()
+    }
+
+    "be eventually consistent" in {
+      val operations = generateOperations(onNode = myself)
+      log.debug(s"random operations on [${myself.name}]: ${operations.mkString(", ")}")
+      try {
+        // perform random operations with both delta and full-state replicators
+        // and compare that the end result is the same
+
+        for (op ← operations) {
+          log.debug("operation: {}", op)
+          op match {
+            case Delay(d) ⇒ Thread.sleep(d)
+            case Incr(key, n, consistency) ⇒
+              fullStateReplicator ! Update(key._1, PNCounterMap.empty[String], WriteLocal)(_ increment (key._2, n))
+              deltaReplicator ! Update(key._1, PNCounterMap.empty[String], WriteLocal)(_ increment (key._2, n))
+            case Decr(key, n, consistency) ⇒
+              fullStateReplicator ! Update(key._1, PNCounterMap.empty[String], WriteLocal)(_ decrement (key._2, n))
+              deltaReplicator ! Update(key._1, PNCounterMap.empty[String], WriteLocal)(_ decrement (key._2, n))
+            case Add(key, elem, consistency) ⇒
+              // to have an deterministic result when mixing add/remove we can only perform
+              // the ORSet operations from one node
+              runOn((if (key == KeyF) List(first) else List(first, second, third)): _*) {
+                fullStateReplicator ! Update(key._1, ORMultiMap.emptyWithValueDeltas[String, String], WriteLocal)(_ addBinding (key._2, elem))
+                deltaReplicator ! Update(key._1, ORMultiMap.emptyWithValueDeltas[String, String], WriteLocal)(_ addBinding (key._2, elem))
+              }
+            case Remove(key, elem, consistency) ⇒
+              runOn(first) {
+                fullStateReplicator ! Update(key._1, ORMultiMap.emptyWithValueDeltas[String, String], WriteLocal)(_ removeBinding (key._2, elem))
+                deltaReplicator ! Update(key._1, ORMultiMap.emptyWithValueDeltas[String, String], WriteLocal)(_ removeBinding (key._2, elem))
+              }
+          }
+        }
+
+        enterBarrier("updated-2")
+
+        List(KeyA, KeyB, KeyC).foreach { key ⇒
+          within(5.seconds) {
+            awaitAssert {
+              val p = TestProbe()
+              fullStateReplicator.tell(Get(key._1, ReadLocal), p.ref)
+              val fullStateValue = p.expectMsgType[GetSuccess[PNCounterMap[String]]].dataValue.get(key._2).get.intValue
+              deltaReplicator.tell(Get(key._1, ReadLocal), p.ref)
+              val deltaValue = p.expectMsgType[GetSuccess[PNCounterMap[String]]].dataValue.get(key._2).get.intValue
+              deltaValue should ===(fullStateValue)
+            }
+          }
+        }
+
+        List(KeyD, KeyE, KeyF).foreach { key ⇒
+          within(5.seconds) {
+            awaitAssert {
+              val p = TestProbe()
+              fullStateReplicator.tell(Get(key._1, ReadLocal), p.ref)
+              val fullStateValue = p.expectMsgType[GetSuccess[ORMultiMap[String, String]]].dataValue.get(key._2)
+              deltaReplicator.tell(Get(key._1, ReadLocal), p.ref)
+              val deltaValue = p.expectMsgType[GetSuccess[ORMultiMap[String, String]]].dataValue.get(key._2)
+              deltaValue should ===(fullStateValue)
+            }
+          }
+        }
+
+        enterBarrierAfterTestStep()
+      } catch {
+        case e: Throwable ⇒
+          info(s"random operations on [${myself.name}]: ${operations.mkString(", ")}")
+          throw e
+      }
+    }
+  }
+
+}
+

akka-distributed-data/src/multi-jvm/scala/akka/cluster/ddata/ReplicatorPruningSpec.scala

@@ -47,6 +47,8 @@ class ReplicatorPruningSpec extends MultiNodeSpec(ReplicatorPruningSpec) with ST
   val KeyA = GCounterKey("A")
   val KeyB = ORSetKey[String]("B")
   val KeyC = PNCounterMapKey[String]("C")
+  val KeyD = ORMultiMapKey[String, String]("D")
+  val KeyE = ORMapKey[String, GSet[String]]("E")
 
   def join(from: RoleName, to: RoleName): Unit = {
     runOn(from) {
@@ -89,6 +91,12 @@ class ReplicatorPruningSpec extends MultiNodeSpec(ReplicatorPruningSpec) with ST
       replicator ! Update(KeyC, PNCounterMap.empty[String], WriteAll(timeout)) { _ increment "x" increment "y" }
       expectMsg(UpdateSuccess(KeyC, None))
 
+      replicator ! Update(KeyD, ORMultiMap.empty[String, String], WriteAll(timeout)) { _ + ("a", Set("A")) }
+      expectMsg(UpdateSuccess(KeyD, None))
+
+      replicator ! Update(KeyE, ORMap.empty[String, GSet[String]], WriteAll(timeout)) { _ + ("a", GSet.empty[String].add("A")) }
+      expectMsg(UpdateSuccess(KeyE, None))
+
       enterBarrier("updates-done")
 
       replicator ! Get(KeyA, ReadLocal)
@@ -104,8 +112,24 @@ class ReplicatorPruningSpec extends MultiNodeSpec(ReplicatorPruningSpec) with ST
       oldMap.get("x") should be(Some(3))
       oldMap.get("y") should be(Some(3))
 
+      replicator ! Get(KeyD, ReadLocal)
+      val oldMultiMap = expectMsgType[GetSuccess[ORMultiMap[String, String]]].dataValue
+      oldMultiMap.get("a") should be(Some(Set("A")))
+
+      replicator ! Get(KeyE, ReadLocal)
+      val oldORMap = expectMsgType[GetSuccess[ORMap[String, GSet[String]]]].dataValue
+      val GSet(d) = oldORMap.entries("a")
+      d should be(Set("A"))
+
       enterBarrier("get-old")
 
+      runOn(third) {
+        replicator ! Update(KeyE, ORMap.empty[String, GSet[String]], WriteLocal) { _ - "a" }
+        expectMsg(UpdateSuccess(KeyE, None))
+      }
+
+      enterBarrier("remove-element")
+
       runOn(first) {
         cluster.leave(node(third).address)
       }
@@ -155,6 +179,25 @@ class ReplicatorPruningSpec extends MultiNodeSpec(ReplicatorPruningSpec) with ST
             }
           }
         }
+        within(5.seconds) {
+          awaitAssert {
+            replicator ! Get(KeyD, ReadLocal)
+            expectMsgPF() {
+              case g @ GetSuccess(KeyD, _) ⇒
+                g.get(KeyD).entries("a") should be(Set("A"))
+                g.get(KeyD).needPruningFrom(thirdUniqueAddress) should be(false)
+            }
+          }
+        }
+        within(5.seconds) {
+          awaitAssert {
+            replicator ! Get(KeyE, ReadLocal)
+            expectMsgPF() {
+              case g @ GetSuccess(KeyE, _) ⇒
+                g.get(KeyE).needPruningFrom(thirdUniqueAddress) should be(false)
+            }
+          }
+        }
       }
       enterBarrier("pruning-done")
 

akka-distributed-data/src/test/scala/akka/cluster/ddata/LWWMapSpec.scala

@@ -47,6 +47,27 @@ class LWWMapSpec extends WordSpec with Matchers {
       (m3 merge m4).entries should be(Map("a" → 1, "b" → 22, "c" → 3))
     }
 
+    "be able to work with deltas" in {
+      val m1 = LWWMap.empty.put(node1, "a", 1, defaultClock[Int]).put(node1, "b", 2, defaultClock[Int])
+      val m2 = LWWMap.empty.put(node2, "c", 3, defaultClock[Int])
+
+      val expected = Map("a" → 1, "b" → 2, "c" → 3)
+      (m1 merge m2).entries should be(expected)
+      (m2 merge m1).entries should be(expected)
+
+      LWWMap.empty.mergeDelta(m1.delta.get).mergeDelta(m2.delta.get).entries should be(expected)
+      LWWMap.empty.mergeDelta(m2.delta.get).mergeDelta(m1.delta.get).entries should be(expected)
+
+      val merged1 = m1 merge m2
+
+      val m3 = merged1.resetDelta.remove(node1, "b")
+      (merged1 mergeDelta m3.delta.get).entries should be(Map("a" → 1, "c" → 3))
+
+      // but if there is a conflicting update the entry is not removed
+      val m4 = merged1.resetDelta.put(node2, "b", 22, defaultClock[Int])
+      (m3 mergeDelta m4.delta.get).entries should be(Map("a" → 1, "b" → 22, "c" → 3))
+    }
+
     "have unapply extractor" in {
       val m1 = LWWMap.empty.put(node1, "a", 1L, defaultClock[Long])
       val LWWMap(entries1) = m1

akka-distributed-data/src/test/scala/akka/cluster/ddata/ORMapSpec.scala

@@ -30,12 +30,43 @@ class ORMapSpec extends WordSpec with Matchers {
 
     }
 
+    "be able to add entries with deltas" in {
+      val m = ORMap().put(node1, "a", GSet() + "A").put(node1, "b", GSet() + "B")
+      val md = m.delta.get
+
+      val m1 = ORMap().mergeDelta(md)
+
+      val GSet(a) = m1.entries("a")
+      a should be(Set("A"))
+      val GSet(b) = m1.entries("b")
+      b should be(Set("B"))
+
+      val m2 = m1.put(node1, "a", GSet() + "C")
+      val GSet(a2) = m2.entries("a")
+      a2 should be(Set("C"))
+
+    }
+
     "be able to remove entry" in {
       val m = ORMap().put(node1, "a", GSet() + "A").put(node1, "b", GSet() + "B").remove(node1, "a")
       m.entries.keySet should not contain ("a")
       m.entries.keySet should contain("b")
     }
 
+    "be able to remove entry using a delta" in {
+      val m = ORMap().put(node1, "a", GSet() + "A").put(node1, "b", GSet() + "B")
+      val addDelta = m.delta.get
+
+      val removeDelta = m.resetDelta.remove(node1, "a").delta.get
+
+      val m1 = ORMap().mergeDelta(addDelta)
+      m1.entries.keySet should contain("a")
+
+      val m2 = m1.mergeDelta(removeDelta)
+      m2.entries.keySet should not contain ("a")
+      m2.entries.keySet should contain("b")
+    }
+
     "be able to add removed" in {
       val m = ORMap().put(node1, "a", GSet() + "A").put(node1, "b", GSet() + "B").remove(node1, "a")
       m.entries.keySet should not contain ("a")
@@ -110,6 +141,330 @@ class ORMapSpec extends WordSpec with Matchers {
       merged2.entries("c").elements should be(Set("C"))
     }
 
+    "not have anomalies for remove+updated scenario and deltas" in {
+      val m1 = ORMap.empty.put(node1, "a", GSet.empty + "A").put(node1, "b", GSet.empty + "B")
+      val m2 = ORMap.empty.put(node2, "c", GSet.empty + "C")
+
+      val merged1 = m1 merge m2
+
+      val m3 = merged1.resetDelta.remove(node1, "b")
+      val m4 = merged1.resetDelta.updated(node1, "b", GSet.empty[String])(_.add("B2"))
+
+      val merged2 = m3 merge m4
+
+      merged2.entries("a").elements should be(Set("A"))
+      // note that B is included, because GSet("B") is merged with GSet("B2")
+      merged2.entries("b").elements should be(Set("B", "B2"))
+      merged2.entries("c").elements should be(Set("C"))
+
+      val merged3 = m3 mergeDelta m4.delta.get
+
+      merged3.entries("a").elements should be(Set("A"))
+      // note that B is included, because GSet("B") is merged with GSet("B2")
+      merged3.entries("b").elements should be(Set("B", "B2"))
+      merged3.entries("c").elements should be(Set("C"))
+    }
+
+    "not have anomalies for remove+updated scenario and deltas 2" in {
+      val m1 = ORMap.empty.put(node1, "a", ORSet.empty.add(node1, "A")).put(node1, "b", ORSet.empty.add(node1, "B"))
+      val m2 = ORMap.empty.put(node2, "c", ORSet.empty.add(node2, "C"))
+
+      val merged1 = m1 merge m2
+
+      val m3 = merged1.resetDelta.remove(node1, "b")
+      val m4 = merged1.resetDelta.remove(node1, "b").updated(node1, "b", ORSet.empty[String])(_.add(node1, "B2"))
+
+      val merged2 = m3 merge m4
+
+      merged2.entries("a").elements should be(Set("A"))
+      // note that B is not included, because it was removed in both timelines
+      merged2.entries("b").elements should be(Set("B2"))
+      merged2.entries("c").elements should be(Set("C"))
+
+      val merged3 = m3 mergeDelta m4.delta.get
+
+      merged3.entries("a").elements should be(Set("A"))
+      // note that B is not included, because it was removed in both timelines
+      merged3.entries("b").elements should be(Set("B2"))
+      merged3.entries("c").elements should be(Set("C"))
+    }
+
+    "not have anomalies for remove+updated scenario and deltas 3" in {
+      val m1 = ORMap.empty.put(node1, "a", ORSet.empty.add(node1, "A")).put(node1, "b", ORSet.empty.add(node1, "B"))
+      val m2 = ORMap.empty.put(node2, "c", ORSet.empty.add(node2, "C"))
+
+      val merged1 = m1 merge m2
+
+      val m3 = merged1.resetDelta.remove(node1, "b")
+      val m4 = merged1.resetDelta.remove(node2, "b").updated(node2, "b", ORSet.empty[String])(_.add(node2, "B2"))
+
+      val merged2 = m3 merge m4
+
+      merged2.entries("a").elements should be(Set("A"))
+      // note that B is not included, because it was removed in both timelines
+      merged2.entries("b").elements should be(Set("B2"))
+      merged2.entries("c").elements should be(Set("C"))
+
+      val merged3 = m3 mergeDelta m4.delta.get
+
+      merged3.entries("a").elements should be(Set("A"))
+      // note that B is not included, because it was removed in both timelines
+      merged3.entries("b").elements should be(Set("B2"))
+      merged3.entries("c").elements should be(Set("C"))
+    }
+
+    "not have anomalies for remove+updated scenario and deltas 4" in {
+      val m1 = ORMap.empty.put(node1, "a", ORSet.empty.add(node1, "A")).put(node1, "b", ORSet.empty.add(node1, "B"))
+      val m2 = ORMap.empty.put(node2, "c", ORSet.empty.add(node2, "C"))
+
+      val merged1 = m1 merge m2
+
+      val m3 = merged1.resetDelta.remove(node1, "b")
+      val m4 = merged1.resetDelta.updated(node1, "b", ORSet.empty[String])(_.add(node1, "B2"))
+
+      val merged2 = m3 merge m4
+
+      merged2.entries("a").elements should be(Set("A"))
+      // note that B is included, because ORSet("B") is merged with ORSet("B2")
+      merged2.entries("b").elements should be(Set("B", "B2"))
+      merged2.entries("c").elements should be(Set("C"))
+
+      val merged3 = m3 mergeDelta m4.delta.get
+
+      merged3.entries("a").elements should be(Set("A"))
+      // note that B is included, because ORSet("B") is merged with ORSet("B2")
+      merged3.entries("b").elements should be(Set("B", "B2"))
+      merged3.entries("c").elements should be(Set("C"))
+    }
+
+    "not have anomalies for remove+updated scenario and deltas 5" in {
+      val m1 = ORMap.empty.put(node1, "a", GSet.empty + "A").put(node1, "b", GSet.empty + "B")
+      val m2 = ORMap.empty.put(node2, "c", GSet.empty + "C")
+
+      val merged1 = m1 merge m2
+
+      val m3 = merged1.resetDelta.remove(node1, "b")
+      val m4 = merged1.resetDelta.put(node2, "b", GSet.empty + "B2")
+
+      val merged2 = m3 merge m4
+
+      merged2.entries("a").elements should be(Set("A"))
+      // note that B is not included, because it was removed in both timelines
+      merged2.entries("b").elements should be(Set("B2"))
+      merged2.entries("c").elements should be(Set("C"))
+
+      val merged3 = m3 mergeDelta m4.delta.get
+
+      merged3.entries("a").elements should be(Set("A"))
+      // note that B is not included, because it was removed in both timelines
+      merged3.entries("b").elements should be(Set("B2"))
+      merged3.entries("c").elements should be(Set("C"))
+    }
+
+    "not have anomalies for remove+updated scenario and deltas 6" in {
+      val m1 = ORMap.empty.put(node1, "a", ORSet.empty.add(node1, "A")).put(node1, "b", ORSet.empty.add(node1, "B"))
+      val m2 = ORMap.empty.put(node2, "b", ORSet.empty.add(node2, "B3"))
+
+      val merged1 = m1 merge m2
+
+      val m3 = merged1.resetDelta.remove(node1, "b")
+      val m4 = merged1.resetDelta.remove(node2, "b").updated(node2, "b", ORSet.empty[String])(_.add(node2, "B1"))
+        .updated(node2, "b", ORSet.empty[String])(_.add(node2, "B2"))
+
+      val merged2 = m3 merge m4
+
+      merged2.entries("a").elements should be(Set("A"))
+      // note that B is not included, because it was removed in both timelines
+      merged2.entries("b").elements should be(Set("B1", "B2"))
+
+      val merged3 = m3 mergeDelta m4.delta.get
+
+      merged3.entries("a").elements should be(Set("A"))
+      // note that B is not included, because it was removed in both timelines
+      merged3.entries("b").elements should be(Set("B1", "B2"))
+    }
+
+    "not have anomalies for remove+updated scenario and deltas 7" in {
+      val m1 = ORMap.empty.put(node1, "a", ORSet.empty.add(node1, "A"))
+        .put(node1, "b", ORSet.empty.add(node1, "B1")).remove(node1, "b")
+      val m2 = ORMap.empty.put(node1, "a", ORSet.empty.add(node1, "A")).put(node1, "b", ORSet.empty.add(node1, "B2"))
+      val m2d = m2.resetDelta.remove(node1, "b")
+      val m2u = m2.resetDelta.updated(node1, "b", ORSet.empty[String])(_.add(node1, "B3"))
+        .updated(node2, "b", ORSet.empty[String])(_.add(node2, "B4"))
+
+      val merged1 = (m1 merge m2d) mergeDelta m2u.delta.get
+
+      merged1.entries("a").elements should be(Set("A"))
+      // note that B1 is lost as it was added and removed earlier in timeline than B2
+      merged1.entries("b").elements should be(Set("B2", "B3", "B4"))
+    }
+
+    "not have anomalies for remove+updated scenario and deltas 8" in {
+      val m1 = ORMap.empty.put(node1, "a", GSet.empty + "A")
+        .put(node1, "b", GSet.empty + "B").put(node2, "b", GSet.empty + "B")
+      val m2 = ORMap.empty.put(node2, "c", GSet.empty + "C")
+
+      val merged1 = m1 merge m2
+
+      val m3 = merged1.resetDelta.remove(node1, "b").remove(node2, "b")
+      val m4 = merged1.resetDelta.put(node2, "b", GSet.empty + "B2").put(node2, "b", GSet.empty + "B3")
+
+      val merged2 = m3 merge m4
+
+      merged2.entries("a").elements should be(Set("A"))
+      merged2.entries("b").elements should be(Set("B3"))
+      merged2.entries("c").elements should be(Set("C"))
+
+      val merged3 = (merged1 mergeDelta m3.delta.get) mergeDelta m4.delta.get
+
+      merged3.entries("a").elements should be(Set("A"))
+      merged3.entries("b").elements should be(Set("B3"))
+      merged3.entries("c").elements should be(Set("C"))
+    }
+
+    "not have anomalies for remove+updated scenario and deltas 9" in {
+      val m1 = ORMap.empty.put(node1, "a", GSet.empty + "A")
+        .put(node1, "b", GSet.empty + "B").put(node2, "b", GSet.empty + "B")
+      val m2 = ORMap.empty.put(node2, "c", GSet.empty + "C")
+
+      val merged1 = m1 merge m2
+
+      val m3 = merged1.resetDelta.remove(node1, "b").remove(node2, "b")
+      val m4 = merged1.resetDelta.updated(node2, "b", GSet.empty[String])(_.add("B2"))
+        .updated(node2, "b", GSet.empty[String])(_.add("B3"))
+
+      val merged2 = m3 merge m4
+
+      merged2.entries("a").elements should be(Set("A"))
+      merged2.entries("b").elements should be(Set("B2", "B3"))
+      merged2.entries("c").elements should be(Set("C"))
+
+      val merged3 = (merged1 mergeDelta m3.delta.get) mergeDelta m4.delta.get
+
+      merged3.entries("a").elements should be(Set("A"))
+      merged3.entries("b").elements should be(Set("B2", "B3"))
+      merged3.entries("c").elements should be(Set("C"))
+    }
+
+    "not have anomalies for remove+updated scenario and deltas 10" in {
+      val m1 = ORMap.empty.put(node2, "a", GSet.empty + "A")
+        .put(node2, "b", GSet.empty + "B")
+
+      val m3 = m1.resetDelta.remove(node2, "b")
+      val m4 = m3.resetDelta.put(node2, "b", GSet.empty + "B2").updated(node2, "b", GSet.empty[String])(_.add("B3"))
+
+      val merged2 = m3 merge m4
+
+      merged2.entries("a").elements should be(Set("A"))
+      merged2.entries("b").elements should be(Set("B2", "B3"))
+
+      val merged3 = m3 mergeDelta m4.delta.get
+
+      merged3.entries("a").elements should be(Set("A"))
+      merged3.entries("b").elements should be(Set("B2", "B3"))
+    }
+
+    "have the usual anomalies for remove+updated scenario" in {
+      // please note that the current ORMultiMap has the same anomaly
+      // because the condition of keeping global vvector is violated
+      // by removal of the whole entry for the removed key "b" which results in removal of it's value's vvector
+      val m1 = ORMap.empty.put(node1, "a", ORSet.empty.add(node1, "A")).put(node1, "b", ORSet.empty.add(node1, "B"))
+      val m2 = ORMap.empty.put(node2, "c", ORSet.empty.add(node2, "C"))
+
+      // m1 - node1 gets the update from m2
+      val merged1 = m1 merge m2
+      // m2 - node2 gets the update from m1
+      val merged2 = m2 merge m1
+
+      // RACE CONDITION ahead!
+      val m3 = merged1.resetDelta.remove(node1, "b")
+      // let's imagine that m3 (node1) update gets propagated here (full state or delta - doesn't matter)
+      // and is in flight, but in the meantime, an element is being added somewhere else (m4 - node2)
+      // and the update is propagated before the update from node1 is merged
+      val m4 = merged2.resetDelta.updated(node2, "b", ORSet.empty[String])(_.add(node2, "B2"))
+      // and later merged on node1
+      val merged3 = m3 merge m4
+      // and the other way round...
+      val merged4 = m4 merge m3
+
+      // result -  the element "B" is kept on both sides...
+      merged3.entries("a").elements should be(Set("A"))
+      merged3.entries("b").elements should be(Set("B", "B2"))
+      merged3.entries("c").elements should be(Set("C"))
+
+      merged4.entries("a").elements should be(Set("A"))
+      merged4.entries("b").elements should be(Set("B", "B2"))
+      merged4.entries("c").elements should be(Set("C"))
+
+      // but if the timing was slightly different, so that the update from node1
+      // would get merged just before update on node2:
+      val merged5 = (m2 merge m3).resetDelta.updated(node2, "b", ORSet.empty[String])(_.add(node2, "B2"))
+      // the update propagated ... and merged on node1:
+      val merged6 = m3 merge merged5
+
+      // then the outcome is different... because the vvector of value("b") was lost...
+      merged5.entries("a").elements should be(Set("A"))
+      // this time it's different...
+      merged5.entries("b").elements should be(Set("B2"))
+      merged5.entries("c").elements should be(Set("C"))
+
+      merged6.entries("a").elements should be(Set("A"))
+      // this time it's different...
+      merged6.entries("b").elements should be(Set("B2"))
+      merged6.entries("c").elements should be(Set("C"))
+    }
+
+    "work with deltas and updated for GSet elements type" in {
+      val m1 = ORMap.empty.put(node1, "a", GSet.empty + "A")
+      val m2 = m1.resetDelta.updated(node1, "a", GSet.empty[String])(_.add("B"))
+      val m3 = ORMap().mergeDelta(m1.delta.get).mergeDelta(m2.delta.get)
+      val GSet(d3) = m3.entries("a")
+      d3 should be(Set("A", "B"))
+    }
+
+    "work with deltas and updated for ORSet elements type" in {
+      val m1 = ORMap.empty.put(node1, "a", ORSet.empty.add(node1, "A"))
+      val m2 = m1.resetDelta.updated(node1, "a", ORSet.empty[String])(_.add(node1, "B"))
+      val m3 = ORMap().mergeDelta(m1.delta.get).mergeDelta(m2.delta.get)
+
+      val ORSet(d3) = m3.entries("a")
+      d3 should be(Set("A", "B"))
+    }
+
+    "work with aggregated deltas and updated for GSet elements type" in {
+      val m1 = ORMap.empty.put(node1, "a", GSet.empty + "A")
+      val m2 = m1.resetDelta.updated(node1, "a", GSet.empty[String])(_.add("B")).updated(node1, "a", GSet.empty[String])(_.add("C"))
+      val m3 = ORMap().mergeDelta(m1.delta.get).mergeDelta(m2.delta.get)
+      val GSet(d3) = m3.entries("a")
+      d3 should be(Set("A", "B", "C"))
+    }
+
+    "work with deltas and updated for GCounter elements type" in {
+      val m1 = ORMap.empty.put(node1, "a", GCounter.empty)
+      val m2 = m1.resetDelta.updated(node1, "a", GCounter.empty)(_.increment(node1, 10))
+      val m3 = m2.resetDelta.updated(node2, "a", GCounter.empty)(_.increment(node2, 10))
+      val m4 = ORMap().mergeDelta(m1.delta.get).mergeDelta(m2.delta.get).mergeDelta(m3.delta.get)
+      val GCounter(num) = m4.entries("a")
+      num should ===(20)
+    }
+
+    "work with deltas and updated for PNCounter elements type" in {
+      val m1 = ORMap.empty.put(node1, "a", PNCounter.empty)
+      val m2 = m1.resetDelta.updated(node1, "a", PNCounter.empty)(_.increment(node1, 10))
+      val m3 = m2.resetDelta.updated(node2, "a", PNCounter.empty)(_.decrement(node2, 10))
+      val m4 = ORMap().mergeDelta(m1.delta.get).mergeDelta(m2.delta.get).mergeDelta(m3.delta.get)
+      val PNCounter(num) = m4.entries("a")
+      num should ===(0)
+    }
+
+    "work with deltas and updated for Flag elements type" in {
+      val m1 = ORMap.empty.put(node1, "a", Flag(false))
+      val m2 = m1.resetDelta.updated(node1, "a", Flag.empty)(_.switchOn)
+      val m3 = ORMap().mergeDelta(m1.delta.get).mergeDelta(m2.delta.get)
+      val Flag(d3) = m3.entries("a")
+      d3 should be(true)
+    }
+
     "not allow put for ORSet elements type" in {
       val m = ORMap().put(node1, "a", ORSet().add(node1, "A"))
 

akka-distributed-data/src/test/scala/akka/cluster/ddata/ORMultiMapSpec.scala

@@ -77,6 +77,12 @@ class ORMultiMapSpec extends WordSpec with Matchers {
 
       val merged2 = m2 merge m1
       merged2.entries should be(expectedMerged)
+
+      val merged3 = m1 mergeDelta m2.delta.get
+      merged3.entries should be(expectedMerged)
+
+      val merged4 = m2 mergeDelta m1.delta.get
+      merged4.entries should be(expectedMerged)
     }
   }
 
@@ -107,6 +113,74 @@ class ORMultiMapSpec extends WordSpec with Matchers {
     m2.entries should be(Map("b" → Set("B1")))
   }
 
+  "not have usual anomalies for remove+addBinding scenario and delta-deltas" in {
+    val m1 = ORMultiMap.emptyWithValueDeltas[String, String].put(node1, "a", Set("A")).put(node1, "b", Set("B"))
+    val m2 = ORMultiMap.emptyWithValueDeltas[String, String].put(node2, "c", Set("C"))
+
+    val merged1 = m1 merge m2
+
+    val m3 = merged1.resetDelta.remove(node1, "b")
+    val m4 = merged1.resetDelta.addBinding(node1, "b", "B2")
+
+    val merged2 = m3 merge m4
+
+    merged2.entries("a") should be(Set("A"))
+    merged2.entries("b") should be(Set("B2"))
+    merged2.entries("c") should be(Set("C"))
+
+    val merged3 = m3 mergeDelta m4.delta.get
+
+    merged3.entries("a") should be(Set("A"))
+    merged3.entries("b") should be(Set("B2"))
+    merged3.entries("c") should be(Set("C"))
+  }
+
+  "not have usual anomalies for remove+addBinding scenario and delta-deltas 2" in {
+    // the new delta-delta ORMultiMap is free from this anomaly
+    val m1 = ORMultiMap.emptyWithValueDeltas[String, String].put(node1, "a", Set("A")).put(node1, "b", Set("B"))
+    val m2 = ORMultiMap.emptyWithValueDeltas[String, String].put(node2, "c", Set("C"))
+
+    // m1 - node1 gets the update from m2
+    val merged1 = m1 merge m2
+    // m2 - node2 gets the update from m1
+    val merged2 = m2 merge m1
+
+    // no race condition
+    val m3 = merged1.resetDelta.remove(node1, "b")
+    // let's imagine that m3 (node1) update gets propagated here (full state or delta - doesn't matter)
+    // and is in flight, but in the meantime, an element is being added somewhere else (m4 - node2)
+    // and the update is propagated before the update from node1 is merged
+    val m4 = merged2.resetDelta.addBinding(node2, "b", "B2")
+    // and later merged on node1
+    val merged3 = m3 merge m4
+    // and the other way round...
+    val merged4 = m4 merge m3
+
+    // result -  the element "B" is kept on both sides...
+    merged3.entries("a") should be(Set("A"))
+    merged3.entries("b") should be(Set("B2"))
+    merged3.entries("c") should be(Set("C"))
+
+    merged4.entries("a") should be(Set("A"))
+    merged4.entries("b") should be(Set("B2"))
+    merged4.entries("c") should be(Set("C"))
+
+    // but if the timing was slightly different, so that the update from node1
+    // would get merged just before update on node2:
+    val merged5 = (m2 merge m3).resetDelta.addBinding(node2, "b", "B2")
+    // the update propagated ... and merged on node1:
+    val merged6 = m3 merge merged5
+
+    // then the outcome would be the same...
+    merged5.entries("a") should be(Set("A"))
+    merged5.entries("b") should be(Set("B2"))
+    merged5.entries("c") should be(Set("C"))
+
+    merged6.entries("a") should be(Set("A"))
+    merged6.entries("b") should be(Set("B2"))
+    merged6.entries("c") should be(Set("C"))
+  }
+
   "have unapply extractor" in {
     val m1 = ORMultiMap.empty.put(node1, "a", Set(1L, 2L)).put(node2, "b", Set(3L))
     val m2: ORMultiMap[String, Long] = m1

akka-distributed-data/src/test/scala/akka/cluster/ddata/PNCounterMapSpec.scala

@@ -46,6 +46,24 @@ class PNCounterMapSpec extends WordSpec with Matchers {
       (m3 merge m4).entries should be(Map("a" → 1, "b" → 13, "c" → 7))
     }
 
+    "be able to work with deltas" in {
+      val m1 = PNCounterMap().increment(node1, "a", 1).increment(node1, "b", 3).increment(node1, "c", 2)
+      val m2 = PNCounterMap().increment(node2, "c", 5)
+
+      val expected = Map("a" → 1, "b" → 3, "c" → 7)
+      (PNCounterMap() mergeDelta m1.delta.get mergeDelta m2.delta.get).entries should be(expected)
+      (PNCounterMap() mergeDelta m2.delta.get mergeDelta m1.delta.get).entries should be(expected)
+
+      val merged1 = m1 merge m2
+
+      val m3 = merged1.resetDelta.remove(node1, "b")
+      (merged1 mergeDelta m3.delta.get).entries should be(Map("a" → 1, "c" → 7))
+
+      // but if there is a conflicting update the entry is not removed
+      val m4 = merged1.resetDelta.increment(node2, "b", 10)
+      (m3 mergeDelta m4.delta.get).entries should be(Map("a" → 1, "b" → 13, "c" → 7))
+    }
+
     "have unapply extractor" in {
       val m1 = PNCounterMap.empty.increment(node1, "a", 1).increment(node2, "b", 2)
       val PNCounterMap(entries1) = m1

akka-distributed-data/src/test/scala/akka/cluster/ddata/protobuf/ReplicatedDataSerializerSpec.scala

@@ -184,6 +184,23 @@ class ReplicatedDataSerializerSpec extends TestKit(ActorSystem(
       checkSerialization(ORMap().put(address1, Flag(), GSet() + "A"))
     }
 
+    "serialize ORMap delta" in {
+      checkSerialization(ORMap().put(address1, "a", GSet() + "A").put(address2, "b", GSet() + "B").delta.get)
+      checkSerialization(ORMap().put(address1, "a", GSet() + "A").resetDelta.remove(address2, "a").delta.get)
+      checkSerialization(ORMap().put(address1, "a", GSet() + "A").remove(address2, "a").delta.get)
+      checkSerialization(ORMap().put(address1, 1, GSet() + "A").delta.get)
+      checkSerialization(ORMap().put(address1, 1L, GSet() + "A").delta.get)
+      checkSerialization(ORMap.empty[String, ORSet[String]]
+        .put(address1, "a", ORSet.empty[String].add(address1, "A"))
+        .put(address2, "b", ORSet.empty[String].add(address2, "B"))
+        .updated(address1, "a", ORSet.empty[String])(_.add(address1, "C")).delta.get)
+      checkSerialization(ORMap.empty[String, ORSet[String]]
+        .resetDelta
+        .updated(address1, "a", ORSet.empty[String])(_.add(address1, "C")).delta.get)
+      // use Flag for this test as object key because it is serializable
+      checkSerialization(ORMap().put(address1, Flag(), GSet() + "A").delta.get)
+    }
+
     "be compatible with old ORMap serialization" in {
       // Below blob was created with previous version of the serializer
       val oldBlobAsBase64 = "H4sIAAAAAAAAAOOax8jlyaXMJc8lzMWXX5KRWqSXkV9copdflC7wXEWUiYGBQRaIGQQkuJS45LiEuHiL83NTUdQwwtWIC6kQpUqVKAulGBOlGJOE+LkYE4W4uJi5GB0FuJUYnUACSRABJ7AAAOLO3C3DAAAA"
@@ -244,6 +261,22 @@ class ReplicatedDataSerializerSpec extends TestKit(ActorSystem(
       checkCompatibility(oldBlobAsBase64, ORMultiMap())
     }
 
+    "serialize ORMultiMap withValueDeltas" in {
+      checkSerialization(ORMultiMap._emptyWithValueDeltas)
+      checkSerialization(ORMultiMap._emptyWithValueDeltas.addBinding(address1, "a", "A"))
+      checkSerialization(ORMultiMap._emptyWithValueDeltas.addBinding(address1, 1, "A"))
+      checkSerialization(ORMultiMap._emptyWithValueDeltas.addBinding(address1, 1L, "A"))
+      checkSerialization(ORMultiMap._emptyWithValueDeltas.addBinding(address1, Flag(), "A"))
+      checkSerialization(ORMultiMap.emptyWithValueDeltas[String, String]
+        .addBinding(address1, "a", "A1")
+        .put(address2, "b", Set("B1", "B2", "B3"))
+        .addBinding(address2, "a", "A2"))
+
+      val m1 = ORMultiMap.emptyWithValueDeltas[String, String].addBinding(address1, "a", "A1").addBinding(address2, "a", "A2")
+      val m2 = ORMultiMap.emptyWithValueDeltas[String, String].put(address2, "b", Set("B1", "B2", "B3"))
+      checkSameContent(m1.merge(m2), m2.merge(m1))
+    }
+
     "serialize DeletedData" in {
       checkSerialization(DeletedData)
     }