Fixed violations on event primary and unique key constraints & code improvement (#1921)

* init commit

* fix-violation-of-primary-key-constraint

* revert thingsboard.yml changes

* remove @Slf4j annotation

* update code

* update Events Dao

* code improvements

* revert changes in logback.xml

* cleaned code

* attributes/on-update-set-null-values

@@ -15,7 +15,6 @@
  */
 package org.thingsboard.server.dao.sql.attributes;
 
-import lombok.extern.slf4j.Slf4j;
 import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.stereotype.Repository;
 import org.thingsboard.server.dao.model.sql.AttributeKvEntity;
@@ -24,7 +23,6 @@ import org.thingsboard.server.dao.util.SqlDao;
 import javax.persistence.EntityManager;
 import javax.persistence.PersistenceContext;
 
-@Slf4j
 @SqlDao
 @Repository
 public abstract class AttributeKvInsertRepository {

+/**
+ * Copyright © 2016-2019 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.dao.sql.attributes;
+
+import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
+import org.thingsboard.server.dao.model.sql.AttributeKvEntity;
+import org.thingsboard.server.dao.util.HsqlDao;
+import org.thingsboard.server.dao.util.SqlDao;
+
+@SqlDao
+@HsqlDao
+@Repository
+@Transactional
+public class HsqlAttributesInsertRepository extends AttributeKvInsertRepository {
+
+    private static final String ON_BOOL_VALUE_UPDATE_SET_NULLS = " attribute_kv.str_v = null, attribute_kv.long_v = null, attribute_kv.dbl_v = null ";
+    private static final String ON_STR_VALUE_UPDATE_SET_NULLS = " attribute_kv.bool_v = null, attribute_kv.long_v = null, attribute_kv.dbl_v = null ";
+    private static final String ON_LONG_VALUE_UPDATE_SET_NULLS = " attribute_kv.str_v = null, attribute_kv.bool_v = null, attribute_kv.dbl_v = null ";
+    private static final String ON_DBL_VALUE_UPDATE_SET_NULLS = " attribute_kv.str_v = null, attribute_kv.long_v = null, attribute_kv.bool_v = null ";
+
+    private static final String INSERT_BOOL_STATEMENT = getInsertOrUpdateString(BOOL_V, ON_BOOL_VALUE_UPDATE_SET_NULLS);
+    private static final String INSERT_STR_STATEMENT = getInsertOrUpdateString(STR_V, ON_STR_VALUE_UPDATE_SET_NULLS);
+    private static final String INSERT_LONG_STATEMENT = getInsertOrUpdateString(LONG_V, ON_LONG_VALUE_UPDATE_SET_NULLS);
+    private static final String INSERT_DBL_STATEMENT = getInsertOrUpdateString(DBL_V, ON_DBL_VALUE_UPDATE_SET_NULLS);
+
+    @Override
+    public void saveOrUpdate(AttributeKvEntity entity) {
+        processSaveOrUpdate(entity, INSERT_BOOL_STATEMENT, INSERT_STR_STATEMENT, INSERT_LONG_STATEMENT, INSERT_DBL_STATEMENT);
+    }
+
+    private static String getInsertOrUpdateString(String value, String nullValues) {
+        return "MERGE INTO attribute_kv USING(VALUES :entity_type, :entity_id, :attribute_type, :attribute_key, :" + value + ", :last_update_ts) A (entity_type, entity_id, attribute_type, attribute_key, " + value + ", last_update_ts) ON (attribute_kv.entity_type=A.entity_type AND attribute_kv.entity_id=A.entity_id AND attribute_kv.attribute_type=A.attribute_type AND attribute_kv.attribute_key=A.attribute_key) WHEN MATCHED THEN UPDATE SET attribute_kv." + value + " = A." + value + ", attribute_kv.last_update_ts = A.last_update_ts," + nullValues + "WHEN NOT MATCHED THEN INSERT (entity_type, entity_id, attribute_type, attribute_key, " + value + ", last_update_ts) VALUES (A.entity_type, A.entity_id, A.attribute_type, A.attribute_key, A." + value + ", A.last_update_ts)";
+    }
+}
\ No newline at end of file

@@ -25,19 +25,24 @@ import org.thingsboard.server.dao.util.SqlDao;
 @PsqlDao
 @Repository
 @Transactional
-public class PsqlInsertRepository extends AttributeKvInsertRepository {
+public class PsqlAttributesInsertRepository extends AttributeKvInsertRepository {
 
-    private static final String INSERT_OR_UPDATE_BOOL_STATEMENT = getInsertOrUpdateString(BOOL_V);
-    private static final String INSERT_OR_UPDATE_STR_STATEMENT = getInsertOrUpdateString(STR_V);
-    private static final String INSERT_OR_UPDATE_LONG_STATEMENT = getInsertOrUpdateString(LONG_V);
-    private static final String INSERT_OR_UPDATE_DBL_STATEMENT = getInsertOrUpdateString(DBL_V);
+    private static final String ON_BOOL_VALUE_UPDATE_SET_NULLS = "str_v = null, long_v = null, dbl_v = null";
+    private static final String ON_STR_VALUE_UPDATE_SET_NULLS = "bool_v = null, long_v = null, dbl_v = null";
+    private static final String ON_LONG_VALUE_UPDATE_SET_NULLS = "str_v = null, bool_v = null, dbl_v = null";
+    private static final String ON_DBL_VALUE_UPDATE_SET_NULLS = "str_v = null, long_v = null, bool_v = null";
+
+    private static final String INSERT_OR_UPDATE_BOOL_STATEMENT = getInsertOrUpdateString(BOOL_V, ON_BOOL_VALUE_UPDATE_SET_NULLS);
+    private static final String INSERT_OR_UPDATE_STR_STATEMENT = getInsertOrUpdateString(STR_V, ON_STR_VALUE_UPDATE_SET_NULLS);
+    private static final String INSERT_OR_UPDATE_LONG_STATEMENT = getInsertOrUpdateString(LONG_V , ON_LONG_VALUE_UPDATE_SET_NULLS);
+    private static final String INSERT_OR_UPDATE_DBL_STATEMENT = getInsertOrUpdateString(DBL_V, ON_DBL_VALUE_UPDATE_SET_NULLS);
 
     @Override
     public void saveOrUpdate(AttributeKvEntity entity) {
         processSaveOrUpdate(entity, INSERT_OR_UPDATE_BOOL_STATEMENT, INSERT_OR_UPDATE_STR_STATEMENT, INSERT_OR_UPDATE_LONG_STATEMENT, INSERT_OR_UPDATE_DBL_STATEMENT);
     }
 
-    private static String getInsertOrUpdateString(String value) {
-        return "INSERT INTO attribute_kv (entity_type, entity_id, attribute_type, attribute_key, " + value + ", last_update_ts) VALUES (:entity_type, :entity_id, :attribute_type, :attribute_key, :" + value + ", :last_update_ts) ON CONFLICT (entity_type, entity_id, attribute_type, attribute_key) DO UPDATE SET " + value + " = :" + value + ", last_update_ts = :last_update_ts";
+    private static String getInsertOrUpdateString(String value, String nullValues) {
+        return "INSERT INTO attribute_kv (entity_type, entity_id, attribute_type, attribute_key, " + value + ", last_update_ts) VALUES (:entity_type, :entity_id, :attribute_type, :attribute_key, :" + value + ", :last_update_ts) ON CONFLICT (entity_type, entity_id, attribute_type, attribute_key) DO UPDATE SET " + value + " = :" + value + ", last_update_ts = :last_update_ts," + nullValues;
     }
 }
\ No newline at end of file

@@ -13,75 +13,84 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.thingsboard.server.dao.sql.attributes;
+package org.thingsboard.server.dao.sql.event;
 
 import lombok.extern.slf4j.Slf4j;
 import org.hibernate.exception.ConstraintViolationException;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.stereotype.Repository;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.TransactionDefinition;
 import org.springframework.transaction.TransactionStatus;
 import org.springframework.transaction.support.DefaultTransactionDefinition;
 import org.thingsboard.server.common.data.UUIDConverter;
-import org.thingsboard.server.dao.model.sql.AttributeKvEntity;
-import org.thingsboard.server.dao.util.HsqlDao;
+import org.thingsboard.server.dao.model.sql.EventEntity;
 import org.thingsboard.server.dao.util.SqlDao;
 
+import javax.persistence.EntityManager;
+import javax.persistence.PersistenceContext;
+import javax.persistence.Query;
+
 @Slf4j
 @SqlDao
-@HsqlDao
 @Repository
-public class HsqlInsertRepository extends AttributeKvInsertRepository {
-
-    @Autowired
-    private PlatformTransactionManager transactionManager;
+public abstract class EventInsertRepository {
 
-    private static final String INSERT_BOOL_STATEMENT = getInsertString(BOOL_V);
-    private static final String INSERT_STR_STATEMENT = getInsertString(STR_V);
-    private static final String INSERT_LONG_STATEMENT = getInsertString(LONG_V);
-    private static final String INSERT_DBL_STATEMENT = getInsertString(DBL_V);
+    @PersistenceContext
+    protected EntityManager entityManager;
 
-    private static final String WHERE_STATEMENT = " WHERE entity_type = :entity_type AND entity_id = :entity_id AND attribute_type = :attribute_type AND attribute_key = :attribute_key";
+    @Autowired
+    protected PlatformTransactionManager transactionManager;
 
-    private static final String UPDATE_BOOL_STATEMENT = getUpdateString(BOOL_V);
-    private static final String UPDATE_STR_STATEMENT = getUpdateString(STR_V);
-    private static final String UPDATE_LONG_STATEMENT = getUpdateString(LONG_V);
-    private static final String UPDATE_DBL_STATEMENT = getUpdateString(DBL_V);
+    public abstract EventEntity saveOrUpdate(EventEntity entity);
 
-    @Override
-    public void saveOrUpdate(AttributeKvEntity entity) {
-        DefaultTransactionDefinition insertDefinition = new DefaultTransactionDefinition();
-        insertDefinition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
-        TransactionStatus insertTransaction = transactionManager.getTransaction(insertDefinition);
+    protected EventEntity saveAndGet(EventEntity entity, String insertOrUpdateOnPrimaryKeyConflict, String insertOrUpdateOnUniqueKeyConflict) {
+        EventEntity eventEntity = null;
+        TransactionStatus insertTransaction = getTransactionStatus(TransactionDefinition.PROPAGATION_REQUIRED);
         try {
-            processSaveOrUpdate(entity, INSERT_BOOL_STATEMENT, INSERT_STR_STATEMENT, INSERT_LONG_STATEMENT, INSERT_DBL_STATEMENT);
+            eventEntity = processSaveOrUpdate(entity, insertOrUpdateOnPrimaryKeyConflict);
             transactionManager.commit(insertTransaction);
-        } catch (Throwable e) {
+        } catch (Throwable throwable) {
             transactionManager.rollback(insertTransaction);
-            if (e.getCause() instanceof ConstraintViolationException) {
-                log.trace("Insert request leaded in a violation of a defined integrity constraint {} for Entity with entityId {} and entityType {}", e.getMessage(), UUIDConverter.fromString(entity.getId().getEntityId()), entity.getId().getEntityType());
-                DefaultTransactionDefinition definition = new DefaultTransactionDefinition();
-                definition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
-                TransactionStatus transaction = transactionManager.getTransaction(definition);
+            if (throwable.getCause() instanceof ConstraintViolationException) {
+                log.trace("Insert request leaded in a violation of a defined integrity constraint {} for Entity with entityId {} and entityType {}", throwable.getMessage(), entity.getEventUid(), entity.getEventType());
+                TransactionStatus transaction = getTransactionStatus(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
                 try {
-                    processSaveOrUpdate(entity, UPDATE_BOOL_STATEMENT, UPDATE_STR_STATEMENT, UPDATE_LONG_STATEMENT, UPDATE_DBL_STATEMENT);
+                    eventEntity = processSaveOrUpdate(entity, insertOrUpdateOnUniqueKeyConflict);
                 } catch (Throwable th) {
-                    log.trace("Could not execute the update statement for Entity with entityId {} and entityType {}", UUIDConverter.fromString(entity.getId().getEntityId()), entity.getId().getEntityType());
+                    log.trace("Could not execute the update statement for Entity with entityId {} and entityType {}", entity.getEventUid(), entity.getEventType());
                     transactionManager.rollback(transaction);
                 }
                 transactionManager.commit(transaction);
             } else {
-                log.trace("Could not execute the insert statement for Entity with entityId {} and entityType {}", UUIDConverter.fromString(entity.getId().getEntityId()), entity.getId().getEntityType());
+                log.trace("Could not execute the insert statement for Entity with entityId {} and entityType {}", entity.getEventUid(), entity.getEventType());
             }
         }
+        return eventEntity;
     }
 
-    private static String getInsertString(String value) {
-        return "INSERT INTO attribute_kv (entity_type, entity_id, attribute_type, attribute_key, " + value + ", last_update_ts) VALUES (:entity_type, :entity_id, :attribute_type, :attribute_key, :" + value + ", :last_update_ts)";
+    @Modifying
+    protected abstract EventEntity doProcessSaveOrUpdate(EventEntity entity, String query);
+
+    protected Query getQuery(EventEntity entity, String query) {
+        return entityManager.createNativeQuery(query, EventEntity.class)
+                .setParameter("id", UUIDConverter.fromTimeUUID(entity.getId()))
+                .setParameter("body", entity.getBody().toString())
+                .setParameter("entity_id", entity.getEntityId())
+                .setParameter("entity_type", entity.getEntityType().name())
+                .setParameter("event_type", entity.getEventType())
+                .setParameter("event_uid", entity.getEventUid())
+                .setParameter("tenant_id", entity.getTenantId());
+    }
+
+    private EventEntity processSaveOrUpdate(EventEntity entity, String query) {
+        return doProcessSaveOrUpdate(entity, query);
     }
 
-    private static String getUpdateString(String value) {
-        return "UPDATE attribute_kv SET " + value + " = :" + value + ", last_update_ts = :last_update_ts" + WHERE_STATEMENT;
+    private TransactionStatus getTransactionStatus(int propagationRequired) {
+        DefaultTransactionDefinition insertDefinition = new DefaultTransactionDefinition();
+        insertDefinition.setPropagationBehavior(propagationRequired);
+        return transactionManager.getTransaction(insertDefinition);
     }
 }
\ No newline at end of file

+/**
+ * Copyright © 2016-2019 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.dao.sql.event;
+
+import org.springframework.stereotype.Repository;
+import org.thingsboard.server.common.data.UUIDConverter;
+import org.thingsboard.server.dao.model.sql.EventEntity;
+import org.thingsboard.server.dao.util.HsqlDao;
+import org.thingsboard.server.dao.util.SqlDao;
+
+@SqlDao
+@HsqlDao
+@Repository
+public class HsqlEventInsertRepository extends EventInsertRepository {
+
+    private static final String P_KEY_CONFLICT_STATEMENT = "(event.id=I.id)";
+    private static final String UNQ_KEY_CONFLICT_STATEMENT = "(event.tenant_id=I.tenant_id AND event.entity_type=I.entity_type AND event.entity_id=I.entity_id AND event.event_type=I.event_type AND event.event_uid=I.event_uid)";
+
+    private static final String INSERT_OR_UPDATE_ON_P_KEY_CONFLICT = getInsertString(P_KEY_CONFLICT_STATEMENT);
+    private static final String INSERT_OR_UPDATE_ON_UNQ_KEY_CONFLICT = getInsertString(UNQ_KEY_CONFLICT_STATEMENT);
+
+    @Override
+    public EventEntity saveOrUpdate(EventEntity entity) {
+        return saveAndGet(entity, INSERT_OR_UPDATE_ON_P_KEY_CONFLICT, INSERT_OR_UPDATE_ON_UNQ_KEY_CONFLICT);
+    }
+
+    @Override
+    protected EventEntity doProcessSaveOrUpdate(EventEntity entity, String query) {
+        getQuery(entity, query).executeUpdate();
+        return entityManager.find(EventEntity.class, UUIDConverter.fromTimeUUID(entity.getId()));
+    }
+
+    private static String getInsertString(String conflictStatement) {
+        return "MERGE INTO event USING (VALUES :id, :body, :entity_id, :entity_type, :event_type, :event_uid, :tenant_id) I (id, body, entity_id, entity_type, event_type, event_uid, tenant_id) ON " + conflictStatement + " WHEN MATCHED THEN UPDATE SET event.id = I.id, event.body = I.body, event.entity_id = I.entity_id, event.entity_type = I.entity_type, event.event_type = I.event_type, event.event_uid = I.event_uid, event.tenant_id = I.tenant_id" +
+                " WHEN NOT MATCHED THEN INSERT (id, body, entity_id, entity_type, event_type, event_uid, tenant_id) VALUES (I.id, I.body, I.entity_id, I.entity_type, I.event_type, I.event_uid, I.tenant_id)";
+    }
+}
\ No newline at end of file

@@ -61,6 +61,9 @@ public class JpaBaseEventDao extends JpaAbstractSearchTimeDao<EventEntity, Event
     @Autowired
     private EventRepository eventRepository;
 
+    @Autowired
+    private EventInsertRepository eventInsertRepository;
+
     @Override
     protected Class<EventEntity> getEntityClass() {
         return EventEntity.class;
@@ -147,7 +150,7 @@ public class JpaBaseEventDao extends JpaAbstractSearchTimeDao<EventEntity, Event
                 eventRepository.findByTenantIdAndEntityTypeAndEntityId(entity.getTenantId(), entity.getEntityType(), entity.getEntityId()) != null) {
             return Optional.empty();
         }
-        return Optional.of(DaoUtil.getData(eventRepository.save(entity)));
+        return Optional.of(DaoUtil.getData(eventInsertRepository.saveOrUpdate(entity)));
     }
 
     private Specification<EventEntity> getEntityFieldsSpec(UUID tenantId, EntityId entityId, String eventType) {

+/**
+ * Copyright © 2016-2019 The Thingsboard Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.thingsboard.server.dao.sql.event;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Repository;
+import org.thingsboard.server.dao.model.sql.EventEntity;
+import org.thingsboard.server.dao.util.PsqlDao;
+import org.thingsboard.server.dao.util.SqlDao;
+
+@Slf4j
+@SqlDao
+@PsqlDao
+@Repository
+public class PsqlEventInsertRepository extends EventInsertRepository {
+
+    private static final String P_KEY_CONFLICT_STATEMENT = "(id)";
+    private static final String UNQ_KEY_CONFLICT_STATEMENT = "(tenant_id, entity_type, entity_id, event_type, event_uid)";
+
+    private static final String UPDATE_P_KEY_STATEMENT = "id = :id";
+    private static final String UPDATE_UNQ_KEY_STATEMENT = "tenant_id = :tenant_id, entity_type = :entity_type, entity_id = :entity_id, event_type = :event_type, event_uid = :event_uid";
+
+    private static final String INSERT_OR_UPDATE_ON_P_KEY_CONFLICT = getInsertOrUpdateString(P_KEY_CONFLICT_STATEMENT, UPDATE_UNQ_KEY_STATEMENT);
+    private static final String INSERT_OR_UPDATE_ON_UNQ_KEY_CONFLICT = getInsertOrUpdateString(UNQ_KEY_CONFLICT_STATEMENT, UPDATE_P_KEY_STATEMENT);
+
+    @Override
+    public EventEntity saveOrUpdate(EventEntity entity) {
+        return saveAndGet(entity, INSERT_OR_UPDATE_ON_P_KEY_CONFLICT, INSERT_OR_UPDATE_ON_UNQ_KEY_CONFLICT);
+    }
+
+    @Override
+    protected EventEntity doProcessSaveOrUpdate(EventEntity entity, String query) {
+        return (EventEntity) getQuery(entity, query).getSingleResult();
+
+    }
+
+    private static String getInsertOrUpdateString(String eventKeyStatement, String updateKeyStatement) {
+        return "INSERT INTO event (id, body, entity_id, entity_type, event_type, event_uid, tenant_id) VALUES (:id, :body, :entity_id, :entity_type, :event_type, :event_uid, :tenant_id) ON CONFLICT " + eventKeyStatement + " DO UPDATE SET body = :body, " + updateKeyStatement + " returning *";
+    }
+}
\ No newline at end of file