Update Latest TS by timestamp

There is a bug when the historic data arrives and overrides the value
with morerecent timestamp in ts_kv_latest table. This PR adds
possibility to update the ts_kv_latest table only if the value that
arrives has the newer timestamp than the one that is already in
ts_kv_latest

@@ -243,6 +243,7 @@ sql:
       batch_max_delay: "${SQL_TS_LATEST_BATCH_MAX_DELAY_MS:100}"
       batch_max_delay: "${SQL_TS_LATEST_BATCH_MAX_DELAY_MS:100}"
       stats_print_interval_ms: "${SQL_TS_LATEST_BATCH_STATS_PRINT_MS:10000}"
       stats_print_interval_ms: "${SQL_TS_LATEST_BATCH_STATS_PRINT_MS:10000}"
       batch_threads: "${SQL_TS_LATEST_BATCH_THREADS:4}"
       batch_threads: "${SQL_TS_LATEST_BATCH_THREADS:4}"
+      update_by_latest_ts: "${SQL_TS_UPDATE_BY_LATEST_TIMESTAMP:true}"
     # Specify whether to sort entities before batch update. Should be enabled for cluster mode to avoid deadlocks
     # Specify whether to sort entities before batch update. Should be enabled for cluster mode to avoid deadlocks
     batch_sort: "${SQL_BATCH_SORT:false}"
     batch_sort: "${SQL_BATCH_SORT:false}"
     # Specify whether to remove null characters from strValue of attributes and timeseries before insert
     # Specify whether to remove null characters from strValue of attributes and timeseries before insert

@@ -15,6 +15,7 @@
  */
  */
 package org.thingsboard.server.dao.sqlts.insert.latest.psql;
 package org.thingsboard.server.dao.sqlts.insert.latest.psql;
 
 
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.jdbc.core.BatchPreparedStatementSetter;
 import org.springframework.jdbc.core.BatchPreparedStatementSetter;
 import org.springframework.stereotype.Repository;
 import org.springframework.stereotype.Repository;
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.TransactionStatus;
@@ -37,20 +38,34 @@ import java.util.List;
 @Transactional
 @Transactional
 public class PsqlLatestInsertTsRepository extends AbstractInsertRepository implements InsertLatestTsRepository {
 public class PsqlLatestInsertTsRepository extends AbstractInsertRepository implements InsertLatestTsRepository {
 
 
+    @Value("${sql.ts_latest.update_by_latest_ts:true}")
+    private Boolean updateByLatestTs;
+
     private static final String BATCH_UPDATE =
     private static final String BATCH_UPDATE =
-            "UPDATE ts_kv_latest SET ts = ?, bool_v = ?, str_v = ?, long_v = ?, dbl_v = ?, json_v = cast(? AS json) WHERE entity_id = ? and key = ?";
+            "UPDATE ts_kv_latest SET ts = ?, bool_v = ?, str_v = ?, long_v = ?, dbl_v = ?, json_v = cast(? AS json) WHERE entity_id = ? AND key = ?;";
 
 
 
 
     private static final String INSERT_OR_UPDATE =
     private static final String INSERT_OR_UPDATE =
             "INSERT INTO ts_kv_latest (entity_id, key, ts, bool_v, str_v, long_v, dbl_v,  json_v) VALUES(?, ?, ?, ?, ?, ?, ?, cast(? AS json)) " +
             "INSERT INTO ts_kv_latest (entity_id, key, ts, bool_v, str_v, long_v, dbl_v,  json_v) VALUES(?, ?, ?, ?, ?, ?, ?, cast(? AS json)) " +
                     "ON CONFLICT (entity_id, key) DO UPDATE SET ts = ?, bool_v = ?, str_v = ?, long_v = ?, dbl_v = ?, json_v = cast(? AS json);";
                     "ON CONFLICT (entity_id, key) DO UPDATE SET ts = ?, bool_v = ?, str_v = ?, long_v = ?, dbl_v = ?, json_v = cast(? AS json);";
 
 
+    private static final String BATCH_UPDATE_BY_LATEST_TS =
+            "UPDATE ts_kv_latest SET ts = ?, bool_v = ?, str_v = ?, long_v = ?, dbl_v = ?, json_v = cast(? AS json) WHERE entity_id = ? AND key = ? AND ts <= ?;";
+
+
+    private static final String INSERT_OR_UPDATE_BY_LATEST_TS =
+            "INSERT INTO ts_kv_latest (entity_id, key, ts, bool_v, str_v, long_v, dbl_v,  json_v) VALUES(?, ?, ?, ?, ?, ?, ?, cast(? AS json)) " +
+                    "ON CONFLICT (entity_id, key) DO UPDATE SET ts = ?, bool_v = ?, str_v = ?, long_v = ?, dbl_v = ?, json_v = cast(? AS json) WHERE ts_kv_latest.ts <= ?;";
+
     @Override
     @Override
     public void saveOrUpdate(List<TsKvLatestEntity> entities) {
     public void saveOrUpdate(List<TsKvLatestEntity> entities) {
         transactionTemplate.execute(new TransactionCallbackWithoutResult() {
         transactionTemplate.execute(new TransactionCallbackWithoutResult() {
             @Override
             @Override
             protected void doInTransactionWithoutResult(TransactionStatus status) {
             protected void doInTransactionWithoutResult(TransactionStatus status) {
-                int[] result = jdbcTemplate.batchUpdate(BATCH_UPDATE, new BatchPreparedStatementSetter() {
+                String batchUpdateQuery = updateByLatestTs ? BATCH_UPDATE_BY_LATEST_TS : BATCH_UPDATE;
+                String insertOrUpdateQuery = updateByLatestTs ? INSERT_OR_UPDATE_BY_LATEST_TS : INSERT_OR_UPDATE;
+
+                int[] result = jdbcTemplate.batchUpdate(batchUpdateQuery, new BatchPreparedStatementSetter() {
                     @Override
                     @Override
                     public void setValues(PreparedStatement ps, int i) throws SQLException {
                     public void setValues(PreparedStatement ps, int i) throws SQLException {
                         TsKvLatestEntity tsKvLatestEntity = entities.get(i);
                         TsKvLatestEntity tsKvLatestEntity = entities.get(i);
@@ -80,6 +95,9 @@ public class PsqlLatestInsertTsRepository extends AbstractInsertRepository imple
 
 
                         ps.setObject(7, tsKvLatestEntity.getEntityId());
                         ps.setObject(7, tsKvLatestEntity.getEntityId());
                         ps.setInt(8, tsKvLatestEntity.getKey());
                         ps.setInt(8, tsKvLatestEntity.getKey());
+                        if (updateByLatestTs) {
+                            ps.setLong(9, tsKvLatestEntity.getTs());
+                        }
                     }
                     }
 
 
                     @Override
                     @Override
@@ -102,7 +120,7 @@ public class PsqlLatestInsertTsRepository extends AbstractInsertRepository imple
                     }
                     }
                 }
                 }
 
 
-                jdbcTemplate.batchUpdate(INSERT_OR_UPDATE, new BatchPreparedStatementSetter() {
+                jdbcTemplate.batchUpdate(insertOrUpdateQuery, new BatchPreparedStatementSetter() {
                     @Override
                     @Override
                     public void setValues(PreparedStatement ps, int i) throws SQLException {
                     public void setValues(PreparedStatement ps, int i) throws SQLException {
                         TsKvLatestEntity tsKvLatestEntity = insertEntities.get(i);
                         TsKvLatestEntity tsKvLatestEntity = insertEntities.get(i);
@@ -111,6 +129,9 @@ public class PsqlLatestInsertTsRepository extends AbstractInsertRepository imple
 
 
                         ps.setLong(3, tsKvLatestEntity.getTs());
                         ps.setLong(3, tsKvLatestEntity.getTs());
                         ps.setLong(9, tsKvLatestEntity.getTs());
                         ps.setLong(9, tsKvLatestEntity.getTs());
+                        if (updateByLatestTs) {
+                            ps.setLong(15, tsKvLatestEntity.getTs());
+                        }
 
 
                         if (tsKvLatestEntity.getBooleanValue() != null) {
                         if (tsKvLatestEntity.getBooleanValue() != null) {
                             ps.setBoolean(4, tsKvLatestEntity.getBooleanValue());
                             ps.setBoolean(4, tsKvLatestEntity.getBooleanValue());