Browse Source
* init commit * cleaned code and add test-properties * cleaned code * psql-update * timescale-update * code-refactoring * fix typo * renamed dao * revert indents * refactored code * fix typo * init-partitioning * code updated * cleaned code * fixed license * fix typo * fixed code after review * add annotation to repository * update psql version for docker * postgres-10 * postgres-10 * update docker compose config * fixed partition saving * change key_id to serial column definition * upgrade psql added * add separate upgrade service * added upgrade script * change image on k8s * change logs * resolve conflict after merge with master * revert datasource url in yml * fix typo * license header fix * remove old methods for the timeseries inserts * clean up code * fix saveOrUpdate for PostgreSQL * refactoring & revert Timescale to use latest table * added PsqlTsAnyDao * duplicated code method removed * remove unused invert dictionary map * change the upgrade directory from 2.4.1 to 2.4.3 * refactor JpaPsqlTimeseriesDaopull/2380/head
ShvaykaD
7 years ago
committed by
Andrew Shvayka
Andrew Shvayka
75 changed files with 2669 additions and 1417 deletions
@ -0,0 +1,179 @@ |
|||
-- |
|||
-- 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. |
|||
-- |
|||
|
|||
-- load function 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 < 100000 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 10!'; |
|||
ELSE |
|||
RAISE NOTICE 'PostgreSQL version is valid!'; |
|||
RAISE NOTICE 'Schema update started...'; |
|||
END IF; |
|||
RETURN valid_version; |
|||
END; |
|||
$$ LANGUAGE 'plpgsql'; |
|||
|
|||
-- load function create_partition_table() |
|||
|
|||
CREATE OR REPLACE FUNCTION create_partition_table() RETURNS VOID AS $$ |
|||
|
|||
BEGIN |
|||
ALTER TABLE ts_kv |
|||
RENAME TO ts_kv_old; |
|||
CREATE TABLE IF NOT EXISTS ts_kv |
|||
( |
|||
LIKE ts_kv_old |
|||
) |
|||
PARTITION BY RANGE (ts); |
|||
ALTER TABLE ts_kv |
|||
DROP COLUMN entity_type; |
|||
ALTER TABLE ts_kv |
|||
ALTER COLUMN entity_id TYPE uuid USING entity_id::uuid; |
|||
ALTER TABLE ts_kv |
|||
ALTER COLUMN key TYPE integer USING key::integer; |
|||
END; |
|||
$$ LANGUAGE 'plpgsql'; |
|||
|
|||
|
|||
-- load function create_partitions() |
|||
|
|||
CREATE OR REPLACE FUNCTION create_partitions() RETURNS VOID AS |
|||
$$ |
|||
DECLARE |
|||
partition_date varchar; |
|||
from_ts bigint; |
|||
to_ts bigint; |
|||
key_cursor CURSOR FOR select SUBSTRING(month_date.first_date, 1, 7) AS partition_date, |
|||
extract(epoch from (month_date.first_date)::timestamp) * 1000 as from_ts, |
|||
extract(epoch from (month_date.first_date::date + INTERVAL '1 MONTH')::timestamp) * |
|||
1000 as to_ts |
|||
FROM (SELECT DISTINCT TO_CHAR(TO_TIMESTAMP(ts / 1000), 'YYYY_MM_01') AS first_date |
|||
FROM ts_kv_old) AS month_date; |
|||
BEGIN |
|||
OPEN key_cursor; |
|||
LOOP |
|||
FETCH key_cursor INTO partition_date, from_ts, to_ts; |
|||
EXIT WHEN NOT FOUND; |
|||
EXECUTE 'CREATE TABLE IF NOT EXISTS ts_kv_' || partition_date || |
|||
' PARTITION OF ts_kv(PRIMARY KEY (entity_id, key, ts)) FOR VALUES FROM (' || from_ts || |
|||
') TO (' || to_ts || ');'; |
|||
RAISE NOTICE 'A partition % has been created!',CONCAT('ts_kv_', partition_date); |
|||
END LOOP; |
|||
|
|||
CLOSE key_cursor; |
|||
END; |
|||
$$ language 'plpgsql'; |
|||
|
|||
-- load function 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'; |
|||
|
|||
-- load function 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 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'; |
|||
|
|||
-- load function insert_into_ts_kv() |
|||
|
|||
CREATE OR REPLACE FUNCTION insert_into_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(first, '-', second, '-1', third, '-', fourth, '-', fifth)::uuid AS entity_id, |
|||
substrings.key AS key, |
|||
substrings.ts AS ts, |
|||
substrings.bool_v AS bool_v, |
|||
substrings.str_v AS str_v, |
|||
substrings.long_v AS long_v, |
|||
substrings.dbl_v AS dbl_v |
|||
FROM (SELECT SUBSTRING(entity_id, 8, 8) AS first, |
|||
SUBSTRING(entity_id, 4, 4) AS second, |
|||
SUBSTRING(entity_id, 1, 3) AS third, |
|||
SUBSTRING(entity_id, 16, 4) AS fourth, |
|||
SUBSTRING(entity_id, 20) AS fifth, |
|||
key_id AS key, |
|||
ts, |
|||
bool_v, |
|||
str_v, |
|||
long_v, |
|||
dbl_v |
|||
FROM ts_kv_old |
|||
INNER JOIN ts_kv_dictionary ON (ts_kv_old.key = ts_kv_dictionary.key)) AS substrings; |
|||
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 partitioned ts_kv!',insert_counter - 1; |
|||
EXIT; |
|||
END IF; |
|||
INSERT INTO ts_kv(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 partitioned ts_kv!',insert_counter; |
|||
END IF; |
|||
END LOOP; |
|||
CLOSE insert_cursor; |
|||
END; |
|||
$$ LANGUAGE 'plpgsql'; |
|||
|
|||
|
|||
@ -0,0 +1,22 @@ |
|||
/** |
|||
* 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; |
|||
|
|||
public interface DatabaseEntitiesUpgradeService { |
|||
|
|||
void upgradeDatabase(String fromVersion) throws Exception; |
|||
|
|||
} |
|||
@ -0,0 +1,32 @@ |
|||
/** |
|||
* 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 org.springframework.context.annotation.Profile; |
|||
import org.springframework.stereotype.Service; |
|||
import org.thingsboard.server.dao.util.PsqlDao; |
|||
import org.thingsboard.server.dao.util.SqlTsDao; |
|||
|
|||
@Service |
|||
@SqlTsDao |
|||
@PsqlDao |
|||
@Profile("install") |
|||
public class PsqlTsDatabaseSchemaService extends SqlAbstractDatabaseSchemaService |
|||
implements TsDatabaseSchemaService { |
|||
public PsqlTsDatabaseSchemaService() { |
|||
super("schema-ts-psql.sql", null); |
|||
} |
|||
} |
|||
@ -0,0 +1,145 @@ |
|||
/** |
|||
* 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.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 { |
|||
|
|||
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 String dbUrl; |
|||
|
|||
@Value("${spring.datasource.username}") |
|||
private String dbUserName; |
|||
|
|||
@Value("${spring.datasource.password}") |
|||
private String dbPassword; |
|||
|
|||
@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 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!"); |
|||
log.info("Please upgrade your PostgreSQL and restart the script!"); |
|||
} else { |
|||
log.info("PostgreSQL version is valid!"); |
|||
log.info("Updating schema ..."); |
|||
executeFunction(conn, CREATE_PARTITION_TABLE); |
|||
executeFunction(conn, CREATE_PARTITIONS); |
|||
executeFunction(conn, CREATE_TS_KV_DICTIONARY_TABLE); |
|||
executeFunction(conn, INSERT_INTO_DICTIONARY); |
|||
executeFunction(conn, INSERT_INTO_TS_KV); |
|||
dropOldTable(conn, DROP_OLD_TABLE); |
|||
log.info("schema timeseries updated!"); |
|||
} |
|||
} |
|||
break; |
|||
default: |
|||
throw new RuntimeException("Unable to upgrade SQL database, unsupported fromVersion: " + fromVersion); |
|||
} |
|||
} |
|||
|
|||
private void loadSql(Connection conn) { |
|||
Path schemaUpdateFile = Paths.get(installScripts.getDataDir(), "upgrade", "2.4.3", LOAD_FUNCTIONS_SQL); |
|||
try { |
|||
loadFunctions(schemaUpdateFile, conn); |
|||
} 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()); |
|||
} |
|||
} |
|||
} |
|||
@ -0,0 +1,23 @@ |
|||
/** |
|||
* 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.util; |
|||
|
|||
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression; |
|||
|
|||
@ConditionalOnExpression("('${database.ts.type}'=='sql' || '${database.entities.type}'=='timescale') " + |
|||
"&& '${spring.jpa.database-platform}'=='org.hibernate.dialect.PostgreSQLDialect'") |
|||
public @interface PsqlTsAnyDao { |
|||
} |
|||
@ -0,0 +1,37 @@ |
|||
/** |
|||
* 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; |
|||
|
|||
import org.springframework.boot.autoconfigure.EnableAutoConfiguration; |
|||
import org.springframework.boot.autoconfigure.domain.EntityScan; |
|||
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.SqlTsDao; |
|||
|
|||
@Configuration |
|||
@EnableAutoConfiguration |
|||
@ComponentScan({"org.thingsboard.server.dao.sqlts.psql", "org.thingsboard.server.dao.sqlts.latest"}) |
|||
@EnableJpaRepositories({"org.thingsboard.server.dao.sqlts.psql", "org.thingsboard.server.dao.sqlts.latest", "org.thingsboard.server.dao.sqlts.dictionary"}) |
|||
@EntityScan({"org.thingsboard.server.dao.model.sqlts.psql", "org.thingsboard.server.dao.model.sqlts.latest", "org.thingsboard.server.dao.model.sqlts.dictionary"}) |
|||
@EnableTransactionManagement |
|||
@SqlTsDao |
|||
@PsqlDao |
|||
public class PsqlTsDaoConfig { |
|||
|
|||
} |
|||
@ -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.model.sqlts.dictionary; |
|||
|
|||
import lombok.Data; |
|||
import org.hibernate.annotations.Generated; |
|||
import org.hibernate.annotations.GenerationTime; |
|||
|
|||
import javax.persistence.Column; |
|||
import javax.persistence.Entity; |
|||
import javax.persistence.Id; |
|||
import javax.persistence.IdClass; |
|||
import javax.persistence.Table; |
|||
|
|||
import static org.thingsboard.server.dao.model.ModelConstants.KEY_COLUMN; |
|||
import static org.thingsboard.server.dao.model.ModelConstants.KEY_ID_COLUMN; |
|||
|
|||
@Data |
|||
@Entity |
|||
@Table(name = "ts_kv_dictionary") |
|||
@IdClass(TsKvDictionaryCompositeKey.class) |
|||
public final class TsKvDictionary { |
|||
|
|||
@Id |
|||
@Column(name = KEY_COLUMN) |
|||
private String key; |
|||
|
|||
@Column(name = KEY_ID_COLUMN, unique = true, columnDefinition="serial") |
|||
@Generated(GenerationTime.INSERT) |
|||
private int keyId; |
|||
|
|||
} |
|||
@ -0,0 +1,34 @@ |
|||
/** |
|||
* 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.model.sqlts.dictionary; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import javax.persistence.Transient; |
|||
import java.io.Serializable; |
|||
|
|||
@Data |
|||
@NoArgsConstructor |
|||
@AllArgsConstructor |
|||
public class TsKvDictionaryCompositeKey implements Serializable{ |
|||
|
|||
@Transient |
|||
private static final long serialVersionUID = -4089175869616037523L; |
|||
|
|||
private String key; |
|||
} |
|||
@ -0,0 +1,37 @@ |
|||
/** |
|||
* 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.model.sqlts.psql; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import lombok.NoArgsConstructor; |
|||
|
|||
import javax.persistence.Transient; |
|||
import java.io.Serializable; |
|||
import java.util.UUID; |
|||
|
|||
@Data |
|||
@AllArgsConstructor |
|||
@NoArgsConstructor |
|||
public class TsKvCompositeKey implements Serializable { |
|||
|
|||
@Transient |
|||
private static final long serialVersionUID = -4089175869616037523L; |
|||
|
|||
private UUID entityId; |
|||
private int key; |
|||
private long ts; |
|||
} |
|||
@ -0,0 +1,135 @@ |
|||
/** |
|||
* 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.model.sqlts.psql; |
|||
|
|||
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 org.thingsboard.server.dao.model.sql.AbstractTsKvEntity; |
|||
|
|||
import javax.persistence.Column; |
|||
import javax.persistence.Entity; |
|||
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.KEY_COLUMN; |
|||
|
|||
@Data |
|||
@Entity |
|||
@Table(name = "ts_kv") |
|||
@IdClass(TsKvCompositeKey.class) |
|||
public final class TsKvEntity extends AbstractTsKvEntity implements ToData<TsKvEntry> { |
|||
|
|||
@Id |
|||
@Column(name = ENTITY_ID_COLUMN, columnDefinition = "uuid") |
|||
protected UUID entityId; |
|||
|
|||
@Id |
|||
@Column(name = KEY_COLUMN) |
|||
protected int key; |
|||
|
|||
@Transient |
|||
protected String strKey; |
|||
|
|||
public TsKvEntity() { |
|||
} |
|||
|
|||
public TsKvEntity(String strValue) { |
|||
this.strValue = strValue; |
|||
} |
|||
|
|||
public TsKvEntity(Long longValue, Double doubleValue, Long longCountValue, Long doubleCountValue, String aggType) { |
|||
if (!isAllNull(longValue, doubleValue, longCountValue, doubleCountValue)) { |
|||
switch (aggType) { |
|||
case AVG: |
|||
double sum = 0.0; |
|||
if (longValue != null) { |
|||
sum += longValue; |
|||
} |
|||
if (doubleValue != null) { |
|||
sum += doubleValue; |
|||
} |
|||
long totalCount = longCountValue + doubleCountValue; |
|||
if (totalCount > 0) { |
|||
this.doubleValue = sum / (longCountValue + doubleCountValue); |
|||
} else { |
|||
this.doubleValue = 0.0; |
|||
} |
|||
break; |
|||
case SUM: |
|||
if (doubleCountValue > 0) { |
|||
this.doubleValue = doubleValue + (longValue != null ? longValue.doubleValue() : 0.0); |
|||
} else { |
|||
this.longValue = longValue; |
|||
} |
|||
break; |
|||
case MIN: |
|||
case MAX: |
|||
if (longCountValue > 0 && doubleCountValue > 0) { |
|||
this.doubleValue = MAX.equals(aggType) ? Math.max(doubleValue, longValue.doubleValue()) : Math.min(doubleValue, longValue.doubleValue()); |
|||
} else if (doubleCountValue > 0) { |
|||
this.doubleValue = doubleValue; |
|||
} else if (longCountValue > 0) { |
|||
this.longValue = longValue; |
|||
} |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
|
|||
public TsKvEntity(Long booleanValueCount, Long strValueCount, Long longValueCount, Long doubleValueCount) { |
|||
if (!isAllNull(booleanValueCount, strValueCount, longValueCount, doubleValueCount)) { |
|||
if (booleanValueCount != 0) { |
|||
this.longValue = booleanValueCount; |
|||
} else if (strValueCount != 0) { |
|||
this.longValue = strValueCount; |
|||
} else { |
|||
this.longValue = longValueCount + doubleValueCount; |
|||
} |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
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); |
|||
} |
|||
|
|||
} |
|||
@ -1,58 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.dao.sqlts; |
|||
|
|||
import org.springframework.data.jpa.repository.Modifying; |
|||
import org.springframework.stereotype.Repository; |
|||
import org.thingsboard.server.dao.model.sqlts.ts.TsKvLatestEntity; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Repository |
|||
public abstract class AbstractLatestInsertRepository extends AbstractInsertRepository { |
|||
|
|||
public abstract void saveOrUpdate(TsKvLatestEntity entity); |
|||
|
|||
public abstract void saveOrUpdate(List<TsKvLatestEntity> entities); |
|||
|
|||
protected void processSaveOrUpdate(TsKvLatestEntity entity, String requestBoolValue, String requestStrValue, String requestLongValue, String requestDblValue) { |
|||
if (entity.getBooleanValue() != null) { |
|||
saveOrUpdateBoolean(entity, requestBoolValue); |
|||
} |
|||
if (entity.getStrValue() != null) { |
|||
saveOrUpdateString(entity, requestStrValue); |
|||
} |
|||
if (entity.getLongValue() != null) { |
|||
saveOrUpdateLong(entity, requestLongValue); |
|||
} |
|||
if (entity.getDoubleValue() != null) { |
|||
saveOrUpdateDouble(entity, requestDblValue); |
|||
} |
|||
} |
|||
|
|||
@Modifying |
|||
protected abstract void saveOrUpdateBoolean(TsKvLatestEntity entity, String query); |
|||
|
|||
@Modifying |
|||
protected abstract void saveOrUpdateString(TsKvLatestEntity entity, String query); |
|||
|
|||
@Modifying |
|||
protected abstract void saveOrUpdateLong(TsKvLatestEntity entity, String query); |
|||
|
|||
@Modifying |
|||
protected abstract void saveOrUpdateDouble(TsKvLatestEntity entity, String query); |
|||
|
|||
} |
|||
@ -0,0 +1,156 @@ |
|||
/** |
|||
* 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; |
|||
|
|||
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; |
|||
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 { |
|||
|
|||
@Autowired |
|||
private 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; |
|||
|
|||
@PostConstruct |
|||
protected void init() { |
|||
super.init(); |
|||
TbSqlBlockingQueueParams tsParams = TbSqlBlockingQueueParams.builder() |
|||
.logName("TS") |
|||
.batchSize(tsBatchSize) |
|||
.maxDelay(tsMaxDelay) |
|||
.statsPrintIntervalMs(tsStatsPrintIntervalMs) |
|||
.build(); |
|||
tsQueue = new TbSqlBlockingQueue<>(tsParams); |
|||
tsQueue.init(logExecutor, v -> insertRepository.saveOrUpdate(v)); |
|||
} |
|||
|
|||
@PreDestroy |
|||
protected void destroy() { |
|||
super.init(); |
|||
if (tsQueue != null) { |
|||
tsQueue.destroy(); |
|||
} |
|||
} |
|||
|
|||
protected ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, ReadTsKvQuery query) { |
|||
if (query.getAggregation() == Aggregation.NONE) { |
|||
return findAllAsyncWithLimit(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(entityId, query.getKey(), startTs, endTs, ts, query.getAggregation())); |
|||
stepTs = endTs; |
|||
} |
|||
return getTskvEntriesFuture(Futures.allAsList(futures)); |
|||
} |
|||
} |
|||
|
|||
protected abstract ListenableFuture<Optional<TsKvEntry>> findAndAggregateAsync(EntityId entityId, String key, long startTs, long endTs, long ts, Aggregation aggregation); |
|||
|
|||
protected abstract ListenableFuture<List<TsKvEntry>> findAllAsyncWithLimit(EntityId entityId, ReadTsKvQuery query); |
|||
|
|||
protected SettableFuture<T> setFutures(List<CompletableFuture<T>> entitiesFutures) { |
|||
SettableFuture<T> listenableFuture = SettableFuture.create(); |
|||
CompletableFuture<List<T>> entities = |
|||
CompletableFuture.allOf(entitiesFutures.toArray(new CompletableFuture[entitiesFutures.size()])) |
|||
.thenApply(v -> entitiesFutures.stream() |
|||
.map(CompletableFuture::join) |
|||
.collect(Collectors.toList())); |
|||
|
|||
entities.whenComplete((tsKvEntities, throwable) -> { |
|||
if (throwable != null) { |
|||
listenableFuture.setException(throwable); |
|||
} else { |
|||
T result = null; |
|||
for (T entity : tsKvEntities) { |
|||
if (entity.isNotEmpty()) { |
|||
result = entity; |
|||
break; |
|||
} |
|||
} |
|||
listenableFuture.set(result); |
|||
} |
|||
}); |
|||
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); |
|||
} |
|||
@ -1,58 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.dao.sqlts; |
|||
|
|||
import org.springframework.data.jpa.repository.Modifying; |
|||
import org.springframework.stereotype.Repository; |
|||
import org.thingsboard.server.dao.model.sql.AbstractTsKvEntity; |
|||
|
|||
import java.util.List; |
|||
|
|||
@Repository |
|||
public abstract class AbstractTimeseriesInsertRepository<T extends AbstractTsKvEntity> extends AbstractInsertRepository { |
|||
|
|||
public abstract void saveOrUpdate(T entity); |
|||
|
|||
public abstract void saveOrUpdate(List<T> entities); |
|||
|
|||
protected void processSaveOrUpdate(T entity, String requestBoolValue, String requestStrValue, String requestLongValue, String requestDblValue) { |
|||
if (entity.getBooleanValue() != null) { |
|||
saveOrUpdateBoolean(entity, requestBoolValue); |
|||
} |
|||
if (entity.getStrValue() != null) { |
|||
saveOrUpdateString(entity, requestStrValue); |
|||
} |
|||
if (entity.getLongValue() != null) { |
|||
saveOrUpdateLong(entity, requestLongValue); |
|||
} |
|||
if (entity.getDoubleValue() != null) { |
|||
saveOrUpdateDouble(entity, requestDblValue); |
|||
} |
|||
} |
|||
|
|||
@Modifying |
|||
protected abstract void saveOrUpdateBoolean(T entity, String query); |
|||
|
|||
@Modifying |
|||
protected abstract void saveOrUpdateString(T entity, String query); |
|||
|
|||
@Modifying |
|||
protected abstract void saveOrUpdateLong(T entity, String query); |
|||
|
|||
@Modifying |
|||
protected abstract void saveOrUpdateDouble(T entity, String query); |
|||
|
|||
} |
|||
@ -0,0 +1,29 @@ |
|||
/** |
|||
* 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; |
|||
|
|||
import lombok.AllArgsConstructor; |
|||
import lombok.Data; |
|||
import org.thingsboard.server.dao.model.sql.AbstractTsKvEntity; |
|||
|
|||
@Data |
|||
@AllArgsConstructor |
|||
public class EntityContainer<T extends AbstractTsKvEntity> { |
|||
|
|||
private T entity; |
|||
private String partitionDate; |
|||
|
|||
} |
|||
@ -0,0 +1,26 @@ |
|||
/** |
|||
* 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; |
|||
|
|||
import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity; |
|||
|
|||
import java.util.List; |
|||
|
|||
public interface InsertLatestRepository { |
|||
|
|||
void saveOrUpdate(List<TsKvLatestEntity> entities); |
|||
|
|||
} |
|||
@ -0,0 +1,26 @@ |
|||
/** |
|||
* 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; |
|||
|
|||
import org.thingsboard.server.dao.model.sql.AbstractTsKvEntity; |
|||
|
|||
import java.util.List; |
|||
|
|||
public interface InsertTsRepository<T extends AbstractTsKvEntity> { |
|||
|
|||
void saveOrUpdate(List<EntityContainer<T>> entities); |
|||
|
|||
} |
|||
@ -0,0 +1,30 @@ |
|||
/** |
|||
* 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.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 java.util.Optional; |
|||
|
|||
@PsqlDao |
|||
public interface TsKvDictionaryRepository extends CrudRepository<TsKvDictionary, TsKvDictionaryCompositeKey> { |
|||
|
|||
Optional<TsKvDictionary> findByKeyId(int keyId); |
|||
|
|||
} |
|||
@ -0,0 +1,89 @@ |
|||
/** |
|||
* 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.hsql; |
|||
|
|||
import org.springframework.jdbc.core.BatchPreparedStatementSetter; |
|||
import org.springframework.stereotype.Repository; |
|||
import org.springframework.transaction.annotation.Transactional; |
|||
import org.thingsboard.server.dao.model.sqlts.hsql.TsKvEntity; |
|||
import org.thingsboard.server.dao.sqlts.AbstractInsertRepository; |
|||
import org.thingsboard.server.dao.sqlts.EntityContainer; |
|||
import org.thingsboard.server.dao.sqlts.InsertTsRepository; |
|||
import org.thingsboard.server.dao.util.HsqlDao; |
|||
import org.thingsboard.server.dao.util.SqlTsDao; |
|||
|
|||
import java.sql.PreparedStatement; |
|||
import java.sql.SQLException; |
|||
import java.sql.Types; |
|||
import java.util.List; |
|||
|
|||
@SqlTsDao |
|||
@HsqlDao |
|||
@Repository |
|||
@Transactional |
|||
public class HsqlTimeseriesInsertRepository extends AbstractInsertRepository implements InsertTsRepository<TsKvEntity> { |
|||
|
|||
private static final String INSERT_OR_UPDATE = |
|||
"MERGE INTO ts_kv USING(VALUES ?, ?, ?, ?, ?, ?, ?, ?) " + |
|||
"T (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " + |
|||
"ON (ts_kv.entity_type=T.entity_type " + |
|||
"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) " + |
|||
"VALUES (T.entity_type, 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) { |
|||
jdbcTemplate.batchUpdate(INSERT_OR_UPDATE, new BatchPreparedStatementSetter() { |
|||
@Override |
|||
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.setString(2, tsKvEntity.getEntityId()); |
|||
ps.setString(3, tsKvEntity.getKey()); |
|||
ps.setLong(4, tsKvEntity.getTs()); |
|||
|
|||
if (tsKvEntity.getBooleanValue() != null) { |
|||
ps.setBoolean(5, tsKvEntity.getBooleanValue()); |
|||
} else { |
|||
ps.setNull(5, Types.BOOLEAN); |
|||
} |
|||
|
|||
ps.setString(6, tsKvEntity.getStrValue()); |
|||
|
|||
if (tsKvEntity.getLongValue() != null) { |
|||
ps.setLong(7, tsKvEntity.getLongValue()); |
|||
} else { |
|||
ps.setNull(7, Types.BIGINT); |
|||
} |
|||
|
|||
if (tsKvEntity.getDoubleValue() != null) { |
|||
ps.setDouble(8, tsKvEntity.getDoubleValue()); |
|||
} else { |
|||
ps.setNull(8, Types.DOUBLE); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public int getBatchSize() { |
|||
return entities.size(); |
|||
} |
|||
}); |
|||
} |
|||
} |
|||
@ -0,0 +1,206 @@ |
|||
/** |
|||
* 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.hsql; |
|||
|
|||
import com.google.common.util.concurrent.Futures; |
|||
import com.google.common.util.concurrent.ListenableFuture; |
|||
import lombok.extern.slf4j.Slf4j; |
|||
import org.springframework.beans.factory.annotation.Autowired; |
|||
import org.springframework.data.domain.PageRequest; |
|||
import org.springframework.data.domain.Sort; |
|||
import org.springframework.stereotype.Component; |
|||
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.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.hsql.TsKvEntity; |
|||
import org.thingsboard.server.dao.sqlts.AbstractSimpleSqlTimeseriesDao; |
|||
import org.thingsboard.server.dao.sqlts.EntityContainer; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesDao; |
|||
import org.thingsboard.server.dao.util.HsqlDao; |
|||
import org.thingsboard.server.dao.util.SqlTsDao; |
|||
|
|||
import java.util.ArrayList; |
|||
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 { |
|||
|
|||
@Autowired |
|||
private TsKvHsqlRepository tsKvRepository; |
|||
|
|||
@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) { |
|||
TsKvEntity entity = new TsKvEntity(); |
|||
entity.setEntityType(entityId.getEntityType()); |
|||
entity.setEntityId(fromTimeUUID(entityId.getId())); |
|||
entity.setTs(tsKvEntry.getTs()); |
|||
entity.setKey(tsKvEntry.getKey()); |
|||
entity.setStrValue(tsKvEntry.getStrValue().orElse(null)); |
|||
entity.setDoubleValue(tsKvEntry.getDoubleValue().orElse(null)); |
|||
entity.setLongValue(tsKvEntry.getLongValue().orElse(null)); |
|||
entity.setBooleanValue(tsKvEntry.getBooleanValue().orElse(null)); |
|||
log.trace("Saving entity: {}", entity); |
|||
return tsQueue.add(new EntityContainer(entity, null)); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> remove(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) { |
|||
return service.submit(() -> { |
|||
tsKvRepository.delete( |
|||
fromTimeUUID(entityId.getId()), |
|||
entityId.getEntityType(), |
|||
query.getKey(), |
|||
query.getStartTs(), |
|||
query.getEndTs()); |
|||
return null; |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> saveLatest(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry) { |
|||
return getSaveLatestFuture(entityId, tsKvEntry); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> removeLatest(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) { |
|||
return getRemoveLatestFuture(tenantId, entityId, query); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<TsKvEntry> findLatest(TenantId tenantId, EntityId entityId, String key) { |
|||
return getFindLatestFuture(entityId, key); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<List<TsKvEntry>> findAllLatest(TenantId tenantId, EntityId entityId) { |
|||
return getFindAllLatestFuture(entityId); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> savePartition(TenantId tenantId, EntityId entityId, long tsKvEntryTs, String key, long ttl) { |
|||
return Futures.immediateFuture(null); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> removePartition(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) { |
|||
return Futures.immediateFuture(null); |
|||
} |
|||
|
|||
protected ListenableFuture<Optional<TsKvEntry>> findAndAggregateAsync(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); |
|||
return Futures.transform(setFutures(entitiesFutures), entity -> { |
|||
if (entity != null && entity.isNotEmpty()) { |
|||
entity.setEntityId(fromTimeUUID(entityId.getId())); |
|||
entity.setEntityType(entityId.getEntityType()); |
|||
entity.setKey(key); |
|||
entity.setTs(ts); |
|||
return Optional.of(DaoUtil.getData(entity)); |
|||
} else { |
|||
return Optional.empty(); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
protected ListenableFuture<List<TsKvEntry>> findAllAsyncWithLimit(EntityId entityId, ReadTsKvQuery query) { |
|||
return Futures.immediateFuture( |
|||
DaoUtil.convertDataList( |
|||
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.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
protected void findSum(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) { |
|||
entitiesFutures.add(tsKvRepository.findSum( |
|||
fromTimeUUID(entityId.getId()), |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
protected void findMin(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) { |
|||
entitiesFutures.add(tsKvRepository.findStringMin( |
|||
fromTimeUUID(entityId.getId()), |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
entitiesFutures.add(tsKvRepository.findNumericMin( |
|||
fromTimeUUID(entityId.getId()), |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
protected void findMax(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) { |
|||
entitiesFutures.add(tsKvRepository.findStringMax( |
|||
fromTimeUUID(entityId.getId()), |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
entitiesFutures.add(tsKvRepository.findNumericMax( |
|||
fromTimeUUID(entityId.getId()), |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
protected void findAvg(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) { |
|||
entitiesFutures.add(tsKvRepository.findAvg( |
|||
fromTimeUUID(entityId.getId()), |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,85 @@ |
|||
/** |
|||
* 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.jdbc.core.BatchPreparedStatementSetter; |
|||
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.util.HsqlDao; |
|||
import org.thingsboard.server.dao.util.SqlTsDao; |
|||
|
|||
import java.sql.PreparedStatement; |
|||
import java.sql.SQLException; |
|||
import java.sql.Types; |
|||
import java.util.List; |
|||
|
|||
@SqlTsDao |
|||
@HsqlDao |
|||
@Repository |
|||
@Transactional |
|||
public class HsqlLatestInsertRepository extends AbstractInsertRepository implements InsertLatestRepository { |
|||
|
|||
private static final String INSERT_OR_UPDATE = |
|||
"MERGE INTO ts_kv_latest USING(VALUES ?, ?, ?, ?, ?, ?, ?, ?) " + |
|||
"T (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " + |
|||
"ON (ts_kv_latest.entity_type=T.entity_type " + |
|||
"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) " + |
|||
"VALUES (T.entity_type, 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.setString(2, entities.get(i).getEntityId()); |
|||
ps.setString(3, entities.get(i).getKey()); |
|||
ps.setLong(4, entities.get(i).getTs()); |
|||
|
|||
if (entities.get(i).getBooleanValue() != null) { |
|||
ps.setBoolean(5, entities.get(i).getBooleanValue()); |
|||
} else { |
|||
ps.setNull(5, Types.BOOLEAN); |
|||
} |
|||
|
|||
ps.setString(6, entities.get(i).getStrValue()); |
|||
|
|||
if (entities.get(i).getLongValue() != null) { |
|||
ps.setLong(7, entities.get(i).getLongValue()); |
|||
} else { |
|||
ps.setNull(7, Types.BIGINT); |
|||
} |
|||
|
|||
if (entities.get(i).getDoubleValue() != null) { |
|||
ps.setDouble(8, entities.get(i).getDoubleValue()); |
|||
} else { |
|||
ps.setNull(8, Types.DOUBLE); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public int getBatchSize() { |
|||
return entities.size(); |
|||
} |
|||
}); |
|||
} |
|||
} |
|||
@ -0,0 +1,312 @@ |
|||
/** |
|||
* 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.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; |
|||
import org.springframework.data.domain.Sort; |
|||
import org.springframework.stereotype.Component; |
|||
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.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.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; |
|||
import org.thingsboard.server.dao.util.PsqlDao; |
|||
import org.thingsboard.server.dao.util.SqlTsDao; |
|||
|
|||
import java.time.Instant; |
|||
import java.time.LocalDateTime; |
|||
import java.time.ZoneOffset; |
|||
import java.util.ArrayList; |
|||
import java.util.List; |
|||
import java.util.Optional; |
|||
import java.util.Set; |
|||
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; |
|||
|
|||
|
|||
@Component |
|||
@Slf4j |
|||
@SqlTsDao |
|||
@PsqlDao |
|||
public class JpaPsqlTimeseriesDao extends AbstractSimpleSqlTimeseriesDao<TsKvEntity> implements TimeseriesDao { |
|||
|
|||
private final ConcurrentMap<String, Integer> tsKvDictionaryMap = new ConcurrentHashMap<>(); |
|||
private final Set<PsqlPartition> partitions = ConcurrentHashMap.newKeySet(); |
|||
|
|||
private static final ReentrantLock tsCreationLock = new ReentrantLock(); |
|||
private static final ReentrantLock partitionCreationLock = new ReentrantLock(); |
|||
|
|||
@Autowired |
|||
private TsKvDictionaryRepository dictionaryRepository; |
|||
|
|||
@Autowired |
|||
private TsKvPsqlRepository tsKvRepository; |
|||
|
|||
@Autowired |
|||
private PsqlPartitioningRepository partitioningRepository; |
|||
|
|||
private SqlTsPartitionDate tsFormat; |
|||
|
|||
@Value("${sql.ts_key_value_partitioning}") |
|||
private String partitioning; |
|||
|
|||
@Override |
|||
protected void init() { |
|||
super.init(); |
|||
Optional<SqlTsPartitionDate> partition = SqlTsPartitionDate.parse(partitioning); |
|||
if (partition.isPresent()) { |
|||
tsFormat = partition.get(); |
|||
} else { |
|||
log.warn("Incorrect configuration of partitioning {}", partitioning); |
|||
throw new RuntimeException("Failed to parse partitioning property: " + partitioning + "!"); |
|||
} |
|||
} |
|||
|
|||
@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(); |
|||
Integer keyId = getOrSaveKeyId(strKey); |
|||
TsKvEntity entity = new TsKvEntity(); |
|||
entity.setEntityId(entityId.getId()); |
|||
entity.setTs(tsKvEntry.getTs()); |
|||
entity.setKey(keyId); |
|||
entity.setStrValue(tsKvEntry.getStrValue().orElse(null)); |
|||
entity.setDoubleValue(tsKvEntry.getDoubleValue().orElse(null)); |
|||
entity.setLongValue(tsKvEntry.getLongValue().orElse(null)); |
|||
entity.setBooleanValue(tsKvEntry.getBooleanValue().orElse(null)); |
|||
PsqlPartition psqlPartition = toPartition(tsKvEntry.getTs()); |
|||
savePartition(psqlPartition); |
|||
log.trace("Saving entity: {}", entity); |
|||
return tsQueue.add(new EntityContainer(entity, psqlPartition.getPartitionDate())); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> remove(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) { |
|||
return service.submit(() -> { |
|||
String strKey = query.getKey(); |
|||
Integer keyId = getOrSaveKeyId(strKey); |
|||
tsKvRepository.delete( |
|||
entityId.getId(), |
|||
keyId, |
|||
query.getStartTs(), |
|||
query.getEndTs()); |
|||
return null; |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> removeLatest(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) { |
|||
return getRemoveLatestFuture(tenantId, entityId, query); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> saveLatest(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry) { |
|||
return getSaveLatestFuture(entityId, tsKvEntry); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<TsKvEntry> findLatest(TenantId tenantId, EntityId entityId, String key) { |
|||
return getFindLatestFuture(entityId, key); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<List<TsKvEntry>> findAllLatest(TenantId tenantId, EntityId entityId) { |
|||
return getFindAllLatestFuture(entityId); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> savePartition(TenantId tenantId, EntityId entityId, long tsKvEntryTs, String key, long ttl) { |
|||
return Futures.immediateFuture(null); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> removePartition(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) { |
|||
return Futures.immediateFuture(null); |
|||
} |
|||
|
|||
protected ListenableFuture<Optional<TsKvEntry>> findAndAggregateAsync(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); |
|||
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(); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
protected ListenableFuture<List<TsKvEntry>> findAllAsyncWithLimit(EntityId entityId, ReadTsKvQuery query) { |
|||
Integer keyId = getOrSaveKeyId(query.getKey()); |
|||
List<TsKvEntity> tsKvEntities = tsKvRepository.findAllWithLimit( |
|||
entityId.getId(), |
|||
keyId, |
|||
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)); |
|||
} |
|||
|
|||
protected void findCount(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) { |
|||
Integer keyId = getOrSaveKeyId(key); |
|||
entitiesFutures.add(tsKvRepository.findCount( |
|||
entityId.getId(), |
|||
keyId, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
protected void findSum(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) { |
|||
Integer keyId = getOrSaveKeyId(key); |
|||
entitiesFutures.add(tsKvRepository.findSum( |
|||
entityId.getId(), |
|||
keyId, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
protected void findMin(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) { |
|||
Integer keyId = getOrSaveKeyId(key); |
|||
entitiesFutures.add(tsKvRepository.findStringMin( |
|||
entityId.getId(), |
|||
keyId, |
|||
startTs, |
|||
endTs)); |
|||
entitiesFutures.add(tsKvRepository.findNumericMin( |
|||
entityId.getId(), |
|||
keyId, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
protected void findMax(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) { |
|||
Integer keyId = getOrSaveKeyId(key); |
|||
entitiesFutures.add(tsKvRepository.findStringMax( |
|||
entityId.getId(), |
|||
keyId, |
|||
startTs, |
|||
endTs)); |
|||
entitiesFutures.add(tsKvRepository.findNumericMax( |
|||
entityId.getId(), |
|||
keyId, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
protected void findAvg(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures) { |
|||
Integer keyId = getOrSaveKeyId(key); |
|||
entitiesFutures.add(tsKvRepository.findAvg( |
|||
entityId.getId(), |
|||
keyId, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
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 void savePartition(PsqlPartition psqlPartition) { |
|||
if (!partitions.contains(psqlPartition)) { |
|||
partitionCreationLock.lock(); |
|||
try { |
|||
log.trace("Saving partition: {}", psqlPartition); |
|||
partitioningRepository.save(psqlPartition); |
|||
log.trace("Adding partition to Set: {}", psqlPartition); |
|||
partitions.add(psqlPartition); |
|||
} finally { |
|||
partitionCreationLock.unlock(); |
|||
} |
|||
} |
|||
} |
|||
|
|||
private PsqlPartition toPartition(long ts) { |
|||
LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(ts), ZoneOffset.UTC); |
|||
LocalDateTime localDateTimeStart = tsFormat.trancateTo(time); |
|||
if (localDateTimeStart == SqlTsPartitionDate.EPOCH_START) { |
|||
return new PsqlPartition(toMills(EPOCH_START), Long.MAX_VALUE, tsFormat.getPattern()); |
|||
} else { |
|||
LocalDateTime localDateTimeEnd = tsFormat.plusTo(localDateTimeStart); |
|||
return new PsqlPartition(toMills(localDateTimeStart), toMills(localDateTimeEnd), tsFormat.getPattern()); |
|||
} |
|||
} |
|||
|
|||
private long toMills(LocalDateTime time) { return time.toInstant(ZoneOffset.UTC).toEpochMilli(); } |
|||
} |
|||
@ -0,0 +1,41 @@ |
|||
/** |
|||
* 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.psql; |
|||
|
|||
import org.springframework.stereotype.Repository; |
|||
import org.springframework.transaction.annotation.Transactional; |
|||
import org.thingsboard.server.dao.timeseries.PsqlPartition; |
|||
import org.thingsboard.server.dao.util.PsqlDao; |
|||
import org.thingsboard.server.dao.util.SqlTsDao; |
|||
|
|||
import javax.persistence.EntityManager; |
|||
import javax.persistence.PersistenceContext; |
|||
|
|||
@SqlTsDao |
|||
@PsqlDao |
|||
@Repository |
|||
@Transactional |
|||
public class PsqlPartitioningRepository { |
|||
|
|||
@PersistenceContext |
|||
private EntityManager entityManager; |
|||
|
|||
public void save(PsqlPartition partition) { |
|||
entityManager.createNativeQuery(partition.getQuery()) |
|||
.executeUpdate(); |
|||
} |
|||
|
|||
} |
|||
@ -0,0 +1,101 @@ |
|||
/** |
|||
* 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.psql; |
|||
|
|||
import org.springframework.jdbc.core.BatchPreparedStatementSetter; |
|||
import org.springframework.stereotype.Repository; |
|||
import org.springframework.transaction.annotation.Transactional; |
|||
import org.thingsboard.server.dao.model.sqlts.psql.TsKvEntity; |
|||
import org.thingsboard.server.dao.sqlts.AbstractInsertRepository; |
|||
import org.thingsboard.server.dao.sqlts.EntityContainer; |
|||
import org.thingsboard.server.dao.sqlts.InsertTsRepository; |
|||
import org.thingsboard.server.dao.util.PsqlDao; |
|||
import org.thingsboard.server.dao.util.SqlTsDao; |
|||
|
|||
import java.sql.PreparedStatement; |
|||
import java.sql.SQLException; |
|||
import java.sql.Types; |
|||
import java.util.ArrayList; |
|||
import java.util.HashMap; |
|||
import java.util.List; |
|||
import java.util.Map; |
|||
|
|||
@SqlTsDao |
|||
@PsqlDao |
|||
@Repository |
|||
@Transactional |
|||
public class PsqlTimeseriesInsertRepository extends AbstractInsertRepository implements InsertTsRepository<TsKvEntity> { |
|||
|
|||
private static final String INSERT_INTO_TS_KV = "INSERT INTO ts_kv_"; |
|||
|
|||
private static final String VALUES_ON_CONFLICT_DO_UPDATE = " (entity_id, key, ts, bool_v, str_v, long_v, dbl_v) VALUES (?, ?, ?, ?, ?, ?, ?) " + |
|||
"ON CONFLICT (entity_id, key, ts) DO UPDATE SET bool_v = ?, str_v = ?, long_v = ?, dbl_v = ?;"; |
|||
|
|||
@Override |
|||
public void saveOrUpdate(List<EntityContainer<TsKvEntity>> entities) { |
|||
Map<String, List<TsKvEntity>> partitionMap = new HashMap<>(); |
|||
for (EntityContainer<TsKvEntity> entityContainer : entities) { |
|||
List<TsKvEntity> tsKvEntities = partitionMap.computeIfAbsent(entityContainer.getPartitionDate(), k -> new ArrayList<>()); |
|||
tsKvEntities.add(entityContainer.getEntity()); |
|||
} |
|||
partitionMap.forEach((partition, entries) -> jdbcTemplate.batchUpdate(getInsertOrUpdateQuery(partition), new BatchPreparedStatementSetter() { |
|||
@Override |
|||
public void setValues(PreparedStatement ps, int i) throws SQLException { |
|||
TsKvEntity tsKvEntity = entries.get(i); |
|||
ps.setObject(1, tsKvEntity.getEntityId()); |
|||
ps.setInt(2, tsKvEntity.getKey()); |
|||
ps.setLong(3, tsKvEntity.getTs()); |
|||
|
|||
if (tsKvEntity.getBooleanValue() != null) { |
|||
ps.setBoolean(4, tsKvEntity.getBooleanValue()); |
|||
ps.setBoolean(8, tsKvEntity.getBooleanValue()); |
|||
} else { |
|||
ps.setNull(4, Types.BOOLEAN); |
|||
ps.setNull(8, Types.BOOLEAN); |
|||
} |
|||
|
|||
ps.setString(5, replaceNullChars(tsKvEntity.getStrValue())); |
|||
ps.setString(9, replaceNullChars(tsKvEntity.getStrValue())); |
|||
|
|||
|
|||
if (tsKvEntity.getLongValue() != null) { |
|||
ps.setLong(6, tsKvEntity.getLongValue()); |
|||
ps.setLong(10, tsKvEntity.getLongValue()); |
|||
} else { |
|||
ps.setNull(6, Types.BIGINT); |
|||
ps.setNull(10, Types.BIGINT); |
|||
} |
|||
|
|||
if (tsKvEntity.getDoubleValue() != null) { |
|||
ps.setDouble(7, tsKvEntity.getDoubleValue()); |
|||
ps.setDouble(11, tsKvEntity.getDoubleValue()); |
|||
} else { |
|||
ps.setNull(7, Types.DOUBLE); |
|||
ps.setNull(11, Types.DOUBLE); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public int getBatchSize() { |
|||
return entries.size(); |
|||
} |
|||
})); |
|||
} |
|||
|
|||
private String getInsertOrUpdateQuery(String partitionDate) { |
|||
return INSERT_INTO_TS_KV + partitionDate + VALUES_ON_CONFLICT_DO_UPDATE; |
|||
} |
|||
} |
|||
@ -0,0 +1,132 @@ |
|||
/** |
|||
* 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.psql; |
|||
|
|||
import org.springframework.data.domain.Pageable; |
|||
import org.springframework.data.jpa.repository.Modifying; |
|||
import org.springframework.data.jpa.repository.Query; |
|||
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.dao.model.sqlts.psql.TsKvCompositeKey; |
|||
import org.thingsboard.server.dao.model.sqlts.psql.TsKvEntity; |
|||
import org.thingsboard.server.dao.util.SqlDao; |
|||
|
|||
import java.util.List; |
|||
import java.util.UUID; |
|||
import java.util.concurrent.CompletableFuture; |
|||
|
|||
@SqlDao |
|||
public interface TsKvPsqlRepository extends CrudRepository<TsKvEntity, TsKvCompositeKey> { |
|||
|
|||
@Query("SELECT tskv FROM TsKvEntity tskv WHERE tskv.entityId = :entityId " + |
|||
"AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs") |
|||
List<TsKvEntity> findAllWithLimit(@Param("entityId") UUID entityId, |
|||
@Param("entityKey") int key, |
|||
@Param("startTs") long startTs, |
|||
@Param("endTs") long endTs, |
|||
Pageable pageable); |
|||
|
|||
@Transactional |
|||
@Modifying |
|||
@Query("DELETE FROM TsKvEntity tskv WHERE tskv.entityId = :entityId " + |
|||
"AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs") |
|||
void delete(@Param("entityId") UUID entityId, |
|||
@Param("entityKey") int 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 " + |
|||
"AND tskv.entityId = :entityId AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs") |
|||
CompletableFuture<TsKvEntity> findStringMax(@Param("entityId") UUID entityId, |
|||
@Param("entityKey") int entityKey, |
|||
@Param("startTs") long startTs, |
|||
@Param("endTs") long endTs); |
|||
|
|||
@Async |
|||
@Query("SELECT new TsKvEntity(MAX(COALESCE(tskv.longValue, -9223372036854775807)), " + |
|||
"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 " + |
|||
"WHERE tskv.entityId = :entityId AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs") |
|||
CompletableFuture<TsKvEntity> findNumericMax(@Param("entityId") UUID entityId, |
|||
@Param("entityKey") int 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 " + |
|||
"AND tskv.entityId = :entityId AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs") |
|||
CompletableFuture<TsKvEntity> findStringMin(@Param("entityId") UUID entityId, |
|||
@Param("entityKey") int entityKey, |
|||
@Param("startTs") long startTs, |
|||
@Param("endTs") long endTs); |
|||
|
|||
@Async |
|||
@Query("SELECT new TsKvEntity(MIN(COALESCE(tskv.longValue, 9223372036854775807)), " + |
|||
"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 " + |
|||
"WHERE tskv.entityId = :entityId AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs") |
|||
CompletableFuture<TsKvEntity> findNumericMin( |
|||
@Param("entityId") UUID entityId, |
|||
@Param("entityKey") int entityKey, |
|||
@Param("startTs") long startTs, |
|||
@Param("endTs") long endTs); |
|||
|
|||
@Async |
|||
@Query("SELECT new TsKvEntity(SUM(CASE WHEN tskv.booleanValue IS NULL THEN 0 ELSE 1 END), " + |
|||
"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.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs") |
|||
CompletableFuture<TsKvEntity> findCount(@Param("entityId") UUID entityId, |
|||
@Param("entityKey") int entityKey, |
|||
@Param("startTs") long startTs, |
|||
@Param("endTs") long endTs); |
|||
|
|||
@Async |
|||
@Query("SELECT new TsKvEntity(SUM(COALESCE(tskv.longValue, 0)), " + |
|||
"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 " + |
|||
"WHERE tskv.entityId = :entityId AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs") |
|||
CompletableFuture<TsKvEntity> findAvg(@Param("entityId") UUID entityId, |
|||
@Param("entityKey") int entityKey, |
|||
@Param("startTs") long startTs, |
|||
@Param("endTs") long endTs); |
|||
|
|||
@Async |
|||
@Query("SELECT new TsKvEntity(SUM(COALESCE(tskv.longValue, 0)), " + |
|||
"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 " + |
|||
"WHERE tskv.entityId = :entityId AND tskv.key = :entityKey AND tskv.ts > :startTs AND tskv.ts <= :endTs") |
|||
CompletableFuture<TsKvEntity> findSum(@Param("entityId") UUID entityId, |
|||
@Param("entityKey") int entityKey, |
|||
@Param("startTs") long startTs, |
|||
@Param("endTs") long endTs); |
|||
|
|||
} |
|||
@ -1,140 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.dao.sqlts.ts; |
|||
|
|||
import org.springframework.jdbc.core.BatchPreparedStatementSetter; |
|||
import org.springframework.stereotype.Repository; |
|||
import org.springframework.transaction.annotation.Transactional; |
|||
import org.thingsboard.server.dao.model.sqlts.ts.TsKvLatestEntity; |
|||
import org.thingsboard.server.dao.sqlts.AbstractLatestInsertRepository; |
|||
import org.thingsboard.server.dao.util.HsqlDao; |
|||
import org.thingsboard.server.dao.util.SqlTsDao; |
|||
|
|||
import java.sql.PreparedStatement; |
|||
import java.sql.SQLException; |
|||
import java.sql.Types; |
|||
import java.util.List; |
|||
|
|||
@SqlTsDao |
|||
@HsqlDao |
|||
@Repository |
|||
@Transactional |
|||
public class HsqlLatestInsertRepository extends AbstractLatestInsertRepository { |
|||
|
|||
private static final String TS_KV_LATEST_CONSTRAINT = "(ts_kv_latest.entity_type=A.entity_type AND ts_kv_latest.entity_id=A.entity_id AND ts_kv_latest.key=A.key)"; |
|||
|
|||
private static final String INSERT_OR_UPDATE_BOOL_STATEMENT = getInsertOrUpdateStringHsql(TS_KV_LATEST_TABLE, TS_KV_LATEST_CONSTRAINT, BOOL_V, HSQL_LATEST_ON_BOOL_VALUE_UPDATE_SET_NULLS); |
|||
private static final String INSERT_OR_UPDATE_STR_STATEMENT = getInsertOrUpdateStringHsql(TS_KV_LATEST_TABLE, TS_KV_LATEST_CONSTRAINT, STR_V, HSQL_LATEST_ON_STR_VALUE_UPDATE_SET_NULLS); |
|||
private static final String INSERT_OR_UPDATE_LONG_STATEMENT = getInsertOrUpdateStringHsql(TS_KV_LATEST_TABLE, TS_KV_LATEST_CONSTRAINT, LONG_V, HSQL_LATEST_ON_LONG_VALUE_UPDATE_SET_NULLS); |
|||
private static final String INSERT_OR_UPDATE_DBL_STATEMENT = getInsertOrUpdateStringHsql(TS_KV_LATEST_TABLE, TS_KV_LATEST_CONSTRAINT, DBL_V, HSQL_LATEST_ON_DBL_VALUE_UPDATE_SET_NULLS); |
|||
|
|||
private static final String INSERT_OR_UPDATE = |
|||
"MERGE INTO ts_kv_latest USING(VALUES ?, ?, ?, ?, ?, ?, ?, ?) " + |
|||
"T (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " + |
|||
"ON (ts_kv_latest.entity_type=T.entity_type " + |
|||
"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) " + |
|||
"VALUES (T.entity_type, T.entity_id, T.key, T.ts, T.bool_v, T.str_v, T.long_v, T.dbl_v);"; |
|||
|
|||
@Override |
|||
public void saveOrUpdate(TsKvLatestEntity entity) { |
|||
processSaveOrUpdate(entity, INSERT_OR_UPDATE_BOOL_STATEMENT, INSERT_OR_UPDATE_STR_STATEMENT, INSERT_OR_UPDATE_LONG_STATEMENT, INSERT_OR_UPDATE_DBL_STATEMENT); |
|||
} |
|||
|
|||
@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.setString(2, entities.get(i).getEntityId()); |
|||
ps.setString(3, entities.get(i).getKey()); |
|||
ps.setLong(4, entities.get(i).getTs()); |
|||
|
|||
if (entities.get(i).getBooleanValue() != null) { |
|||
ps.setBoolean(5, entities.get(i).getBooleanValue()); |
|||
} else { |
|||
ps.setNull(5, Types.BOOLEAN); |
|||
} |
|||
|
|||
ps.setString(6, entities.get(i).getStrValue()); |
|||
|
|||
if (entities.get(i).getLongValue() != null) { |
|||
ps.setLong(7, entities.get(i).getLongValue()); |
|||
} else { |
|||
ps.setNull(7, Types.BIGINT); |
|||
} |
|||
|
|||
if (entities.get(i).getDoubleValue() != null) { |
|||
ps.setDouble(8, entities.get(i).getDoubleValue()); |
|||
} else { |
|||
ps.setNull(8, Types.DOUBLE); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public int getBatchSize() { |
|||
return entities.size(); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
protected void saveOrUpdateBoolean(TsKvLatestEntity entity, String query) { |
|||
entityManager.createNativeQuery(query) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("key", entity.getKey()) |
|||
.setParameter("ts", entity.getTs()) |
|||
.setParameter("bool_v", entity.getBooleanValue()) |
|||
.executeUpdate(); |
|||
} |
|||
|
|||
@Override |
|||
protected void saveOrUpdateString(TsKvLatestEntity entity, String query) { |
|||
entityManager.createNativeQuery(query) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("key", entity.getKey()) |
|||
.setParameter("ts", entity.getTs()) |
|||
.setParameter("str_v", entity.getStrValue()) |
|||
.executeUpdate(); |
|||
} |
|||
|
|||
@Override |
|||
protected void saveOrUpdateLong(TsKvLatestEntity entity, String query) { |
|||
entityManager.createNativeQuery(query) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("key", entity.getKey()) |
|||
.setParameter("ts", entity.getTs()) |
|||
.setParameter("long_v", entity.getLongValue()) |
|||
.executeUpdate(); |
|||
} |
|||
|
|||
@Override |
|||
protected void saveOrUpdateDouble(TsKvLatestEntity entity, String query) { |
|||
entityManager.createNativeQuery(query) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("key", entity.getKey()) |
|||
.setParameter("ts", entity.getTs()) |
|||
.setParameter("dbl_v", entity.getDoubleValue()) |
|||
.executeUpdate(); |
|||
} |
|||
} |
|||
@ -1,141 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.dao.sqlts.ts; |
|||
|
|||
import org.springframework.jdbc.core.BatchPreparedStatementSetter; |
|||
import org.springframework.stereotype.Repository; |
|||
import org.springframework.transaction.annotation.Transactional; |
|||
import org.thingsboard.server.dao.model.sqlts.ts.TsKvEntity; |
|||
import org.thingsboard.server.dao.sqlts.AbstractTimeseriesInsertRepository; |
|||
import org.thingsboard.server.dao.util.HsqlDao; |
|||
import org.thingsboard.server.dao.util.SqlTsDao; |
|||
|
|||
import java.sql.PreparedStatement; |
|||
import java.sql.SQLException; |
|||
import java.sql.Types; |
|||
import java.util.List; |
|||
|
|||
@SqlTsDao |
|||
@HsqlDao |
|||
@Repository |
|||
@Transactional |
|||
public class HsqlTimeseriesInsertRepository extends AbstractTimeseriesInsertRepository<TsKvEntity> { |
|||
|
|||
private static final String TS_KV_CONSTRAINT = "(ts_kv.entity_type=A.entity_type AND ts_kv.entity_id=A.entity_id AND ts_kv.key=A.key AND ts_kv.ts=A.ts)"; |
|||
|
|||
private static final String INSERT_OR_UPDATE_BOOL_STATEMENT = getInsertOrUpdateStringHsql(TS_KV_TABLE, TS_KV_CONSTRAINT, BOOL_V, HSQL_ON_BOOL_VALUE_UPDATE_SET_NULLS); |
|||
private static final String INSERT_OR_UPDATE_STR_STATEMENT = getInsertOrUpdateStringHsql(TS_KV_TABLE, TS_KV_CONSTRAINT, STR_V, HSQL_ON_STR_VALUE_UPDATE_SET_NULLS); |
|||
private static final String INSERT_OR_UPDATE_LONG_STATEMENT = getInsertOrUpdateStringHsql(TS_KV_TABLE, TS_KV_CONSTRAINT, LONG_V, HSQL_ON_LONG_VALUE_UPDATE_SET_NULLS); |
|||
private static final String INSERT_OR_UPDATE_DBL_STATEMENT = getInsertOrUpdateStringHsql(TS_KV_TABLE, TS_KV_CONSTRAINT, DBL_V, HSQL_ON_DBL_VALUE_UPDATE_SET_NULLS); |
|||
|
|||
private static final String INSERT_OR_UPDATE = |
|||
"MERGE INTO ts_kv USING(VALUES ?, ?, ?, ?, ?, ?, ?, ?) " + |
|||
"T (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) " + |
|||
"ON (ts_kv.entity_type=T.entity_type " + |
|||
"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) " + |
|||
"VALUES (T.entity_type, T.entity_id, T.key, T.ts, T.bool_v, T.str_v, T.long_v, T.dbl_v);"; |
|||
|
|||
@Override |
|||
public void saveOrUpdate(TsKvEntity entity) { |
|||
processSaveOrUpdate(entity, INSERT_OR_UPDATE_BOOL_STATEMENT, INSERT_OR_UPDATE_STR_STATEMENT, INSERT_OR_UPDATE_LONG_STATEMENT, INSERT_OR_UPDATE_DBL_STATEMENT); |
|||
} |
|||
|
|||
@Override |
|||
public void saveOrUpdate(List<TsKvEntity> 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.setString(2, entities.get(i).getEntityId()); |
|||
ps.setString(3, entities.get(i).getKey()); |
|||
ps.setLong(4, entities.get(i).getTs()); |
|||
|
|||
if (entities.get(i).getBooleanValue() != null) { |
|||
ps.setBoolean(5, entities.get(i).getBooleanValue()); |
|||
} else { |
|||
ps.setNull(5, Types.BOOLEAN); |
|||
} |
|||
|
|||
ps.setString(6, entities.get(i).getStrValue()); |
|||
|
|||
if (entities.get(i).getLongValue() != null) { |
|||
ps.setLong(7, entities.get(i).getLongValue()); |
|||
} else { |
|||
ps.setNull(7, Types.BIGINT); |
|||
} |
|||
|
|||
if (entities.get(i).getDoubleValue() != null) { |
|||
ps.setDouble(8, entities.get(i).getDoubleValue()); |
|||
} else { |
|||
ps.setNull(8, Types.DOUBLE); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public int getBatchSize() { |
|||
return entities.size(); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
protected void saveOrUpdateBoolean(TsKvEntity entity, String query) { |
|||
entityManager.createNativeQuery(query) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("key", entity.getKey()) |
|||
.setParameter("ts", entity.getTs()) |
|||
.setParameter("bool_v", entity.getBooleanValue()) |
|||
.executeUpdate(); |
|||
} |
|||
|
|||
@Override |
|||
protected void saveOrUpdateString(TsKvEntity entity, String query) { |
|||
entityManager.createNativeQuery(query) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("key", entity.getKey()) |
|||
.setParameter("ts", entity.getTs()) |
|||
.setParameter("str_v", entity.getStrValue()) |
|||
.executeUpdate(); |
|||
} |
|||
|
|||
@Override |
|||
protected void saveOrUpdateLong(TsKvEntity entity, String query) { |
|||
entityManager.createNativeQuery(query) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("key", entity.getKey()) |
|||
.setParameter("ts", entity.getTs()) |
|||
.setParameter("long_v", entity.getLongValue()) |
|||
.executeUpdate(); |
|||
} |
|||
|
|||
@Override |
|||
protected void saveOrUpdateDouble(TsKvEntity entity, String query) { |
|||
entityManager.createNativeQuery(query) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("key", entity.getKey()) |
|||
.setParameter("ts", entity.getTs()) |
|||
.setParameter("dbl_v", entity.getDoubleValue()) |
|||
.executeUpdate(); |
|||
} |
|||
} |
|||
@ -1,436 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.dao.sqlts.ts; |
|||
|
|||
import com.google.common.collect.Lists; |
|||
import com.google.common.util.concurrent.FutureCallback; |
|||
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.springframework.data.domain.PageRequest; |
|||
import org.springframework.data.domain.Sort; |
|||
import org.springframework.stereotype.Component; |
|||
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; |
|||
import org.thingsboard.server.common.data.kv.BasicTsKvEntry; |
|||
import org.thingsboard.server.common.data.kv.DeleteTsKvQuery; |
|||
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.ts.TsKvEntity; |
|||
import org.thingsboard.server.dao.model.sqlts.ts.TsKvLatestCompositeKey; |
|||
import org.thingsboard.server.dao.model.sqlts.ts.TsKvLatestEntity; |
|||
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.AbstractLatestInsertRepository; |
|||
import org.thingsboard.server.dao.sqlts.AbstractSqlTimeseriesDao; |
|||
import org.thingsboard.server.dao.sqlts.AbstractTimeseriesInsertRepository; |
|||
import org.thingsboard.server.dao.timeseries.SimpleListenableFuture; |
|||
import org.thingsboard.server.dao.timeseries.TimeseriesDao; |
|||
import org.thingsboard.server.dao.util.SqlTsDao; |
|||
|
|||
import javax.annotation.Nullable; |
|||
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.concurrent.ExecutionException; |
|||
import java.util.stream.Collectors; |
|||
|
|||
import static org.thingsboard.server.common.data.UUIDConverter.fromTimeUUID; |
|||
|
|||
|
|||
@Component |
|||
@Slf4j |
|||
@SqlTsDao |
|||
public class JpaTimeseriesDao extends AbstractSqlTimeseriesDao implements TimeseriesDao { |
|||
|
|||
@Autowired |
|||
private TsKvRepository tsKvRepository; |
|||
|
|||
@Autowired |
|||
private TsKvLatestRepository tsKvLatestRepository; |
|||
|
|||
@Autowired |
|||
private AbstractTimeseriesInsertRepository insertRepository; |
|||
|
|||
@Autowired |
|||
private AbstractLatestInsertRepository insertLatestRepository; |
|||
|
|||
@Autowired |
|||
ScheduledLogExecutorComponent logExecutor; |
|||
|
|||
@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; |
|||
|
|||
@Value("${sql.ts_latest.batch_size:1000}") |
|||
private int tsLatestBatchSize; |
|||
|
|||
@Value("${sql.ts_latest.batch_max_delay:100}") |
|||
private long tsLatestMaxDelay; |
|||
|
|||
@Value("${sql.ts_latest.stats_print_interval_ms:1000}") |
|||
private long tsLatestStatsPrintIntervalMs; |
|||
|
|||
private TbSqlBlockingQueue<TsKvEntity> tsQueue; |
|||
private TbSqlBlockingQueue<TsKvLatestEntity> tsLatestQueue; |
|||
|
|||
|
|||
@PostConstruct |
|||
private void init() { |
|||
TbSqlBlockingQueueParams tsParams = TbSqlBlockingQueueParams.builder() |
|||
.logName("TS") |
|||
.batchSize(tsBatchSize) |
|||
.maxDelay(tsMaxDelay) |
|||
.statsPrintIntervalMs(tsStatsPrintIntervalMs) |
|||
.build(); |
|||
tsQueue = new TbSqlBlockingQueue<>(tsParams); |
|||
tsQueue.init(logExecutor, v -> insertRepository.saveOrUpdate(v)); |
|||
|
|||
TbSqlBlockingQueueParams tsLatestParams = TbSqlBlockingQueueParams.builder() |
|||
.logName("TS Latest") |
|||
.batchSize(tsLatestBatchSize) |
|||
.maxDelay(tsLatestMaxDelay) |
|||
.statsPrintIntervalMs(tsLatestStatsPrintIntervalMs) |
|||
.build(); |
|||
tsLatestQueue = new TbSqlBlockingQueue<>(tsLatestParams); |
|||
tsLatestQueue.init(logExecutor, v -> insertLatestRepository.saveOrUpdate(v)); |
|||
} |
|||
|
|||
@PreDestroy |
|||
private void destroy() { |
|||
if (tsQueue != null) { |
|||
tsQueue.destroy(); |
|||
} |
|||
|
|||
if (tsLatestQueue != null) { |
|||
tsLatestQueue.destroy(); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, List<ReadTsKvQuery> queries) { |
|||
return processFindAllAsync(tenantId, entityId, queries); |
|||
} |
|||
|
|||
protected ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, ReadTsKvQuery query) { |
|||
if (query.getAggregation() == Aggregation.NONE) { |
|||
return findAllAsyncWithLimit(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)); |
|||
} |
|||
} |
|||
|
|||
private ListenableFuture<Optional<TsKvEntry>> findAndAggregateAsync(TenantId tenantId, EntityId entityId, String key, long startTs, long endTs, long ts, Aggregation aggregation) { |
|||
List<CompletableFuture<TsKvEntity>> entitiesFutures = new ArrayList<>(); |
|||
String entityIdStr = fromTimeUUID(entityId.getId()); |
|||
switchAgregation(entityId, key, startTs, endTs, aggregation, entitiesFutures, entityIdStr); |
|||
|
|||
SettableFuture<TsKvEntity> listenableFuture = SettableFuture.create(); |
|||
|
|||
CompletableFuture<List<TsKvEntity>> entities = |
|||
CompletableFuture.allOf(entitiesFutures.toArray(new CompletableFuture[entitiesFutures.size()])) |
|||
.thenApply(v -> entitiesFutures.stream() |
|||
.map(CompletableFuture::join) |
|||
.collect(Collectors.toList())); |
|||
|
|||
entities.whenComplete((tsKvEntities, throwable) -> { |
|||
if (throwable != null) { |
|||
listenableFuture.setException(throwable); |
|||
} else { |
|||
TsKvEntity result = null; |
|||
for (TsKvEntity entity : tsKvEntities) { |
|||
if (entity.isNotEmpty()) { |
|||
result = entity; |
|||
break; |
|||
} |
|||
} |
|||
listenableFuture.set(result); |
|||
} |
|||
}); |
|||
return Futures.transform(listenableFuture, entity -> { |
|||
if (entity != null && entity.isNotEmpty()) { |
|||
entity.setEntityId(entityIdStr); |
|||
entity.setEntityType(entityId.getEntityType()); |
|||
entity.setKey(key); |
|||
entity.setTs(ts); |
|||
return Optional.of(DaoUtil.getData(entity)); |
|||
} else { |
|||
return Optional.empty(); |
|||
} |
|||
}); |
|||
} |
|||
|
|||
private void switchAgregation(EntityId entityId, String key, long startTs, long endTs, Aggregation aggregation, List<CompletableFuture<TsKvEntity>> entitiesFutures, String entityIdStr) { |
|||
switch (aggregation) { |
|||
case AVG: |
|||
findAvg(entityId, key, startTs, endTs, entitiesFutures, entityIdStr); |
|||
break; |
|||
case MAX: |
|||
findMax(entityId, key, startTs, endTs, entitiesFutures, entityIdStr); |
|||
break; |
|||
case MIN: |
|||
findMin(entityId, key, startTs, endTs, entitiesFutures, entityIdStr); |
|||
break; |
|||
case SUM: |
|||
findSum(entityId, key, startTs, endTs, entitiesFutures, entityIdStr); |
|||
break; |
|||
case COUNT: |
|||
findCount(entityId, key, startTs, endTs, entitiesFutures, entityIdStr); |
|||
break; |
|||
default: |
|||
throw new IllegalArgumentException("Not supported aggregation type: " + aggregation); |
|||
} |
|||
} |
|||
|
|||
private void findCount(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures, String entityIdStr) { |
|||
entitiesFutures.add(tsKvRepository.findCount( |
|||
entityIdStr, |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
private void findSum(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures, String entityIdStr) { |
|||
entitiesFutures.add(tsKvRepository.findSum( |
|||
entityIdStr, |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
private void findMin(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures, String entityIdStr) { |
|||
entitiesFutures.add(tsKvRepository.findStringMin( |
|||
entityIdStr, |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
entitiesFutures.add(tsKvRepository.findNumericMin( |
|||
entityIdStr, |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
private void findMax(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures, String entityIdStr) { |
|||
entitiesFutures.add(tsKvRepository.findStringMax( |
|||
entityIdStr, |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
entitiesFutures.add(tsKvRepository.findNumericMax( |
|||
entityIdStr, |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
private void findAvg(EntityId entityId, String key, long startTs, long endTs, List<CompletableFuture<TsKvEntity>> entitiesFutures, String entityIdStr) { |
|||
entitiesFutures.add(tsKvRepository.findAvg( |
|||
entityIdStr, |
|||
entityId.getEntityType(), |
|||
key, |
|||
startTs, |
|||
endTs)); |
|||
} |
|||
|
|||
private ListenableFuture<List<TsKvEntry>> findAllAsyncWithLimit(EntityId entityId, ReadTsKvQuery query) { |
|||
return Futures.immediateFuture( |
|||
DaoUtil.convertDataList( |
|||
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"))))); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<TsKvEntry> findLatest(TenantId tenantId, EntityId entityId, String key) { |
|||
TsKvLatestCompositeKey compositeKey = |
|||
new TsKvLatestCompositeKey( |
|||
entityId.getEntityType(), |
|||
fromTimeUUID(entityId.getId()), |
|||
key); |
|||
Optional<TsKvLatestEntity> entry = tsKvLatestRepository.findById(compositeKey); |
|||
TsKvEntry result; |
|||
if (entry.isPresent()) { |
|||
result = DaoUtil.getData(entry.get()); |
|||
} else { |
|||
result = new BasicTsKvEntry(System.currentTimeMillis(), new StringDataEntry(key, null)); |
|||
} |
|||
return Futures.immediateFuture(result); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<List<TsKvEntry>> findAllLatest(TenantId tenantId, EntityId entityId) { |
|||
return Futures.immediateFuture( |
|||
DaoUtil.convertDataList(Lists.newArrayList( |
|||
tsKvLatestRepository.findAllByEntityTypeAndEntityId( |
|||
entityId.getEntityType(), |
|||
UUIDConverter.fromTimeUUID(entityId.getId()))))); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> save(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry, long ttl) { |
|||
TsKvEntity entity = new TsKvEntity(); |
|||
entity.setEntityType(entityId.getEntityType()); |
|||
entity.setEntityId(fromTimeUUID(entityId.getId())); |
|||
entity.setTs(tsKvEntry.getTs()); |
|||
entity.setKey(tsKvEntry.getKey()); |
|||
entity.setStrValue(tsKvEntry.getStrValue().orElse(null)); |
|||
entity.setDoubleValue(tsKvEntry.getDoubleValue().orElse(null)); |
|||
entity.setLongValue(tsKvEntry.getLongValue().orElse(null)); |
|||
entity.setBooleanValue(tsKvEntry.getBooleanValue().orElse(null)); |
|||
log.trace("Saving entity: {}", entity); |
|||
return tsQueue.add(entity); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> savePartition(TenantId tenantId, EntityId entityId, long tsKvEntryTs, String key, long ttl) { |
|||
return Futures.immediateFuture(null); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> saveLatest(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry) { |
|||
TsKvLatestEntity latestEntity = new TsKvLatestEntity(); |
|||
latestEntity.setEntityType(entityId.getEntityType()); |
|||
latestEntity.setEntityId(fromTimeUUID(entityId.getId())); |
|||
latestEntity.setTs(tsKvEntry.getTs()); |
|||
latestEntity.setKey(tsKvEntry.getKey()); |
|||
latestEntity.setStrValue(tsKvEntry.getStrValue().orElse(null)); |
|||
latestEntity.setDoubleValue(tsKvEntry.getDoubleValue().orElse(null)); |
|||
latestEntity.setLongValue(tsKvEntry.getLongValue().orElse(null)); |
|||
latestEntity.setBooleanValue(tsKvEntry.getBooleanValue().orElse(null)); |
|||
return tsLatestQueue.add(latestEntity); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> remove(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) { |
|||
return service.submit(() -> { |
|||
tsKvRepository.delete( |
|||
fromTimeUUID(entityId.getId()), |
|||
entityId.getEntityType(), |
|||
query.getKey(), |
|||
query.getStartTs(), |
|||
query.getEndTs()); |
|||
return null; |
|||
}); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> removeLatest(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) { |
|||
ListenableFuture<TsKvEntry> latestFuture = findLatest(tenantId, entityId, query.getKey()); |
|||
|
|||
ListenableFuture<Boolean> booleanFuture = Futures.transform(latestFuture, tsKvEntry -> { |
|||
long ts = tsKvEntry.getTs(); |
|||
return ts > query.getStartTs() && ts <= query.getEndTs(); |
|||
}, service); |
|||
|
|||
ListenableFuture<Void> removedLatestFuture = Futures.transformAsync(booleanFuture, isRemove -> { |
|||
if (isRemove) { |
|||
TsKvLatestEntity latestEntity = new TsKvLatestEntity(); |
|||
latestEntity.setEntityType(entityId.getEntityType()); |
|||
latestEntity.setEntityId(fromTimeUUID(entityId.getId())); |
|||
latestEntity.setKey(query.getKey()); |
|||
return service.submit(() -> { |
|||
tsKvLatestRepository.delete(latestEntity); |
|||
return null; |
|||
}); |
|||
} |
|||
return Futures.immediateFuture(null); |
|||
}, service); |
|||
|
|||
final SimpleListenableFuture<Void> resultFuture = new SimpleListenableFuture<>(); |
|||
Futures.addCallback(removedLatestFuture, new FutureCallback<Void>() { |
|||
@Override |
|||
public void onSuccess(@Nullable Void result) { |
|||
if (query.getRewriteLatestIfDeleted()) { |
|||
ListenableFuture<Void> savedLatestFuture = Futures.transformAsync(booleanFuture, isRemove -> { |
|||
if (isRemove) { |
|||
return getNewLatestEntryFuture(tenantId, entityId, query); |
|||
} |
|||
return Futures.immediateFuture(null); |
|||
}, service); |
|||
|
|||
try { |
|||
resultFuture.set(savedLatestFuture.get()); |
|||
} catch (InterruptedException | ExecutionException e) { |
|||
log.warn("Could not get latest saved value for [{}], {}", entityId, query.getKey(), e); |
|||
} |
|||
} else { |
|||
resultFuture.set(null); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public void onFailure(Throwable t) { |
|||
log.warn("[{}] Failed to process remove of the latest value", entityId, t); |
|||
} |
|||
}); |
|||
return resultFuture; |
|||
} |
|||
|
|||
private ListenableFuture<Void> getNewLatestEntryFuture(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) { |
|||
ListenableFuture<List<TsKvEntry>> future = findNewLatestEntryFuture(tenantId, entityId, query); |
|||
return Futures.transformAsync(future, entryList -> { |
|||
if (entryList.size() == 1) { |
|||
return saveLatest(tenantId, entityId, entryList.get(0)); |
|||
} else { |
|||
log.trace("Could not find new latest value for [{}], key - {}", entityId, query.getKey()); |
|||
} |
|||
return Futures.immediateFuture(null); |
|||
}, service); |
|||
} |
|||
|
|||
@Override |
|||
public ListenableFuture<Void> removePartition(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) { |
|||
return service.submit(() -> null); |
|||
} |
|||
} |
|||
@ -1,143 +0,0 @@ |
|||
/** |
|||
* Copyright © 2016-2020 The Thingsboard Authors |
|||
* |
|||
* Licensed under the Apache License, Version 2.0 (the "License"); |
|||
* you may not use this file except in compliance with the License. |
|||
* You may obtain a copy of the License at |
|||
* |
|||
* http://www.apache.org/licenses/LICENSE-2.0
|
|||
* |
|||
* Unless required by applicable law or agreed to in writing, software |
|||
* distributed under the License is distributed on an "AS IS" BASIS, |
|||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
|||
* See the License for the specific language governing permissions and |
|||
* limitations under the License. |
|||
*/ |
|||
package org.thingsboard.server.dao.sqlts.ts; |
|||
|
|||
import org.springframework.jdbc.core.BatchPreparedStatementSetter; |
|||
import org.springframework.stereotype.Repository; |
|||
import org.springframework.transaction.annotation.Transactional; |
|||
import org.thingsboard.server.dao.model.sqlts.ts.TsKvEntity; |
|||
import org.thingsboard.server.dao.sqlts.AbstractTimeseriesInsertRepository; |
|||
import org.thingsboard.server.dao.util.PsqlDao; |
|||
import org.thingsboard.server.dao.util.SqlTsDao; |
|||
|
|||
import java.sql.PreparedStatement; |
|||
import java.sql.SQLException; |
|||
import java.sql.Types; |
|||
import java.util.List; |
|||
|
|||
@SqlTsDao |
|||
@PsqlDao |
|||
@Repository |
|||
@Transactional |
|||
public class PsqlTimeseriesInsertRepository extends AbstractTimeseriesInsertRepository<TsKvEntity> { |
|||
|
|||
private static final String TS_KV_CONSTRAINT = "(entity_type, entity_id, key, ts)"; |
|||
|
|||
private static final String INSERT_OR_UPDATE_BOOL_STATEMENT = getInsertOrUpdateStringPsql(TS_KV_TABLE, TS_KV_CONSTRAINT, BOOL_V, PSQL_ON_BOOL_VALUE_UPDATE_SET_NULLS); |
|||
private static final String INSERT_OR_UPDATE_STR_STATEMENT = getInsertOrUpdateStringPsql(TS_KV_TABLE, TS_KV_CONSTRAINT, STR_V, PSQL_ON_STR_VALUE_UPDATE_SET_NULLS); |
|||
private static final String INSERT_OR_UPDATE_LONG_STATEMENT = getInsertOrUpdateStringPsql(TS_KV_TABLE, TS_KV_CONSTRAINT, LONG_V, PSQL_ON_LONG_VALUE_UPDATE_SET_NULLS); |
|||
private static final String INSERT_OR_UPDATE_DBL_STATEMENT = getInsertOrUpdateStringPsql(TS_KV_TABLE, TS_KV_CONSTRAINT, DBL_V, PSQL_ON_DBL_VALUE_UPDATE_SET_NULLS); |
|||
|
|||
private static final String INSERT_OR_UPDATE = |
|||
"INSERT INTO ts_kv (entity_type, entity_id, key, ts, bool_v, str_v, long_v, dbl_v) VALUES(?, ?, ?, ?, ?, ?, ?, ?) " + |
|||
"ON CONFLICT (entity_type, entity_id, key, ts) DO UPDATE SET bool_v = ?, str_v = ?, long_v = ?, dbl_v = ?;"; |
|||
|
|||
@Override |
|||
public void saveOrUpdate(TsKvEntity entity) { |
|||
processSaveOrUpdate(entity, INSERT_OR_UPDATE_BOOL_STATEMENT, INSERT_OR_UPDATE_STR_STATEMENT, INSERT_OR_UPDATE_LONG_STATEMENT, INSERT_OR_UPDATE_DBL_STATEMENT); |
|||
} |
|||
|
|||
@Override |
|||
protected void saveOrUpdateBoolean(TsKvEntity entity, String query) { |
|||
entityManager.createNativeQuery(query) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("key", entity.getKey()) |
|||
.setParameter("ts", entity.getTs()) |
|||
.setParameter("bool_v", entity.getBooleanValue()) |
|||
.executeUpdate(); |
|||
} |
|||
|
|||
@Override |
|||
protected void saveOrUpdateString(TsKvEntity entity, String query) { |
|||
entityManager.createNativeQuery(query) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("key", entity.getKey()) |
|||
.setParameter("ts", entity.getTs()) |
|||
.setParameter("str_v", replaceNullChars(entity.getStrValue())) |
|||
.executeUpdate(); |
|||
} |
|||
|
|||
@Override |
|||
protected void saveOrUpdateLong(TsKvEntity entity, String query) { |
|||
entityManager.createNativeQuery(query) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("key", entity.getKey()) |
|||
.setParameter("ts", entity.getTs()) |
|||
.setParameter("long_v", entity.getLongValue()) |
|||
.executeUpdate(); |
|||
} |
|||
|
|||
@Override |
|||
protected void saveOrUpdateDouble(TsKvEntity entity, String query) { |
|||
entityManager.createNativeQuery(query) |
|||
.setParameter("entity_type", entity.getEntityType().name()) |
|||
.setParameter("entity_id", entity.getEntityId()) |
|||
.setParameter("key", entity.getKey()) |
|||
.setParameter("ts", entity.getTs()) |
|||
.setParameter("dbl_v", entity.getDoubleValue()) |
|||
.executeUpdate(); |
|||
} |
|||
|
|||
@Override |
|||
public void saveOrUpdate(List<TsKvEntity> entities) { |
|||
jdbcTemplate.batchUpdate(INSERT_OR_UPDATE, new BatchPreparedStatementSetter() { |
|||
@Override |
|||
public void setValues(PreparedStatement ps, int i) throws SQLException { |
|||
TsKvEntity tsKvEntity = entities.get(i); |
|||
ps.setString(1, tsKvEntity.getEntityType().name()); |
|||
ps.setString(2, tsKvEntity.getEntityId()); |
|||
ps.setString(3, tsKvEntity.getKey()); |
|||
ps.setLong(4, tsKvEntity.getTs()); |
|||
|
|||
if (tsKvEntity.getBooleanValue() != null) { |
|||
ps.setBoolean(5, tsKvEntity.getBooleanValue()); |
|||
ps.setBoolean(9, tsKvEntity.getBooleanValue()); |
|||
} else { |
|||
ps.setNull(5, Types.BOOLEAN); |
|||
ps.setNull(9, Types.BOOLEAN); |
|||
} |
|||
|
|||
ps.setString(6, replaceNullChars(tsKvEntity.getStrValue())); |
|||
ps.setString(10, replaceNullChars(tsKvEntity.getStrValue())); |
|||
|
|||
|
|||
if (tsKvEntity.getLongValue() != null) { |
|||
ps.setLong(7, tsKvEntity.getLongValue()); |
|||
ps.setLong(11, tsKvEntity.getLongValue()); |
|||
} else { |
|||
ps.setNull(7, Types.BIGINT); |
|||
ps.setNull(11, Types.BIGINT); |
|||
} |
|||
|
|||
if (tsKvEntity.getDoubleValue() != null) { |
|||
ps.setDouble(8, tsKvEntity.getDoubleValue()); |
|||
ps.setDouble(12, tsKvEntity.getDoubleValue()); |
|||
} else { |
|||
ps.setNull(8, Types.DOUBLE); |
|||
ps.setNull(12, Types.DOUBLE); |
|||
} |
|||
} |
|||
|
|||
@Override |
|||
public int getBatchSize() { |
|||
return entities.size(); |
|||
} |
|||
}); |
|||
} |
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
/** |
|||
* 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.timeseries; |
|||
|
|||
import lombok.Data; |
|||
|
|||
import java.text.SimpleDateFormat; |
|||
import java.util.Date; |
|||
|
|||
@Data |
|||
public class PsqlPartition { |
|||
|
|||
private static final String TABLE_REGEX = "ts_kv_"; |
|||
|
|||
private long start; |
|||
private long end; |
|||
private String partitionDate; |
|||
private String query; |
|||
|
|||
public PsqlPartition(long start, long end, String pattern) { |
|||
this.start = start; |
|||
this.end = end; |
|||
this.partitionDate = new SimpleDateFormat(pattern).format(new Date(start)); |
|||
this.query = createStatement(start, end, partitionDate); |
|||
} |
|||
|
|||
private String createStatement(long start, long end, String partitionDate) { |
|||
return "CREATE TABLE IF NOT EXISTS " + TABLE_REGEX + partitionDate + " PARTITION OF ts_kv(PRIMARY KEY (entity_id, key, ts)) FOR VALUES FROM (" + start + ") TO (" + end + ")"; |
|||
} |
|||
} |
|||
@ -0,0 +1,93 @@ |
|||
/** |
|||
* 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.timeseries; |
|||
|
|||
import java.time.LocalDateTime; |
|||
import java.time.ZoneOffset; |
|||
import java.time.temporal.ChronoUnit; |
|||
import java.time.temporal.TemporalUnit; |
|||
import java.util.Optional; |
|||
|
|||
public enum SqlTsPartitionDate { |
|||
|
|||
MINUTES("yyyy_MM_dd_HH_mm", ChronoUnit.MINUTES), HOURS("yyyy_MM_dd_HH", ChronoUnit.HOURS), DAYS("yyyy_MM_dd", ChronoUnit.DAYS), MONTHS("yyyy_MM", ChronoUnit.MONTHS), YEARS("yyyy", ChronoUnit.YEARS), INDEFINITE("indefinite", ChronoUnit.FOREVER); |
|||
|
|||
private final String pattern; |
|||
private final transient TemporalUnit truncateUnit; |
|||
public final static LocalDateTime EPOCH_START = LocalDateTime.ofEpochSecond(0, 0, ZoneOffset.UTC); |
|||
|
|||
SqlTsPartitionDate(String pattern, TemporalUnit truncateUnit) { |
|||
this.pattern = pattern; |
|||
this.truncateUnit = truncateUnit; |
|||
} |
|||
|
|||
public String getPattern() { |
|||
return pattern; |
|||
} |
|||
|
|||
public TemporalUnit getTruncateUnit() { |
|||
return truncateUnit; |
|||
} |
|||
|
|||
public LocalDateTime trancateTo(LocalDateTime time) { |
|||
switch (this) { |
|||
case MINUTES: |
|||
return time.truncatedTo(ChronoUnit.MINUTES); |
|||
case HOURS: |
|||
return time.truncatedTo(ChronoUnit.HOURS); |
|||
case DAYS: |
|||
return time.truncatedTo(ChronoUnit.DAYS); |
|||
case MONTHS: |
|||
return time.truncatedTo(ChronoUnit.DAYS).withDayOfMonth(1); |
|||
case YEARS: |
|||
return time.truncatedTo(ChronoUnit.DAYS).withDayOfYear(1); |
|||
case INDEFINITE: |
|||
return EPOCH_START; |
|||
default: |
|||
throw new RuntimeException("Failed to parse partitioning property!"); |
|||
} |
|||
} |
|||
|
|||
public LocalDateTime plusTo(LocalDateTime time) { |
|||
switch (this) { |
|||
case MINUTES: |
|||
return time.plusMinutes(1); |
|||
case HOURS: |
|||
return time.plusHours(1); |
|||
case DAYS: |
|||
return time.plusDays(1); |
|||
case MONTHS: |
|||
return time.plusMonths(1); |
|||
case YEARS: |
|||
return time.plusYears(1); |
|||
default: |
|||
throw new RuntimeException("Failed to parse partitioning property!"); |
|||
} |
|||
} |
|||
|
|||
public static Optional<SqlTsPartitionDate> parse(String name) { |
|||
SqlTsPartitionDate partition = null; |
|||
if (name != null) { |
|||
for (SqlTsPartitionDate partitionDate : SqlTsPartitionDate.values()) { |
|||
if (partitionDate.name().equalsIgnoreCase(name)) { |
|||
partition = partitionDate; |
|||
break; |
|||
} |
|||
} |
|||
} |
|||
return Optional.of(partition); |
|||
} |
|||
} |
|||
@ -0,0 +1,43 @@ |
|||
-- |
|||
-- 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. |
|||
-- |
|||
|
|||
CREATE TABLE IF NOT EXISTS ts_kv ( |
|||
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 |
|||
) 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 varchar(31) 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) |
|||
); |
|||
Loading…
Reference in new issue