Fixed violations on attributes primary keys constraints (#1909)

* init commit

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

* revert thingsboard.yml changes

* remove @Slf4j annotation

* update code

common/dao-api/src/main/java/org/thingsboard/server/dao/util/HsqlDao.java

@@ -0,0 +1,22 @@
+/**
+ * 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.util;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+@ConditionalOnProperty(prefix = "spring.jpa", value = "database-platform", havingValue = "org.hibernate.dialect.HSQLDialect")
+public @interface HsqlDao {
+}

common/dao-api/src/main/java/org/thingsboard/server/dao/util/PsqlDao.java

@@ -0,0 +1,22 @@
+/**
+ * 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.util;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+@ConditionalOnProperty(prefix = "spring.jpa", value = "database-platform", havingValue = "org.hibernate.dialect.PostgreSQLDialect")
+public @interface PsqlDao {
+}

dao/src/main/java/org/thingsboard/server/dao/model/sql/AttributeKvEntity.java

@@ -16,7 +16,6 @@
 package org.thingsboard.server.dao.model.sql;
 
 import lombok.Data;
-import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
 import org.thingsboard.server.common.data.kv.BooleanDataEntry;
@@ -29,19 +28,11 @@ import org.thingsboard.server.dao.model.ToData;
 import javax.persistence.Column;
 import javax.persistence.EmbeddedId;
 import javax.persistence.Entity;
-import javax.persistence.EnumType;
-import javax.persistence.Enumerated;
-import javax.persistence.Id;
-import javax.persistence.IdClass;
 import javax.persistence.Table;
 import java.io.Serializable;
 
-import static org.thingsboard.server.dao.model.ModelConstants.ATTRIBUTE_KEY_COLUMN;
-import static org.thingsboard.server.dao.model.ModelConstants.ATTRIBUTE_TYPE_COLUMN;
 import static org.thingsboard.server.dao.model.ModelConstants.BOOLEAN_VALUE_COLUMN;
 import static org.thingsboard.server.dao.model.ModelConstants.DOUBLE_VALUE_COLUMN;
-import static org.thingsboard.server.dao.model.ModelConstants.ENTITY_ID_COLUMN;
-import static org.thingsboard.server.dao.model.ModelConstants.ENTITY_TYPE_COLUMN;
 import static org.thingsboard.server.dao.model.ModelConstants.LAST_UPDATE_TS_COLUMN;
 import static org.thingsboard.server.dao.model.ModelConstants.LONG_VALUE_COLUMN;
 import static org.thingsboard.server.dao.model.ModelConstants.STRING_VALUE_COLUMN;

dao/src/main/java/org/thingsboard/server/dao/sql/attributes/AttributeKvInsertRepository.java

@@ -0,0 +1,104 @@
+/**
+ * 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 lombok.extern.slf4j.Slf4j;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.stereotype.Repository;
+import org.thingsboard.server.dao.model.sql.AttributeKvEntity;
+import org.thingsboard.server.dao.util.SqlDao;
+
+import javax.persistence.EntityManager;
+import javax.persistence.PersistenceContext;
+
+@Slf4j
+@SqlDao
+@Repository
+public abstract class AttributeKvInsertRepository {
+
+    protected static final String BOOL_V = "bool_v";
+    protected static final String STR_V = "str_v";
+    protected static final String LONG_V = "long_v";
+    protected static final String DBL_V = "dbl_v";
+
+    @PersistenceContext
+    protected EntityManager entityManager;
+
+    public abstract void saveOrUpdate(AttributeKvEntity entity);
+
+    protected void processSaveOrUpdate(AttributeKvEntity entity, String requestBoolValue, String requestStrValue, String requestLongValue, String requestDblValue) {
+        if (entity.getBooleanValue() != null) {
+            saveOrUpdateBoolean(entity, requestBoolValue);
+        }
+        if (entity.getStrValue() != null) {
+            saveOrUpdateString(entity, requestStrValue);
+        }
+        if (entity.getLongValue() != null) {
+            saveOrUpdateLong(entity, requestLongValue);
+        }
+        if (entity.getDoubleValue() != null) {
+            saveOrUpdateDouble(entity, requestDblValue);
+        }
+    }
+
+    @Modifying
+    private void saveOrUpdateBoolean(AttributeKvEntity entity, String query) {
+        entityManager.createNativeQuery(query)
+                .setParameter("entity_type", entity.getId().getEntityType().name())
+                .setParameter("entity_id", entity.getId().getEntityId())
+                .setParameter("attribute_type", entity.getId().getAttributeType())
+                .setParameter("attribute_key", entity.getId().getAttributeKey())
+                .setParameter("bool_v", entity.getBooleanValue())
+                .setParameter("last_update_ts", entity.getLastUpdateTs())
+                .executeUpdate();
+    }
+
+    @Modifying
+    private void saveOrUpdateString(AttributeKvEntity entity, String query) {
+        entityManager.createNativeQuery(query)
+                .setParameter("entity_type", entity.getId().getEntityType().name())
+                .setParameter("entity_id", entity.getId().getEntityId())
+                .setParameter("attribute_type", entity.getId().getAttributeType())
+                .setParameter("attribute_key", entity.getId().getAttributeKey())
+                .setParameter("str_v", entity.getStrValue())
+                .setParameter("last_update_ts", entity.getLastUpdateTs())
+                .executeUpdate();
+    }
+
+    @Modifying
+    private void saveOrUpdateLong(AttributeKvEntity entity, String query) {
+        entityManager.createNativeQuery(query)
+                .setParameter("entity_type", entity.getId().getEntityType().name())
+                .setParameter("entity_id", entity.getId().getEntityId())
+                .setParameter("attribute_type", entity.getId().getAttributeType())
+                .setParameter("attribute_key", entity.getId().getAttributeKey())
+                .setParameter("long_v", entity.getLongValue())
+                .setParameter("last_update_ts", entity.getLastUpdateTs())
+                .executeUpdate();
+    }
+
+    @Modifying
+    private void saveOrUpdateDouble(AttributeKvEntity entity, String query) {
+        entityManager.createNativeQuery(query)
+                .setParameter("entity_type", entity.getId().getEntityType().name())
+                .setParameter("entity_id", entity.getId().getEntityId())
+                .setParameter("attribute_type", entity.getId().getAttributeType())
+                .setParameter("attribute_key", entity.getId().getAttributeKey())
+                .setParameter("dbl_v", entity.getDoubleValue())
+                .setParameter("last_update_ts", entity.getLastUpdateTs())
+                .executeUpdate();
+    }
+}

dao/src/main/java/org/thingsboard/server/dao/sql/attributes/HsqlInsertRepository.java

@@ -0,0 +1,87 @@
+/**
+ * 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 lombok.extern.slf4j.Slf4j;
+import org.hibernate.exception.ConstraintViolationException;
+import org.springframework.beans.factory.annotation.Autowired;
+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.util.SqlDao;
+
+@Slf4j
+@SqlDao
+@HsqlDao
+@Repository
+public class HsqlInsertRepository extends AttributeKvInsertRepository {
+
+    @Autowired
+    private PlatformTransactionManager transactionManager;
+
+    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);
+
+    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";
+
+    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);
+
+    @Override
+    public void saveOrUpdate(AttributeKvEntity entity) {
+        DefaultTransactionDefinition insertDefinition = new DefaultTransactionDefinition();
+        insertDefinition.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
+        TransactionStatus insertTransaction = transactionManager.getTransaction(insertDefinition);
+        try {
+            processSaveOrUpdate(entity, INSERT_BOOL_STATEMENT, INSERT_STR_STATEMENT, INSERT_LONG_STATEMENT, INSERT_DBL_STATEMENT);
+            transactionManager.commit(insertTransaction);
+        } catch (Throwable e) {
+            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);
+                try {
+                    processSaveOrUpdate(entity, UPDATE_BOOL_STATEMENT, UPDATE_STR_STATEMENT, UPDATE_LONG_STATEMENT, UPDATE_DBL_STATEMENT);
+                } 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());
+                    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());
+            }
+        }
+    }
+
+    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)";
+    }
+
+    private static String getUpdateString(String value) {
+        return "UPDATE attribute_kv SET " + value + " = :" + value + ", last_update_ts = :last_update_ts" + WHERE_STATEMENT;
+    }
+}

dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java

@@ -47,6 +47,9 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl
     @Autowired
     private AttributeKvRepository attributeKvRepository;
 
+    @Autowired
+    private AttributeKvInsertRepository attributeKvInsertRepository;
+
     @Override
     public ListenableFuture<Optional<AttributeKvEntry>> find(TenantId tenantId, EntityId entityId, String attributeType, String attributeKey) {
         AttributeKvCompositeKey compositeKey =
@@ -87,11 +90,12 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl
         entity.setLongValue(attribute.getLongValue().orElse(null));
         entity.setBooleanValue(attribute.getBooleanValue().orElse(null));
         return service.submit(() -> {
-            attributeKvRepository.save(entity);
+            attributeKvInsertRepository.saveOrUpdate(entity);
             return null;
         });
     }
 
+
     @Override
     public ListenableFuture<List<Void>> removeAll(TenantId tenantId, EntityId entityId, String attributeType, List<String> keys) {
         List<AttributeKvEntity> entitiesToDelete = keys

dao/src/main/java/org/thingsboard/server/dao/sql/attributes/PsqlInsertRepository.java

@@ -0,0 +1,43 @@
+/**
+ * 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.PsqlDao;
+import org.thingsboard.server.dao.util.SqlDao;
+
+@SqlDao
+@PsqlDao
+@Repository
+@Transactional
+public class PsqlInsertRepository 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);
+
+    @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";
+    }
+}