Merge branch 'develop/3.0' of https://github.com/thingsboard/thingsboard into map/3.0

-/**
- * Copyright © 2016-2020 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.service.install;
-
-import com.datastax.driver.core.KeyspaceMetadata;
-import com.datastax.driver.core.exceptions.InvalidQueryException;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.context.annotation.Profile;
-import org.springframework.stereotype.Service;
-import org.thingsboard.server.dao.dashboard.DashboardService;
-import org.thingsboard.server.dao.util.NoSqlDao;
-import org.thingsboard.server.service.install.cql.CassandraDbHelper;
-
-import java.nio.file.Files;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-
-import static org.thingsboard.server.service.install.DatabaseHelper.ADDITIONAL_INFO;
-import static org.thingsboard.server.service.install.DatabaseHelper.ASSET;
-import static org.thingsboard.server.service.install.DatabaseHelper.ASSIGNED_CUSTOMERS;
-import static org.thingsboard.server.service.install.DatabaseHelper.CONFIGURATION;
-import static org.thingsboard.server.service.install.DatabaseHelper.CUSTOMER_ID;
-import static org.thingsboard.server.service.install.DatabaseHelper.DASHBOARD;
-import static org.thingsboard.server.service.install.DatabaseHelper.DEVICE;
-import static org.thingsboard.server.service.install.DatabaseHelper.END_TS;
-import static org.thingsboard.server.service.install.DatabaseHelper.ENTITY_ID;
-import static org.thingsboard.server.service.install.DatabaseHelper.ENTITY_TYPE;
-import static org.thingsboard.server.service.install.DatabaseHelper.ENTITY_VIEW;
-import static org.thingsboard.server.service.install.DatabaseHelper.ENTITY_VIEWS;
-import static org.thingsboard.server.service.install.DatabaseHelper.ID;
-import static org.thingsboard.server.service.install.DatabaseHelper.KEYS;
-import static org.thingsboard.server.service.install.DatabaseHelper.NAME;
-import static org.thingsboard.server.service.install.DatabaseHelper.SEARCH_TEXT;
-import static org.thingsboard.server.service.install.DatabaseHelper.START_TS;
-import static org.thingsboard.server.service.install.DatabaseHelper.TENANT_ID;
-import static org.thingsboard.server.service.install.DatabaseHelper.TITLE;
-import static org.thingsboard.server.service.install.DatabaseHelper.TYPE;
-
-@Service
-@NoSqlDao
-@Profile("install")
-@Slf4j
-public class CassandraDatabaseUpgradeService extends AbstractCassandraDatabaseUpgradeService implements DatabaseEntitiesUpgradeService {
-
-    private static final String SCHEMA_UPDATE_CQL = "schema_update.cql";
-
-    @Autowired
-    private DashboardService dashboardService;
-
-    @Autowired
-    private InstallScripts installScripts;
-
-    @Override
-    public void upgradeDatabase(String fromVersion) throws Exception {
-
-        switch (fromVersion) {
-            case "1.2.3":
-
-                log.info("Upgrading Cassandara DataBase from version {} to 1.3.0 ...", fromVersion);
-
-                //Dump devices, assets and relations
-
-                cluster.getSession();
-
-                KeyspaceMetadata ks = cluster.getCluster().getMetadata().getKeyspace(cluster.getKeyspaceName());
-
-                log.info("Dumping devices ...");
-                Path devicesDump = CassandraDbHelper.dumpCfIfExists(ks, cluster.getSession(), DEVICE,
-                        new String[]{"id", TENANT_ID, CUSTOMER_ID, "name", SEARCH_TEXT, ADDITIONAL_INFO, "type"},
-                        new String[]{"", "", "", "", "", "", "default"},
-                        "tb-devices");
-                log.info("Devices dumped.");
-
-                log.info("Dumping assets ...");
-                Path assetsDump = CassandraDbHelper.dumpCfIfExists(ks, cluster.getSession(), ASSET,
-                        new String[]{"id", TENANT_ID, CUSTOMER_ID, "name", SEARCH_TEXT, ADDITIONAL_INFO, "type"},
-                        new String[]{"", "", "", "", "", "", "default"},
-                        "tb-assets");
-                log.info("Assets dumped.");
-
-                log.info("Dumping relations ...");
-                Path relationsDump = CassandraDbHelper.dumpCfIfExists(ks, cluster.getSession(), "relation",
-                        new String[]{"from_id", "from_type", "to_id", "to_type", "relation_type", ADDITIONAL_INFO, "relation_type_group"},
-                        new String[]{"", "", "", "", "", "", "COMMON"},
-                        "tb-relations");
-                log.info("Relations dumped.");
-
-                log.info("Updating schema ...");
-                Path schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "1.3.0", SCHEMA_UPDATE_CQL);
-                loadCql(schemaUpdateFile);
-                log.info("Schema updated.");
-
-                //Restore devices, assets and relations
-
-                log.info("Restoring devices ...");
-                if (devicesDump != null) {
-                    CassandraDbHelper.loadCf(ks, cluster.getSession(), DEVICE,
-                            new String[]{"id", TENANT_ID, CUSTOMER_ID, "name", SEARCH_TEXT, ADDITIONAL_INFO, "type"}, devicesDump);
-                    Files.deleteIfExists(devicesDump);
-                }
-                log.info("Devices restored.");
-
-                log.info("Dumping device types ...");
-                Path deviceTypesDump = CassandraDbHelper.dumpCfIfExists(ks, cluster.getSession(), DEVICE,
-                        new String[]{TENANT_ID, "type"},
-                        new String[]{"", ""},
-                        "tb-device-types");
-                if (deviceTypesDump != null) {
-                    CassandraDbHelper.appendToEndOfLine(deviceTypesDump, "DEVICE");
-                }
-                log.info("Device types dumped.");
-                log.info("Loading device types ...");
-                if (deviceTypesDump != null) {
-                    CassandraDbHelper.loadCf(ks, cluster.getSession(), "entity_subtype",
-                            new String[]{TENANT_ID, "type", "entity_type"}, deviceTypesDump);
-                    Files.deleteIfExists(deviceTypesDump);
-                }
-                log.info("Device types loaded.");
-
-                log.info("Restoring assets ...");
-                if (assetsDump != null) {
-                    CassandraDbHelper.loadCf(ks, cluster.getSession(), ASSET,
-                            new String[]{"id", TENANT_ID, CUSTOMER_ID, "name", SEARCH_TEXT, ADDITIONAL_INFO, "type"}, assetsDump);
-                    Files.deleteIfExists(assetsDump);
-                }
-                log.info("Assets restored.");
-
-                log.info("Dumping asset types ...");
-                Path assetTypesDump = CassandraDbHelper.dumpCfIfExists(ks, cluster.getSession(), ASSET,
-                        new String[]{TENANT_ID, "type"},
-                        new String[]{"", ""},
-                        "tb-asset-types");
-                if (assetTypesDump != null) {
-                    CassandraDbHelper.appendToEndOfLine(assetTypesDump, "ASSET");
-                }
-                log.info("Asset types dumped.");
-                log.info("Loading asset types ...");
-                if (assetTypesDump != null) {
-                    CassandraDbHelper.loadCf(ks, cluster.getSession(), "entity_subtype",
-                            new String[]{TENANT_ID, "type", "entity_type"}, assetTypesDump);
-                    Files.deleteIfExists(assetTypesDump);
-                }
-                log.info("Asset types loaded.");
-
-                log.info("Restoring relations ...");
-                if (relationsDump != null) {
-                    CassandraDbHelper.loadCf(ks, cluster.getSession(), "relation",
-                            new String[]{"from_id", "from_type", "to_id", "to_type", "relation_type", ADDITIONAL_INFO, "relation_type_group"}, relationsDump);
-                    Files.deleteIfExists(relationsDump);
-                }
-                log.info("Relations restored.");
-
-                break;
-            case "1.3.0":
-                break;
-            case "1.3.1":
-
-                cluster.getSession();
-
-                ks = cluster.getCluster().getMetadata().getKeyspace(cluster.getKeyspaceName());
-
-                log.info("Dumping dashboards ...");
-                Path dashboardsDump = CassandraDbHelper.dumpCfIfExists(ks, cluster.getSession(), DASHBOARD,
-                        new String[]{ID, TENANT_ID, CUSTOMER_ID, TITLE, SEARCH_TEXT, ASSIGNED_CUSTOMERS, CONFIGURATION},
-                        new String[]{"", "", "", "", "", "", ""},
-                        "tb-dashboards", true);
-                log.info("Dashboards dumped.");
-
-
-                log.info("Updating schema ...");
-                schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "1.4.0", SCHEMA_UPDATE_CQL);
-                loadCql(schemaUpdateFile);
-                log.info("Schema updated.");
-
-                log.info("Restoring dashboards ...");
-                if (dashboardsDump != null) {
-                    CassandraDbHelper.loadCf(ks, cluster.getSession(), DASHBOARD,
-                            new String[]{ID, TENANT_ID, TITLE, SEARCH_TEXT, CONFIGURATION}, dashboardsDump, true);
-                    DatabaseHelper.upgradeTo40_assignDashboards(dashboardsDump, dashboardService, false);
-                    Files.deleteIfExists(dashboardsDump);
-                }
-                log.info("Dashboards restored.");
-                break;
-            case "1.4.0":
-
-                log.info("Updating schema ...");
-                schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "2.0.0", SCHEMA_UPDATE_CQL);
-                loadCql(schemaUpdateFile);
-                log.info("Schema updated.");
-
-                break;
-
-            case "2.0.0":
-
-                log.info("Updating schema ...");
-                schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "2.1.1", SCHEMA_UPDATE_CQL);
-                loadCql(schemaUpdateFile);
-                log.info("Schema updated.");
-
-                break;
-
-            case "2.1.1":
-
-                log.info("Upgrading Cassandra DataBase from version {} to 2.1.2 ...", fromVersion);
-
-                cluster.getSession();
-
-                ks = cluster.getCluster().getMetadata().getKeyspace(cluster.getKeyspaceName());
-
-                log.info("Dumping entity views ...");
-                Path entityViewsDump = CassandraDbHelper.dumpCfIfExists(ks, cluster.getSession(), ENTITY_VIEWS,
-                        new String[]{ID, ENTITY_ID, ENTITY_TYPE, TENANT_ID, CUSTOMER_ID, NAME, TYPE, KEYS, START_TS, END_TS, SEARCH_TEXT, ADDITIONAL_INFO},
-                        new String[]{"", "", "", "", "", "", "default", "", "0", "0", "", ""},
-                        "tb-entity-views");
-                log.info("Entity views dumped.");
-
-                log.info("Updating schema ...");
-                schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "2.1.2", SCHEMA_UPDATE_CQL);
-                loadCql(schemaUpdateFile);
-                log.info("Schema updated.");
-
-                log.info("Restoring entity views ...");
-                if (entityViewsDump != null) {
-                    CassandraDbHelper.loadCf(ks, cluster.getSession(), ENTITY_VIEW,
-                            new String[]{ID, ENTITY_ID, ENTITY_TYPE, TENANT_ID, CUSTOMER_ID, NAME, TYPE, KEYS, START_TS, END_TS, SEARCH_TEXT, ADDITIONAL_INFO}, entityViewsDump);
-                    Files.deleteIfExists(entityViewsDump);
-                }
-                log.info("Entity views restored.");
-
-                break;
-            case "2.1.3":
-                break;
-            case "2.3.0":
-                break;
-            case "2.3.1":
-                log.info("Updating schema ...");
-                String updateDeviceTableStmt = "alter table device add label text";
-                try {
-                    cluster.getSession().execute(updateDeviceTableStmt);
-                    Thread.sleep(2500);
-                } catch (InvalidQueryException e) {
-                }
-                log.info("Schema updated.");
-                break;
-            case "2.4.1":
-                log.info("Updating schema ...");
-                String updateAssetTableStmt = "alter table asset add label text";
-                try {
-                    log.info("Updating assets ...");
-                    cluster.getSession().execute(updateAssetTableStmt);
-                    Thread.sleep(2500);
-                    log.info("Assets updated.");
-                } catch (InvalidQueryException e) {
-                }
-                log.info("Schema updated.");
-                break;
-            case "2.4.2":
-                log.info("Updating schema ...");
-                String updateAlarmTableStmt = "alter table alarm add propagate_relation_types text";
-                try {
-                    log.info("Updating alarms ...");
-                    cluster.getSession().execute(updateAlarmTableStmt);
-                    Thread.sleep(2500);
-                    log.info("Alarms updated.");
-                } catch (InvalidQueryException e) {
-                }
-                log.info("Schema updated.");
-                break;
-            case "2.4.3":
-                log.info("Updating schema ...");
-                String updateAttributeKvTableStmt = "alter table attributes_kv_cf add json_v text";
-                try {
-                    log.info("Updating attributes ...");
-                    cluster.getSession().execute(updateAttributeKvTableStmt);
-                    Thread.sleep(2500);
-                    log.info("Attributes updated.");
-                } catch (InvalidQueryException e) {
-                }
-                log.info("Schema updated.");
-                break;
-            default:
-                throw new RuntimeException("Unable to upgrade Cassandra database, unsupported fromVersion: " + fromVersion);
-        }
-    }
-
-}

+/**
+ * Copyright © 2016-2020 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.service.install.migrate;
 package org.thingsboard.server.service.install.migrate;
 
 
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;
@@ -6,7 +21,9 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Profile;
 import org.springframework.context.annotation.Profile;
 import org.springframework.stereotype.Service;
 import org.springframework.stereotype.Service;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.EntityType;
+import org.thingsboard.server.common.data.UUIDConverter;
 import org.thingsboard.server.dao.cassandra.CassandraCluster;
 import org.thingsboard.server.dao.cassandra.CassandraCluster;
+import org.thingsboard.server.dao.util.NoSqlAnyDao;
 import org.thingsboard.server.dao.util.SqlDao;
 import org.thingsboard.server.dao.util.SqlDao;
 import org.thingsboard.server.service.install.EntityDatabaseSchemaService;
 import org.thingsboard.server.service.install.EntityDatabaseSchemaService;
 
 
@@ -20,11 +37,13 @@ import static org.thingsboard.server.service.install.migrate.CassandraToSqlColum
 import static org.thingsboard.server.service.install.migrate.CassandraToSqlColumn.doubleColumn;
 import static org.thingsboard.server.service.install.migrate.CassandraToSqlColumn.doubleColumn;
 import static org.thingsboard.server.service.install.migrate.CassandraToSqlColumn.enumToIntColumn;
 import static org.thingsboard.server.service.install.migrate.CassandraToSqlColumn.enumToIntColumn;
 import static org.thingsboard.server.service.install.migrate.CassandraToSqlColumn.idColumn;
 import static org.thingsboard.server.service.install.migrate.CassandraToSqlColumn.idColumn;
+import static org.thingsboard.server.service.install.migrate.CassandraToSqlColumn.jsonColumn;
 import static org.thingsboard.server.service.install.migrate.CassandraToSqlColumn.stringColumn;
 import static org.thingsboard.server.service.install.migrate.CassandraToSqlColumn.stringColumn;
 
 
 @Service
 @Service
 @Profile("install")
 @Profile("install")
 @SqlDao
 @SqlDao
+@NoSqlAnyDao
 @Slf4j
 @Slf4j
 public class CassandraEntitiesToSqlMigrateService implements EntitiesMigrateService {
 public class CassandraEntitiesToSqlMigrateService implements EntitiesMigrateService {
 
 
@@ -49,7 +68,7 @@ public class CassandraEntitiesToSqlMigrateService implements EntitiesMigrateServ
         log.info("Performing migration of entities data from cassandra to SQL database ...");
         log.info("Performing migration of entities data from cassandra to SQL database ...");
         entityDatabaseSchemaService.createDatabaseSchema();
         entityDatabaseSchemaService.createDatabaseSchema();
         try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
         try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
-            conn.setAutoCommit(true);
+            conn.setAutoCommit(false);
             for (CassandraToSqlTable table: tables) {
             for (CassandraToSqlTable table: tables) {
                 table.migrateToSql(cluster.getSession(), conn);
                 table.migrateToSql(cluster.getSession(), conn);
             }
             }
@@ -60,7 +79,7 @@ public class CassandraEntitiesToSqlMigrateService implements EntitiesMigrateServ
     }
     }
 
 
     private static List<CassandraToSqlTable> tables = Arrays.asList(
     private static List<CassandraToSqlTable> tables = Arrays.asList(
-        new CassandraToSqlTable("admin_settings",
+       new CassandraToSqlTable("admin_settings",
                 idColumn("id"),
                 idColumn("id"),
                 stringColumn("key"),
                 stringColumn("key"),
                 stringColumn("json_value")),
                 stringColumn("json_value")),
@@ -87,7 +106,17 @@ public class CassandraEntitiesToSqlMigrateService implements EntitiesMigrateServ
                 stringColumn("type"),
                 stringColumn("type"),
                 stringColumn("label"),
                 stringColumn("label"),
                 stringColumn("search_text"),
                 stringColumn("search_text"),
-                stringColumn("additional_info")),
+                stringColumn("additional_info")) {
+            @Override
+            protected boolean onConstraintViolation(List<CassandraToSqlColumnData[]> batchData,
+                                                    CassandraToSqlColumnData[] data, String constraint) {
+                if (constraint.equalsIgnoreCase("asset_name_unq_key")) {
+                    this.handleUniqueNameViolation(data, "asset");
+                    return true;
+                }
+                return super.onConstraintViolation(batchData, data, constraint);
+            }
+        },
         new CassandraToSqlTable("audit_log_by_tenant_id", "audit_log",
         new CassandraToSqlTable("audit_log_by_tenant_id", "audit_log",
                 idColumn("id"),
                 idColumn("id"),
                 idColumn("tenant_id"),
                 idColumn("tenant_id"),
@@ -110,7 +139,7 @@ public class CassandraEntitiesToSqlMigrateService implements EntitiesMigrateServ
                 stringColumn("str_v"),
                 stringColumn("str_v"),
                 bigintColumn("long_v"),
                 bigintColumn("long_v"),
                 doubleColumn("dbl_v"),
                 doubleColumn("dbl_v"),
-                stringColumn("json_v"),
+                jsonColumn("json_v"),
                 bigintColumn("last_update_ts")),
                 bigintColumn("last_update_ts")),
         new CassandraToSqlTable("component_descriptor",
         new CassandraToSqlTable("component_descriptor",
                 idColumn("id"),
                 idColumn("id"),
@@ -150,7 +179,17 @@ public class CassandraEntitiesToSqlMigrateService implements EntitiesMigrateServ
                 stringColumn("type"),
                 stringColumn("type"),
                 stringColumn("label"),
                 stringColumn("label"),
                 stringColumn("search_text"),
                 stringColumn("search_text"),
-                stringColumn("additional_info")),
+                stringColumn("additional_info")) {
+            @Override
+            protected boolean onConstraintViolation(List<CassandraToSqlColumnData[]> batchData,
+                                                    CassandraToSqlColumnData[] data, String constraint) {
+                if (constraint.equalsIgnoreCase("device_name_unq_key")) {
+                    this.handleUniqueNameViolation(data, "device");
+                    return true;
+                }
+                return super.onConstraintViolation(batchData, data, constraint);
+            }
+        },
         new CassandraToSqlTable("device_credentials",
         new CassandraToSqlTable("device_credentials",
                 idColumn("id"),
                 idColumn("id"),
                 idColumn("device_id"),
                 idColumn("device_id"),
@@ -182,7 +221,17 @@ public class CassandraEntitiesToSqlMigrateService implements EntitiesMigrateServ
                 stringColumn("authority"),
                 stringColumn("authority"),
                 stringColumn("first_name"),
                 stringColumn("first_name"),
                 stringColumn("last_name"),
                 stringColumn("last_name"),
-                stringColumn("additional_info")),
+                stringColumn("additional_info")) {
+            @Override
+            protected boolean onConstraintViolation(List<CassandraToSqlColumnData[]> batchData,
+                                                    CassandraToSqlColumnData[] data, String constraint) {
+                if (constraint.equalsIgnoreCase("tb_user_email_key")) {
+                    this.handleUniqueEmailViolation(data);
+                    return true;
+                }
+                return super.onConstraintViolation(batchData, data, constraint);
+            }
+        },
         new CassandraToSqlTable("tenant",
         new CassandraToSqlTable("tenant",
                 idColumn("id"),
                 idColumn("id"),
                 stringColumn("title"),
                 stringColumn("title"),
@@ -203,7 +252,19 @@ public class CassandraEntitiesToSqlMigrateService implements EntitiesMigrateServ
                 booleanColumn("enabled"),
                 booleanColumn("enabled"),
                 stringColumn("password"),
                 stringColumn("password"),
                 stringColumn("activate_token"),
                 stringColumn("activate_token"),
-                stringColumn("reset_token")),
+                stringColumn("reset_token")) {
+            @Override
+            protected boolean onConstraintViolation(List<CassandraToSqlColumnData[]> batchData,
+                                                    CassandraToSqlColumnData[] data, String constraint) {
+                if (constraint.equalsIgnoreCase("user_credentials_user_id_key")) {
+                    String id = UUIDConverter.fromString(this.getColumnData(data, "id").getValue()).toString();
+                    log.warn("Found user credentials record with duplicate user_id [id:[{}]]. Record will be ignored!", id);
+                    this.ignoreRecord(batchData, data);
+                    return true;
+                }
+                return super.onConstraintViolation(batchData, data, constraint);
+            }
+        },
         new CassandraToSqlTable("widget_type",
         new CassandraToSqlTable("widget_type",
                 idColumn("id"),
                 idColumn("id"),
                 idColumn("tenant_id"),
                 idColumn("tenant_id"),

+/**
+ * Copyright © 2016-2020 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.service.install.migrate;
 package org.thingsboard.server.service.install.migrate;
 
 
 import com.datastax.driver.core.Row;
 import com.datastax.driver.core.Row;
@@ -7,14 +22,21 @@ import org.thingsboard.server.common.data.UUIDConverter;
 import java.sql.PreparedStatement;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.sql.SQLException;
 import java.sql.Types;
 import java.sql.Types;
+import java.util.regex.Pattern;
 
 
 @Data
 @Data
 public class CassandraToSqlColumn {
 public class CassandraToSqlColumn {
 
 
+    private static final ThreadLocal<Pattern> PATTERN_THREAD_LOCAL = ThreadLocal.withInitial(() -> Pattern.compile(String.valueOf(Character.MIN_VALUE)));
+    private static final String EMPTY_STR = "";
+
+    private int index;
+    private int sqlIndex;
     private String cassandraColumnName;
     private String cassandraColumnName;
     private String sqlColumnName;
     private String sqlColumnName;
     private CassandraToSqlColumnType type;
     private CassandraToSqlColumnType type;
     private int sqlType;
     private int sqlType;
+    private int size;
     private Class<? extends Enum> enumClass;
     private Class<? extends Enum> enumClass;
 
 
     public static CassandraToSqlColumn idColumn(String name) {
     public static CassandraToSqlColumn idColumn(String name) {
@@ -41,6 +63,10 @@ public class CassandraToSqlColumn {
         return new CassandraToSqlColumn(name, CassandraToSqlColumnType.BOOLEAN);
         return new CassandraToSqlColumn(name, CassandraToSqlColumnType.BOOLEAN);
     }
     }
 
 
+    public static CassandraToSqlColumn jsonColumn(String name) {
+        return new CassandraToSqlColumn(name, CassandraToSqlColumnType.JSON);
+    }
+
     public static CassandraToSqlColumn enumToIntColumn(String name, Class<? extends Enum> enumClass) {
     public static CassandraToSqlColumn enumToIntColumn(String name, Class<? extends Enum> enumClass) {
         return new CassandraToSqlColumn(name, CassandraToSqlColumnType.ENUM_TO_INT, enumClass);
         return new CassandraToSqlColumn(name, CassandraToSqlColumnType.ENUM_TO_INT, enumClass);
     }
     }
@@ -67,36 +93,9 @@ public class CassandraToSqlColumn {
         this.sqlColumnName = sqlColumnName;
         this.sqlColumnName = sqlColumnName;
         this.type = type;
         this.type = type;
         this.enumClass = enumClass;
         this.enumClass = enumClass;
-        switch (this.type) {
-            case ID:
-            case STRING:
-                this.sqlType = Types.VARCHAR;
-                break;
-            case DOUBLE:
-                this.sqlType = Types.DOUBLE;
-                break;
-            case INTEGER:
-            case ENUM_TO_INT:
-                this.sqlType = Types.INTEGER;
-                break;
-            case FLOAT:
-                this.sqlType = Types.FLOAT;
-                break;
-            case BIGINT:
-                this.sqlType = Types.BIGINT;
-                break;
-            case BOOLEAN:
-                this.sqlType = Types.BOOLEAN;
-                break;
-        }
     }
     }
 
 
-    public void prepareColumnValue(Row row, PreparedStatement sqlInsertStatement, int index) throws SQLException {
-        String value = this.getColumnValue(row, index);
-        this.setColumnValue(sqlInsertStatement, index, value);
-    }
-
-    private String getColumnValue(Row row, int index) {
+    public String getColumnValue(Row row) {
         if (row.isNull(index)) {
         if (row.isNull(index)) {
             return null;
             return null;
         } else {
         } else {
@@ -114,46 +113,56 @@ public class CassandraToSqlColumn {
                 case BOOLEAN:
                 case BOOLEAN:
                     return Boolean.toString(row.getBool(index));
                     return Boolean.toString(row.getBool(index));
                 case STRING:
                 case STRING:
+                case JSON:
                 case ENUM_TO_INT:
                 case ENUM_TO_INT:
                 default:
                 default:
-                   return row.getString(index);
+                    String value = row.getString(index);
+                    return this.replaceNullChars(value);
             }
             }
         }
         }
     }
     }
 
 
-    private void setColumnValue(PreparedStatement sqlInsertStatement, int index, String value) throws SQLException {
+    public void setColumnValue(PreparedStatement sqlInsertStatement, String value) throws SQLException {
         if (value == null) {
         if (value == null) {
-            sqlInsertStatement.setNull(index, this.sqlType);
+            sqlInsertStatement.setNull(this.sqlIndex, this.sqlType);
         } else {
         } else {
             switch (this.type) {
             switch (this.type) {
                 case DOUBLE:
                 case DOUBLE:
-                    sqlInsertStatement.setDouble(index, Double.parseDouble(value));
+                    sqlInsertStatement.setDouble(this.sqlIndex, Double.parseDouble(value));
                     break;
                     break;
                 case INTEGER:
                 case INTEGER:
-                    sqlInsertStatement.setInt(index, Integer.parseInt(value));
+                    sqlInsertStatement.setInt(this.sqlIndex, Integer.parseInt(value));
                     break;
                     break;
                 case FLOAT:
                 case FLOAT:
-                    sqlInsertStatement.setFloat(index, Float.parseFloat(value));
+                    sqlInsertStatement.setFloat(this.sqlIndex, Float.parseFloat(value));
                     break;
                     break;
                 case BIGINT:
                 case BIGINT:
-                    sqlInsertStatement.setLong(index, Long.parseLong(value));
+                    sqlInsertStatement.setLong(this.sqlIndex, Long.parseLong(value));
                     break;
                     break;
                 case BOOLEAN:
                 case BOOLEAN:
-                    sqlInsertStatement.setBoolean(index, Boolean.parseBoolean(value));
+                    sqlInsertStatement.setBoolean(this.sqlIndex, Boolean.parseBoolean(value));
                     break;
                     break;
                 case ENUM_TO_INT:
                 case ENUM_TO_INT:
                     Enum enumVal = Enum.valueOf(this.enumClass, value);
                     Enum enumVal = Enum.valueOf(this.enumClass, value);
                     int intValue = enumVal.ordinal();
                     int intValue = enumVal.ordinal();
-                    sqlInsertStatement.setInt(index, intValue);
+                    sqlInsertStatement.setInt(this.sqlIndex, intValue);
                     break;
                     break;
+                case JSON:
                 case STRING:
                 case STRING:
                 case ID:
                 case ID:
                 default:
                 default:
-                    sqlInsertStatement.setString(index, value);
+                    sqlInsertStatement.setString(this.sqlIndex, value);
                     break;
                     break;
             }
             }
         }
         }
     }
     }
 
 
+    private String replaceNullChars(String strValue) {
+        if (strValue != null) {
+            return PATTERN_THREAD_LOCAL.get().matcher(strValue).replaceAll(EMPTY_STR);
+        }
+        return strValue;
+    }
+
 }
 }
 
 

+/**
+ * Copyright © 2016-2020 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.service.install.migrate;
+
+import lombok.Data;
+
+@Data
+public class CassandraToSqlColumnData {
+
+    private String value;
+    private String originalValue;
+    private int constraintCounter = 0;
+
+    public CassandraToSqlColumnData(String value) {
+        this.value = value;
+        this.originalValue = value;
+    }
+
+    public int nextContraintCounter() {
+        return ++constraintCounter;
+    }
+
+    public String getNextConstraintStringValue(CassandraToSqlColumn column) {
+        int counter = this.nextContraintCounter();
+        String newValue = this.originalValue + counter;
+        int overflow = newValue.length() - column.getSize();
+        if (overflow > 0) {
+            newValue = this.originalValue.substring(0, this.originalValue.length()-overflow) + counter;
+        }
+        return newValue;
+    }
+
+    public String getNextConstraintEmailValue(CassandraToSqlColumn column) {
+        int counter = this.nextContraintCounter();
+        String[] emailValues = this.originalValue.split("@");
+        String newValue = emailValues[0] + "+" + counter + "@" + emailValues[1];
+        int overflow = newValue.length() - column.getSize();
+        if (overflow > 0) {
+            newValue = emailValues[0].substring(0, emailValues[0].length()-overflow) + "+" + counter + "@" + emailValues[1];
+        }
+        return newValue;
+    }
+
+    public String getLogValue() {
+        if (this.value != null && this.value.length() > 255) {
+            return this.value.substring(0, 255) + "...[truncated " + (this.value.length() - 255) + " symbols]";
+        }
+        return this.value;
+    }
+
+}

+/**
+ * Copyright © 2016-2020 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.service.install.migrate;
 package org.thingsboard.server.service.install.migrate;
 
 
 public enum CassandraToSqlColumnType {
 public enum CassandraToSqlColumnType {
@@ -8,5 +23,6 @@ public enum CassandraToSqlColumnType {
     BIGINT,
     BIGINT,
     BOOLEAN,
     BOOLEAN,
     STRING,
     STRING,
+    JSON,
     ENUM_TO_INT
     ENUM_TO_INT
 }
 }

+/**
+ * Copyright © 2016-2020 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.service.install.migrate;
 package org.thingsboard.server.service.install.migrate;
 
 
 import com.datastax.driver.core.ResultSet;
 import com.datastax.driver.core.ResultSet;
@@ -7,66 +22,255 @@ import com.datastax.driver.core.SimpleStatement;
 import com.datastax.driver.core.Statement;
 import com.datastax.driver.core.Statement;
 import lombok.Data;
 import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 import lombok.extern.slf4j.Slf4j;
+import org.hibernate.exception.ConstraintViolationException;
+import org.hibernate.internal.util.JdbcExceptionHelper;
+import org.postgresql.util.PSQLException;
+import org.thingsboard.server.common.data.UUIDConverter;
+import org.thingsboard.server.dao.exception.DataValidationException;
 
 
+import java.sql.BatchUpdateException;
 import java.sql.Connection;
 import java.sql.Connection;
+import java.sql.DatabaseMetaData;
 import java.sql.PreparedStatement;
 import java.sql.PreparedStatement;
 import java.sql.SQLException;
 import java.sql.SQLException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Arrays;
 import java.util.Iterator;
 import java.util.Iterator;
 import java.util.List;
 import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
 
 
 @Data
 @Data
 @Slf4j
 @Slf4j
 public class CassandraToSqlTable {
 public class CassandraToSqlTable {
 
 
+    private static final int DEFAULT_BATCH_SIZE = 10000;
+
     private String cassandraCf;
     private String cassandraCf;
     private String sqlTableName;
     private String sqlTableName;
 
 
     private List<CassandraToSqlColumn> columns;
     private List<CassandraToSqlColumn> columns;
 
 
+    private int batchSize = DEFAULT_BATCH_SIZE;
+
+    private PreparedStatement sqlInsertStatement;
+
     public CassandraToSqlTable(String tableName, CassandraToSqlColumn... columns) {
     public CassandraToSqlTable(String tableName, CassandraToSqlColumn... columns) {
-        this(tableName, tableName, columns);
+        this(tableName, tableName, DEFAULT_BATCH_SIZE, columns);
+    }
+
+    public CassandraToSqlTable(String tableName, String sqlTableName, CassandraToSqlColumn... columns) {
+        this(tableName, sqlTableName, DEFAULT_BATCH_SIZE, columns);
+    }
+
+    public CassandraToSqlTable(String tableName, int batchSize, CassandraToSqlColumn... columns) {
+        this(tableName, tableName, batchSize, columns);
     }
     }
 
 
-    public CassandraToSqlTable(String cassandraCf, String sqlTableName, CassandraToSqlColumn... columns) {
+    public CassandraToSqlTable(String cassandraCf, String sqlTableName, int batchSize, CassandraToSqlColumn... columns) {
         this.cassandraCf = cassandraCf;
         this.cassandraCf = cassandraCf;
         this.sqlTableName = sqlTableName;
         this.sqlTableName = sqlTableName;
+        this.batchSize = batchSize;
         this.columns = Arrays.asList(columns);
         this.columns = Arrays.asList(columns);
+        for (int i=0;i<columns.length;i++) {
+            this.columns.get(i).setIndex(i);
+            this.columns.get(i).setSqlIndex(i+1);
+        }
     }
     }
 
 
     public void migrateToSql(Session session, Connection conn) throws SQLException {
     public void migrateToSql(Session session, Connection conn) throws SQLException {
-        log.info("Migrating data from cassandra '{}' Column Family to '{}' SQL table...", this.cassandraCf, this.sqlTableName);
-        PreparedStatement sqlInsertStatement = createSqlInsertStatement(conn);
+        log.info("[{}] Migrating data from cassandra '{}' Column Family to '{}' SQL table...", this.sqlTableName, this.cassandraCf, this.sqlTableName);
+        DatabaseMetaData metadata = conn.getMetaData();
+        java.sql.ResultSet resultSet = metadata.getColumns(null, null, this.sqlTableName, null);
+        while (resultSet.next()) {
+            String name = resultSet.getString("COLUMN_NAME");
+            int sqlType = resultSet.getInt("DATA_TYPE");
+            int size = resultSet.getInt("COLUMN_SIZE");
+            CassandraToSqlColumn column = this.getColumn(name);
+            column.setSize(size);
+            column.setSqlType(sqlType);
+        }
+        this.sqlInsertStatement = createSqlInsertStatement(conn);
         Statement cassandraSelectStatement = createCassandraSelectStatement();
         Statement cassandraSelectStatement = createCassandraSelectStatement();
         cassandraSelectStatement.setFetchSize(100);
         cassandraSelectStatement.setFetchSize(100);
         ResultSet rs = session.execute(cassandraSelectStatement);
         ResultSet rs = session.execute(cassandraSelectStatement);
         Iterator<Row> iter = rs.iterator();
         Iterator<Row> iter = rs.iterator();
         int rowCounter = 0;
         int rowCounter = 0;
-        while (iter.hasNext()) {
+        List<CassandraToSqlColumnData[]> batchData;
+        boolean hasNext;
+        do {
+            batchData = this.extractBatchData(iter);
+            hasNext = batchData.size() == this.batchSize;
+            this.batchInsert(batchData, conn);
+            rowCounter += batchData.size();
+            log.info("[{}] {} records migrated so far...", this.sqlTableName, rowCounter);
+        } while (hasNext);
+        this.sqlInsertStatement.close();
+        log.info("[{}] {} total records migrated.", this.sqlTableName, rowCounter);
+        log.info("[{}] Finished migration data from cassandra '{}' Column Family to '{}' SQL table.",
+                this.sqlTableName, this.cassandraCf, this.sqlTableName);
+    }
+
+    private List<CassandraToSqlColumnData[]> extractBatchData(Iterator<Row> iter) {
+        List<CassandraToSqlColumnData[]> batchData = new ArrayList<>();
+        while (iter.hasNext() && batchData.size() < this.batchSize) {
             Row row = iter.next();
             Row row = iter.next();
             if (row != null) {
             if (row != null) {
-                this.migrateRowToSql(row, sqlInsertStatement);
-                rowCounter++;
-                if (rowCounter % 100 == 0) {
-                    sqlInsertStatement.executeBatch();
-                    log.info("{} records migrated so far...", rowCounter);
-                }
+                CassandraToSqlColumnData[] data = this.extractRowData(row);
+                batchData.add(data);
             }
             }
         }
         }
-        if (rowCounter % 100 > 0) {
-            sqlInsertStatement.executeBatch();
+        return batchData;
+    }
+
+    private CassandraToSqlColumnData[] extractRowData(Row row) {
+        CassandraToSqlColumnData[] data = new CassandraToSqlColumnData[this.columns.size()];
+        for (CassandraToSqlColumn column: this.columns) {
+            String value = column.getColumnValue(row);
+            data[column.getIndex()] = new CassandraToSqlColumnData(value);
         }
         }
-        sqlInsertStatement.close();
-        log.info("{} total records migrated.", rowCounter);
-        log.info("Finished migration data from cassandra '{}' Column Family to '{}' SQL table.", this.cassandraCf, this.sqlTableName);
+        return this.validateColumnData(data);
     }
     }
 
 
-    private void migrateRowToSql(Row row, PreparedStatement sqlInsertStatement) throws SQLException {
-        for (int i=0; i<this.columns.size();i++) {
+    private CassandraToSqlColumnData[] validateColumnData(CassandraToSqlColumnData[] data) {
+        for (int i=0;i<data.length;i++) {
             CassandraToSqlColumn column = this.columns.get(i);
             CassandraToSqlColumn column = this.columns.get(i);
-            column.prepareColumnValue(row, sqlInsertStatement, i);
+            if (column.getType() == CassandraToSqlColumnType.STRING) {
+                CassandraToSqlColumnData columnData = data[i];
+                String value = columnData.getValue();
+                if (value != null && value.length() > column.getSize()) {
+                    log.warn("[{}] Value size [{}] exceeds maximum size [{}] of column [{}] and will be truncated!",
+                            this.sqlTableName,
+                            value.length(), column.getSize(), column.getSqlColumnName());
+                    log.warn("[{}] Affected data:\n{}", this.sqlTableName, this.dataToString(data));
+                    value = value.substring(0, column.getSize());
+                    columnData.setOriginalValue(value);
+                    columnData.setValue(value);
+                }
+            }
+        }
+        return data;
+    }
+
+    private void batchInsert(List<CassandraToSqlColumnData[]> batchData, Connection conn) throws SQLException {
+        boolean retry = false;
+        for (CassandraToSqlColumnData[] data : batchData) {
+            for (CassandraToSqlColumn column: this.columns) {
+                column.setColumnValue(this.sqlInsertStatement, data[column.getIndex()].getValue());
+            }
+            try {
+                this.sqlInsertStatement.executeUpdate();
+            } catch (SQLException e) {
+                if (this.handleInsertException(batchData, data, conn, e)) {
+                    retry = true;
+                    break;
+                } else {
+                    throw e;
+                }
+            }
+        }
+        if (retry) {
+            this.batchInsert(batchData, conn);
+        } else {
+            conn.commit();
+        }
+    }
+
+    private boolean handleInsertException(List<CassandraToSqlColumnData[]> batchData,
+                                          CassandraToSqlColumnData[] data,
+                                          Connection conn, SQLException ex) throws SQLException {
+        conn.commit();
+        String constraint = extractConstraintName(ex).orElse(null);
+        if (constraint != null) {
+            if (this.onConstraintViolation(batchData, data, constraint)) {
+                return true;
+            } else {
+                log.error("[{}] Unhandled constraint violation [{}] during insert!", this.sqlTableName, constraint);
+                log.error("[{}] Affected data:\n{}", this.sqlTableName, this.dataToString(data));
+            }
+        } else {
+            log.error("[{}] Unhandled exception during insert!", this.sqlTableName);
+            log.error("[{}] Affected data:\n{}", this.sqlTableName, this.dataToString(data));
         }
         }
-        sqlInsertStatement.addBatch();
+        return false;
+    }
+
+    private String dataToString(CassandraToSqlColumnData[] data) {
+        StringBuffer stringData = new StringBuffer("{\n");
+        for (int i=0;i<data.length;i++) {
+            String columnName = this.columns.get(i).getSqlColumnName();
+            String value = data[i].getLogValue();
+            stringData.append("\"").append(columnName).append("\": ").append("[").append(value).append("]\n");
+        }
+        stringData.append("}");
+        return stringData.toString();
+    }
+
+    protected boolean onConstraintViolation(List<CassandraToSqlColumnData[]> batchData,
+                                            CassandraToSqlColumnData[] data, String constraint) {
+        return false;
+    }
+
+    protected void handleUniqueNameViolation(CassandraToSqlColumnData[] data, String entityType) {
+        CassandraToSqlColumn nameColumn = this.getColumn("name");
+        CassandraToSqlColumn searchTextColumn = this.getColumn("search_text");
+        CassandraToSqlColumnData nameColumnData = data[nameColumn.getIndex()];
+        CassandraToSqlColumnData searchTextColumnData = data[searchTextColumn.getIndex()];
+        String prevName = nameColumnData.getValue();
+        String newName = nameColumnData.getNextConstraintStringValue(nameColumn);
+        nameColumnData.setValue(newName);
+        searchTextColumnData.setValue(searchTextColumnData.getNextConstraintStringValue(searchTextColumn));
+        String id = UUIDConverter.fromString(this.getColumnData(data, "id").getValue()).toString();
+        log.warn("Found {} with duplicate name [id:[{}]]. Attempting to rename {} from '{}' to '{}'...", entityType, id, entityType, prevName, newName);
+    }
+
+    protected void handleUniqueEmailViolation(CassandraToSqlColumnData[] data) {
+        CassandraToSqlColumn emailColumn = this.getColumn("email");
+        CassandraToSqlColumn searchTextColumn = this.getColumn("search_text");
+        CassandraToSqlColumnData emailColumnData = data[emailColumn.getIndex()];
+        CassandraToSqlColumnData searchTextColumnData = data[searchTextColumn.getIndex()];
+        String prevEmail = emailColumnData.getValue();
+        String newEmail = emailColumnData.getNextConstraintEmailValue(emailColumn);
+        emailColumnData.setValue(newEmail);
+        searchTextColumnData.setValue(searchTextColumnData.getNextConstraintEmailValue(searchTextColumn));
+        String id = UUIDConverter.fromString(this.getColumnData(data, "id").getValue()).toString();
+        log.warn("Found user with duplicate email [id:[{}]]. Attempting to rename email from '{}' to '{}'...", id, prevEmail, newEmail);
+    }
+
+    protected void ignoreRecord(List<CassandraToSqlColumnData[]> batchData, CassandraToSqlColumnData[] data) {
+        log.warn("[{}] Affected data:\n{}", this.sqlTableName, this.dataToString(data));
+        int index = batchData.indexOf(data);
+        if (index > 0) {
+            batchData.remove(index);
+        }
+    }
+
+    protected CassandraToSqlColumn getColumn(String sqlColumnName) {
+        return this.columns.stream().filter(col -> col.getSqlColumnName().equals(sqlColumnName)).findFirst().get();
+    }
+
+    protected CassandraToSqlColumnData getColumnData(CassandraToSqlColumnData[] data, String sqlColumnName) {
+        CassandraToSqlColumn column = this.getColumn(sqlColumnName);
+        return data[column.getIndex()];
+    }
+
+    private Optional<String> extractConstraintName(SQLException ex) {
+        final String sqlState = JdbcExceptionHelper.extractSqlState( ex );
+        if (sqlState != null) {
+            String sqlStateClassCode = JdbcExceptionHelper.determineSqlStateClassCode( sqlState );
+            if ( sqlStateClassCode != null ) {
+                if (Arrays.asList(
+                        "23",	// "integrity constraint violation"
+                        "27",	// "triggered data change violation"
+                        "44"	// "with check option violation"
+                ).contains(sqlStateClassCode)) {
+                    if (ex instanceof PSQLException) {
+                        return Optional.of(((PSQLException)ex).getServerErrorMessage().getConstraint());
+                    }
+                }
+            }
+        }
+        return Optional.empty();
     }
     }
 
 
     private Statement createCassandraSelectStatement() {
     private Statement createCassandraSelectStatement() {
@@ -88,8 +292,13 @@ public class CassandraToSqlTable {
         }
         }
         insertStatementBuilder.deleteCharAt(insertStatementBuilder.length() - 1);
         insertStatementBuilder.deleteCharAt(insertStatementBuilder.length() - 1);
         insertStatementBuilder.append(") VALUES (");
         insertStatementBuilder.append(") VALUES (");
-        for (CassandraToSqlColumn ignored : columns) {
-            insertStatementBuilder.append("?").append(",");
+        for (CassandraToSqlColumn column : columns) {
+            if (column.getType() == CassandraToSqlColumnType.JSON) {
+                insertStatementBuilder.append("cast(? AS json)");
+            } else {
+                insertStatementBuilder.append("?");
+            }
+            insertStatementBuilder.append(",");
         }
         }
         insertStatementBuilder.deleteCharAt(insertStatementBuilder.length() - 1);
         insertStatementBuilder.deleteCharAt(insertStatementBuilder.length() - 1);
         insertStatementBuilder.append(")");
         insertStatementBuilder.append(")");

+/**
+ * Copyright © 2016-2020 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.service.install.migrate;
 package org.thingsboard.server.service.install.migrate;
 
 
 public interface EntitiesMigrateService {
 public interface EntitiesMigrateService {

@@ -20,7 +20,7 @@ import BaseGauge = CanvasGauges.BaseGauge;
 import { FontStyle, FontWeight } from '@home/components/widget/lib/settings.models';
 import { FontStyle, FontWeight } from '@home/components/widget/lib/settings.models';
 import * as tinycolor_ from 'tinycolor2';
 import * as tinycolor_ from 'tinycolor2';
 import { ColorFormats } from 'tinycolor2';
 import { ColorFormats } from 'tinycolor2';
-import { isDefined, isUndefined } from '@core/utils';
+import { isDefined, isString, isUndefined } from '@core/utils';
 
 
 const tinycolor = tinycolor_;
 const tinycolor = tinycolor_;
 
 
@@ -32,13 +32,20 @@ export interface DigitalGaugeColorRange {
   rgbString: string;
   rgbString: string;
 }
 }
 
 
+export interface colorLevelSetting {
+  value: number;
+  color: string;
+}
+
+export type levelColors = Array<string | colorLevelSetting>;
+
 export interface CanvasDigitalGaugeOptions extends GenericOptions {
 export interface CanvasDigitalGaugeOptions extends GenericOptions {
   gaugeType?: GaugeType;
   gaugeType?: GaugeType;
   gaugeWithScale?: number;
   gaugeWithScale?: number;
   dashThickness?: number;
   dashThickness?: number;
   roundedLineCap?: boolean;
   roundedLineCap?: boolean;
   gaugeColor?: string;
   gaugeColor?: string;
-  levelColors?: string[];
+  levelColors?: levelColors;
   symbol?: string;
   symbol?: string;
   label?: string;
   label?: string;
   hideValue?: boolean;
   hideValue?: boolean;
@@ -229,26 +236,30 @@ export class CanvasDigitalGauge extends BaseGauge {
     }
     }
 
 
     const colorsCount = options.levelColors.length;
     const colorsCount = options.levelColors.length;
+    const isColorProperty = isString(options.levelColors[0]);
     const inc = colorsCount > 1 ? (1 / (colorsCount - 1)) : 1;
     const inc = colorsCount > 1 ? (1 / (colorsCount - 1)) : 1;
     options.colorsRange = [];
     options.colorsRange = [];
     if (options.neonGlowBrightness) {
     if (options.neonGlowBrightness) {
       options.neonColorsRange = [];
       options.neonColorsRange = [];
     }
     }
     for (let i = 0; i < options.levelColors.length; i++) {
     for (let i = 0; i < options.levelColors.length; i++) {
-      const percentage = inc * i;
-      let tColor = tinycolor(options.levelColors[i]);
-      options.colorsRange[i] = {
-        pct: percentage,
-        color: tColor.toRgb(),
-        rgbString: tColor.toRgbString()
-      };
-      if (options.neonGlowBrightness) {
-        tColor = tinycolor(options.levelColors[i]).brighten(options.neonGlowBrightness);
-        options.neonColorsRange[i] = {
+      let levelColor: any = options.levelColors[i];
+      if (levelColor !== null) {
+        let percentage = isColorProperty ? inc * i : CanvasDigitalGauge.normalizeValue(levelColor.value, options.minValue, options.maxValue);
+        let tColor = tinycolor(isColorProperty ? levelColor : levelColor.color);
+        options.colorsRange.push({
           pct: percentage,
           pct: percentage,
           color: tColor.toRgb(),
           color: tColor.toRgb(),
           rgbString: tColor.toRgbString()
           rgbString: tColor.toRgbString()
-        };
+        });
+        if (options.neonGlowBrightness) {
+          tColor = tinycolor(isColorProperty ? levelColor : levelColor.color).brighten(options.neonGlowBrightness);
+          options.neonColorsRange.push({
+            pct: percentage,
+            color: tColor.toRgb(),
+            rgbString: tColor.toRgbString()
+          });
+        }
       }
       }
     }
     }
 
 
@@ -262,6 +273,17 @@ export class CanvasDigitalGauge extends BaseGauge {
     return options;
     return options;
   }
   }
 
 
+  static normalizeValue (value: number, min: number, max: number): number {
+    let normalValue = (value - min) / (max - min);
+    if (normalValue <= 0) {
+      return 0;
+    }
+    if (normalValue >= 1) {
+      return 1;
+    }
+    return normalValue;
+  }
+
   private initValueClone() {
   private initValueClone() {
     const canvas = this.canvas;
     const canvas = this.canvas;
     this.elementValueClone = canvas.element.cloneNode(true) as HTMLCanvasElementClone;
     this.elementValueClone = canvas.element.cloneNode(true) as HTMLCanvasElementClone;

@@ -19,6 +19,26 @@ import { GaugeType } from '@home/components/widget/lib/canvas-digital-gauge';
 import { AnimationRule } from '@home/components/widget/lib/analogue-gauge.models';
 import { AnimationRule } from '@home/components/widget/lib/analogue-gauge.models';
 import { FontSettings } from '@home/components/widget/lib/settings.models';
 import { FontSettings } from '@home/components/widget/lib/settings.models';
 
 
+export interface colorLevelProperty {
+  valueSource: string;
+  entityAlias?: string;
+  attribute?: string;
+  value?: number;
+}
+
+export interface fixedLevelColors {
+  from?: colorLevelProperty;
+  to?: colorLevelProperty;
+  color: string;
+}
+
+export interface colorLevelSetting {
+  value: number;
+  color: string;
+}
+
+export type colorLevel = Array<string | colorLevelSetting>;
+
 export interface DigitalGaugeSettings {
 export interface DigitalGaugeSettings {
   minValue?: number;
   minValue?: number;
   maxValue?: number;
   maxValue?: number;
@@ -38,7 +58,9 @@ export interface DigitalGaugeSettings {
   gaugeWidthScale?: number;
   gaugeWidthScale?: number;
   defaultColor?: string;
   defaultColor?: string;
   gaugeColor?: string;
   gaugeColor?: string;
-  levelColors?: string[];
+  useFixedLevelColor?: boolean;
+  levelColors?: colorLevel;
+  fixedLevelColors?: fixedLevelColors[];
   animation?: boolean;
   animation?: boolean;
   animationDuration?: number;
   animationDuration?: number;
   animationRule?: AnimationRule;
   animationRule?: AnimationRule;
@@ -147,6 +169,11 @@ export const digitalGaugeSettingsSchema: JsonSettingsSchema = {
         type: 'string',
         type: 'string',
         default: null
         default: null
       },
       },
+      useFixedLevelColor: {
+        title: 'Use precise value for the color indicator',
+        type: 'boolean',
+        default: false
+      },
       levelColors: {
       levelColors: {
         title: 'Colors of indicator, from lower to upper',
         title: 'Colors of indicator, from lower to upper',
         type: 'array',
         type: 'array',
@@ -155,6 +182,66 @@ export const digitalGaugeSettingsSchema: JsonSettingsSchema = {
           type: 'string'
           type: 'string'
         }
         }
       },
       },
+      fixedLevelColors: {
+        title: 'The colors for the indicator using boundary values',
+        type: 'array',
+        items: {
+          title: 'levelColor',
+          type: 'object',
+          properties: {
+            from: {
+              title: 'From',
+              type: 'object',
+              properties: {
+                valueSource: {
+                  title: '[From] Value source',
+                  type: 'string',
+                  default: 'predefinedValue'
+                },
+                entityAlias: {
+                  title: '[From] Source entity alias',
+                  type: 'string'
+                },
+                attribute: {
+                  title: '[From] Source entity attribute',
+                  type: 'string'
+                },
+                value: {
+                  title: '[From] Value (if predefined value is selected)',
+                  type: 'number'
+                }
+              }
+            },
+            to: {
+              title: 'To',
+              type: 'object',
+              properties: {
+                valueSource: {
+                  title: '[To] Value source',
+                  type: 'string',
+                  default: 'predefinedValue'
+                },
+                entityAlias: {
+                  title: '[To] Source entity alias',
+                  type: 'string'
+                },
+                attribute: {
+                  title: '[To] Source entity attribute',
+                  type: 'string'
+                },
+                value: {
+                  title: '[To] Value (if predefined value is selected)',
+                  type: 'number'
+                }
+              }
+            },
+            color: {
+              title: 'Color',
+              type: 'string'
+            }
+          }
+        }
+      },
       animation: {
       animation: {
         title: 'Enable animation',
         title: 'Enable animation',
         type: 'boolean',
         type: 'boolean',
@@ -343,8 +430,10 @@ export const digitalGaugeSettingsSchema: JsonSettingsSchema = {
       key: 'gaugeColor',
       key: 'gaugeColor',
       type: 'color'
       type: 'color'
     },
     },
+    'useFixedLevelColor',
     {
     {
       key: 'levelColors',
       key: 'levelColors',
+      condition: 'model.useFixedLevelColor !== true',
       items: [
       items: [
         {
         {
           key: 'levelColors[]',
           key: 'levelColors[]',
@@ -352,6 +441,52 @@ export const digitalGaugeSettingsSchema: JsonSettingsSchema = {
         }
         }
       ]
       ]
     },
     },
+    {
+      key: 'fixedLevelColors',
+      condition: 'model.useFixedLevelColor === true',
+      items: [
+        {
+          key: 'fixedLevelColors[].from.valueSource',
+          type: 'rc-select',
+          multiple: false,
+          items: [
+            {
+              value: 'predefinedValue',
+              label: 'Predefined value (Default)'
+            },
+            {
+              value: 'entityAttribute',
+              label: 'Value taken from entity attribute'
+            }
+          ]
+        },
+        'fixedLevelColors[].from.value',
+        'fixedLevelColors[].from.entityAlias',
+        'fixedLevelColors[].from.attribute',
+        {
+          key: 'fixedLevelColors[].to.valueSource',
+          type: 'rc-select',
+          multiple: false,
+          items: [
+            {
+              value: 'predefinedValue',
+              label: 'Predefined value (Default)'
+            },
+            {
+              value: 'entityAttribute',
+              label: 'Value taken from entity attribute'
+            }
+          ]
+        },
+        'fixedLevelColors[].to.value',
+        'fixedLevelColors[].to.entityAlias',
+        'fixedLevelColors[].to.attribute',
+        {
+          key: 'fixedLevelColors[].color',
+          type: 'color'
+        }
+      ]
+    },
     'animation',
     'animation',
     'animationDuration',
     'animationDuration',
     {
     {

@@ -16,14 +16,20 @@
 
 
 import * as CanvasGauges from 'canvas-gauges';
 import * as CanvasGauges from 'canvas-gauges';
 import { WidgetContext } from '@home/models/widget-component.models';
 import { WidgetContext } from '@home/models/widget-component.models';
-import { DigitalGaugeSettings, digitalGaugeSettingsSchema } from '@home/components/widget/lib/digital-gauge.models';
+import {
+  colorLevelSetting,
+  DigitalGaugeSettings,
+  digitalGaugeSettingsSchema
+} from '@home/components/widget/lib/digital-gauge.models';
 import * as tinycolor_ from 'tinycolor2';
 import * as tinycolor_ from 'tinycolor2';
 import { isDefined } from '@core/utils';
 import { isDefined } from '@core/utils';
 import { prepareFontSettings } from '@home/components/widget/lib/settings.models';
 import { prepareFontSettings } from '@home/components/widget/lib/settings.models';
 import { CanvasDigitalGauge, CanvasDigitalGaugeOptions } from '@home/components/widget/lib/canvas-digital-gauge';
 import { CanvasDigitalGauge, CanvasDigitalGaugeOptions } from '@home/components/widget/lib/canvas-digital-gauge';
 import { DatePipe } from '@angular/common';
 import { DatePipe } from '@angular/common';
-import { JsonSettingsSchema } from '@shared/models/widget.models';
+import {DataKey, Datasource, DatasourceType, JsonSettingsSchema, widgetType} from '@shared/models/widget.models';
 import GenericOptions = CanvasGauges.GenericOptions;
 import GenericOptions = CanvasGauges.GenericOptions;
+import {IWidgetSubscription, WidgetSubscriptionOptions} from "@core/api/widget-api.models";
+import {DataKeyType} from "@shared/models/telemetry/telemetry.models";
 
 
 const tinycolor = tinycolor_;
 const tinycolor = tinycolor_;
 
 
@@ -32,6 +38,7 @@ const digitalGaugeSettingsSchemaValue = digitalGaugeSettingsSchema;
 export class TbCanvasDigitalGauge {
 export class TbCanvasDigitalGauge {
 
 
   private localSettings: DigitalGaugeSettings;
   private localSettings: DigitalGaugeSettings;
+  private levelColorsSourcesSubscription: IWidgetSubscription;
 
 
   private gauge: CanvasDigitalGauge;
   private gauge: CanvasDigitalGauge;
 
 
@@ -65,10 +72,16 @@ export class TbCanvasDigitalGauge {
     this.localSettings.gaugeWidthScale = settings.gaugeWidthScale || 0.75;
     this.localSettings.gaugeWidthScale = settings.gaugeWidthScale || 0.75;
     this.localSettings.gaugeColor = settings.gaugeColor || tinycolor(keyColor).setAlpha(0.2).toRgbString();
     this.localSettings.gaugeColor = settings.gaugeColor || tinycolor(keyColor).setAlpha(0.2).toRgbString();
 
 
-    if (!settings.levelColors || settings.levelColors.length <= 0) {
-      this.localSettings.levelColors = [keyColor];
+    this.localSettings.useFixedLevelColor = settings.useFixedLevelColor || false;
+    if (!settings.useFixedLevelColor) {
+      if (!settings.levelColors || settings.levelColors.length <= 0) {
+        this.localSettings.levelColors = [keyColor];
+      } else {
+        this.localSettings.levelColors = settings.levelColors.slice();
+      }
     } else {
     } else {
-      this.localSettings.levelColors = settings.levelColors.slice();
+      this.localSettings.levelColors = [keyColor];
+      this.localSettings.fixedLevelColors = settings.fixedLevelColors || [];
     }
     }
 
 
     this.localSettings.decimals = isDefined(dataKey.decimals) ? dataKey.decimals :
     this.localSettings.decimals = isDefined(dataKey.decimals) ? dataKey.decimals :
@@ -176,6 +189,130 @@ export class TbCanvasDigitalGauge {
     };
     };
 
 
     this.gauge = new CanvasDigitalGauge(gaugeData).draw();
     this.gauge = new CanvasDigitalGauge(gaugeData).draw();
+    this.init();
+  }
+
+  init() {
+    if (this.localSettings.useFixedLevelColor) {
+      if (this.localSettings.fixedLevelColors && this.localSettings.fixedLevelColors.length > 0) {
+        this.localSettings.levelColors = this.settingLevelColorsSubscribe(this.localSettings.fixedLevelColors);
+        this.updateLevelColors(this.localSettings.levelColors);
+      }
+    }
+  }
+
+  settingLevelColorsSubscribe(options) {
+    let levelColorsDatasource: Datasource[] = [];
+    let predefineLevelColors: colorLevelSetting[] = [];
+
+    function setLevelColor(levelSetting, color) {
+      if (levelSetting.valueSource === 'predefinedValue' && isFinite(levelSetting.value)) {
+        predefineLevelColors.push({
+          value: levelSetting.value,
+          color: color
+        })
+      } else if (levelSetting.entityAlias && levelSetting.attribute) {
+        let entityAliasId = this.ctx.aliasController.getEntityAliasId(levelSetting.entityAlias);
+        if (!entityAliasId) {
+          return;
+        }
+
+        let datasource = levelColorsDatasource.find((datasource) => {
+          return datasource.entityAliasId === entityAliasId;
+        });
+
+        let dataKey: DataKey = {
+          type: DataKeyType.attribute,
+          name: levelSetting.attribute,
+          label: levelSetting.attribute,
+          settings: [{
+            color: color,
+            index: predefineLevelColors.length
+          }],
+          _hash: Math.random()
+        };
+
+        if (datasource) {
+          let findDataKey = datasource.dataKeys.find((dataKey) => {
+            return dataKey.name === levelSetting.attribute;
+          });
+
+          if (findDataKey) {
+            findDataKey.settings.push({
+              color: color,
+              index: predefineLevelColors.length
+            });
+          } else {
+            datasource.dataKeys.push(dataKey)
+          }
+        } else {
+          let datasource: Datasource = {
+            type: DatasourceType.entity,
+            name: levelSetting.entityAlias,
+            aliasName: levelSetting.entityAlias,
+            entityAliasId: entityAliasId,
+            dataKeys: [dataKey]
+          };
+          levelColorsDatasource.push(datasource);
+        }
+
+        predefineLevelColors.push(null);
+      }
+    }
+
+    for (let i = 0; i < options.length; i++) {
+      let levelColor = options[i];
+      if (levelColor.from) {
+        setLevelColor.call(this, levelColor.from, levelColor.color);
+      }
+      if (levelColor.to) {
+        setLevelColor.call(this, levelColor.to, levelColor.color);
+      }
+    }
+
+    this.subscribeLevelColorsAttributes(levelColorsDatasource);
+
+    return predefineLevelColors;
+  }
+
+  updateLevelColors(levelColors) {
+    (this.gauge.options as CanvasDigitalGaugeOptions).levelColors = levelColors;
+    this.gauge.options = CanvasDigitalGauge.configure(this.gauge.options);
+    this.gauge.update({} as CanvasDigitalGaugeOptions);
+  }
+
+  subscribeLevelColorsAttributes(datasources: Datasource[]) {
+    let TbCanvasDigitalGauge = this;
+    let levelColorsSourcesSubscriptionOptions: WidgetSubscriptionOptions = {
+      datasources: datasources,
+      useDashboardTimewindow: false,
+      type: widgetType.latest,
+      callbacks: {
+        onDataUpdated: (subscription) => {
+          for (let i = 0; i < subscription.data.length; i++) {
+            let keyData = subscription.data[i];
+            if (keyData && keyData.data && keyData.data[0]) {
+              let attrValue = keyData.data[0][1];
+              if (isFinite(attrValue)) {
+                for (let i = 0; i < keyData.dataKey.settings.length; i++) {
+                  let setting = keyData.dataKey.settings[i];
+                  this.localSettings.levelColors[setting.index] = {
+                    value: attrValue,
+                    color: setting.color
+                  };
+                }
+              }
+            }
+          }
+          this.updateLevelColors(this.localSettings.levelColors);
+        }
+      }
+    };
+    this.ctx.subscriptionApi.createSubscription(levelColorsSourcesSubscriptionOptions, true).subscribe(
+      (subscription) => {
+        TbCanvasDigitalGauge.levelColorsSourcesSubscription = subscription;
+      }
+    );
   }
   }
 
 
   update() {
   update() {

@@ -37,7 +37,7 @@ export class MailServerComponent extends PageComponent implements OnInit, HasCon
   adminSettings: AdminSettings<MailServerSettings>;
   adminSettings: AdminSettings<MailServerSettings>;
   smtpProtocols = ['smtp', 'smtps'];
   smtpProtocols = ['smtp', 'smtps'];
 
 
-  tlsVersions = ['TLSv1.0', 'TLSv1.1', 'TLSv1.2', 'TLSv1.3'];
+  tlsVersions = ['TLSv1', 'TLSv1.1', 'TLSv1.2', 'TLSv1.3'];
 
 
   constructor(protected store: Store<AppState>,
   constructor(protected store: Store<AppState>,
               private router: Router,
               private router: Router,

@@ -186,7 +186,6 @@ export class JsonFormComponent implements OnInit, ControlValueAccessor, Validato
       val = undefined;
       val = undefined;
     }
     }
     if (JsonFormUtils.updateValue(key, this.model, val) || forceUpdate) {
     if (JsonFormUtils.updateValue(key, this.model, val) || forceUpdate) {
-      this.formProps.model = this.model;
       this.isModelValid = this.validateModel();
       this.isModelValid = this.validateModel();
       this.updateView();
       this.updateView();
     }
     }
@@ -233,7 +232,7 @@ export class JsonFormComponent implements OnInit, ControlValueAccessor, Validato
     this.formProps.schema = this.schema;
     this.formProps.schema = this.schema;
     this.formProps.form = this.form;
     this.formProps.form = this.form;
     this.formProps.groupInfoes = this.groupInfoes;
     this.formProps.groupInfoes = this.groupInfoes;
-    this.formProps.model = deepClone(this.model);
+    this.formProps.model = this.model;
     this.renderReactSchemaForm();
     this.renderReactSchemaForm();
   }
   }