Merge branch 'master' of https://github.com/thingsboard/thingsboard into gatewayStateVlad

application/src/main/data/upgrade/2.4.3/schema_update_psql_ts.sql

@@ -14,7 +14,7 @@
 -- limitations under the License.
 --
 
--- load function check_version()
+-- select check_version();
 
 CREATE OR REPLACE FUNCTION check_version() RETURNS boolean AS $$
 DECLARE
@@ -38,9 +38,9 @@ BEGIN
 END;
 $$ LANGUAGE 'plpgsql';
 
--- load function create_partition_table()
+-- select create_partition_ts_kv_table();
 
-CREATE OR REPLACE FUNCTION create_partition_table() RETURNS VOID AS $$
+CREATE OR REPLACE FUNCTION create_partition_ts_kv_table() RETURNS VOID AS $$
 
 BEGIN
   ALTER TABLE ts_kv
@@ -59,8 +59,32 @@ BEGIN
 END;
 $$ LANGUAGE 'plpgsql';
 
+-- select create_new_ts_kv_latest_table();
 
--- load function create_partitions()
+CREATE OR REPLACE FUNCTION create_new_ts_kv_latest_table() RETURNS VOID AS $$
+
+BEGIN
+  ALTER TABLE ts_kv_latest
+    RENAME TO ts_kv_latest_old;
+  ALTER TABLE ts_kv_latest_old
+     RENAME CONSTRAINT ts_kv_latest_pkey TO ts_kv_latest_pkey_old;
+  CREATE TABLE IF NOT EXISTS ts_kv_latest
+  (
+    LIKE ts_kv_latest_old
+  );
+  ALTER TABLE ts_kv_latest
+    DROP COLUMN entity_type;
+  ALTER TABLE ts_kv_latest
+    ALTER COLUMN entity_id TYPE uuid USING entity_id::uuid;
+  ALTER TABLE ts_kv_latest
+    ALTER COLUMN key TYPE integer USING key::integer;
+  ALTER TABLE ts_kv_latest
+    ADD CONSTRAINT ts_kv_latest_pkey PRIMARY KEY (entity_id, key);
+END;
+$$ LANGUAGE 'plpgsql';
+
+
+-- select create_partitions();
 
 CREATE OR REPLACE FUNCTION create_partitions() RETURNS VOID AS
 $$
@@ -89,7 +113,7 @@ BEGIN
 END;
 $$ language 'plpgsql';
 
--- load function create_ts_kv_dictionary_table()
+-- select create_ts_kv_dictionary_table();
 
 CREATE OR REPLACE FUNCTION create_ts_kv_dictionary_table() RETURNS VOID AS $$
 
@@ -103,7 +127,7 @@ BEGIN
 END;
 $$ LANGUAGE 'plpgsql';
 
--- load function insert_into_dictionary()
+-- select insert_into_dictionary();
 
 CREATE OR REPLACE FUNCTION insert_into_dictionary() RETURNS VOID AS
 $$
@@ -128,7 +152,7 @@ BEGIN
 END;
 $$ language 'plpgsql';
 
--- load function insert_into_ts_kv()
+-- select insert_into_ts_kv();
 
 CREATE OR REPLACE FUNCTION insert_into_ts_kv() RETURNS void AS
 $$
@@ -136,18 +160,18 @@ DECLARE
     insert_size CONSTANT integer := 10000;
     insert_counter       integer DEFAULT 0;
     insert_record        RECORD;
-    insert_cursor CURSOR FOR SELECT CONCAT(first, '-', second, '-1', third, '-', fourth, '-', fifth)::uuid AS entity_id,
+    insert_cursor CURSOR FOR SELECT CONCAT(first_part_uuid, '-', second_part_uuid, '-1', third_part_uuid, '-', fourth_part_uuid, '-', fifth_part_uuid)::uuid AS entity_id,
-                                    substrings.key                                                         AS key,
+                                    ts_kv_records.key                                                         AS key,
-                                    substrings.ts                                                          AS ts,
+                                    ts_kv_records.ts                                                          AS ts,
-                                    substrings.bool_v                                                      AS bool_v,
+                                    ts_kv_records.bool_v                                                      AS bool_v,
-                                    substrings.str_v                                                       AS str_v,
+                                    ts_kv_records.str_v                                                       AS str_v,
-                                    substrings.long_v                                                      AS long_v,
+                                    ts_kv_records.long_v                                                      AS long_v,
-                                    substrings.dbl_v                                                       AS dbl_v
+                                    ts_kv_records.dbl_v                                                       AS dbl_v
-                             FROM (SELECT SUBSTRING(entity_id, 8, 8)  AS first,
+                             FROM (SELECT SUBSTRING(entity_id, 8, 8)  AS first_part_uuid,
-                                          SUBSTRING(entity_id, 4, 4)  AS second,
+                                          SUBSTRING(entity_id, 4, 4)  AS second_part_uuid,
-                                          SUBSTRING(entity_id, 1, 3)  AS third,
+                                          SUBSTRING(entity_id, 1, 3)  AS third_part_uuid,
-                                          SUBSTRING(entity_id, 16, 4) AS fourth,
+                                          SUBSTRING(entity_id, 16, 4) AS fourth_part_uuid,
-                                          SUBSTRING(entity_id, 20)    AS fifth,
+                                          SUBSTRING(entity_id, 20)    AS fifth_part_uuid,
                                           key_id                      AS key,
                                           ts,
                                           bool_v,
@@ -155,7 +179,7 @@ DECLARE
                                           long_v,
                                           dbl_v
                                    FROM ts_kv_old
-                                            INNER JOIN ts_kv_dictionary ON (ts_kv_old.key = ts_kv_dictionary.key)) AS substrings;
+                                            INNER JOIN ts_kv_dictionary ON (ts_kv_old.key = ts_kv_dictionary.key)) AS ts_kv_records;
 BEGIN
     OPEN insert_cursor;
     LOOP
@@ -176,4 +200,52 @@ BEGIN
 END;
 $$ LANGUAGE 'plpgsql';
 
+-- select insert_into_ts_kv_latest();
+
+CREATE OR REPLACE FUNCTION insert_into_ts_kv_latest() RETURNS void AS
+$$
+DECLARE
+    insert_size CONSTANT integer := 10000;
+    insert_counter       integer DEFAULT 0;
+    insert_record        RECORD;
+    insert_cursor CURSOR FOR SELECT CONCAT(first_part_uuid, '-', second_part_uuid, '-1', third_part_uuid, '-', fourth_part_uuid, '-', fifth_part_uuid)::uuid AS entity_id,
+                                    ts_kv_latest_records.key                                                         AS key,
+                                    ts_kv_latest_records.ts                                                          AS ts,
+                                    ts_kv_latest_records.bool_v                                                      AS bool_v,
+                                    ts_kv_latest_records.str_v                                                       AS str_v,
+                                    ts_kv_latest_records.long_v                                                      AS long_v,
+                                    ts_kv_latest_records.dbl_v                                                       AS dbl_v
+                             FROM (SELECT SUBSTRING(entity_id, 8, 8)  AS first_part_uuid,
+                                          SUBSTRING(entity_id, 4, 4)  AS second_part_uuid,
+                                          SUBSTRING(entity_id, 1, 3)  AS third_part_uuid,
+                                          SUBSTRING(entity_id, 16, 4) AS fourth_part_uuid,
+                                          SUBSTRING(entity_id, 20)    AS fifth_part_uuid,
+                                          key_id                      AS key,
+                                          ts,
+                                          bool_v,
+                                          str_v,
+                                          long_v,
+                                          dbl_v
+                                   FROM ts_kv_latest_old
+                                            INNER JOIN ts_kv_dictionary ON (ts_kv_latest_old.key = ts_kv_dictionary.key)) AS ts_kv_latest_records;
+BEGIN
+    OPEN insert_cursor;
+    LOOP
+        insert_counter := insert_counter + 1;
+        FETCH insert_cursor INTO insert_record;
+        IF NOT FOUND THEN
+            RAISE NOTICE '% records have been inserted into the ts_kv_latest!',insert_counter - 1;
+            EXIT;
+        END IF;
+        INSERT INTO ts_kv_latest(entity_id, key, ts, bool_v, str_v, long_v, dbl_v)
+        VALUES (insert_record.entity_id, insert_record.key, insert_record.ts, insert_record.bool_v, insert_record.str_v,
+                insert_record.long_v, insert_record.dbl_v);
+        IF MOD(insert_counter, insert_size) = 0 THEN
+            RAISE NOTICE '% records have been inserted into the ts_kv_latest!',insert_counter;
+        END IF;
+    END LOOP;
+    CLOSE insert_cursor;
+END;
+$$ LANGUAGE 'plpgsql';
+
 

application/src/main/data/upgrade/2.4.3/schema_update_timescale_ts.sql

@@ -0,0 +1,213 @@
+--
+-- 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.
+--
+
+-- select check_version();
+
+CREATE OR REPLACE FUNCTION check_version() RETURNS boolean AS $$
+DECLARE
+    current_version integer;
+    valid_version boolean;
+BEGIN
+    RAISE NOTICE 'Check the current installed PostgreSQL version...';
+    SELECT current_setting('server_version_num') INTO current_version;
+    IF current_version < 90600 THEN
+        valid_version := FALSE;
+    ELSE
+        valid_version := TRUE;
+    END IF;
+    IF valid_version = FALSE THEN
+        RAISE NOTICE 'Postgres version should be at least more than 9.6!';
+    ELSE
+        RAISE NOTICE 'PostgreSQL version is valid!';
+        RAISE NOTICE 'Schema update started...';
+    END IF;
+    RETURN valid_version;
+END;
+$$ LANGUAGE 'plpgsql';
+
+-- select create_new_tenant_ts_kv_table();
+
+CREATE OR REPLACE FUNCTION create_new_tenant_ts_kv_table() RETURNS VOID AS $$
+
+BEGIN
+  ALTER TABLE tenant_ts_kv
+    RENAME TO tenant_ts_kv_old;
+  CREATE TABLE IF NOT EXISTS tenant_ts_kv
+  (
+    LIKE tenant_ts_kv_old
+  );
+  ALTER TABLE tenant_ts_kv
+    ALTER COLUMN tenant_id TYPE uuid USING tenant_id::uuid;
+  ALTER TABLE tenant_ts_kv
+    ALTER COLUMN entity_id TYPE uuid USING entity_id::uuid;
+  ALTER TABLE tenant_ts_kv
+    ALTER COLUMN key TYPE integer USING key::integer;
+  ALTER TABLE tenant_ts_kv
+    ADD CONSTRAINT tenant_ts_kv_pkey PRIMARY KEY(tenant_id, entity_id, key, ts);
+  ALTER INDEX idx_tenant_ts_kv RENAME TO idx_tenant_ts_kv_old;
+  ALTER INDEX tenant_ts_kv_ts_idx RENAME TO tenant_ts_kv_ts_idx_old;
+--   PERFORM create_hypertable('tenant_ts_kv', 'ts', chunk_time_interval => 86400000, if_not_exists => true);
+  CREATE INDEX IF NOT EXISTS idx_tenant_ts_kv ON tenant_ts_kv(tenant_id, entity_id, key, ts);
+END;
+$$ LANGUAGE 'plpgsql';
+
+
+-- select create_ts_kv_latest_table();
+
+CREATE OR REPLACE FUNCTION create_ts_kv_latest_table() RETURNS VOID AS $$
+
+BEGIN
+    CREATE TABLE IF NOT EXISTS ts_kv_latest
+    (
+        entity_id uuid NOT NULL,
+        key int NOT NULL,
+        ts bigint NOT NULL,
+        bool_v boolean,
+        str_v varchar(10000000),
+        long_v bigint,
+        dbl_v double precision,
+        CONSTRAINT ts_kv_latest_pkey PRIMARY KEY (entity_id, key)
+    );
+END;
+$$ LANGUAGE 'plpgsql';
+
+
+-- select create_ts_kv_dictionary_table();
+
+CREATE OR REPLACE FUNCTION create_ts_kv_dictionary_table() RETURNS VOID AS $$
+
+BEGIN
+  CREATE TABLE IF NOT EXISTS ts_kv_dictionary
+  (
+    key    varchar(255) NOT NULL,
+    key_id serial UNIQUE,
+    CONSTRAINT ts_key_id_pkey PRIMARY KEY (key)
+  );
+END;
+$$ LANGUAGE 'plpgsql';
+
+-- select insert_into_dictionary();
+
+CREATE OR REPLACE FUNCTION insert_into_dictionary() RETURNS VOID AS
+$$
+DECLARE
+    insert_record RECORD;
+    key_cursor CURSOR FOR SELECT DISTINCT key
+                          FROM tenant_ts_kv_old
+                          ORDER BY key;
+BEGIN
+    OPEN key_cursor;
+    LOOP
+        FETCH key_cursor INTO insert_record;
+        EXIT WHEN NOT FOUND;
+        IF NOT EXISTS(SELECT key FROM ts_kv_dictionary WHERE key = insert_record.key) THEN
+            INSERT INTO ts_kv_dictionary(key) VALUES (insert_record.key);
+            RAISE NOTICE 'Key: % has been inserted into the dictionary!',insert_record.key;
+        ELSE
+            RAISE NOTICE 'Key: % already exists in the dictionary!',insert_record.key;
+        END IF;
+    END LOOP;
+    CLOSE key_cursor;
+END;
+$$ language 'plpgsql';
+
+-- select insert_into_tenant_ts_kv();
+
+CREATE OR REPLACE FUNCTION insert_into_tenant_ts_kv() RETURNS void AS
+$$
+DECLARE
+    insert_size CONSTANT integer := 10000;
+    insert_counter       integer DEFAULT 0;
+    insert_record        RECORD;
+    insert_cursor CURSOR FOR SELECT CONCAT(tenant_id_first_part_uuid, '-', tenant_id_second_part_uuid, '-1', tenant_id_third_part_uuid, '-', tenant_id_fourth_part_uuid, '-', tenant_id_fifth_part_uuid)::uuid AS tenant_id,
+                                    CONCAT(entity_id_first_part_uuid, '-', entity_id_second_part_uuid, '-1', entity_id_third_part_uuid, '-', entity_id_fourth_part_uuid, '-', entity_id_fifth_part_uuid)::uuid AS entity_id,
+                                    tenant_ts_kv_records.key                                                         AS key,
+                                    tenant_ts_kv_records.ts                                                          AS ts,
+                                    tenant_ts_kv_records.bool_v                                                      AS bool_v,
+                                    tenant_ts_kv_records.str_v                                                       AS str_v,
+                                    tenant_ts_kv_records.long_v                                                      AS long_v,
+                                    tenant_ts_kv_records.dbl_v                                                       AS dbl_v
+                             FROM (SELECT SUBSTRING(tenant_id, 8, 8)  AS tenant_id_first_part_uuid,
+                                          SUBSTRING(tenant_id, 4, 4)  AS tenant_id_second_part_uuid,
+                                          SUBSTRING(tenant_id, 1, 3)  AS tenant_id_third_part_uuid,
+                                          SUBSTRING(tenant_id, 16, 4) AS tenant_id_fourth_part_uuid,
+                                          SUBSTRING(tenant_id, 20)    AS tenant_id_fifth_part_uuid,
+                                          SUBSTRING(entity_id, 8, 8)  AS entity_id_first_part_uuid,
+                                          SUBSTRING(entity_id, 4, 4)  AS entity_id_second_part_uuid,
+                                          SUBSTRING(entity_id, 1, 3)  AS entity_id_third_part_uuid,
+                                          SUBSTRING(entity_id, 16, 4) AS entity_id_fourth_part_uuid,
+                                          SUBSTRING(entity_id, 20)    AS entity_id_fifth_part_uuid,
+                                          key_id                      AS key,
+                                          ts,
+                                          bool_v,
+                                          str_v,
+                                          long_v,
+                                          dbl_v
+                                   FROM tenant_ts_kv_old
+                                            INNER JOIN ts_kv_dictionary ON (tenant_ts_kv_old.key = ts_kv_dictionary.key)) AS tenant_ts_kv_records;
+BEGIN
+    OPEN insert_cursor;
+    LOOP
+        insert_counter := insert_counter + 1;
+        FETCH insert_cursor INTO insert_record;
+        IF NOT FOUND THEN
+            RAISE NOTICE '% records have been inserted into the new tenant_ts_kv table!',insert_counter - 1;
+            EXIT;
+        END IF;
+        INSERT INTO tenant_ts_kv(tenant_id, entity_id, key, ts, bool_v, str_v, long_v, dbl_v)
+        VALUES (insert_record.tenant_id, insert_record.entity_id, insert_record.key, insert_record.ts, insert_record.bool_v, insert_record.str_v,
+                insert_record.long_v, insert_record.dbl_v);
+        IF MOD(insert_counter, insert_size) = 0 THEN
+            RAISE NOTICE '% records have been inserted into the new tenant_ts_kv table!',insert_counter;
+        END IF;
+    END LOOP;
+    CLOSE insert_cursor;
+END;
+$$ LANGUAGE 'plpgsql';
+
+-- select insert_into_ts_kv_latest();
+
+CREATE OR REPLACE FUNCTION insert_into_ts_kv_latest() RETURNS void AS
+$$
+DECLARE
+    insert_size CONSTANT integer := 10000;
+    insert_counter       integer DEFAULT 0;
+    latest_record        RECORD;
+    insert_record        RECORD;
+    insert_cursor CURSOR FOR SELECT
+                                    latest_records.key          AS key,
+                                    latest_records.entity_id    AS entity_id,
+                                    latest_records.ts           AS ts
+                             FROM (SELECT DISTINCT key AS key, entity_id AS entity_id, MAX(ts) AS ts FROM tenant_ts_kv GROUP BY key, entity_id) AS latest_records;
+BEGIN
+    OPEN insert_cursor;
+    LOOP
+        insert_counter := insert_counter + 1;
+        FETCH insert_cursor INTO latest_record;
+        IF NOT FOUND THEN
+            RAISE NOTICE '% records have been inserted into the ts_kv_latest table!',insert_counter - 1;
+            EXIT;
+        END IF;
+        SELECT entity_id AS entity_id, key AS key, ts AS ts, bool_v AS bool_v, str_v AS str_v, long_v AS long_v, dbl_v AS dbl_v INTO insert_record FROM tenant_ts_kv WHERE entity_id = latest_record.entity_id AND key = latest_record.key AND ts = latest_record.ts;
+        INSERT INTO ts_kv_latest(entity_id, key, ts, bool_v, str_v, long_v, dbl_v)
+        VALUES (insert_record.entity_id, insert_record.key, insert_record.ts, insert_record.bool_v, insert_record.str_v, insert_record.long_v, insert_record.dbl_v);
+        IF MOD(insert_counter, insert_size) = 0 THEN
+            RAISE NOTICE '% records have been inserted into the ts_kv_latest table!',insert_counter;
+        END IF;
+    END LOOP;
+    CLOSE insert_cursor;
+END;
+$$ LANGUAGE 'plpgsql';

application/src/main/java/org/thingsboard/server/install/ThingsboardInstallService.java

@@ -53,7 +53,7 @@ public class ThingsboardInstallService {
     @Autowired
     private DatabaseEntitiesUpgradeService databaseEntitiesUpgradeService;
 
-    @Autowired
+    @Autowired(required = false)
     private DatabaseTsUpgradeService databaseTsUpgradeService;
 
     @Autowired

application/src/main/java/org/thingsboard/server/service/install/AbstractSqlTsDatabaseUpgradeService.java

@@ -0,0 +1,124 @@
+/**
+ * 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 lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.sql.CallableStatement;
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.sql.Types;
+
+@Slf4j
+public abstract class AbstractSqlTsDatabaseUpgradeService {
+
+    protected static final String CALL_REGEX = "call ";
+    protected static final String CHECK_VERSION = "check_version()";
+    protected static final String DROP_TABLE = "DROP TABLE ";
+    protected static final String DROP_FUNCTION_IF_EXISTS = "DROP FUNCTION IF EXISTS ";
+
+    private static final String CALL_CHECK_VERSION = CALL_REGEX + CHECK_VERSION;
+
+
+    private static final String FUNCTION = "function: {}";
+    private static final String DROP_STATEMENT = "drop statement: {}";
+    private static final String QUERY = "query: {}";
+    private static final String SUCCESSFULLY_EXECUTED = "Successfully executed ";
+    private static final String FAILED_TO_EXECUTE = "Failed to execute ";
+    private static final String FAILED_DUE_TO = " due to: {}";
+
+    protected static final String SUCCESSFULLY_EXECUTED_FUNCTION = SUCCESSFULLY_EXECUTED + FUNCTION;
+    protected static final String FAILED_TO_EXECUTE_FUNCTION_DUE_TO = FAILED_TO_EXECUTE + FUNCTION + FAILED_DUE_TO;
+
+    protected static final String SUCCESSFULLY_EXECUTED_DROP_STATEMENT = SUCCESSFULLY_EXECUTED + DROP_STATEMENT;
+    protected static final String FAILED_TO_EXECUTE_DROP_STATEMENT = FAILED_TO_EXECUTE + DROP_STATEMENT + FAILED_DUE_TO;
+
+    protected static final String SUCCESSFULLY_EXECUTED_QUERY = SUCCESSFULLY_EXECUTED + QUERY;
+    protected static final String FAILED_TO_EXECUTE_QUERY = FAILED_TO_EXECUTE + QUERY + FAILED_DUE_TO;
+
+    @Value("${spring.datasource.url}")
+    protected String dbUrl;
+
+    @Value("${spring.datasource.username}")
+    protected String dbUserName;
+
+    @Value("${spring.datasource.password}")
+    protected String dbPassword;
+
+    @Autowired
+    protected InstallScripts installScripts;
+
+    protected abstract void loadSql(Connection conn);
+
+    protected void loadFunctions(Path sqlFile, Connection conn) throws Exception {
+        String sql = new String(Files.readAllBytes(sqlFile), StandardCharsets.UTF_8);
+        conn.createStatement().execute(sql); //NOSONAR, ignoring because method used to execute thingsboard database upgrade script
+    }
+
+    protected boolean checkVersion(Connection conn) {
+        log.info("Check the current PostgreSQL version...");
+        boolean versionValid = false;
+        try {
+            CallableStatement callableStatement = conn.prepareCall("{? = " + CALL_CHECK_VERSION + " }");
+            callableStatement.registerOutParameter(1, Types.BOOLEAN);
+            callableStatement.execute();
+            versionValid = callableStatement.getBoolean(1);
+            callableStatement.close();
+        } catch (Exception e) {
+            log.info("Failed to check current PostgreSQL version due to: {}", e.getMessage());
+        }
+        return versionValid;
+    }
+
+    protected void executeFunction(Connection conn, String query) {
+        log.info("{} ... ", query);
+        try {
+            CallableStatement callableStatement = conn.prepareCall("{" + query + "}");
+            callableStatement.execute();
+            callableStatement.close();
+            log.info(SUCCESSFULLY_EXECUTED_FUNCTION, query.replace(CALL_REGEX, ""));
+            Thread.sleep(2000);
+        } catch (Exception e) {
+            log.info(FAILED_TO_EXECUTE_FUNCTION_DUE_TO, query, e.getMessage());
+        }
+    }
+
+    protected void executeDropStatement(Connection conn, String query) {
+        try {
+            conn.createStatement().execute(query); //NOSONAR, ignoring because method used to execute thingsboard database upgrade script
+            log.info(SUCCESSFULLY_EXECUTED_DROP_STATEMENT, query);
+            Thread.sleep(5000);
+        } catch (InterruptedException | SQLException e) {
+            log.info(FAILED_TO_EXECUTE_DROP_STATEMENT, query, e.getMessage());
+        }
+    }
+
+    protected void executeQuery(Connection conn, String query) {
+        try {
+            conn.createStatement().execute(query); //NOSONAR, ignoring because method used to execute thingsboard database upgrade script
+            log.info(SUCCESSFULLY_EXECUTED_QUERY, query);
+            Thread.sleep(5000);
+        } catch (InterruptedException | SQLException e) {
+            log.info(FAILED_TO_EXECUTE_QUERY, query, e.getMessage());
+        }
+    }
+
+}

application/src/main/java/org/thingsboard/server/service/install/PsqlTsDatabaseUpgradeService.java

@@ -16,51 +16,55 @@
 package org.thingsboard.server.service.install;
 
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Profile;
 import org.springframework.stereotype.Service;
 import org.thingsboard.server.dao.util.PsqlDao;
 import org.thingsboard.server.dao.util.SqlTsDao;
 
-import java.nio.charset.StandardCharsets;
-import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.sql.CallableStatement;
 import java.sql.Connection;
 import java.sql.DriverManager;
-import java.sql.SQLException;
-import java.sql.Types;
 
 @Service
 @Profile("install")
 @Slf4j
 @SqlTsDao
 @PsqlDao
-public class PsqlTsDatabaseUpgradeService implements DatabaseTsUpgradeService {
+public class PsqlTsDatabaseUpgradeService extends AbstractSqlTsDatabaseUpgradeService implements DatabaseTsUpgradeService {
 
-    private static final String CALL_REGEX = "call ";
     private static final String LOAD_FUNCTIONS_SQL = "schema_update_psql_ts.sql";
-    private static final String CHECK_VERSION = CALL_REGEX + "check_version()";
-    private static final String CREATE_PARTITION_TABLE = CALL_REGEX + "create_partition_table()";
-    private static final String CREATE_PARTITIONS = CALL_REGEX + "create_partitions()";
-    private static final String CREATE_TS_KV_DICTIONARY_TABLE = CALL_REGEX + "create_ts_kv_dictionary_table()";
-    private static final String INSERT_INTO_DICTIONARY = CALL_REGEX + "insert_into_dictionary()";
-    private static final String INSERT_INTO_TS_KV = CALL_REGEX + "insert_into_ts_kv()";
-    private static final String DROP_OLD_TABLE = "DROP TABLE ts_kv_old;";
 
-    @Value("${spring.datasource.url}")
+    private static final String TS_KV_OLD = "ts_kv_old;";
-    private String dbUrl;
+    private static final String TS_KV_LATEST_OLD = "ts_kv_latest_old;";
 
-    @Value("${spring.datasource.username}")
+    private static final String CREATE_PARTITION_TS_KV_TABLE = "create_partition_ts_kv_table()";
-    private String dbUserName;
+    private static final String CREATE_NEW_TS_KV_LATEST_TABLE = "create_new_ts_kv_latest_table()";
+    private static final String CREATE_PARTITIONS = "create_partitions()";
+    private static final String CREATE_TS_KV_DICTIONARY_TABLE = "create_ts_kv_dictionary_table()";
+    private static final String INSERT_INTO_DICTIONARY = "insert_into_dictionary()";
+    private static final String INSERT_INTO_TS_KV = "insert_into_ts_kv()";
+    private static final String INSERT_INTO_TS_KV_LATEST = "insert_into_ts_kv_latest()";
 
-    @Value("${spring.datasource.password}")
+    private static final String CALL_CREATE_PARTITION_TS_KV_TABLE = CALL_REGEX + CREATE_PARTITION_TS_KV_TABLE;
-    private String dbPassword;
+    private static final String CALL_CREATE_NEW_TS_KV_LATEST_TABLE = CALL_REGEX + CREATE_NEW_TS_KV_LATEST_TABLE;
+    private static final String CALL_CREATE_PARTITIONS = CALL_REGEX + CREATE_PARTITIONS;
+    private static final String CALL_CREATE_TS_KV_DICTIONARY_TABLE = CALL_REGEX + CREATE_TS_KV_DICTIONARY_TABLE;
+    private static final String CALL_INSERT_INTO_DICTIONARY = CALL_REGEX + INSERT_INTO_DICTIONARY;
+    private static final String CALL_INSERT_INTO_TS_KV = CALL_REGEX + INSERT_INTO_TS_KV;
+    private static final String CALL_INSERT_INTO_TS_KV_LATEST = CALL_REGEX + INSERT_INTO_TS_KV_LATEST;
 
-    @Autowired
+    private static final String DROP_TABLE_TS_KV_OLD = DROP_TABLE + TS_KV_OLD;
-    private InstallScripts installScripts;
+    private static final String DROP_TABLE_TS_KV_LATEST_OLD = DROP_TABLE + TS_KV_LATEST_OLD;
+
+    private static final String DROP_FUNCTION_CHECK_VERSION = DROP_FUNCTION_IF_EXISTS + CHECK_VERSION;
+    private static final String DROP_FUNCTION_CREATE_PARTITION_TS_KV_TABLE = DROP_FUNCTION_IF_EXISTS + CREATE_PARTITION_TS_KV_TABLE;
+    private static final String DROP_FUNCTION_CREATE_NEW_TS_KV_LATEST_TABLE = DROP_FUNCTION_IF_EXISTS + CREATE_NEW_TS_KV_LATEST_TABLE;
+    private static final String DROP_FUNCTION_CREATE_PARTITIONS = DROP_FUNCTION_IF_EXISTS + CREATE_PARTITIONS;
+    private static final String DROP_FUNCTION_CREATE_TS_KV_DICTIONARY_TABLE = DROP_FUNCTION_IF_EXISTS + CREATE_TS_KV_DICTIONARY_TABLE;
+    private static final String DROP_FUNCTION_INSERT_INTO_DICTIONARY = DROP_FUNCTION_IF_EXISTS + INSERT_INTO_DICTIONARY;
+    private static final String DROP_FUNCTION_INSERT_INTO_TS_KV = DROP_FUNCTION_IF_EXISTS + INSERT_INTO_TS_KV;
+    private static final String DROP_FUNCTION_INSERT_INTO_TS_KV_LATEST = DROP_FUNCTION_IF_EXISTS + INSERT_INTO_TS_KV_LATEST;
 
     @Override
     public void upgradeDatabase(String fromVersion) throws Exception {
@@ -70,7 +74,6 @@ public class PsqlTsDatabaseUpgradeService implements DatabaseTsUpgradeService {
                     log.info("Updating timeseries schema ...");
                     log.info("Load upgrade functions ...");
                     loadSql(conn);
-                    log.info("Upgrade functions successfully loaded!");
                     boolean versionValid = checkVersion(conn);
                     if (!versionValid) {
                         log.info("PostgreSQL version should be at least more than 10!");
@@ -78,12 +81,26 @@ public class PsqlTsDatabaseUpgradeService implements DatabaseTsUpgradeService {
                     } else {
                         log.info("PostgreSQL version is valid!");
                         log.info("Updating schema ...");
-                        executeFunction(conn, CREATE_PARTITION_TABLE);
+                        executeFunction(conn, CALL_CREATE_PARTITION_TS_KV_TABLE);
-                        executeFunction(conn, CREATE_PARTITIONS);
+                        executeFunction(conn, CALL_CREATE_PARTITIONS);
-                        executeFunction(conn, CREATE_TS_KV_DICTIONARY_TABLE);
+                        executeFunction(conn, CALL_CREATE_TS_KV_DICTIONARY_TABLE);
-                        executeFunction(conn, INSERT_INTO_DICTIONARY);
+                        executeFunction(conn, CALL_INSERT_INTO_DICTIONARY);
-                        executeFunction(conn, INSERT_INTO_TS_KV);
+                        executeFunction(conn, CALL_INSERT_INTO_TS_KV);
-                        dropOldTable(conn, DROP_OLD_TABLE);
+                        executeFunction(conn, CALL_CREATE_NEW_TS_KV_LATEST_TABLE);
+                        executeFunction(conn, CALL_INSERT_INTO_TS_KV_LATEST);
+
+                        executeDropStatement(conn, DROP_TABLE_TS_KV_OLD);
+                        executeDropStatement(conn, DROP_TABLE_TS_KV_LATEST_OLD);
+
+                        executeDropStatement(conn, DROP_FUNCTION_CHECK_VERSION);
+                        executeDropStatement(conn, DROP_FUNCTION_CREATE_PARTITION_TS_KV_TABLE);
+                        executeDropStatement(conn, DROP_FUNCTION_CREATE_PARTITIONS);
+                        executeDropStatement(conn, DROP_FUNCTION_CREATE_TS_KV_DICTIONARY_TABLE);
+                        executeDropStatement(conn, DROP_FUNCTION_INSERT_INTO_DICTIONARY);
+                        executeDropStatement(conn, DROP_FUNCTION_INSERT_INTO_TS_KV);
+                        executeDropStatement(conn, DROP_FUNCTION_CREATE_NEW_TS_KV_LATEST_TABLE);
+                        executeDropStatement(conn, DROP_FUNCTION_INSERT_INTO_TS_KV_LATEST);
+
                         log.info("schema timeseries updated!");
                     }
                 }
@@ -93,53 +110,13 @@ public class PsqlTsDatabaseUpgradeService implements DatabaseTsUpgradeService {
         }
     }
 
-    private void loadSql(Connection conn) {
+    protected void loadSql(Connection conn) {
         Path schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "2.4.3", LOAD_FUNCTIONS_SQL);
         try {
             loadFunctions(schemaUpdateFile, conn);
+            log.info("Upgrade functions successfully loaded!");
         } catch (Exception e) {
             log.info("Failed to load PostgreSQL upgrade functions due to: {}", e.getMessage());
         }
     }
-
-    private void loadFunctions(Path sqlFile, Connection conn) throws Exception {
-        String sql = new String(Files.readAllBytes(sqlFile), StandardCharsets.UTF_8);
-        conn.createStatement().execute(sql); //NOSONAR, ignoring because method used to execute thingsboard database upgrade script
-    }
-
-    private boolean checkVersion(Connection conn) {
-        log.info("Check the current PostgreSQL version...");
-        boolean versionValid = false;
-        try {
-            CallableStatement callableStatement = conn.prepareCall("{? = " + CHECK_VERSION + " }");
-            callableStatement.registerOutParameter(1, Types.BOOLEAN);
-            callableStatement.execute();
-            versionValid = callableStatement.getBoolean(1);
-            callableStatement.close();
-        } catch (Exception e) {
-            log.info("Failed to check current PostgreSQL version due to: {}", e.getMessage());
-        }
-        return versionValid;
-    }
-
-    private void executeFunction(Connection conn, String query) {
-        log.info("{} ... ", query);
-        try {
-            CallableStatement callableStatement = conn.prepareCall("{" + query + "}");
-            callableStatement.execute();
-            callableStatement.close();
-            log.info("Successfully executed: {}", query.replace(CALL_REGEX, ""));
-        } catch (Exception e) {
-            log.info("Failed to execute {} due to: {}", query, e.getMessage());
-        }
-    }
-
-    private void dropOldTable(Connection conn, String query) {
-        try {
-            conn.createStatement().execute(query); //NOSONAR, ignoring because method used to execute thingsboard database upgrade script
-            Thread.sleep(5000);
-        } catch (InterruptedException | SQLException e) {
-            log.info("Failed to drop table {} due to: {}", query.replace("DROP TABLE ", ""), e.getMessage());
-        }
-    }
 }

application/src/main/java/org/thingsboard/server/service/install/SqlAbstractDatabaseSchemaService.java

@@ -32,13 +32,13 @@ public abstract class SqlAbstractDatabaseSchemaService implements DatabaseSchema
     private static final String SQL_DIR = "sql";
 
     @Value("${spring.datasource.url}")
-    private String dbUrl;
+    protected String dbUrl;
 
     @Value("${spring.datasource.username}")
-    private String dbUserName;
+    protected String dbUserName;
 
     @Value("${spring.datasource.password}")
-    private String dbPassword;
+    protected String dbPassword;
 
     @Autowired
     private InstallScripts installScripts;

application/src/main/java/org/thingsboard/server/service/install/TimescaleTsDatabaseSchemaService.java

@@ -0,0 +1,68 @@
+/**
+ * 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 lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Service;
+import org.thingsboard.server.dao.util.PsqlDao;
+import org.thingsboard.server.dao.util.TimescaleDBTsDao;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+
+@Service
+@TimescaleDBTsDao
+@PsqlDao
+@Profile("install")
+@Slf4j
+public class TimescaleTsDatabaseSchemaService extends SqlAbstractDatabaseSchemaService implements TsDatabaseSchemaService {
+
+    private static final String QUERY = "query: {}";
+    private static final String SUCCESSFULLY_EXECUTED = "Successfully executed ";
+    private static final String FAILED_TO_EXECUTE = "Failed to execute ";
+    private static final String FAILED_DUE_TO = " due to: {}";
+
+    private static final String SUCCESSFULLY_EXECUTED_QUERY = SUCCESSFULLY_EXECUTED + QUERY;
+    private static final String FAILED_TO_EXECUTE_QUERY = FAILED_TO_EXECUTE + QUERY + FAILED_DUE_TO;
+
+    @Value("${sql.timescale.chunk_time_interval:86400000}")
+    private long chunkTimeInterval;
+
+    public TimescaleTsDatabaseSchemaService() {
+        super("schema-timescale.sql", "schema-timescale-idx.sql");
+    }
+
+    @Override
+    public void createDatabaseSchema() throws Exception {
+        super.createDatabaseSchema();
+        executeQuery("SELECT create_hypertable('tenant_ts_kv', 'ts', chunk_time_interval => " + chunkTimeInterval + ", if_not_exists => true);");
+    }
+
+    private void executeQuery(String query) {
+        try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
+            conn.createStatement().execute(query); //NOSONAR, ignoring because method used to execute thingsboard database upgrade script
+            log.info(SUCCESSFULLY_EXECUTED_QUERY, query);
+            Thread.sleep(5000);
+        } catch (InterruptedException | SQLException e) {
+            log.info(FAILED_TO_EXECUTE_QUERY, query, e.getMessage());
+        }
+    }
+
+
+}

application/src/main/java/org/thingsboard/server/service/install/TimescaleTsDatabaseUpgradeService.java

@@ -0,0 +1,125 @@
+/**
+ * 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 lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Profile;
+import org.springframework.stereotype.Service;
+import org.thingsboard.server.dao.util.PsqlDao;
+import org.thingsboard.server.dao.util.TimescaleDBTsDao;
+
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.DriverManager;
+
+@Service
+@Profile("install")
+@Slf4j
+@TimescaleDBTsDao
+@PsqlDao
+public class TimescaleTsDatabaseUpgradeService extends AbstractSqlTsDatabaseUpgradeService implements DatabaseTsUpgradeService {
+
+    @Value("${sql.timescale.chunk_time_interval:86400000}")
+    private long chunkTimeInterval;
+
+    private static final String LOAD_FUNCTIONS_SQL = "schema_update_timescale_ts.sql";
+
+    private static final String TENANT_TS_KV_OLD_TABLE = "tenant_ts_kv_old;";
+
+    private static final String CREATE_TS_KV_LATEST_TABLE = "create_ts_kv_latest_table()";
+    private static final String CREATE_NEW_TENANT_TS_KV_TABLE = "create_new_tenant_ts_kv_table()";
+    private static final String CREATE_TS_KV_DICTIONARY_TABLE = "create_ts_kv_dictionary_table()";
+    private static final String INSERT_INTO_DICTIONARY = "insert_into_dictionary()";
+    private static final String INSERT_INTO_TENANT_TS_KV = "insert_into_tenant_ts_kv()";
+    private static final String INSERT_INTO_TS_KV_LATEST = "insert_into_ts_kv_latest()";
+
+    private static final String CALL_CREATE_TS_KV_LATEST_TABLE = CALL_REGEX + CREATE_TS_KV_LATEST_TABLE;
+    private static final String CALL_CREATE_NEW_TENANT_TS_KV_TABLE = CALL_REGEX + CREATE_NEW_TENANT_TS_KV_TABLE;
+    private static final String CALL_CREATE_TS_KV_DICTIONARY_TABLE = CALL_REGEX + CREATE_TS_KV_DICTIONARY_TABLE;
+    private static final String CALL_INSERT_INTO_DICTIONARY = CALL_REGEX + INSERT_INTO_DICTIONARY;
+    private static final String CALL_INSERT_INTO_TS_KV = CALL_REGEX + INSERT_INTO_TENANT_TS_KV;
+    private static final String CALL_INSERT_INTO_TS_KV_LATEST = CALL_REGEX + INSERT_INTO_TS_KV_LATEST;
+
+    private static final String DROP_OLD_TENANT_TS_KV_TABLE = DROP_TABLE + TENANT_TS_KV_OLD_TABLE;
+
+    private static final String DROP_FUNCTION_CREATE_TS_KV_LATEST_TABLE = DROP_FUNCTION_IF_EXISTS + CREATE_TS_KV_LATEST_TABLE;
+    private static final String DROP_FUNCTION_CREATE_TENANT_TS_KV_TABLE_COPY = DROP_FUNCTION_IF_EXISTS + CREATE_NEW_TENANT_TS_KV_TABLE;
+    private static final String DROP_FUNCTION_CREATE_TS_KV_DICTIONARY_TABLE = DROP_FUNCTION_IF_EXISTS + CREATE_TS_KV_DICTIONARY_TABLE;
+    private static final String DROP_FUNCTION_INSERT_INTO_DICTIONARY = DROP_FUNCTION_IF_EXISTS + INSERT_INTO_DICTIONARY;
+    private static final String DROP_FUNCTION_INSERT_INTO_TENANT_TS_KV = DROP_FUNCTION_IF_EXISTS + INSERT_INTO_TENANT_TS_KV;
+    private static final String DROP_FUNCTION_INSERT_INTO_TS_KV_LATEST = DROP_FUNCTION_IF_EXISTS + INSERT_INTO_TS_KV_LATEST;
+
+    @Autowired
+    private InstallScripts installScripts;
+
+    @Override
+    public void upgradeDatabase(String fromVersion) throws Exception {
+        switch (fromVersion) {
+            case "2.4.3":
+                try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
+                    log.info("Updating timescale schema ...");
+                    log.info("Load upgrade functions ...");
+                    loadSql(conn);
+                    boolean versionValid = checkVersion(conn);
+                    if (!versionValid) {
+                        log.info("PostgreSQL version should be at least more than 9.6!");
+                        log.info("Please upgrade your PostgreSQL and restart the script!");
+                    } else {
+                        log.info("PostgreSQL version is valid!");
+                        log.info("Updating schema ...");
+                        executeFunction(conn, CALL_CREATE_TS_KV_LATEST_TABLE);
+                        executeFunction(conn, CALL_CREATE_NEW_TENANT_TS_KV_TABLE);
+
+                        executeQuery(conn, "SELECT create_hypertable('tenant_ts_kv', 'ts', chunk_time_interval => " + chunkTimeInterval + ", if_not_exists => true);");
+
+                        executeFunction(conn, CALL_CREATE_TS_KV_DICTIONARY_TABLE);
+                        executeFunction(conn, CALL_INSERT_INTO_DICTIONARY);
+                        executeFunction(conn, CALL_INSERT_INTO_TS_KV);
+                        executeFunction(conn, CALL_INSERT_INTO_TS_KV_LATEST);
+
+                        //executeQuery(conn, "SELECT set_chunk_time_interval('tenant_ts_kv', " + chunkTimeInterval +");");
+
+                        executeDropStatement(conn, DROP_OLD_TENANT_TS_KV_TABLE);
+
+                        executeDropStatement(conn, DROP_FUNCTION_CREATE_TS_KV_LATEST_TABLE);
+                        executeDropStatement(conn, DROP_FUNCTION_CREATE_TENANT_TS_KV_TABLE_COPY);
+                        executeDropStatement(conn, DROP_FUNCTION_CREATE_TS_KV_DICTIONARY_TABLE);
+                        executeDropStatement(conn, DROP_FUNCTION_INSERT_INTO_DICTIONARY);
+                        executeDropStatement(conn, DROP_FUNCTION_INSERT_INTO_TENANT_TS_KV);
+                        executeDropStatement(conn, DROP_FUNCTION_INSERT_INTO_TS_KV_LATEST);
+
+                        log.info("schema timeseries updated!");
+                    }
+                }
+                break;
+            default:
+                throw new RuntimeException("Unable to upgrade SQL database, unsupported fromVersion: " + fromVersion);
+        }
+    }
+
+    protected void loadSql(Connection conn) {
+        Path schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "2.4.3", LOAD_FUNCTIONS_SQL);
+        try {
+            loadFunctions(schemaUpdateFile, conn);
+            log.info("Upgrade functions successfully loaded!");
+        } catch (Exception e) {
+            log.info("Failed to load Timescale upgrade functions due to: {}", e.getMessage());
+        }
+    }
+}

application/src/main/resources/thingsboard.yml

@@ -208,14 +208,14 @@ sql:
       batch_size: "${SQL_TS_LATEST_BATCH_SIZE:10000}"
       batch_max_delay: "${SQL_TS_LATEST_BATCH_MAX_DELAY_MS:100}"
       stats_print_interval_ms: "${SQL_TS_LATEST_BATCH_STATS_PRINT_MS:10000}"
-    ts_timescale:
-      batch_size: "${SQL_TS_TIMESCALE_BATCH_SIZE:10000}"
-      batch_max_delay: "${SQL_TS_TIMESCALE_BATCH_MAX_DELAY_MS:100}"
-      stats_print_interval_ms: "${SQL_TS_TIMESCALE_BATCH_STATS_PRINT_MS:10000}"
     # Specify whether to remove null characters from strValue of attributes and timeseries before insert
     remove_null_chars: "${SQL_REMOVE_NULL_CHARS:true}"
-    # Specify partitioning size for timestamp key-value storage. Example: DAYS, MONTHS, YEARS, INDEFINITE
+    postgres:
-    ts_key_value_partitioning: "${TS_KV_PARTITIONING:MONTHS}"
+      # Specify partitioning size for timestamp key-value storage. Example: DAYS, MONTHS, YEARS, INDEFINITE.
+      ts_key_value_partitioning: "${SQL_POSTGRES_TS_KV_PARTITIONING:MONTHS}"
+    timescale:
+      # Specify Interval size for new data chunks storage.
+      chunk_time_interval: "${SQL_TIMESCALE_CHUNK_TIME_INTERVAL:604800000}"
 
 # Actor system parameters
 actors:

application/src/main/java/org/thingsboard/server/service/install/SqlTimescaleDatabaseSchemaService.java → common/dao-api/src/main/java/org/thingsboard/server/dao/util/SqlTsAnyDao.java

@@ -13,19 +13,10 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-package org.thingsboard.server.service.install;
+package org.thingsboard.server.dao.util;
 
-import org.springframework.context.annotation.Profile;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
-import org.springframework.stereotype.Service;
-import org.thingsboard.server.dao.util.SqlDao;
-import org.thingsboard.server.dao.util.TimescaleDBTsDao;
 
-@Service
+@ConditionalOnExpression("'${database.ts.type}'=='sql' || '${database.ts.type}'=='timescale'")
-@TimescaleDBTsDao
+public @interface SqlTsAnyDao {
-@Profile("install")
+}
-public class SqlTimescaleDatabaseSchemaService extends SqlAbstractDatabaseSchemaService
-        implements TsDatabaseSchemaService {
-    public SqlTimescaleDatabaseSchemaService() {
-        super("schema-timescale.sql", "schema-timescale-idx.sql");
-    }
-}

dao/src/main/java/org/thingsboard/server/dao/HsqlTsDaoConfig.java

@@ -27,8 +27,8 @@ import org.thingsboard.server.dao.util.SqlTsDao;
 @Configuration
 @EnableAutoConfiguration
 @ComponentScan({"org.thingsboard.server.dao.sqlts.hsql", "org.thingsboard.server.dao.sqlts.latest"})
-@EnableJpaRepositories({"org.thingsboard.server.dao.sqlts.hsql", "org.thingsboard.server.dao.sqlts.latest"})
+@EnableJpaRepositories({"org.thingsboard.server.dao.sqlts.hsql", "org.thingsboard.server.dao.sqlts.latest", "org.thingsboard.server.dao.sqlts.dictionary"})
-@EntityScan({"org.thingsboard.server.dao.model.sqlts.hsql", "org.thingsboard.server.dao.model.sqlts.latest"})
+@EntityScan({"org.thingsboard.server.dao.model.sqlts.hsql", "org.thingsboard.server.dao.model.sqlts.latest", "org.thingsboard.server.dao.model.sqlts.dictionary"})
 @EnableTransactionManagement
 @SqlTsDao
 @HsqlDao

dao/src/main/java/org/thingsboard/server/dao/JpaDaoConfig.java

@@ -22,7 +22,6 @@ import org.springframework.context.annotation.Configuration;
 import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
 import org.springframework.transaction.annotation.EnableTransactionManagement;
 import org.thingsboard.server.dao.util.SqlDao;
-import org.thingsboard.server.dao.util.TimescaleDBTsDao;
 
 /**
  * @author Valerii Sosliuk

dao/src/main/java/org/thingsboard/server/dao/TimescaleDaoConfig.java

@@ -21,6 +21,7 @@ import org.springframework.context.annotation.ComponentScan;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
 import org.springframework.transaction.annotation.EnableTransactionManagement;
+import org.thingsboard.server.dao.util.PsqlDao;
 import org.thingsboard.server.dao.util.TimescaleDBTsDao;
 
 @Configuration
@@ -30,6 +31,7 @@ import org.thingsboard.server.dao.util.TimescaleDBTsDao;
 @EntityScan({"org.thingsboard.server.dao.model.sqlts.timescale", "org.thingsboard.server.dao.model.sqlts.dictionary", "org.thingsboard.server.dao.model.sqlts.latest"})
 @EnableTransactionManagement
 @TimescaleDBTsDao
+@PsqlDao
 public class TimescaleDaoConfig {
 
 }

dao/src/main/java/org/thingsboard/server/dao/alarm/BaseAlarmService.java

@@ -386,13 +386,7 @@ public class BaseAlarmService extends AbstractEntityService implements AlarmServ
     private void updateRelations(Alarm alarm, AlarmStatus oldStatus, AlarmStatus newStatus) {
         try {
             List<EntityRelation> relations = relationService.findByToAsync(alarm.getTenantId(), alarm.getId(), RelationTypeGroup.ALARM).get();
-
+            Set<EntityId> parents = relations.stream().map(EntityRelation::getFrom).collect(Collectors.toSet());
-            List<String> propagateRelationTypes = alarm.getPropagateRelationTypes();
-            Stream<EntityRelation> relationStream = relations.stream();
-            if (!CollectionUtils.isEmpty(propagateRelationTypes)) {
-                relationStream = relationStream.filter(entityRelation -> propagateRelationTypes.contains(entityRelation.getType()));
-            }
-            Set<EntityId> parents = relationStream.map(EntityRelation::getFrom).collect(Collectors.toSet());
             for (EntityId parentId : parents) {
                 updateAlarmRelation(alarm.getTenantId(), parentId, alarm.getId(), oldStatus, newStatus);
             }

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

@@ -16,26 +16,42 @@
 package org.thingsboard.server.dao.model.sql;
 
 import lombok.Data;
+import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
+import org.thingsboard.server.common.data.kv.BooleanDataEntry;
+import org.thingsboard.server.common.data.kv.DoubleDataEntry;
+import org.thingsboard.server.common.data.kv.KvEntry;
+import org.thingsboard.server.common.data.kv.LongDataEntry;
+import org.thingsboard.server.common.data.kv.StringDataEntry;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.dao.model.ToData;
 
 import javax.persistence.Column;
 import javax.persistence.Id;
 import javax.persistence.MappedSuperclass;
+import javax.persistence.Transient;
+
+import java.util.UUID;
 
 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.LONG_VALUE_COLUMN;
 import static org.thingsboard.server.dao.model.ModelConstants.STRING_VALUE_COLUMN;
 import static org.thingsboard.server.dao.model.ModelConstants.TS_COLUMN;
 
 @Data
 @MappedSuperclass
-public abstract class AbstractTsKvEntity {
+public abstract class AbstractTsKvEntity implements ToData<TsKvEntry> {
 
     protected static final String SUM = "SUM";
     protected static final String AVG = "AVG";
     protected static final String MIN = "MIN";
     protected static final String MAX = "MAX";
 
+    @Id
+    @Column(name = ENTITY_ID_COLUMN, columnDefinition = "uuid")
+    protected UUID entityId;
+
     @Id
     @Column(name = TS_COLUMN)
     protected Long ts;
@@ -52,6 +68,9 @@ public abstract class AbstractTsKvEntity {
     @Column(name = DOUBLE_VALUE_COLUMN)
     protected Double doubleValue;
 
+    @Transient
+    protected String strKey;
+
     public abstract boolean isNotEmpty();
 
     protected static boolean isAllNull(Object... args) {
@@ -62,4 +81,20 @@ public abstract class AbstractTsKvEntity {
         }
         return true;
     }
+
+    @Override
+    public TsKvEntry toData() {
+        KvEntry kvEntry = null;
+        if (strValue != null) {
+            kvEntry = new StringDataEntry(strKey, strValue);
+        } else if (longValue != null) {
+            kvEntry = new LongDataEntry(strKey, longValue);
+        } else if (doubleValue != null) {
+            kvEntry = new DoubleDataEntry(strKey, doubleValue);
+        } else if (booleanValue != null) {
+            kvEntry = new BooleanDataEntry(strKey, booleanValue);
+        }
+        return new BasicTsKvEntry(ts, kvEntry);
+    }
+
 }

dao/src/main/java/org/thingsboard/server/dao/model/sqlts/dictionary/TsKvDictionary.java

@@ -38,7 +38,7 @@ public final class TsKvDictionary {
     @Column(name = KEY_COLUMN)
     private String key;
 
-    @Column(name = KEY_ID_COLUMN, unique = true, columnDefinition="serial")
+    @Column(name = KEY_ID_COLUMN, unique = true, columnDefinition="int")
     @Generated(GenerationTime.INSERT)
     private int keyId;
 

dao/src/main/java/org/thingsboard/server/dao/model/sqlts/hsql/TsKvCompositeKey.java

@@ -22,6 +22,7 @@ import org.thingsboard.server.common.data.EntityType;
 
 import javax.persistence.Transient;
 import java.io.Serializable;
+import java.util.UUID;
 
 @Data
 @AllArgsConstructor
@@ -31,9 +32,8 @@ public class TsKvCompositeKey implements Serializable {
     @Transient
     private static final long serialVersionUID = -4089175869616037523L;
 
-    private EntityType entityType;
+    private UUID entityId;
-    private String entityId;
+    private int key;
-    private String key;
     private long ts;
 
 }

dao/src/main/java/org/thingsboard/server/dao/model/sqlts/hsql/TsKvEntity.java

@@ -34,6 +34,9 @@ import javax.persistence.Enumerated;
 import javax.persistence.Id;
 import javax.persistence.IdClass;
 import javax.persistence.Table;
+import javax.persistence.Transient;
+
+import java.util.UUID;
 
 import static org.thingsboard.server.dao.model.ModelConstants.ENTITY_ID_COLUMN;
 import static org.thingsboard.server.dao.model.ModelConstants.ENTITY_TYPE_COLUMN;
@@ -45,18 +48,9 @@ import static org.thingsboard.server.dao.model.ModelConstants.KEY_COLUMN;
 @IdClass(TsKvCompositeKey.class)
 public final class TsKvEntity extends AbstractTsKvEntity implements ToData<TsKvEntry> {
 
-    @Id
-    @Enumerated(EnumType.STRING)
-    @Column(name = ENTITY_TYPE_COLUMN)
-    private EntityType entityType;
-
-    @Id
-    @Column(name = ENTITY_ID_COLUMN)
-    private String entityId;
-
     @Id
     @Column(name = KEY_COLUMN)
-    private String key;
+    private int key;
 
     public TsKvEntity() {
     }
@@ -120,19 +114,4 @@ public final class TsKvEntity extends AbstractTsKvEntity implements ToData<TsKvE
     public boolean isNotEmpty() {
         return strValue != null || longValue != null || doubleValue != null || booleanValue != null;
     }
-
-    @Override
-    public TsKvEntry toData() {
-        KvEntry kvEntry = null;
-        if (strValue != null) {
-            kvEntry = new StringDataEntry(key, strValue);
-        } else if (longValue != null) {
-            kvEntry = new LongDataEntry(key, longValue);
-        } else if (doubleValue != null) {
-            kvEntry = new DoubleDataEntry(key, doubleValue);
-        } else if (booleanValue != null) {
-            kvEntry = new BooleanDataEntry(key, booleanValue);
-        }
-        return new BasicTsKvEntry(ts, kvEntry);
-    }
 }

dao/src/main/java/org/thingsboard/server/dao/model/sqlts/latest/TsKvLatestCompositeKey.java

@@ -22,6 +22,7 @@ import org.thingsboard.server.common.data.EntityType;
 
 import javax.persistence.Transient;
 import java.io.Serializable;
+import java.util.UUID;
 
 @Data
 @NoArgsConstructor
@@ -31,7 +32,6 @@ public class TsKvLatestCompositeKey implements Serializable{
     @Transient
     private static final long serialVersionUID = -4089175869616037523L;
 
-    private EntityType entityType;
+    private UUID entityId;
-    private String entityId;
+    private int key;
-    private String key;
 }

dao/src/main/java/org/thingsboard/server/dao/model/sqlts/latest/TsKvLatestEntity.java

@@ -16,66 +16,78 @@
 package org.thingsboard.server.dao.model.sqlts.latest;
 
 import lombok.Data;
-import org.thingsboard.server.common.data.EntityType;
-import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
-import org.thingsboard.server.common.data.kv.BooleanDataEntry;
-import org.thingsboard.server.common.data.kv.DoubleDataEntry;
-import org.thingsboard.server.common.data.kv.KvEntry;
-import org.thingsboard.server.common.data.kv.LongDataEntry;
-import org.thingsboard.server.common.data.kv.StringDataEntry;
-import org.thingsboard.server.common.data.kv.TsKvEntry;
-import org.thingsboard.server.dao.model.ToData;
 import org.thingsboard.server.dao.model.sql.AbstractTsKvEntity;
+import org.thingsboard.server.dao.sqlts.latest.SearchTsKvLatestRepository;
 
 import javax.persistence.Column;
+import javax.persistence.ColumnResult;
+import javax.persistence.ConstructorResult;
 import javax.persistence.Entity;
-import javax.persistence.EnumType;
-import javax.persistence.Enumerated;
 import javax.persistence.Id;
 import javax.persistence.IdClass;
+import javax.persistence.NamedNativeQueries;
+import javax.persistence.NamedNativeQuery;
+import javax.persistence.SqlResultSetMapping;
+import javax.persistence.SqlResultSetMappings;
 import javax.persistence.Table;
+import java.util.UUID;
 
-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.KEY_COLUMN;
 
 @Data
 @Entity
 @Table(name = "ts_kv_latest")
 @IdClass(TsKvLatestCompositeKey.class)
-public final class TsKvLatestEntity extends AbstractTsKvEntity implements ToData<TsKvEntry> {
+@SqlResultSetMappings({
+        @SqlResultSetMapping(
+                name = "tsKvLatestFindMapping",
+                classes = {
+                        @ConstructorResult(
+                                targetClass = TsKvLatestEntity.class,
+                                columns = {
+                                        @ColumnResult(name = "entityId", type = UUID.class),
+                                        @ColumnResult(name = "key", type = Integer.class),
+                                        @ColumnResult(name = "strKey", type = String.class),
+                                        @ColumnResult(name = "strValue", type = String.class),
+                                        @ColumnResult(name = "boolValue", type = Boolean.class),
+                                        @ColumnResult(name = "longValue", type = Long.class),
+                                        @ColumnResult(name = "doubleValue", type = Double.class),
+                                        @ColumnResult(name = "ts", type = Long.class),
 
-    @Id
+                                }
-    @Enumerated(EnumType.STRING)
+                        ),
-    @Column(name = ENTITY_TYPE_COLUMN)
+                })
-    private EntityType entityType;
+})
-
+@NamedNativeQueries({
-    @Id
+        @NamedNativeQuery(
-    @Column(name = ENTITY_ID_COLUMN)
+                name = SearchTsKvLatestRepository.FIND_ALL_BY_ENTITY_ID,
-    private String entityId;
+                query = SearchTsKvLatestRepository.FIND_ALL_BY_ENTITY_ID_QUERY,
+                resultSetMapping = "tsKvLatestFindMapping",
+                resultClass = TsKvLatestEntity.class
+        )
+})
+public final class TsKvLatestEntity extends AbstractTsKvEntity {
 
     @Id
     @Column(name = KEY_COLUMN)
-    private String key;
+    private int key;
 
     @Override
     public boolean isNotEmpty() {
         return strValue != null || longValue != null || doubleValue != null || booleanValue != null;
     }
 
-    @Override
+    public TsKvLatestEntity() {
-    public TsKvEntry toData() {
-        KvEntry kvEntry = null;
-        if (strValue != null) {
-            kvEntry = new StringDataEntry(key, strValue);
-        } else if (longValue != null) {
-            kvEntry = new LongDataEntry(key, longValue);
-        } else if (doubleValue != null) {
-            kvEntry = new DoubleDataEntry(key, doubleValue);
-        } else if (booleanValue != null) {
-            kvEntry = new BooleanDataEntry(key, booleanValue);
-        }
-        return new BasicTsKvEntry(ts, kvEntry);
     }
 
+    public TsKvLatestEntity(UUID entityId, Integer key, String strKey, String strValue, Boolean boolValue, Long longValue, Double doubleValue, Long ts) {
+        this.entityId = entityId;
+        this.key = key;
+        this.ts = ts;
+        this.longValue = longValue;
+        this.doubleValue = doubleValue;
+        this.strValue = strValue;
+        this.booleanValue = boolValue;
+        this.strKey = strKey;
+    }
 }

dao/src/main/java/org/thingsboard/server/dao/model/sqlts/psql/TsKvEntity.java

@@ -41,18 +41,11 @@ import static org.thingsboard.server.dao.model.ModelConstants.KEY_COLUMN;
 @Entity
 @Table(name = "ts_kv")
 @IdClass(TsKvCompositeKey.class)
-public final class TsKvEntity extends AbstractTsKvEntity implements ToData<TsKvEntry> {
+public final class TsKvEntity extends AbstractTsKvEntity {
-
-    @Id
-    @Column(name = ENTITY_ID_COLUMN, columnDefinition = "uuid")
-    protected UUID entityId;
 
     @Id
     @Column(name = KEY_COLUMN)
-    protected int key;
+    private int key;
-
-    @Transient
-    protected String strKey;
 
     public TsKvEntity() {
     }
@@ -116,20 +109,4 @@ public final class TsKvEntity extends AbstractTsKvEntity implements ToData<TsKvE
     public boolean isNotEmpty() {
         return strValue != null || longValue != null || doubleValue != null || booleanValue != null;
     }
-
-    @Override
-    public TsKvEntry toData() {
-        KvEntry kvEntry = null;
-        if (strValue != null) {
-            kvEntry = new StringDataEntry(strKey, strValue);
-        } else if (longValue != null) {
-            kvEntry = new LongDataEntry(strKey, longValue);
-        } else if (doubleValue != null) {
-            kvEntry = new DoubleDataEntry(strKey, doubleValue);
-        } else if (booleanValue != null) {
-            kvEntry = new BooleanDataEntry(strKey, booleanValue);
-        }
-        return new BasicTsKvEntry(ts, kvEntry);
-    }
-
 }

dao/src/main/java/org/thingsboard/server/dao/model/sqlts/timescale/TimescaleTsKvEntity.java

@@ -129,17 +129,9 @@ public final class TimescaleTsKvEntity extends AbstractTsKvEntity implements ToD
     @Column(name = TENANT_ID_COLUMN, columnDefinition = "uuid")
     private UUID tenantId;
 
-    @Id
-    @Column(name = ENTITY_ID_COLUMN, columnDefinition = "uuid")
-    protected UUID entityId;
-
     @Id
     @Column(name = KEY_COLUMN)
-    protected int key;
+    private int key;
-
-    @Transient
-    protected String strKey;
-
 
     public TimescaleTsKvEntity() {
     }
@@ -204,20 +196,4 @@ public final class TimescaleTsKvEntity extends AbstractTsKvEntity implements ToD
     public boolean isNotEmpty() {
         return ts != null && (strValue != null || longValue != null || doubleValue != null || booleanValue != null);
     }
-
-    @Override
-    public TsKvEntry toData() {
-        KvEntry kvEntry = null;
-        if (strValue != null) {
-            kvEntry = new StringDataEntry(strKey, strValue);
-        } else if (longValue != null) {
-            kvEntry = new LongDataEntry(strKey, longValue);
-        } else if (doubleValue != null) {
-            kvEntry = new DoubleDataEntry(strKey, doubleValue);
-        } else if (booleanValue != null) {
-            kvEntry = new BooleanDataEntry(strKey, booleanValue);
-        }
-        return new BasicTsKvEntry(ts, kvEntry);
-    }
-
 }

dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSimpleSqlTimeseriesDao.java → dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java

@@ -15,16 +15,13 @@
  */
 package org.thingsboard.server.dao.sqlts;
 
-import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.Aggregation;
-import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.dao.model.sql.AbstractTsKvEntity;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueue;
@@ -32,26 +29,16 @@ import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams;
 
 import javax.annotation.PostConstruct;
 import javax.annotation.PreDestroy;
-import java.util.ArrayList;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.stream.Collectors;
 
 @Slf4j
-public abstract class AbstractSimpleSqlTimeseriesDao<T extends AbstractTsKvEntity> extends AbstractSqlTimeseriesDao {
+public abstract class AbstractChunkedAggregationTimeseriesDao<T extends AbstractTsKvEntity> extends AbstractSqlTimeseriesDao {
 
     @Autowired
-    private InsertTsRepository<T> insertRepository;
+    protected InsertTsRepository<T> insertRepository;
-
-    @Value("${sql.ts.batch_size:1000}")
-    private int tsBatchSize;
-
-    @Value("${sql.ts.batch_max_delay:100}")
-    private long tsMaxDelay;
-
-    @Value("${sql.ts.stats_print_interval_ms:1000}")
-    private long tsStatsPrintIntervalMs;
 
     protected TbSqlBlockingQueue<EntityContainer<T>> tsQueue;
 
@@ -76,26 +63,39 @@ public abstract class AbstractSimpleSqlTimeseriesDao<T extends AbstractTsKvEntit
         }
     }
 
-    protected ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, ReadTsKvQuery query) {
+    protected abstract ListenableFuture<Optional<TsKvEntry>> findAndAggregateAsync(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, long ts, Aggregation aggregation);
-        if (query.getAggregation() == Aggregation.NONE) {
+
-            return findAllAsyncWithLimit(entityId, query);
+    protected void switchAggregation(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, Aggregation aggregation, List<CompletableFuture<T>> entitiesFutures) {
-        } else {
+        switch (aggregation) {
-            long stepTs = query.getStartTs();
+            case AVG:
-            List<ListenableFuture<Optional<TsKvEntry>>> futures = new ArrayList<>();
+                findAvg(tenantId, entityId, key, startTs, endTs, entitiesFutures);
-            while (stepTs < query.getEndTs()) {
+                break;
-                long startTs = stepTs;
+            case MAX:
-                long endTs = stepTs + query.getInterval();
+                findMax(tenantId, entityId, key, startTs, endTs, entitiesFutures);
-                long ts = startTs + (endTs - startTs) / 2;
+                break;
-                futures.add(findAndAggregateAsync(entityId, query.getKey(), startTs, endTs, ts, query.getAggregation()));
+            case MIN:
-                stepTs = endTs;
+                findMin(tenantId, entityId, key, startTs, endTs, entitiesFutures);
-            }
+                break;
-            return getTskvEntriesFuture(Futures.allAsList(futures));
+            case SUM:
+                findSum(tenantId, entityId, key, startTs, endTs, entitiesFutures);
+                break;
+            case COUNT:
+                findCount(tenantId, entityId, key, startTs, endTs, entitiesFutures);
+                break;
+            default:
+                throw new IllegalArgumentException("Not supported aggregation type: " + aggregation);
         }
     }
 
-    protected abstract ListenableFuture<Optional<TsKvEntry>> findAndAggregateAsync(EntityId entityId, String key, long startTs, long endTs, long ts, Aggregation aggregation);
+    protected abstract void findCount(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<T>> entitiesFutures);
+
+    protected abstract void findSum(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<T>> entitiesFutures);
+
+    protected abstract void findMin(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<T>> entitiesFutures);
+
+    protected abstract void findMax(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<T>> entitiesFutures);
 
-    protected abstract ListenableFuture<List<TsKvEntry>> findAllAsyncWithLimit(EntityId entityId, ReadTsKvQuery query);
+    protected abstract void findAvg(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<T>> entitiesFutures);
 
     protected SettableFuture<T> setFutures(List<CompletableFuture<T>> entitiesFutures) {
         SettableFuture<T> listenableFuture = SettableFuture.create();
@@ -121,36 +121,4 @@ public abstract class AbstractSimpleSqlTimeseriesDao<T extends AbstractTsKvEntit
         });
         return listenableFuture;
     }
-
+}
-    protected void switchAgregation(EntityId entityId, String key, long startTs, long endTs, Aggregation aggregation, List<CompletableFuture<T>> entitiesFutures) {
-        switch (aggregation) {
-            case AVG:
-                findAvg(entityId, key, startTs, endTs, entitiesFutures);
-                break;
-            case MAX:
-                findMax(entityId, key, startTs, endTs, entitiesFutures);
-                break;
-            case MIN:
-                findMin(entityId, key, startTs, endTs, entitiesFutures);
-                break;
-            case SUM:
-                findSum(entityId, key, startTs, endTs, entitiesFutures);
-                break;
-            case COUNT:
-                findCount(entityId, key, startTs, endTs, entitiesFutures);
-                break;
-            default:
-                throw new IllegalArgumentException("Not supported aggregation type: " + aggregation);
-        }
-    }
-
-    protected abstract void findCount(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<T>> entitiesFutures);
-
-    protected abstract void findSum(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<T>> entitiesFutures);
-
-    protected abstract void findMin(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<T>> entitiesFutures);
-
-    protected abstract void findMax(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<T>> entitiesFutures);
-
-    protected abstract void findAvg(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<T>> entitiesFutures);
-}

dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java

@@ -21,9 +21,9 @@ import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import lombok.extern.slf4j.Slf4j;
+import org.hibernate.exception.ConstraintViolationException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
-import org.thingsboard.server.common.data.UUIDConverter;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.Aggregation;
@@ -34,12 +34,16 @@ import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.StringDataEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.dao.DaoUtil;
+import org.thingsboard.server.dao.model.sqlts.dictionary.TsKvDictionary;
+import org.thingsboard.server.dao.model.sqlts.dictionary.TsKvDictionaryCompositeKey;
 import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestCompositeKey;
 import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity;
 import org.thingsboard.server.dao.sql.JpaAbstractDaoListeningExecutorService;
 import org.thingsboard.server.dao.sql.ScheduledLogExecutorComponent;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueue;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams;
+import org.thingsboard.server.dao.sqlts.dictionary.TsKvDictionaryRepository;
+import org.thingsboard.server.dao.sqlts.latest.SearchTsKvLatestRepository;
 import org.thingsboard.server.dao.sqlts.latest.TsKvLatestRepository;
 import org.thingsboard.server.dao.timeseries.SimpleListenableFuture;
 
@@ -49,24 +53,34 @@ import javax.annotation.PreDestroy;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Collectors;
 
-import static org.thingsboard.server.common.data.UUIDConverter.fromTimeUUID;
-
 @Slf4j
 public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningExecutorService {
 
     private static final String DESC_ORDER = "DESC";
 
+    private final ConcurrentMap<String, Integer> tsKvDictionaryMap = new ConcurrentHashMap<>();
+
+    private static final ReentrantLock tsCreationLock = new ReentrantLock();
+
     @Autowired
     private TsKvLatestRepository tsKvLatestRepository;
 
     @Autowired
-    private InsertLatestRepository insertLatestRepository;
+    private SearchTsKvLatestRepository searchTsKvLatestRepository;
 
     @Autowired
-    protected ScheduledLogExecutorComponent logExecutor;
+    private InsertLatestTsRepository insertLatestTsRepository;
+
+    @Autowired
+    private TsKvDictionaryRepository dictionaryRepository;
+
+    private TbSqlBlockingQueue<TsKvLatestEntity> tsLatestQueue;
 
     @Value("${sql.ts_latest.batch_size:1000}")
     private int tsLatestBatchSize;
@@ -77,7 +91,17 @@ public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningEx
     @Value("${sql.ts_latest.stats_print_interval_ms:1000}")
     private long tsLatestStatsPrintIntervalMs;
 
-    private TbSqlBlockingQueue<TsKvLatestEntity> tsLatestQueue;
+    @Autowired
+    protected ScheduledLogExecutorComponent logExecutor;
+
+    @Value("${sql.ts.batch_size:1000}")
+    protected int tsBatchSize;
+
+    @Value("${sql.ts.batch_max_delay:100}")
+    protected long tsMaxDelay;
+
+    @Value("${sql.ts.stats_print_interval_ms:1000}")
+    protected long tsStatsPrintIntervalMs;
 
     @PostConstruct
     protected void init() {
@@ -88,7 +112,7 @@ public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningEx
                 .statsPrintIntervalMs(tsLatestStatsPrintIntervalMs)
                 .build();
         tsLatestQueue = new TbSqlBlockingQueue<>(tsLatestParams);
-        tsLatestQueue.init(logExecutor, v -> insertLatestRepository.saveOrUpdate(v));
+        tsLatestQueue.init(logExecutor, v -> insertLatestTsRepository.saveOrUpdate(v));
     }
 
     @PreDestroy
@@ -120,6 +144,8 @@ public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningEx
 
     protected abstract ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, ReadTsKvQuery query);
 
+    protected abstract ListenableFuture<List<TsKvEntry>> findAllAsyncWithLimit(TenantId tenantId, EntityId entityId, ReadTsKvQuery query);
+
     protected ListenableFuture<List<TsKvEntry>> getTskvEntriesFuture(ListenableFuture<List<Optional<TsKvEntry>>> future) {
         return Futures.transform(future, new Function<List<Optional<TsKvEntry>>, List<TsKvEntry>>() {
             @Nullable
@@ -147,13 +173,14 @@ public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningEx
     protected ListenableFuture<TsKvEntry> getFindLatestFuture(EntityId entityId, String key) {
         TsKvLatestCompositeKey compositeKey =
                 new TsKvLatestCompositeKey(
-                        entityId.getEntityType(),
+                        entityId.getId(),
-                        fromTimeUUID(entityId.getId()),
+                        getOrSaveKeyId(key));
-                        key);
         Optional<TsKvLatestEntity> entry = tsKvLatestRepository.findById(compositeKey);
         TsKvEntry result;
         if (entry.isPresent()) {
-            result = DaoUtil.getData(entry.get());
+            TsKvLatestEntity tsKvLatestEntity = entry.get();
+            tsKvLatestEntity.setStrKey(key);
+            result = DaoUtil.getData(tsKvLatestEntity);
         } else {
             result = new BasicTsKvEntry(System.currentTimeMillis(), new StringDataEntry(key, null));
         }
@@ -171,9 +198,8 @@ public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningEx
         ListenableFuture<Void> removedLatestFuture = Futures.transformAsync(booleanFuture, isRemove -> {
             if (isRemove) {
                 TsKvLatestEntity latestEntity = new TsKvLatestEntity();
-                latestEntity.setEntityType(entityId.getEntityType());
+                latestEntity.setEntityId(entityId.getId());
-                latestEntity.setEntityId(fromTimeUUID(entityId.getId()));
+                latestEntity.setKey(getOrSaveKeyId(query.getKey()));
-                latestEntity.setKey(query.getKey());
                 return service.submit(() -> {
                     tsKvLatestRepository.delete(latestEntity);
                     return null;
@@ -215,17 +241,14 @@ public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningEx
     protected ListenableFuture<List<TsKvEntry>> getFindAllLatestFuture(EntityId entityId) {
         return Futures.immediateFuture(
                 DaoUtil.convertDataList(Lists.newArrayList(
-                        tsKvLatestRepository.findAllByEntityTypeAndEntityId(
+                        searchTsKvLatestRepository.findAllByEntityId(entityId.getId()))));
-                                entityId.getEntityType(),
-                                UUIDConverter.fromTimeUUID(entityId.getId())))));
     }
 
     protected ListenableFuture<Void> getSaveLatestFuture(EntityId entityId, TsKvEntry tsKvEntry) {
         TsKvLatestEntity latestEntity = new TsKvLatestEntity();
-        latestEntity.setEntityType(entityId.getEntityType());
+        latestEntity.setEntityId(entityId.getId());
-        latestEntity.setEntityId(fromTimeUUID(entityId.getId()));
         latestEntity.setTs(tsKvEntry.getTs());
-        latestEntity.setKey(tsKvEntry.getKey());
+        latestEntity.setKey(getOrSaveKeyId(tsKvEntry.getKey()));
         latestEntity.setStrValue(tsKvEntry.getStrValue().orElse(null));
         latestEntity.setDoubleValue(tsKvEntry.getDoubleValue().orElse(null));
         latestEntity.setLongValue(tsKvEntry.getLongValue().orElse(null));
@@ -233,6 +256,42 @@ public abstract class AbstractSqlTimeseriesDao extends JpaAbstractDaoListeningEx
         return tsLatestQueue.add(latestEntity);
     }
 
+    protected Integer getOrSaveKeyId(String strKey) {
+        Integer keyId = tsKvDictionaryMap.get(strKey);
+        if (keyId == null) {
+            Optional<TsKvDictionary> tsKvDictionaryOptional;
+            tsKvDictionaryOptional = dictionaryRepository.findById(new TsKvDictionaryCompositeKey(strKey));
+            if (!tsKvDictionaryOptional.isPresent()) {
+                tsCreationLock.lock();
+                try {
+                    tsKvDictionaryOptional = dictionaryRepository.findById(new TsKvDictionaryCompositeKey(strKey));
+                    if (!tsKvDictionaryOptional.isPresent()) {
+                        TsKvDictionary tsKvDictionary = new TsKvDictionary();
+                        tsKvDictionary.setKey(strKey);
+                        try {
+                            TsKvDictionary saved = dictionaryRepository.save(tsKvDictionary);
+                            tsKvDictionaryMap.put(saved.getKey(), saved.getKeyId());
+                            keyId = saved.getKeyId();
+                        } catch (ConstraintViolationException e) {
+                            tsKvDictionaryOptional = dictionaryRepository.findById(new TsKvDictionaryCompositeKey(strKey));
+                            TsKvDictionary dictionary = tsKvDictionaryOptional.orElseThrow(() -> new RuntimeException("Failed to get TsKvDictionary entity from DB!"));
+                            tsKvDictionaryMap.put(dictionary.getKey(), dictionary.getKeyId());
+                            keyId = dictionary.getKeyId();
+                        }
+                    } else {
+                        keyId = tsKvDictionaryOptional.get().getKeyId();
+                    }
+                } finally {
+                    tsCreationLock.unlock();
+                }
+            } else {
+                keyId = tsKvDictionaryOptional.get().getKeyId();
+                tsKvDictionaryMap.put(strKey, keyId);
+            }
+        }
+        return keyId;
+    }
+
     private ListenableFuture<Void> getNewLatestEntryFuture(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) {
         ListenableFuture<List<TsKvEntry>> future = findNewLatestEntryFuture(tenantId, entityId, query);
         return Futures.transformAsync(future, entryList -> {

dao/src/main/java/org/thingsboard/server/dao/sqlts/InsertLatestRepository.java → dao/src/main/java/org/thingsboard/server/dao/sqlts/InsertLatestTsRepository.java

@@ -19,7 +19,7 @@ import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity;
 
 import java.util.List;
 
-public interface InsertLatestRepository {
+public interface InsertLatestTsRepository {
 
     void saveOrUpdate(List<TsKvLatestEntity> entities);
 

dao/src/main/java/org/thingsboard/server/dao/sqlts/dictionary/TsKvDictionaryRepository.java

@@ -18,11 +18,11 @@ package org.thingsboard.server.dao.sqlts.dictionary;
 import org.springframework.data.repository.CrudRepository;
 import org.thingsboard.server.dao.model.sqlts.dictionary.TsKvDictionary;
 import org.thingsboard.server.dao.model.sqlts.dictionary.TsKvDictionaryCompositeKey;
-import org.thingsboard.server.dao.util.PsqlDao;
+import org.thingsboard.server.dao.util.SqlTsAnyDao;
 
 import java.util.Optional;
 
-@PsqlDao
+@SqlTsAnyDao
 public interface TsKvDictionaryRepository extends CrudRepository<TsKvDictionary, TsKvDictionaryCompositeKey> {
 
     Optional<TsKvDictionary> findByKeyId(int keyId);

dao/src/main/java/org/thingsboard/server/dao/sqlts/hsql/HsqlTimeseriesInsertRepository.java → dao/src/main/java/org/thingsboard/server/dao/sqlts/hsql/HsqlInsertTsRepository.java

@@ -34,18 +34,17 @@ import java.util.List;
 @HsqlDao
 @Repository
 @Transactional
-public class HsqlTimeseriesInsertRepository extends AbstractInsertRepository implements InsertTsRepository<TsKvEntity> {
+public class HsqlInsertTsRepository extends AbstractInsertRepository implements InsertTsRepository<TsKvEntity> {
 
     private static final String INSERT_OR_UPDATE =
-            "MERGE INTO ts_kv USING(VALUES ?, ?, ?, ?, ?, ?, ?, ?) " +
+            "MERGE INTO ts_kv USING(VALUES ?, ?, ?, ?, ?, ?, ?) " +
-                    "T (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " +
+                    "T (entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " +
-                    "ON (ts_kv.entity_type=T.entity_type " +
+                    "ON (ts_kv.entity_id=T.entity_id " +
-                    "AND ts_kv.entity_id=T.entity_id " +
                     "AND ts_kv.key=T.key " +
                     "AND ts_kv.ts=T.ts) " +
                     "WHEN MATCHED THEN UPDATE SET ts_kv.bool_v = T.bool_v, ts_kv.str_v = T.str_v, ts_kv.long_v = T.long_v, ts_kv.dbl_v = T.dbl_v " +
-                    "WHEN NOT MATCHED THEN INSERT (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " +
+                    "WHEN NOT MATCHED THEN INSERT (entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " +
-                    "VALUES (T.entity_type, T.entity_id, T.key, T.ts, T.bool_v, T.str_v, T.long_v, T.dbl_v);";
+                    "VALUES (T.entity_id, T.key, T.ts, T.bool_v, T.str_v, T.long_v, T.dbl_v);";
 
     @Override
     public void saveOrUpdate(List<EntityContainer<TsKvEntity>> entities) {
@@ -54,29 +53,28 @@ public class HsqlTimeseriesInsertRepository extends AbstractInsertRepository imp
             public void setValues(PreparedStatement ps, int i) throws SQLException {
                 EntityContainer<TsKvEntity> tsKvEntityEntityContainer = entities.get(i);
                 TsKvEntity tsKvEntity = tsKvEntityEntityContainer.getEntity();
-                ps.setString(1, tsKvEntity.getEntityType().name());
+                ps.setObject(1, tsKvEntity.getEntityId());
-                ps.setString(2, tsKvEntity.getEntityId());
+                ps.setInt(2, tsKvEntity.getKey());
-                ps.setString(3, tsKvEntity.getKey());
+                ps.setLong(3, tsKvEntity.getTs());
-                ps.setLong(4, tsKvEntity.getTs());
 
                 if (tsKvEntity.getBooleanValue() != null) {
-                    ps.setBoolean(5, tsKvEntity.getBooleanValue());
+                    ps.setBoolean(4, tsKvEntity.getBooleanValue());
                 } else {
-                    ps.setNull(5, Types.BOOLEAN);
+                    ps.setNull(4, Types.BOOLEAN);
                 }
 
-                ps.setString(6, tsKvEntity.getStrValue());
+                ps.setString(5, tsKvEntity.getStrValue());
 
                 if (tsKvEntity.getLongValue() != null) {
-                    ps.setLong(7, tsKvEntity.getLongValue());
+                    ps.setLong(6, tsKvEntity.getLongValue());
                 } else {
-                    ps.setNull(7, Types.BIGINT);
+                    ps.setNull(6, Types.BIGINT);
                 }
 
                 if (tsKvEntity.getDoubleValue() != null) {
-                    ps.setDouble(8, tsKvEntity.getDoubleValue());
+                    ps.setDouble(7, tsKvEntity.getDoubleValue());
                 } else {
-                    ps.setNull(8, Types.DOUBLE);
+                    ps.setNull(7, Types.DOUBLE);
                 }
             }
 

dao/src/main/java/org/thingsboard/server/dao/sqlts/hsql/JpaHsqlTimeseriesDao.java

@@ -30,7 +30,7 @@ import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.dao.DaoUtil;
 import org.thingsboard.server.dao.model.sqlts.hsql.TsKvEntity;
-import org.thingsboard.server.dao.sqlts.AbstractSimpleSqlTimeseriesDao;
+import org.thingsboard.server.dao.sqlts.AbstractChunkedAggregationTimeseriesDao;
 import org.thingsboard.server.dao.sqlts.EntityContainer;
 import org.thingsboard.server.dao.timeseries.TimeseriesDao;
 import org.thingsboard.server.dao.util.HsqlDao;
@@ -41,14 +41,12 @@ import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 
-import static org.thingsboard.server.common.data.UUIDConverter.fromTimeUUID;
-
 
 @Component
 @Slf4j
 @SqlTsDao
 @HsqlDao
-public class JpaHsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEntity> implements TimeseriesDao {
+public class JpaHsqlTimeseriesDao extends AbstractChunkedAggregationTimeseriesDao<TsKvEntity> implements TimeseriesDao {
 
     @Autowired
     private TsKvHsqlRepository tsKvRepository;
@@ -60,11 +58,12 @@ public class JpaHsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
 
     @Override
     public ListenableFuture<Void> save(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry, long ttl) {
+        String strKey = tsKvEntry.getKey();
+        Integer keyId = getOrSaveKeyId(strKey);
         TsKvEntity entity = new TsKvEntity();
-        entity.setEntityType(entityId.getEntityType());
+        entity.setEntityId(entityId.getId());
-        entity.setEntityId(fromTimeUUID(entityId.getId()));
         entity.setTs(tsKvEntry.getTs());
-        entity.setKey(tsKvEntry.getKey());
+        entity.setKey(keyId);
         entity.setStrValue(tsKvEntry.getStrValue().orElse(null));
         entity.setDoubleValue(tsKvEntry.getDoubleValue().orElse(null));
         entity.setLongValue(tsKvEntry.getLongValue().orElse(null));
@@ -77,9 +76,8 @@ public class JpaHsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
     public ListenableFuture<Void> remove(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) {
         return service.submit(() -> {
             tsKvRepository.delete(
-                    fromTimeUUID(entityId.getId()),
+                    entityId.getId(),
-                    entityId.getEntityType(),
+                    getOrSaveKeyId(query.getKey()),
-                    query.getKey(),
                     query.getStartTs(),
                     query.getEndTs());
             return null;
@@ -116,14 +114,47 @@ public class JpaHsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
         return Futures.immediateFuture(null);
     }
 
-    protected ListenableFuture<Optional<TsKvEntry>> findAndAggregateAsync(EntityId entityId, String key, long startTs, long endTs, long ts, Aggregation aggregation) {
+    protected ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, ReadTsKvQuery query) {
+        if (query.getAggregation() == Aggregation.NONE) {
+            return findAllAsyncWithLimit(tenantId, entityId, query);
+        } else {
+            long stepTs = query.getStartTs();
+            List<ListenableFuture<Optional<TsKvEntry>>> futures = new ArrayList<>();
+            while (stepTs < query.getEndTs()) {
+                long startTs = stepTs;
+                long endTs = stepTs + query.getInterval();
+                long ts = startTs + (endTs - startTs) / 2;
+                futures.add(findAndAggregateAsync(tenantId, entityId, query.getKey(), startTs, endTs, ts, query.getAggregation()));
+                stepTs = endTs;
+            }
+            return getTskvEntriesFuture(Futures.allAsList(futures));
+        }
+    }
+
+    @Override
+    protected ListenableFuture<List<TsKvEntry>> findAllAsyncWithLimit(TenantId tenantId, EntityId entityId, ReadTsKvQuery query) {
+        List<TsKvEntity> tsKvEntities = tsKvRepository.findAllWithLimit(
+                entityId.getId(),
+                getOrSaveKeyId(query.getKey()),
+                query.getStartTs(),
+                query.getEndTs(),
+                new PageRequest(0, query.getLimit(),
+                        new Sort(Sort.Direction.fromString(
+                                query.getOrderBy()), "ts")));
+        tsKvEntities.forEach(tsKvEntity -> tsKvEntity.setStrKey(query.getKey()));
+        return Futures.immediateFuture(
+                DaoUtil.convertDataList(
+                        tsKvEntities));
+    }
+
+    @Override
+    protected ListenableFuture<Optional<TsKvEntry>> findAndAggregateAsync(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, long ts, Aggregation aggregation) {
         List<CompletableFuture<TsKvEntity>> entitiesFutures = new ArrayList<>();
-        switchAgregation(entityId, key, startTs, endTs, aggregation, entitiesFutures);
+        switchAggregation(tenantId, entityId, key, startTs, endTs, aggregation, entitiesFutures);
         return Futures.transform(setFutures(entitiesFutures), entity -> {
             if (entity != null && entity.isNotEmpty()) {
-                entity.setEntityId(fromTimeUUID(entityId.getId()));
+                entity.setEntityId(entityId.getId());
-                entity.setEntityType(entityId.getEntityType());
+                entity.setKey(getOrSaveKeyId(key));
-                entity.setKey(key);
                 entity.setTs(ts);
                 return Optional.of(DaoUtil.getData(entity));
             } else {
@@ -132,75 +163,63 @@ public class JpaHsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
         });
     }
 
-    protected ListenableFuture<List<TsKvEntry>> findAllAsyncWithLimit(EntityId entityId, ReadTsKvQuery query) {
+    @Override
-        return Futures.immediateFuture(
+    protected void findCount(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
-                DaoUtil.convertDataList(
+        Integer keyId = getOrSaveKeyId(key);
-                        tsKvRepository.findAllWithLimit(
-                                fromTimeUUID(entityId.getId()),
-                                entityId.getEntityType(),
-                                query.getKey(),
-                                query.getStartTs(),
-                                query.getEndTs(),
-                                new PageRequest(0, query.getLimit(),
-                                        new Sort(Sort.Direction.fromString(
-                                                query.getOrderBy()), "ts")))));
-    }
-
-    protected void findCount(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
         entitiesFutures.add(tsKvRepository.findCount(
-                fromTimeUUID(entityId.getId()),
+                entityId.getId(),
-                entityId.getEntityType(),
+                keyId,
-                key,
                 startTs,
                 endTs));
     }
 
-    protected void findSum(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+    @Override
+    protected void findSum(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+        Integer keyId = getOrSaveKeyId(key);
         entitiesFutures.add(tsKvRepository.findSum(
-                fromTimeUUID(entityId.getId()),
+                entityId.getId(),
-                entityId.getEntityType(),
+                keyId,
-                key,
                 startTs,
                 endTs));
     }
 
-    protected void findMin(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+    @Override
+    protected void findMin(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+        Integer keyId = getOrSaveKeyId(key);
         entitiesFutures.add(tsKvRepository.findStringMin(
-                fromTimeUUID(entityId.getId()),
+                entityId.getId(),
-                entityId.getEntityType(),
+                keyId,
-                key,
                 startTs,
                 endTs));
         entitiesFutures.add(tsKvRepository.findNumericMin(
-                fromTimeUUID(entityId.getId()),
+                entityId.getId(),
-                entityId.getEntityType(),
+                keyId,
-                key,
                 startTs,
                 endTs));
     }
 
-    protected void findMax(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+    @Override
+    protected void findMax(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+        Integer keyId = getOrSaveKeyId(key);
         entitiesFutures.add(tsKvRepository.findStringMax(
-                fromTimeUUID(entityId.getId()),
+                entityId.getId(),
-                entityId.getEntityType(),
+                keyId,
-                key,
                 startTs,
                 endTs));
         entitiesFutures.add(tsKvRepository.findNumericMax(
-                fromTimeUUID(entityId.getId()),
+                entityId.getId(),
-                entityId.getEntityType(),
+                keyId,
-                key,
                 startTs,
                 endTs));
     }
 
-    protected void findAvg(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+    @Override
+    protected void findAvg(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+        Integer keyId = getOrSaveKeyId(key);
         entitiesFutures.add(tsKvRepository.findAvg(
-                fromTimeUUID(entityId.getId()),
+                entityId.getId(),
-                entityId.getEntityType(),
+                keyId,
-                key,
                 startTs,
                 endTs));
     }
-
 }

dao/src/main/java/org/thingsboard/server/dao/sqlts/hsql/TsKvHsqlRepository.java

@@ -22,23 +22,21 @@ import org.springframework.data.repository.CrudRepository;
 import org.springframework.data.repository.query.Param;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.transaction.annotation.Transactional;
-import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.dao.model.sqlts.hsql.TsKvCompositeKey;
 import org.thingsboard.server.dao.model.sqlts.hsql.TsKvEntity;
 import org.thingsboard.server.dao.util.SqlDao;
 
 import java.util.List;
+import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 
 @SqlDao
 public interface TsKvHsqlRepository extends CrudRepository<TsKvEntity, TsKvCompositeKey> {
 
     @Query("SELECT tskv FROM TsKvEntity tskv WHERE tskv.entityId = :entityId " +
-            "AND tskv.entityType = :entityType AND tskv.key = :entityKey " +
+            "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs")
-            "AND tskv.ts > :startTs AND tskv.ts <= :endTs")
+    List<TsKvEntity> findAllWithLimit(@Param("entityId") UUID entityId,
-    List<TsKvEntity> findAllWithLimit(@Param("entityId") String entityId,
+                                      @Param("entityKey") int key,
-                                      @Param("entityType") EntityType entityType,
-                                      @Param("entityKey") String key,
                                       @Param("startTs") long startTs,
                                       @Param("endTs") long endTs,
                                       Pageable pageable);
@@ -46,22 +44,18 @@ public interface TsKvHsqlRepository extends CrudRepository<TsKvEntity, TsKvCompo
     @Transactional
     @Modifying
     @Query("DELETE FROM TsKvEntity tskv WHERE tskv.entityId = :entityId " +
-            "AND tskv.entityType = :entityType AND tskv.key = :entityKey " +
+            "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs")
-            "AND tskv.ts > :startTs AND tskv.ts <= :endTs")
+    void delete(@Param("entityId") UUID entityId,
-    void delete(@Param("entityId") String entityId,
+                @Param("entityKey") int key,
-                @Param("entityType") EntityType entityType,
-                @Param("entityKey") String key,
                 @Param("startTs") long startTs,
                 @Param("endTs") long endTs);
 
     @Async
     @Query("SELECT new TsKvEntity(MAX(tskv.strValue)) FROM TsKvEntity tskv " +
-            "WHERE tskv.strValue IS NOT NULL " +
+            "WHERE tskv.strValue IS NOT NULL AND tskv.entityId = :entityId AND tskv.key = :entityKey" +
-            "AND tskv.entityId = :entityId AND tskv.entityType = :entityType " +
+            " AND tskv.ts > :startTs AND tskv.ts <= :endTs")
-            "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs")
+    CompletableFuture<TsKvEntity> findStringMax(@Param("entityId") UUID entityId,
-    CompletableFuture<TsKvEntity> findStringMax(@Param("entityId") String entityId,
+                                                @Param("entityKey") int entityKey,
-                                                @Param("entityType") EntityType entityType,
-                                                @Param("entityKey") String entityKey,
                                                 @Param("startTs") long startTs,
                                                 @Param("endTs") long endTs);
 
@@ -70,24 +64,20 @@ public interface TsKvHsqlRepository extends CrudRepository<TsKvEntity, TsKvCompo
             "MAX(COALESCE(tskv.doubleValue, -1.79769E+308)), " +
             "SUM(CASE WHEN tskv.longValue IS NULL THEN 0 ELSE 1 END), " +
             "SUM(CASE WHEN tskv.doubleValue IS NULL THEN 0 ELSE 1 END), " +
-            "'MAX') FROM TsKvEntity tskv " +
+            "'MAX') FROM TsKvEntity tskv WHERE tskv.entityId = :entityId " +
-            "WHERE tskv.entityId = :entityId AND tskv.entityType = :entityType " +
             "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs")
-    CompletableFuture<TsKvEntity> findNumericMax(@Param("entityId") String entityId,
+    CompletableFuture<TsKvEntity> findNumericMax(@Param("entityId") UUID entityId,
-                                          @Param("entityType") EntityType entityType,
+                                          @Param("entityKey") int entityKey,
-                                          @Param("entityKey") String entityKey,
                                           @Param("startTs") long startTs,
                                           @Param("endTs") long endTs);
 
 
     @Async
     @Query("SELECT new TsKvEntity(MIN(tskv.strValue)) FROM TsKvEntity tskv " +
-            "WHERE tskv.strValue IS NOT NULL " +
+            "WHERE tskv.strValue IS NOT NULL AND tskv.entityId = :entityId " +
-            "AND tskv.entityId = :entityId AND tskv.entityType = :entityType " +
             "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs")
-    CompletableFuture<TsKvEntity> findStringMin(@Param("entityId") String entityId,
+    CompletableFuture<TsKvEntity> findStringMin(@Param("entityId") UUID entityId,
-                                          @Param("entityType") EntityType entityType,
+                                          @Param("entityKey") int entityKey,
-                                          @Param("entityKey") String entityKey,
                                           @Param("startTs") long startTs,
                                           @Param("endTs") long endTs);
 
@@ -96,12 +86,10 @@ public interface TsKvHsqlRepository extends CrudRepository<TsKvEntity, TsKvCompo
             "MIN(COALESCE(tskv.doubleValue, 1.79769E+308)), " +
             "SUM(CASE WHEN tskv.longValue IS NULL THEN 0 ELSE 1 END), " +
             "SUM(CASE WHEN tskv.doubleValue IS NULL THEN 0 ELSE 1 END), " +
-            "'MIN') FROM TsKvEntity tskv " +
+            "'MIN') FROM TsKvEntity tskv WHERE tskv.entityId = :entityId " +
-            "WHERE tskv.entityId = :entityId AND tskv.entityType = :entityType " +
             "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs")
-    CompletableFuture<TsKvEntity> findNumericMin(@Param("entityId") String entityId,
+    CompletableFuture<TsKvEntity> findNumericMin(@Param("entityId") UUID entityId,
-                                          @Param("entityType") EntityType entityType,
+                                          @Param("entityKey") int entityKey,
-                                          @Param("entityKey") String entityKey,
                                           @Param("startTs") long startTs,
                                           @Param("endTs") long endTs);
 
@@ -110,11 +98,9 @@ public interface TsKvHsqlRepository extends CrudRepository<TsKvEntity, TsKvCompo
             "SUM(CASE WHEN tskv.strValue IS NULL THEN 0 ELSE 1 END), " +
             "SUM(CASE WHEN tskv.longValue IS NULL THEN 0 ELSE 1 END), " +
             "SUM(CASE WHEN tskv.doubleValue IS NULL THEN 0 ELSE 1 END)) FROM TsKvEntity tskv " +
-            "WHERE tskv.entityId = :entityId AND tskv.entityType = :entityType " +
+            "WHERE tskv.entityId = :entityId AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs")
-            "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs")
+    CompletableFuture<TsKvEntity> findCount(@Param("entityId") UUID entityId,
-    CompletableFuture<TsKvEntity> findCount(@Param("entityId") String entityId,
+                                            @Param("entityKey") int entityKey,
-                                            @Param("entityType") EntityType entityType,
-                                            @Param("entityKey") String entityKey,
                                             @Param("startTs") long startTs,
                                             @Param("endTs") long endTs);
 
@@ -123,12 +109,10 @@ public interface TsKvHsqlRepository extends CrudRepository<TsKvEntity, TsKvCompo
             "SUM(COALESCE(tskv.doubleValue, 0.0)), " +
             "SUM(CASE WHEN tskv.longValue IS NULL THEN 0 ELSE 1 END), " +
             "SUM(CASE WHEN tskv.doubleValue IS NULL THEN 0 ELSE 1 END), " +
-            "'AVG') FROM TsKvEntity tskv " +
+            "'AVG') FROM TsKvEntity tskv WHERE tskv.entityId = :entityId " +
-            "WHERE tskv.entityId = :entityId AND tskv.entityType = :entityType " +
             "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs")
-    CompletableFuture<TsKvEntity> findAvg(@Param("entityId") String entityId,
+    CompletableFuture<TsKvEntity> findAvg(@Param("entityId") UUID entityId,
-                                          @Param("entityType") EntityType entityType,
+                                          @Param("entityKey") int entityKey,
-                                          @Param("entityKey") String entityKey,
                                           @Param("startTs") long startTs,
                                           @Param("endTs") long endTs);
 
@@ -137,12 +121,10 @@ public interface TsKvHsqlRepository extends CrudRepository<TsKvEntity, TsKvCompo
             "SUM(COALESCE(tskv.doubleValue, 0.0)), " +
             "SUM(CASE WHEN tskv.longValue IS NULL THEN 0 ELSE 1 END), " +
             "SUM(CASE WHEN tskv.doubleValue IS NULL THEN 0 ELSE 1 END), " +
-            "'SUM') FROM TsKvEntity tskv " +
+            "'SUM') FROM TsKvEntity tskv WHERE tskv.entityId = :entityId " +
-            "WHERE tskv.entityId = :entityId AND tskv.entityType = :entityType " +
             "AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs")
-    CompletableFuture<TsKvEntity> findSum(@Param("entityId") String entityId,
+    CompletableFuture<TsKvEntity> findSum(@Param("entityId") UUID entityId,
-                                          @Param("entityType") EntityType entityType,
+                                          @Param("entityKey") int entityKey,
-                                          @Param("entityKey") String entityKey,
                                           @Param("startTs") long startTs,
                                           @Param("endTs") long endTs);
 

dao/src/main/java/org/thingsboard/server/dao/sqlts/latest/HsqlLatestInsertRepository.java → dao/src/main/java/org/thingsboard/server/dao/sqlts/latest/HsqlLatestInsertTsRepository.java

@@ -20,7 +20,7 @@ import org.springframework.stereotype.Repository;
 import org.springframework.transaction.annotation.Transactional;
 import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity;
 import org.thingsboard.server.dao.sqlts.AbstractInsertRepository;
-import org.thingsboard.server.dao.sqlts.InsertLatestRepository;
+import org.thingsboard.server.dao.sqlts.InsertLatestTsRepository;
 import org.thingsboard.server.dao.util.HsqlDao;
 import org.thingsboard.server.dao.util.SqlTsDao;
 
@@ -33,46 +33,44 @@ import java.util.List;
 @HsqlDao
 @Repository
 @Transactional
-public class HsqlLatestInsertRepository extends AbstractInsertRepository implements InsertLatestRepository {
+public class HsqlLatestInsertTsRepository extends AbstractInsertRepository implements InsertLatestTsRepository {
 
     private static final String INSERT_OR_UPDATE =
-            "MERGE INTO ts_kv_latest USING(VALUES ?, ?, ?, ?, ?, ?, ?, ?) " +
+            "MERGE INTO ts_kv_latest USING(VALUES ?, ?, ?, ?, ?, ?, ?) " +
-                    "T (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " +
+                    "T (entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " +
-                    "ON (ts_kv_latest.entity_type=T.entity_type " +
+                    "ON (ts_kv_latest.entity_id=T.entity_id " +
-                    "AND ts_kv_latest.entity_id=T.entity_id " +
                     "AND ts_kv_latest.key=T.key) " +
                     "WHEN MATCHED THEN UPDATE SET ts_kv_latest.ts = T.ts, ts_kv_latest.bool_v = T.bool_v, ts_kv_latest.str_v = T.str_v, ts_kv_latest.long_v = T.long_v, ts_kv_latest.dbl_v = T.dbl_v " +
-                    "WHEN NOT MATCHED THEN INSERT (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " +
+                    "WHEN NOT MATCHED THEN INSERT (entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " +
-                    "VALUES (T.entity_type, T.entity_id, T.key, T.ts, T.bool_v, T.str_v, T.long_v, T.dbl_v);";
+                    "VALUES (T.entity_id, T.key, T.ts, T.bool_v, T.str_v, T.long_v, T.dbl_v);";
 
     @Override
     public void saveOrUpdate(List<TsKvLatestEntity> entities) {
         jdbcTemplate.batchUpdate(INSERT_OR_UPDATE, new BatchPreparedStatementSetter() {
             @Override
             public void setValues(PreparedStatement ps, int i) throws SQLException {
-                ps.setString(1, entities.get(i).getEntityType().name());
+                ps.setObject(1, entities.get(i).getEntityId());
-                ps.setString(2, entities.get(i).getEntityId());
+                ps.setInt(2, entities.get(i).getKey());
-                ps.setString(3, entities.get(i).getKey());
+                ps.setLong(3, entities.get(i).getTs());
-                ps.setLong(4, entities.get(i).getTs());
 
                 if (entities.get(i).getBooleanValue() != null) {
-                    ps.setBoolean(5, entities.get(i).getBooleanValue());
+                    ps.setBoolean(4, entities.get(i).getBooleanValue());
                 } else {
-                    ps.setNull(5, Types.BOOLEAN);
+                    ps.setNull(4, Types.BOOLEAN);
                 }
 
-                ps.setString(6, entities.get(i).getStrValue());
+                ps.setString(5, entities.get(i).getStrValue());
 
                 if (entities.get(i).getLongValue() != null) {
-                    ps.setLong(7, entities.get(i).getLongValue());
+                    ps.setLong(6, entities.get(i).getLongValue());
                 } else {
-                    ps.setNull(7, Types.BIGINT);
+                    ps.setNull(6, Types.BIGINT);
                 }
 
                 if (entities.get(i).getDoubleValue() != null) {
-                    ps.setDouble(8, entities.get(i).getDoubleValue());
+                    ps.setDouble(7, entities.get(i).getDoubleValue());
                 } else {
-                    ps.setNull(8, Types.DOUBLE);
+                    ps.setNull(7, Types.DOUBLE);
                 }
             }
 

dao/src/main/java/org/thingsboard/server/dao/sqlts/latest/PsqlLatestInsertRepository.java → dao/src/main/java/org/thingsboard/server/dao/sqlts/latest/PsqlLatestInsertTsRepository.java

@@ -22,7 +22,7 @@ import org.springframework.transaction.annotation.Transactional;
 import org.springframework.transaction.support.TransactionCallbackWithoutResult;
 import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity;
 import org.thingsboard.server.dao.sqlts.AbstractInsertRepository;
-import org.thingsboard.server.dao.sqlts.InsertLatestRepository;
+import org.thingsboard.server.dao.sqlts.InsertLatestTsRepository;
 import org.thingsboard.server.dao.util.PsqlTsAnyDao;
 
 import java.sql.PreparedStatement;
@@ -35,15 +35,15 @@ import java.util.List;
 @PsqlTsAnyDao
 @Repository
 @Transactional
-public class PsqlLatestInsertRepository extends AbstractInsertRepository implements InsertLatestRepository {
+public class PsqlLatestInsertTsRepository extends AbstractInsertRepository implements InsertLatestTsRepository {
 
     private static final String BATCH_UPDATE =
-            "UPDATE ts_kv_latest SET ts = ?, bool_v = ?, str_v = ?, long_v = ?, dbl_v = ? WHERE entity_type = ? AND entity_id = ? and key = ?";
+            "UPDATE ts_kv_latest SET ts = ?, bool_v = ?, str_v = ?, long_v = ?, dbl_v = ? WHERE entity_id = ? and key = ?";
 
 
     private static final String INSERT_OR_UPDATE =
-            "INSERT INTO ts_kv_latest (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) VALUES(?, ?, ?, ?, ?, ?, ?, ?) " +
+            "INSERT INTO ts_kv_latest (entity_id, key, ts, bool_v, str_v, long_v, dbl_v) VALUES(?, ?, ?, ?, ?, ?, ?) " +
-                    "ON CONFLICT (entity_type, entity_id, key) DO UPDATE SET ts = ?, bool_v = ?, str_v = ?, long_v = ?, dbl_v = ?;";
+                    "ON CONFLICT (entity_id, key) DO UPDATE SET ts = ?, bool_v = ?, str_v = ?, long_v = ?, dbl_v = ?;";
 
     @Override
     public void saveOrUpdate(List<TsKvLatestEntity> entities) {
@@ -76,9 +76,8 @@ public class PsqlLatestInsertRepository extends AbstractInsertRepository impleme
                             ps.setNull(5, Types.DOUBLE);
                         }
 
-                        ps.setString(6, tsKvLatestEntity.getEntityType().name());
+                        ps.setObject(6, tsKvLatestEntity.getEntityId());
-                        ps.setString(7, tsKvLatestEntity.getEntityId());
+                        ps.setInt(7, tsKvLatestEntity.getKey());
-                        ps.setString(8, tsKvLatestEntity.getKey());
                     }
 
                     @Override
@@ -105,38 +104,37 @@ public class PsqlLatestInsertRepository extends AbstractInsertRepository impleme
                     @Override
                     public void setValues(PreparedStatement ps, int i) throws SQLException {
                         TsKvLatestEntity tsKvLatestEntity = insertEntities.get(i);
-                        ps.setString(1, tsKvLatestEntity.getEntityType().name());
+                        ps.setObject(1, tsKvLatestEntity.getEntityId());
-                        ps.setString(2, tsKvLatestEntity.getEntityId());
+                        ps.setInt(2, tsKvLatestEntity.getKey());
-                        ps.setString(3, tsKvLatestEntity.getKey());
+                        ps.setLong(3, tsKvLatestEntity.getTs());
-                        ps.setLong(4, tsKvLatestEntity.getTs());
+                        ps.setLong(8, tsKvLatestEntity.getTs());
-                        ps.setLong(9, tsKvLatestEntity.getTs());
 
                         if (tsKvLatestEntity.getBooleanValue() != null) {
-                            ps.setBoolean(5, tsKvLatestEntity.getBooleanValue());
+                            ps.setBoolean(4, tsKvLatestEntity.getBooleanValue());
-                            ps.setBoolean(10, tsKvLatestEntity.getBooleanValue());
+                            ps.setBoolean(9, tsKvLatestEntity.getBooleanValue());
                         } else {
-                            ps.setNull(5, Types.BOOLEAN);
+                            ps.setNull(4, Types.BOOLEAN);
-                            ps.setNull(10, Types.BOOLEAN);
+                            ps.setNull(9, Types.BOOLEAN);
                         }
 
-                        ps.setString(6, replaceNullChars(tsKvLatestEntity.getStrValue()));
+                        ps.setString(5, replaceNullChars(tsKvLatestEntity.getStrValue()));
-                        ps.setString(11, replaceNullChars(tsKvLatestEntity.getStrValue()));
+                        ps.setString(10, replaceNullChars(tsKvLatestEntity.getStrValue()));
 
 
                         if (tsKvLatestEntity.getLongValue() != null) {
-                            ps.setLong(7, tsKvLatestEntity.getLongValue());
+                            ps.setLong(6, tsKvLatestEntity.getLongValue());
-                            ps.setLong(12, tsKvLatestEntity.getLongValue());
+                            ps.setLong(11, tsKvLatestEntity.getLongValue());
                         } else {
-                            ps.setNull(7, Types.BIGINT);
+                            ps.setNull(6, Types.BIGINT);
-                            ps.setNull(12, Types.BIGINT);
+                            ps.setNull(11, Types.BIGINT);
                         }
 
                         if (tsKvLatestEntity.getDoubleValue() != null) {
-                            ps.setDouble(8, tsKvLatestEntity.getDoubleValue());
+                            ps.setDouble(7, tsKvLatestEntity.getDoubleValue());
-                            ps.setDouble(13, tsKvLatestEntity.getDoubleValue());
+                            ps.setDouble(12, tsKvLatestEntity.getDoubleValue());
                         } else {
-                            ps.setNull(8, Types.DOUBLE);
+                            ps.setNull(7, Types.DOUBLE);
-                            ps.setNull(13, Types.DOUBLE);
+                            ps.setNull(12, Types.DOUBLE);
                         }
                     }
 

dao/src/main/java/org/thingsboard/server/dao/sqlts/latest/SearchTsKvLatestRepository.java

@@ -0,0 +1,45 @@
+/**
+ * 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.dao.sqlts.latest;
+
+import org.springframework.stereotype.Repository;
+import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity;
+import org.thingsboard.server.dao.util.SqlTsAnyDao;
+
+import javax.persistence.EntityManager;
+import javax.persistence.PersistenceContext;
+import java.util.List;
+import java.util.UUID;
+
+@SqlTsAnyDao
+@Repository
+public class SearchTsKvLatestRepository {
+
+    public static final String FIND_ALL_BY_ENTITY_ID = "findAllByEntityId";
+    public static final String FIND_ALL_BY_ENTITY_ID_QUERY = "SELECT ts_kv_latest.entity_id AS entityId, ts_kv_latest.key AS key, ts_kv_dictionary.key AS strKey, ts_kv_latest.str_v AS strValue," +
+            " ts_kv_latest.bool_v AS boolValue, ts_kv_latest.long_v AS longValue, ts_kv_latest.dbl_v AS doubleValue, ts_kv_latest.ts AS ts FROM ts_kv_latest " +
+            "INNER JOIN ts_kv_dictionary ON ts_kv_latest.key = ts_kv_dictionary.key_id WHERE ts_kv_latest.entity_id = cast(:id AS uuid)";
+
+    @PersistenceContext
+    private EntityManager entityManager;
+
+    public List<TsKvLatestEntity> findAllByEntityId(UUID entityId) {
+        return entityManager.createNamedQuery(FIND_ALL_BY_ENTITY_ID, TsKvLatestEntity.class)
+                .setParameter("id", entityId)
+                .getResultList();
+    }
+
+}

dao/src/main/java/org/thingsboard/server/dao/sqlts/latest/TsKvLatestRepository.java

@@ -16,15 +16,15 @@
 package org.thingsboard.server.dao.sqlts.latest;
 
 import org.springframework.data.repository.CrudRepository;
-import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestCompositeKey;
 import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity;
 import org.thingsboard.server.dao.util.SqlDao;
 
 import java.util.List;
+import java.util.UUID;
 
 @SqlDao
 public interface TsKvLatestRepository extends CrudRepository<TsKvLatestEntity, TsKvLatestCompositeKey> {
 
-    List<TsKvLatestEntity> findAllByEntityTypeAndEntityId(EntityType entityType, String entityId);
+    List<TsKvLatestEntity> findAllByEntityId(UUID entityId);
 }

dao/src/main/java/org/thingsboard/server/dao/sqlts/psql/JpaPsqlTimeseriesDao.java

@@ -18,7 +18,6 @@ package org.thingsboard.server.dao.sqlts.psql;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import lombok.extern.slf4j.Slf4j;
-import org.hibernate.exception.ConstraintViolationException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.domain.PageRequest;
@@ -31,12 +30,9 @@ import org.thingsboard.server.common.data.kv.DeleteTsKvQuery;
 import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.dao.DaoUtil;
-import org.thingsboard.server.dao.model.sqlts.dictionary.TsKvDictionary;
-import org.thingsboard.server.dao.model.sqlts.dictionary.TsKvDictionaryCompositeKey;
 import org.thingsboard.server.dao.model.sqlts.psql.TsKvEntity;
-import org.thingsboard.server.dao.sqlts.AbstractSimpleSqlTimeseriesDao;
+import org.thingsboard.server.dao.sqlts.AbstractChunkedAggregationTimeseriesDao;
 import org.thingsboard.server.dao.sqlts.EntityContainer;
-import org.thingsboard.server.dao.sqlts.dictionary.TsKvDictionaryRepository;
 import org.thingsboard.server.dao.timeseries.PsqlPartition;
 import org.thingsboard.server.dao.timeseries.SqlTsPartitionDate;
 import org.thingsboard.server.dao.timeseries.TimeseriesDao;
@@ -48,10 +44,12 @@ import java.time.LocalDateTime;
 import java.time.ZoneOffset;
 import java.time.ZonedDateTime;
 import java.time.format.DateTimeFormatter;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.locks.ReentrantLock;
 
 import static org.thingsboard.server.dao.timeseries.SqlTsPartitionDate.EPOCH_START;
@@ -61,17 +59,11 @@ import static org.thingsboard.server.dao.timeseries.SqlTsPartitionDate.EPOCH_STA
 @Slf4j
 @SqlTsDao
 @PsqlDao
-public class JpaPsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEntity> implements TimeseriesDao {
+public class JpaPsqlTimeseriesDao extends AbstractChunkedAggregationTimeseriesDao<TsKvEntity> implements TimeseriesDao {
 
-    private final ConcurrentMap<String, Integer> tsKvDictionaryMap = new ConcurrentHashMap<>();
     private final Map<Long, PsqlPartition> partitions = new ConcurrentHashMap<>();
-
-    private static final ReentrantLock tsCreationLock = new ReentrantLock();
     private static final ReentrantLock partitionCreationLock = new ReentrantLock();
 
-    @Autowired
-    private TsKvDictionaryRepository dictionaryRepository;
-
     @Autowired
     private TsKvPsqlRepository tsKvRepository;
 
@@ -81,7 +73,7 @@ public class JpaPsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
     private SqlTsPartitionDate tsFormat;
     private PsqlPartition indefinitePartition;
 
-    @Value("${sql.ts_key_value_partitioning}")
+    @Value("${sql.postgres.ts_key_value_partitioning:MONTHS}")
     private String partitioning;
 
     @Override
@@ -100,11 +92,6 @@ public class JpaPsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
         }
     }
 
-    @Override
-    public ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, List<ReadTsKvQuery> queries) {
-        return processFindAllAsync(tenantId, entityId, queries);
-    }
-
     @Override
     public ListenableFuture<Void> save(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry, long ttl) {
         String strKey = tsKvEntry.getKey();
@@ -166,22 +153,25 @@ public class JpaPsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
         return Futures.immediateFuture(null);
     }
 
-    protected ListenableFuture<Optional<TsKvEntry>> findAndAggregateAsync(EntityId entityId, String key, long startTs, long endTs, long ts, Aggregation aggregation) {
+    protected ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, ReadTsKvQuery query) {
-        List<CompletableFuture<TsKvEntity>> entitiesFutures = new ArrayList<>();
+        if (query.getAggregation() == Aggregation.NONE) {
-        switchAgregation(entityId, key, startTs, endTs, aggregation, entitiesFutures);
+            return findAllAsyncWithLimit(tenantId, entityId, query);
-        return Futures.transform(setFutures(entitiesFutures), entity -> {
+        } else {
-            if (entity != null && entity.isNotEmpty()) {
+            long stepTs = query.getStartTs();
-                entity.setEntityId(entityId.getId());
+            List<ListenableFuture<Optional<TsKvEntry>>> futures = new ArrayList<>();
-                entity.setStrKey(key);
+            while (stepTs < query.getEndTs()) {
-                entity.setTs(ts);
+                long startTs = stepTs;
-                return Optional.of(DaoUtil.getData(entity));
+                long endTs = stepTs + query.getInterval();
-            } else {
+                long ts = startTs + (endTs - startTs) / 2;
-                return Optional.empty();
+                futures.add(findAndAggregateAsync(tenantId, entityId, query.getKey(), startTs, endTs, ts, query.getAggregation()));
+                stepTs = endTs;
             }
-        });
+            return getTskvEntriesFuture(Futures.allAsList(futures));
+        }
     }
 
-    protected ListenableFuture<List<TsKvEntry>> findAllAsyncWithLimit(EntityId entityId, ReadTsKvQuery query) {
+    @Override
+    protected ListenableFuture<List<TsKvEntry>> findAllAsyncWithLimit(TenantId tenantId, EntityId entityId, ReadTsKvQuery query) {
         Integer keyId = getOrSaveKeyId(query.getKey());
         List<TsKvEntity> tsKvEntities = tsKvRepository.findAllWithLimit(
                 entityId.getId(),
@@ -195,7 +185,24 @@ public class JpaPsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
         return Futures.immediateFuture(DaoUtil.convertDataList(tsKvEntities));
     }
 
-    protected void findCount(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+    @Override
+    protected ListenableFuture<Optional<TsKvEntry>> findAndAggregateAsync(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, long ts, Aggregation aggregation) {
+        List<CompletableFuture<TsKvEntity>> entitiesFutures = new ArrayList<>();
+        switchAggregation(tenantId, entityId, key, startTs, endTs, aggregation, entitiesFutures);
+        return Futures.transform(setFutures(entitiesFutures), entity -> {
+            if (entity != null && entity.isNotEmpty()) {
+                entity.setEntityId(entityId.getId());
+                entity.setStrKey(key);
+                entity.setTs(ts);
+                return Optional.of(DaoUtil.getData(entity));
+            } else {
+                return Optional.empty();
+            }
+        });
+    }
+
+    @Override
+    protected void findCount(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
         Integer keyId = getOrSaveKeyId(key);
         entitiesFutures.add(tsKvRepository.findCount(
                 entityId.getId(),
@@ -204,7 +211,8 @@ public class JpaPsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
                 endTs));
     }
 
-    protected void findSum(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+    @Override
+    protected void findSum(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
         Integer keyId = getOrSaveKeyId(key);
         entitiesFutures.add(tsKvRepository.findSum(
                 entityId.getId(),
@@ -213,7 +221,8 @@ public class JpaPsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
                 endTs));
     }
 
-    protected void findMin(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+    @Override
+    protected void findMin(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
         Integer keyId = getOrSaveKeyId(key);
         entitiesFutures.add(tsKvRepository.findStringMin(
                 entityId.getId(),
@@ -227,7 +236,8 @@ public class JpaPsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
                 endTs));
     }
 
-    protected void findMax(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+    @Override
+    protected void findMax(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
         Integer keyId = getOrSaveKeyId(key);
         entitiesFutures.add(tsKvRepository.findStringMax(
                 entityId.getId(),
@@ -241,7 +251,8 @@ public class JpaPsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
                 endTs));
     }
 
-    protected void findAvg(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
+    @Override
+    protected void findAvg(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) {
         Integer keyId = getOrSaveKeyId(key);
         entitiesFutures.add(tsKvRepository.findAvg(
                 entityId.getId(),
@@ -250,40 +261,9 @@ public class JpaPsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEnt
                 endTs));
     }
 
-    private Integer getOrSaveKeyId(String strKey) {
+    @Override
-        Integer keyId = tsKvDictionaryMap.get(strKey);
+    public ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, List<ReadTsKvQuery> queries) {
-        if (keyId == null) {
+        return processFindAllAsync(tenantId, entityId, queries);
-            Optional<TsKvDictionary> tsKvDictionaryOptional;
-            tsKvDictionaryOptional = dictionaryRepository.findById(new TsKvDictionaryCompositeKey(strKey));
-            if (!tsKvDictionaryOptional.isPresent()) {
-                tsCreationLock.lock();
-                try {
-                    tsKvDictionaryOptional = dictionaryRepository.findById(new TsKvDictionaryCompositeKey(strKey));
-                    if (!tsKvDictionaryOptional.isPresent()) {
-                        TsKvDictionary tsKvDictionary = new TsKvDictionary();
-                        tsKvDictionary.setKey(strKey);
-                        try {
-                            TsKvDictionary saved = dictionaryRepository.save(tsKvDictionary);
-                            tsKvDictionaryMap.put(saved.getKey(), saved.getKeyId());
-                            keyId = saved.getKeyId();
-                        } catch (ConstraintViolationException e) {
-                            tsKvDictionaryOptional = dictionaryRepository.findById(new TsKvDictionaryCompositeKey(strKey));
-                            TsKvDictionary dictionary = tsKvDictionaryOptional.orElseThrow(() -> new RuntimeException("Failed to get TsKvDictionary entity from DB!"));
-                            tsKvDictionaryMap.put(dictionary.getKey(), dictionary.getKeyId());
-                            keyId = dictionary.getKeyId();
-                        }
-                    } else {
-                        keyId = tsKvDictionaryOptional.get().getKeyId();
-                    }
-                } finally {
-                    tsCreationLock.unlock();
-                }
-            } else {
-                keyId = tsKvDictionaryOptional.get().getKeyId();
-                tsKvDictionaryMap.put(strKey, keyId);
-            }
-        }
-        return keyId;
     }
 
     private void savePartition(PsqlPartition psqlPartition) {

dao/src/main/java/org/thingsboard/server/dao/sqlts/psql/PsqlTimeseriesInsertRepository.java → dao/src/main/java/org/thingsboard/server/dao/sqlts/psql/PsqlInsertTsRepository.java

@@ -37,7 +37,7 @@ import java.util.Map;
 @PsqlDao
 @Repository
 @Transactional
-public class PsqlTimeseriesInsertRepository extends AbstractInsertRepository implements InsertTsRepository<TsKvEntity> {
+public class PsqlInsertTsRepository extends AbstractInsertRepository implements InsertTsRepository<TsKvEntity> {
 
     private static final String INSERT_INTO_TS_KV = "INSERT INTO ts_kv_";
 

dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleInsertRepository.java → dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleInsertTsRepository.java

@@ -34,7 +34,7 @@ import java.util.List;
 @PsqlDao
 @Repository
 @Transactional
-public class TimescaleInsertRepository extends AbstractInsertRepository implements InsertTsRepository<TimescaleTsKvEntity> {
+public class TimescaleInsertTsRepository extends AbstractInsertRepository implements InsertTsRepository<TimescaleTsKvEntity> {
 
     private static final String INSERT_OR_UPDATE =
             "INSERT INTO tenant_ts_kv (tenant_id, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) VALUES(?, ?, ?, ?, ?, ?, ?, ?) " +

dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleTimeseriesDao.java

@@ -19,9 +19,7 @@ import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import lombok.extern.slf4j.Slf4j;
-import org.hibernate.exception.ConstraintViolationException;
 import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Value;
 import org.springframework.data.domain.PageRequest;
 import org.springframework.data.domain.Sort;
 import org.springframework.stereotype.Component;
@@ -33,15 +31,12 @@ import org.thingsboard.server.common.data.kv.DeleteTsKvQuery;
 import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.dao.DaoUtil;
-import org.thingsboard.server.dao.model.sqlts.dictionary.TsKvDictionary;
-import org.thingsboard.server.dao.model.sqlts.dictionary.TsKvDictionaryCompositeKey;
 import org.thingsboard.server.dao.model.sqlts.timescale.TimescaleTsKvEntity;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueue;
 import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams;
 import org.thingsboard.server.dao.sqlts.AbstractSqlTimeseriesDao;
 import org.thingsboard.server.dao.sqlts.EntityContainer;
 import org.thingsboard.server.dao.sqlts.InsertTsRepository;
-import org.thingsboard.server.dao.sqlts.dictionary.TsKvDictionaryRepository;
 import org.thingsboard.server.dao.timeseries.TimeseriesDao;
 import org.thingsboard.server.dao.util.TimescaleDBTsDao;
 
@@ -53,25 +48,12 @@ import java.util.List;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.locks.ReentrantLock;
-
 
 @Component
 @Slf4j
 @TimescaleDBTsDao
 public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements TimeseriesDao {
 
-    private static final String TS = "ts";
-
-    private final ConcurrentMap<String, Integer> tsKvDictionaryMap = new ConcurrentHashMap<>();
-
-    private static final ReentrantLock tsCreationLock = new ReentrantLock();
-
-    @Autowired
-    private TsKvDictionaryRepository dictionaryRepository;
-
     @Autowired
     private TsKvTimescaleRepository tsKvRepository;
 
@@ -79,40 +61,32 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
     private AggregationRepository aggregationRepository;
 
     @Autowired
-    private InsertTsRepository<TimescaleTsKvEntity> insertRepository;
+    protected InsertTsRepository<TimescaleTsKvEntity> insertRepository;
-
-    @Value("${sql.ts_timescale.batch_size:1000}")
-    private int batchSize;
-
-    @Value("${sql.ts_timescale.batch_max_delay:100}")
-    private long maxDelay;
 
-    @Value("${sql.ts_timescale.stats_print_interval_ms:1000}")
+    protected TbSqlBlockingQueue<EntityContainer<TimescaleTsKvEntity>> tsQueue;
-    private long statsPrintIntervalMs;
-
-    private TbSqlBlockingQueue<EntityContainer<TimescaleTsKvEntity>> queue;
 
     @PostConstruct
     protected void init() {
         super.init();
-        TbSqlBlockingQueueParams params = TbSqlBlockingQueueParams.builder()
+        TbSqlBlockingQueueParams tsParams = TbSqlBlockingQueueParams.builder()
                 .logName("TS Timescale")
-                .batchSize(batchSize)
+                .batchSize(tsBatchSize)
-                .maxDelay(maxDelay)
+                .maxDelay(tsMaxDelay)
-                .statsPrintIntervalMs(statsPrintIntervalMs)
+                .statsPrintIntervalMs(tsStatsPrintIntervalMs)
                 .build();
-        queue = new TbSqlBlockingQueue<>(params);
+        tsQueue = new TbSqlBlockingQueue<>(tsParams);
-        queue.init(logExecutor, v -> insertRepository.saveOrUpdate(v));
+        tsQueue.init(logExecutor, v -> insertRepository.saveOrUpdate(v));
     }
 
     @PreDestroy
     protected void destroy() {
         super.destroy();
-        if (queue != null) {
+        if (tsQueue != null) {
-            queue.destroy();
+            tsQueue.destroy();
         }
     }
 
+    @Override
     protected ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, ReadTsKvQuery query) {
         if (query.getAggregation() == Aggregation.NONE) {
             return findAllAsyncWithLimit(tenantId, entityId, query);
@@ -120,11 +94,58 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
             long startTs = query.getStartTs();
             long endTs = query.getEndTs();
             long timeBucket = query.getInterval();
-            ListenableFuture<List<Optional<TsKvEntry>>> future = findAndAggregateAsync(tenantId, entityId, query.getKey(), startTs, endTs, timeBucket, query.getAggregation());
+            ListenableFuture<List<Optional<TsKvEntry>>> future = findAllAndAggregateAsync(tenantId, entityId, query.getKey(), startTs, endTs, timeBucket, query.getAggregation());
             return getTskvEntriesFuture(future);
         }
     }
 
+    @Override
+    protected ListenableFuture<List<TsKvEntry>> findAllAsyncWithLimit(TenantId tenantId, EntityId entityId, ReadTsKvQuery query) {
+        String strKey = query.getKey();
+        Integer keyId = getOrSaveKeyId(strKey);
+        List<TimescaleTsKvEntity> timescaleTsKvEntities = tsKvRepository.findAllWithLimit(
+                tenantId.getId(),
+                entityId.getId(),
+                keyId,
+                query.getStartTs(),
+                query.getEndTs(),
+                new PageRequest(0, query.getLimit(),
+                        new Sort(Sort.Direction.fromString(
+                                query.getOrderBy()), "ts")));
+        timescaleTsKvEntities.forEach(tsKvEntity -> tsKvEntity.setStrKey(strKey));
+        return Futures.immediateFuture(DaoUtil.convertDataList(timescaleTsKvEntities));
+    }
+
+    private ListenableFuture<List<Optional<TsKvEntry>>> findAllAndAggregateAsync(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, long timeBucket, Aggregation aggregation) {
+        CompletableFuture<List<TimescaleTsKvEntity>> listCompletableFuture = switchAggregation(key, startTs, endTs, timeBucket, aggregation, entityId.getId(), tenantId.getId());
+        SettableFuture<List<TimescaleTsKvEntity>> listenableFuture = SettableFuture.create();
+        listCompletableFuture.whenComplete((timescaleTsKvEntities, throwable) -> {
+            if (throwable != null) {
+                listenableFuture.setException(throwable);
+            } else {
+                listenableFuture.set(timescaleTsKvEntities);
+            }
+        });
+        return Futures.transform(listenableFuture, timescaleTsKvEntities -> {
+            if (!CollectionUtils.isEmpty(timescaleTsKvEntities)) {
+                List<Optional<TsKvEntry>> result = new ArrayList<>();
+                timescaleTsKvEntities.forEach(entity -> {
+                    if (entity != null && entity.isNotEmpty()) {
+                        entity.setEntityId(entityId.getId());
+                        entity.setTenantId(tenantId.getId());
+                        entity.setStrKey(key);
+                        result.add(Optional.of(DaoUtil.getData(entity)));
+                    } else {
+                        result.add(Optional.empty());
+                    }
+                });
+                return result;
+            } else {
+                return Collections.emptyList();
+            }
+        });
+    }
+
     @Override
     public ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, List<ReadTsKvQuery> queries) {
         return processFindAllAsync(tenantId, entityId, queries);
@@ -154,7 +175,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
         entity.setLongValue(tsKvEntry.getLongValue().orElse(null));
         entity.setBooleanValue(tsKvEntry.getBooleanValue().orElse(null));
         log.trace("Saving entity to timescale db: {}", entity);
-        return queue.add(new EntityContainer(entity, null));
+        return tsQueue.add(new EntityContainer(entity, null));
     }
 
     @Override
@@ -192,89 +213,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
         return service.submit(() -> null);
     }
 
-    private ListenableFuture<List<TsKvEntry>> findAllAsyncWithLimit(TenantId tenantId, EntityId entityId, ReadTsKvQuery query) {
+    private CompletableFuture<List<TimescaleTsKvEntity>> switchAggregation(String key, long startTs, long endTs, long timeBucket, Aggregation aggregation, UUID entityId, UUID tenantId) {
-        String strKey = query.getKey();
-        Integer keyId = getOrSaveKeyId(strKey);
-        List<TimescaleTsKvEntity> timescaleTsKvEntities = tsKvRepository.findAllWithLimit(
-                tenantId.getId(),
-                entityId.getId(),
-                keyId,
-                query.getStartTs(),
-                query.getEndTs(),
-                new PageRequest(0, query.getLimit(),
-                        new Sort(Sort.Direction.fromString(
-                                query.getOrderBy()), TS)));
-        timescaleTsKvEntities.forEach(tsKvEntity -> tsKvEntity.setStrKey(strKey));
-        return Futures.immediateFuture(DaoUtil.convertDataList(timescaleTsKvEntities));
-    }
-
-    private Integer getOrSaveKeyId(String strKey) {
-        Integer keyId = tsKvDictionaryMap.get(strKey);
-        if (keyId == null) {
-            Optional<TsKvDictionary> tsKvDictionaryOptional;
-            tsKvDictionaryOptional = dictionaryRepository.findById(new TsKvDictionaryCompositeKey(strKey));
-            if (!tsKvDictionaryOptional.isPresent()) {
-                tsCreationLock.lock();
-                try {
-                    tsKvDictionaryOptional = dictionaryRepository.findById(new TsKvDictionaryCompositeKey(strKey));
-                    if (!tsKvDictionaryOptional.isPresent()) {
-                        TsKvDictionary tsKvDictionary = new TsKvDictionary();
-                        tsKvDictionary.setKey(strKey);
-                        try {
-                            TsKvDictionary saved = dictionaryRepository.save(tsKvDictionary);
-                            tsKvDictionaryMap.put(saved.getKey(), saved.getKeyId());
-                            keyId = saved.getKeyId();
-                        } catch (ConstraintViolationException e) {
-                            tsKvDictionaryOptional = dictionaryRepository.findById(new TsKvDictionaryCompositeKey(strKey));
-                            TsKvDictionary dictionary = tsKvDictionaryOptional.orElseThrow(() -> new RuntimeException("Failed to get TsKvDictionary entity from DB!"));
-                            tsKvDictionaryMap.put(dictionary.getKey(), dictionary.getKeyId());
-                            keyId = dictionary.getKeyId();
-                        }
-                    } else {
-                        keyId = tsKvDictionaryOptional.get().getKeyId();
-                    }
-                } finally {
-                    tsCreationLock.unlock();
-                }
-            } else {
-                keyId = tsKvDictionaryOptional.get().getKeyId();
-                tsKvDictionaryMap.put(strKey, keyId);
-            }
-        }
-        return keyId;
-    }
-
-    private ListenableFuture<List<Optional<TsKvEntry>>> findAndAggregateAsync(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, long timeBucket, Aggregation aggregation) {
-        CompletableFuture<List<TimescaleTsKvEntity>> listCompletableFuture = switchAgregation(key, startTs, endTs, timeBucket, aggregation, entityId.getId(), tenantId.getId());
-        SettableFuture<List<TimescaleTsKvEntity>> listenableFuture = SettableFuture.create();
-        listCompletableFuture.whenComplete((timescaleTsKvEntities, throwable) -> {
-            if (throwable != null) {
-                listenableFuture.setException(throwable);
-            } else {
-                listenableFuture.set(timescaleTsKvEntities);
-            }
-        });
-        return Futures.transform(listenableFuture, timescaleTsKvEntities -> {
-            if (!CollectionUtils.isEmpty(timescaleTsKvEntities)) {
-                List<Optional<TsKvEntry>> result = new ArrayList<>();
-                timescaleTsKvEntities.forEach(entity -> {
-                    if (entity != null && entity.isNotEmpty()) {
-                        entity.setEntityId(entityId.getId());
-                        entity.setTenantId(tenantId.getId());
-                        entity.setStrKey(key);
-                        result.add(Optional.of(DaoUtil.getData(entity)));
-                    } else {
-                        result.add(Optional.empty());
-                    }
-                });
-                return result;
-            } else {
-                return Collections.emptyList();
-            }
-        });
-    }
-
-    private CompletableFuture<List<TimescaleTsKvEntity>> switchAgregation(String key, long startTs, long endTs, long timeBucket, Aggregation aggregation, UUID entityId, UUID tenantId) {
         switch (aggregation) {
             case AVG:
                 return findAvg(key, startTs, endTs, timeBucket, entityId, tenantId);
@@ -291,9 +230,9 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
         }
     }
 
-    private CompletableFuture<List<TimescaleTsKvEntity>> findAvg(String key, long startTs, long endTs, long timeBucket, UUID entityId, UUID tenantId) {
+    private CompletableFuture<List<TimescaleTsKvEntity>> findCount(String key, long startTs, long endTs, long timeBucket, UUID entityId, UUID tenantId) {
         Integer keyId = getOrSaveKeyId(key);
-        return aggregationRepository.findAvg(
+        return aggregationRepository.findCount(
                 tenantId,
                 entityId,
                 keyId,
@@ -302,9 +241,9 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
                 endTs);
     }
 
-    private CompletableFuture<List<TimescaleTsKvEntity>> findMax(String key, long startTs, long endTs, long timeBucket, UUID entityId, UUID tenantId) {
+    private CompletableFuture<List<TimescaleTsKvEntity>> findSum(String key, long startTs, long endTs, long timeBucket, UUID entityId, UUID tenantId) {
         Integer keyId = getOrSaveKeyId(key);
-        return aggregationRepository.findMax(
+        return aggregationRepository.findSum(
                 tenantId,
                 entityId,
                 keyId,
@@ -322,12 +261,11 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
                 timeBucket,
                 startTs,
                 endTs);
-
     }
 
-    private CompletableFuture<List<TimescaleTsKvEntity>> findSum(String key, long startTs, long endTs, long timeBucket, UUID entityId, UUID tenantId) {
+    private CompletableFuture<List<TimescaleTsKvEntity>> findMax(String key, long startTs, long endTs, long timeBucket, UUID entityId, UUID tenantId) {
         Integer keyId = getOrSaveKeyId(key);
-        return aggregationRepository.findSum(
+        return aggregationRepository.findMax(
                 tenantId,
                 entityId,
                 keyId,
@@ -336,9 +274,9 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements
                 endTs);
     }
 
-    private CompletableFuture<List<TimescaleTsKvEntity>> findCount(String key, long startTs, long endTs, long timeBucket, UUID entityId, UUID tenantId) {
+    private CompletableFuture<List<TimescaleTsKvEntity>> findAvg(String key, long startTs, long endTs, long timeBucket, UUID entityId, UUID tenantId) {
         Integer keyId = getOrSaveKeyId(key);
-        return aggregationRepository.findCount(
+        return aggregationRepository.findAvg(
                 tenantId,
                 entityId,
                 keyId,

dao/src/main/resources/sql/schema-timescale.sql

@@ -25,7 +25,7 @@ CREATE TABLE IF NOT EXISTS tenant_ts_kv (
     str_v varchar(10000000),
     long_v bigint,
     dbl_v double precision,
-    CONSTRAINT ts_kv_pkey PRIMARY KEY (tenant_id, entity_id, key, ts)
+    CONSTRAINT tenant_ts_kv_pkey PRIMARY KEY (tenant_id, entity_id, key, ts)
 );
 
 CREATE TABLE IF NOT EXISTS ts_kv_dictionary (
@@ -35,15 +35,12 @@ CREATE TABLE IF NOT EXISTS ts_kv_dictionary (
 );
 
 CREATE TABLE IF NOT EXISTS ts_kv_latest (
-    entity_type varchar(255) NOT NULL,
+    entity_id uuid NOT NULL,
-    entity_id varchar(31) NOT NULL,
+    key int NOT NULL,
-    key varchar(255) NOT NULL,
     ts bigint NOT NULL,
     bool_v boolean,
     str_v varchar(10000000),
     long_v bigint,
     dbl_v double precision,
-    CONSTRAINT ts_kv_latest_pkey PRIMARY KEY (entity_type, entity_id, key)
+    CONSTRAINT ts_kv_latest_pkey PRIMARY KEY (entity_id, key)
-);
+);
-
-SELECT create_hypertable('tenant_ts_kv', 'ts', chunk_time_interval => 86400000, if_not_exists => true);

dao/src/main/resources/sql/schema-ts-hsql.sql

@@ -14,26 +14,32 @@
 -- limitations under the License.
 --
 
+SET DATABASE SQL SYNTAX PGS TRUE;
+
 CREATE TABLE IF NOT EXISTS ts_kv (
-    entity_type varchar(255) NOT NULL,
+    entity_id uuid NOT NULL,
-    entity_id varchar(31) NOT NULL,
+    key int NOT NULL,
-    key varchar(255) NOT NULL,
     ts bigint NOT NULL,
     bool_v boolean,
     str_v varchar(10000000),
     long_v bigint,
     dbl_v double precision,
-    CONSTRAINT ts_kv_pkey PRIMARY KEY (entity_type, entity_id, key, ts)
+    CONSTRAINT ts_kv_pkey PRIMARY KEY (entity_id, key, ts)
 );
 
 CREATE TABLE IF NOT EXISTS ts_kv_latest (
-    entity_type varchar(255) NOT NULL,
+    entity_id uuid NOT NULL,
-    entity_id varchar(31) NOT NULL,
+    key int NOT NULL,
-    key varchar(255) NOT NULL,
     ts bigint NOT NULL,
     bool_v boolean,
     str_v varchar(10000000),
     long_v bigint,
     dbl_v double precision,
-    CONSTRAINT ts_kv_latest_pkey PRIMARY KEY (entity_type, entity_id, key)
+    CONSTRAINT ts_kv_latest_pkey PRIMARY KEY (entity_id, key)
+);
+
+CREATE TABLE IF NOT EXISTS ts_kv_dictionary (
+    key varchar(255) NOT NULL,
+    key_id int GENERATED BY DEFAULT AS IDENTITY(start with 0 increment by 1) UNIQUE,
+    CONSTRAINT ts_key_id_pkey PRIMARY KEY (key)
 );

dao/src/main/resources/sql/schema-ts-psql.sql

@@ -24,20 +24,19 @@ CREATE TABLE IF NOT EXISTS ts_kv (
     dbl_v double precision
 ) PARTITION BY RANGE (ts);
 
-CREATE TABLE IF NOT EXISTS ts_kv_dictionary (
-    key varchar(255) NOT NULL,
-    key_id serial UNIQUE,
-    CONSTRAINT ts_key_id_pkey PRIMARY KEY (key)
-);
-
 CREATE TABLE IF NOT EXISTS ts_kv_latest (
-    entity_type varchar(255) NOT NULL,
+    entity_id uuid NOT NULL,
-    entity_id varchar(31) NOT NULL,
+    key int NOT NULL,
-    key varchar(255) NOT NULL,
     ts bigint NOT NULL,
     bool_v boolean,
     str_v varchar(10000000),
     long_v bigint,
     dbl_v double precision,
-    CONSTRAINT ts_kv_latest_pkey PRIMARY KEY (entity_type, entity_id, key)
+    CONSTRAINT ts_kv_latest_pkey PRIMARY KEY (entity_id, key)
+);
+
+CREATE TABLE IF NOT EXISTS ts_kv_dictionary (
+    key varchar(255) NOT NULL,
+    key_id serial UNIQUE,
+    CONSTRAINT ts_key_id_pkey PRIMARY KEY (key)
 );

k8s/README.md

@@ -9,6 +9,15 @@ You need to have a Kubernetes cluster, and the kubectl command-line tool must be
 If you do not already have a cluster, you can create one by using [Minikube](https://kubernetes.io/docs/setup/minikube), 
 or you can choose any other available [Kubernetes cluster deployment solutions](https://kubernetes.io/docs/setup/pick-right-solution/).
 
+### Enable ingress addon
+
+By default ingress addon is disable in the Minikube, and available only in cluster providers.
+To enable ingress, please execute next command:
+
+`
+$ minikube addons enable ingress
+` 
+
 ## Installation
 
 Before performing initial installation you can configure the type of database to be used with ThingsBoard.