Pass the semigroup to create Mergeable online

summingbird-online/src/main/scala/com/twitter/summingbird/online/CombinedServiceStoreFactory.scala

@@ -28,22 +28,19 @@ object CombinedServiceStoreFactory {
   def apply[K, V](onlineStore: => MergeableStore[(K, BatchID), V], batchesToKeep: Int)(implicit b: Batcher) = {
 
     lazy val clientStore = ClientStore[K, V](onlineStore, batchesToKeep)(b, onlineStore.semigroup)
-
-    new CombinedServiceStoreFactory[K, V] {
-      def mergeableStore = () => onlineStore
-      def mergeableBatcher = b
-      def serviceStore = () => clientStore
-    }
+    from[K, V](clientStore, onlineStore, b)
   }
 
   def apply[K, V](offlineStore: => ReadableStore[K, (BatchID, V)], onlineStore: => MergeableStore[(K, BatchID), V], batchesToKeep: Int)(implicit b: Batcher) = {
 
     lazy val clientStore = ClientStore[K, V](offlineStore, onlineStore, batchesToKeep)(b, onlineStore.semigroup)
+    from[K, V](clientStore, onlineStore, b)
+  }
 
+  private[this] def from[K, V](cs: => ReadableStore[K, V], online: => MergeableStore[(K, BatchID), V], b: Batcher): CombinedServiceStoreFactory[K, V] =
     new CombinedServiceStoreFactory[K, V] {
-      def mergeableStore = () => onlineStore
+      def mergeableStore = _ => online
       def mergeableBatcher = b
-      def serviceStore = () => clientStore
+      def serviceStore = () => cs
     }
-  }
-}
+}

summingbird-online/src/main/scala/com/twitter/summingbird/online/MergeableStoreFactory.scala

@@ -16,6 +16,7 @@
 
 package com.twitter.summingbird.online
 
+import com.twitter.algebird.Semigroup
 import com.twitter.storehaus.algebra.{ MergeableStore, Mergeable, StoreAlgebra }
 import com.twitter.summingbird.batch.{ Batcher, BatchID }
 
@@ -26,23 +27,28 @@ import com.twitter.summingbird.batch.{ Batcher, BatchID }
  */
 object MergeableStoreFactory {
 
-  def apply[K, V](store: () => Mergeable[K, V], batcher: Batcher) = {
+  def apply[K, V](store: () => Mergeable[K, V], batcher: Batcher) =
+    fromWithSemigroup { _: Semigroup[V] => store() }(batcher)
+
+  def from[K, V](store: => Mergeable[(K, BatchID), V])(implicit batcher: Batcher): MergeableStoreFactory[(K, BatchID), V] =
+    fromWithSemigroup(_ => store)
+
+  def fromWithSemigroup[K, V](fn: Semigroup[V] => Mergeable[K, V])(implicit batcher: Batcher): MergeableStoreFactory[K, V] =
     new MergeableStoreFactory[K, V] {
-      def mergeableStore = store
+      def mergeableStore = fn
       def mergeableBatcher = batcher
     }
-  }
-
-  def from[K, V](store: => Mergeable[(K, BatchID), V])(implicit batcher: Batcher): MergeableStoreFactory[(K, BatchID), V] =
-    apply({ () => store }, batcher)
 
   def fromOnlineOnly[K, V](store: => MergeableStore[K, V]): MergeableStoreFactory[(K, BatchID), V] = {
     implicit val batcher = Batcher.unit
     from(store.convert { k: (K, BatchID) => k._1 })
   }
+
+  def fromOnlineOnlyWithSemigroup[K, V](fn: Semigroup[V] => MergeableStore[K, V]): MergeableStoreFactory[(K, BatchID), V] =
+    fromWithSemigroup(fn.andThen(_.convert { k: (K, BatchID) => k._1 }))(Batcher.unit)
 }
 
 trait MergeableStoreFactory[-K, V] extends java.io.Serializable {
-  def mergeableStore: () => Mergeable[K, V]
+  def mergeableStore: Semigroup[V] => Mergeable[K, V]
   def mergeableBatcher: Batcher
 }

summingbird-online/src/main/scala/com/twitter/summingbird/online/MergeableStoreFactoryAlgebra.scala → summingbird-online/src/main/scala/com/twitter/summingbird/online/WrappedTSInMergeable.scala

@@ -41,22 +41,3 @@ class WrappedTSInMergeable[K, V](self: Mergeable[K, V]) extends Mergeable[K, (Ti
         })
     }
 }
-
-object MergeableStoreFactoryAlgebra {
-  /*
-      Our tuples that we hand to the store are of the form ((K, BatchID), (Timestamp, V))
-      but in our store we only store ((K, BatchID), V). That is we don't include the timestamp.
-      We need these timestamps to continue processing downstream however, so we use a Right timestamp to say
-      the last value is taken. (Which may not be the max(TS)).
-
-      The merge operation here takes the inbound value of (Timestamp, V), performs the inner merge from the store.
-      Then looks back up the timestamp handed from the stream and outputs with that.
-      */
-  def wrapOnlineFactory[K, V](supplier: MergeableStoreFactory[K, V]): MergeableStoreFactory[K, (Timestamp, V)] =
-    {
-      val mergeable: () => Mergeable[K, (Timestamp, V)] =
-        () => { new WrappedTSInMergeable(supplier.mergeableStore()) }
-
-      MergeableStoreFactory[K, (Timestamp, V)](mergeable, supplier.mergeableBatcher)
-    }
-}

summingbird-online/src/main/scala/com/twitter/summingbird/online/executor/Summer.scala

@@ -55,7 +55,7 @@ import scala.util.control.NonFatal
  */
 
 class Summer[Key, Value: Semigroup, Event, S, D, RC](
-  @transient storeSupplier: MergeableStoreFactory[Key, Value],
+  @transient storeSupplier: () => Mergeable[Key, Value],
   @transient flatMapOp: FlatMapOperation[(Key, (Option[Value], Value)), Event],
   @transient successHandler: OnlineSuccessHandler,
   @transient exceptionHandler: OnlineExceptionHandler,
@@ -86,7 +86,7 @@ class Summer[Key, Value: Semigroup, Event, S, D, RC](
 
   override def init(runtimeContext: RC) {
     super.init(runtimeContext)
-    storePromise.setValue(storeBox.get.mergeableStore())
+    storePromise.setValue(storeBox.get())
     store.toString // Do the lazy evaluation now so we can connect before tuples arrive.
 
     successHandlerOpt = if (includeSuccessHandler.get) Some(successHandlerBox.get) else None

summingbird-storm/src/main/scala/com/twitter/summingbird/storm/StormPlatform.scala

@@ -186,14 +186,6 @@ abstract class Storm(options: Map[String, Options], transformConfig: Summingbird
     type ExecutorValueType = (Timestamp, V)
     type ExecutorOutputType = (Timestamp, (K, (Option[V], V)))
 
-    val supplier: MergeableStoreFactory[ExecutorKeyType, V] = summer.store match {
-      case m: MergeableStoreFactory[ExecutorKeyType, V] => m
-      case _ => sys.error("Should never be able to get here, looking for a MergeableStoreFactory from %s".format(summer.store))
-    }
-
-    val wrappedStore: MergeableStoreFactory[ExecutorKeyType, ExecutorValueType] =
-      MergeableStoreFactoryAlgebra.wrapOnlineFactory(supplier)
-
     val anchorTuples = getOrElse(stormDag, node, AnchorTuples.default)
     val metrics = getOrElse(stormDag, node, DEFAULT_SUMMER_STORM_METRICS)
     val shouldEmit = stormDag.dependantsOf(node).size > 0
@@ -218,6 +210,8 @@ abstract class Storm(options: Map[String, Options], transformConfig: Summingbird
     val flatmapOp: FlatMapOperation[(ExecutorKeyType, (Option[ExecutorValueType], ExecutorValueType)), ExecutorOutputType] =
       FlatMapOperation.apply(storeBaseFMOp)
 
+    val supplier: MergeableStoreFactory[ExecutorKeyType, V] = summer.store
+
     val sinkBolt = BaseBolt(
       jobID,
       metrics.metrics,
@@ -227,7 +221,7 @@ abstract class Storm(options: Map[String, Options], transformConfig: Summingbird
       ackOnEntry,
       maxExecutePerSec,
       new executor.Summer(
-        wrappedStore,
+        () => new WrappedTSInMergeable(supplier.mergeableStore(semigroup)),
         flatmapOp,
         getOrElse(stormDag, node, DEFAULT_ONLINE_SUCCESS_HANDLER),
         getOrElse(stormDag, node, DEFAULT_ONLINE_EXCEPTION_HANDLER),