use upsert native SQL queries to update stats + do not update lastTs & period on late measures

- this is way quicker, and won't throw duplicate exception :)
- also, do not update last timestamp or sample period if timestamp is late (less than current last_ts)

Author: derlin

src/main/kotlin/ch/derlin/bbdata/common/stats/SqlStats.kt

@@ -36,6 +36,7 @@ data class SqlStats(
         @Column(name = "avg_sample_period")
         var avgSamplePeriod: Double = .0
 ) {
+    /** THIS IS NOT USED ANYMORE see SqlStatsRepository for more details
     fun updateWithNewValue(v: NewValue) {
         if (nWrites > 0L) {
             val deltaMs = abs(v.timestamp!!.millis - lastTs!!.millis)
@@ -47,4 +48,6 @@ data class SqlStats(
 
         nWrites += 1
     }
+    */
+
 }

src/main/kotlin/ch/derlin/bbdata/common/stats/SqlStatsRepository.kt

@@ -1,6 +1,7 @@
 package ch.derlin.bbdata.common.stats
 
 import ch.derlin.bbdata.Profiles
+import org.joda.time.DateTime
 import org.springframework.context.annotation.Profile
 import org.springframework.data.jpa.repository.JpaRepository
 import org.springframework.data.jpa.repository.Modifying
@@ -15,4 +16,25 @@ import org.springframework.transaction.annotation.Transactional
 
 @Profile(Profiles.SQL_STATS)
 @Repository
-interface SqlStatsRepository : JpaRepository<SqlStats, Long>
+interface SqlStatsRepository : JpaRepository<SqlStats, Long> {
+
+    @Transactional
+    @Modifying
+    @Query("""
+        INSERT INTO stats (object_id, last_ts, n_writes)
+        VALUES (:objectId, :ts, 1) ON DUPLICATE KEY UPDATE
+        avg_sample_period = CASE WHEN VALUES(last_ts) > last_ts 
+            THEN ( (avg_sample_period * (n_writes-1)) + ABS((UNIX_TIMESTAMP(last_ts) * 1000) - (UNIX_TIMESTAMP(VALUES(last_ts)) * 1000)) ) / n_writes 
+            ELSE avg_sample_period 
+        END,
+        n_writes = n_writes + 1,
+        last_ts = GREATEST(last_ts, VALUES(last_ts))
+    """, nativeQuery = true)
+    fun updateWriteStats(objectId: Long, ts: DateTime)
+
+    @Transactional
+    @Modifying
+    @Query("INSERT INTO stats (object_id, n_reads) VALUES (:objectId, 1) ON DUPLICATE KEY UPDATE n_reads = n_reads + 1", nativeQuery = true)
+    fun updateReadCounter(objectId: Long)
+
+}

src/main/kotlin/ch/derlin/bbdata/common/stats/StatsLogic.kt

@@ -52,13 +52,16 @@ class CassandraStatsLogic(private val objectStatsRepository: ObjectStatsReposito
         val objectStats = objectStatsRepository.findById(objectId).orElse(ObjectStats())
         val objectStatsCounter = objectStatsCounterRepository.findById(objectId).orElse(ObjectStatsCounter())
 
-        // compute new stats
-        val deltaMs = abs(v.timestamp!!.millis - (objectStats.lastTimestamp ?: v.timestamp).millis)
-        val nRecords = objectStatsCounter.nValues
-        val newSamplePeriod = if (nRecords > 0) (objectStats.avgSamplePeriod * (nRecords - 1) + deltaMs) / nRecords else .0f
+        // update object stats
+        if (objectStats.lastTimestamp == null || v.timestamp!! > objectStats.lastTimestamp) {
+            // update lastTs and compute sample period only if new value is not a late timestamp
+            val deltaMs = abs(v.timestamp!!.millis - (objectStats.lastTimestamp ?: v.timestamp).millis)
+            val nRecords = objectStatsCounter.nValues
+            val newSamplePeriod = if (nRecords > 0) (objectStats.avgSamplePeriod * (nRecords - 1) + deltaMs) / nRecords else .0f
+            objectStatsRepository.update(v.objectId.toInt(), newSamplePeriod, v.timestamp)
+        }
 
-        // save new stats
-        objectStatsRepository.update(v.objectId.toInt(), newSamplePeriod, v.timestamp)
+        // increment counter
         objectStatsCounterRepository.updateWriteCounter(v.objectId.toInt())
     }
 
@@ -88,30 +91,11 @@ class CassandraStatsLogic(private val objectStatsRepository: ObjectStatsReposito
 class SqlStatsLogic(private val statsRepository: SqlStatsRepository) : StatsLogic {
 
     override fun updateStats(v: NewValue) {
-        val stats = statsRepository.findById(v.objectId!!).orElse(SqlStats(objectId = v.objectId))
-        stats.updateWithNewValue(v)
-        statsRepository.save(stats)
-    }
-
-    override fun updateAllStats(vs: List<NewValue>) {
-        // get all unique objects whose objectIds are in vs
-        val stats = vs.map { it.objectId!! }.toSet().map{ objectId ->
-            objectId to statsRepository.findById(objectId).orElse(SqlStats(objectId = objectId))
-        }.toMap()
-
-        // update each: this will also work when multiple values target the same object
-        vs.forEach {
-            stats[it.objectId]!!.updateWithNewValue(it)
-        }
-
-        // use the bulk save option of MySQL repositories to speed up the process
-        statsRepository.saveAll(stats.values)
+        statsRepository.updateWriteStats(v.objectId!!, v.timestamp!!)
     }
 
     override fun incrementReadCounter(objectId: Long) {
-        val stats = statsRepository.findById(objectId).orElse(SqlStats(objectId = objectId))
-        stats.nReads += 1
-        statsRepository.save(stats)
+        statsRepository.updateReadCounter(objectId)
     }
 
     override fun getStats(objectId: Long): Stats {

src/test/kotlin/ch/derlin/bbdata/input/InputApiTest.kt

@@ -33,7 +33,7 @@ class InputApiTest {
     private lateinit var restTemplate: TestRestTemplate
 
     // asynchronous stats update, hence wait a bit (ms)
-    val statsWait: Long = 5000
+    val statsWait: Long = 100
 
     companion object {
         val OBJ = 1