diff --git a/dao/src/main/java/org/thingsboard/server/dao/dictionary/KeyDictionaryDao.java b/dao/src/main/java/org/thingsboard/server/dao/dictionary/KeyDictionaryDao.java new file mode 100644 index 0000000000..aa663bf208 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/dictionary/KeyDictionaryDao.java @@ -0,0 +1,25 @@ +/** + * Copyright © 2016-2023 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.dictionary; + + +public interface KeyDictionaryDao { + + Integer getOrSaveKeyId(String strKey); + + String getKey(Integer attributeKey); + +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/JpaAbstractDaoListeningExecutorService.java b/dao/src/main/java/org/thingsboard/server/dao/sql/JpaAbstractDaoListeningExecutorService.java index 3a10b7f619..5646aa1b6c 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/JpaAbstractDaoListeningExecutorService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/JpaAbstractDaoListeningExecutorService.java @@ -16,29 +16,17 @@ package org.thingsboard.server.dao.sql; import lombok.extern.slf4j.Slf4j; -import org.hibernate.exception.ConstraintViolationException; import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.dao.DataIntegrityViolationException; import org.springframework.jdbc.core.JdbcTemplate; -import org.thingsboard.server.dao.model.sqlts.dictionary.KeyDictionaryCompositeKey; -import org.thingsboard.server.dao.model.sqlts.dictionary.KeyDictionaryEntry; -import org.thingsboard.server.dao.sqlts.dictionary.KeyDictionaryRepository; import javax.sql.DataSource; import java.sql.SQLException; import java.sql.SQLWarning; import java.sql.Statement; -import java.util.Optional; -import java.util.concurrent.ConcurrentHashMap; -import java.util.concurrent.ConcurrentMap; -import java.util.concurrent.locks.ReentrantLock; @Slf4j public abstract class JpaAbstractDaoListeningExecutorService { - private final ConcurrentMap keyDictionaryMap = new ConcurrentHashMap<>(); - protected static final ReentrantLock creationLock = new ReentrantLock(); - @Autowired protected JpaExecutorService service; @@ -48,49 +36,6 @@ public abstract class JpaAbstractDaoListeningExecutorService { @Autowired protected JdbcTemplate jdbcTemplate; - @Autowired - protected KeyDictionaryRepository keyDictionaryRepository; - - protected Integer getOrSaveKeyId(String strKey) { - Integer keyId = keyDictionaryMap.get(strKey); - if (keyId == null) { - Optional tsKvDictionaryOptional; - tsKvDictionaryOptional = keyDictionaryRepository.findById(new KeyDictionaryCompositeKey(strKey)); - if (tsKvDictionaryOptional.isEmpty()) { - creationLock.lock(); - try { - keyId = keyDictionaryMap.get(strKey); - if (keyId != null) { - return keyId; - } - tsKvDictionaryOptional = keyDictionaryRepository.findById(new KeyDictionaryCompositeKey(strKey)); - if (tsKvDictionaryOptional.isEmpty()) { - KeyDictionaryEntry keyDictionaryEntry = new KeyDictionaryEntry(); - keyDictionaryEntry.setKey(strKey); - try { - KeyDictionaryEntry saved = keyDictionaryRepository.save(keyDictionaryEntry); - keyDictionaryMap.put(saved.getKey(), saved.getKeyId()); - keyId = saved.getKeyId(); - } catch (DataIntegrityViolationException | ConstraintViolationException e) { - tsKvDictionaryOptional = keyDictionaryRepository.findById(new KeyDictionaryCompositeKey(strKey)); - KeyDictionaryEntry dictionary = tsKvDictionaryOptional.orElseThrow(() -> new RuntimeException("Failed to get KeyDictionaryEntry entity from DB!")); - keyDictionaryMap.put(dictionary.getKey(), dictionary.getKeyId()); - keyId = dictionary.getKeyId(); - } - } else { - keyId = tsKvDictionaryOptional.get().getKeyId(); - } - } finally { - creationLock.unlock(); - } - } else { - keyId = tsKvDictionaryOptional.get().getKeyId(); - keyDictionaryMap.put(strKey, keyId); - } - } - return keyId; - } - protected void printWarnings(Statement statement) throws SQLException { SQLWarning warnings = statement.getWarnings(); if (warnings != null) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java index 54d0bf9495..7114d6db3e 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/attributes/JpaAttributeDao.java @@ -33,9 +33,9 @@ import org.thingsboard.server.common.data.kv.AttributeKvEntry; import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.dao.DaoUtil; import org.thingsboard.server.dao.attributes.AttributesDao; +import org.thingsboard.server.dao.dictionary.KeyDictionaryDao; import org.thingsboard.server.dao.model.sql.AttributeKvCompositeKey; import org.thingsboard.server.dao.model.sql.AttributeKvEntity; -import org.thingsboard.server.dao.model.sqlts.dictionary.KeyDictionaryEntry; import org.thingsboard.server.dao.sql.JpaAbstractDaoListeningExecutorService; import org.thingsboard.server.dao.sql.ScheduledLogExecutorComponent; import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams; @@ -67,6 +67,9 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl @Autowired private StatsFactory statsFactory; + @Autowired + private KeyDictionaryDao keyDictionaryDao; + @Value("${sql.attributes.batch_size:1000}") private int batchSize; @@ -114,7 +117,7 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl @Override public Optional find(TenantId tenantId, EntityId entityId, AttributeScope attributeScope, String attributeKey) { AttributeKvCompositeKey compositeKey = - getAttributeKvCompositeKey(entityId, attributeScope.getId(), getOrSaveKeyId(attributeKey)); + getAttributeKvCompositeKey(entityId, attributeScope.getId(), keyDictionaryDao.getOrSaveKeyId(attributeKey)); Optional attributeKvEntityOptional = attributeKvRepository.findById(compositeKey); if (attributeKvEntityOptional.isPresent()) { AttributeKvEntity attributeKvEntity = attributeKvEntityOptional.get(); @@ -130,10 +133,10 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl attributeKeys .stream() .map(attributeKey -> - getAttributeKvCompositeKey(entityId, attributeScope.getId(), getOrSaveKeyId(attributeKey))) + getAttributeKvCompositeKey(entityId, attributeScope.getId(), keyDictionaryDao.getOrSaveKeyId(attributeKey))) .collect(Collectors.toList()); List attributes = attributeKvRepository.findAllById(compositeKeys); - attributes.forEach(attributeKvEntity -> attributeKvEntity.setStrKey(getKey(attributeKvEntity.getId().getAttributeKey()))); + attributes.forEach(attributeKvEntity -> attributeKvEntity.setStrKey(keyDictionaryDao.getKey(attributeKvEntity.getId().getAttributeKey()))); return DaoUtil.convertDataList(Lists.newArrayList(attributes)); } @@ -142,7 +145,7 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl List attributes = attributeKvRepository.findAllEntityIdAndAttributeType( entityId.getId(), attributeScope.getId()); - attributes.forEach(attributeKvEntity -> attributeKvEntity.setStrKey(getKey(attributeKvEntity.getId().getAttributeKey()))); + attributes.forEach(attributeKvEntity -> attributeKvEntity.setStrKey(keyDictionaryDao.getKey(attributeKvEntity.getId().getAttributeKey()))); return DaoUtil.convertDataList(Lists.newArrayList( attributes)); } @@ -151,10 +154,10 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl public List findAllKeysByDeviceProfileId(TenantId tenantId, DeviceProfileId deviceProfileId) { if (deviceProfileId != null) { return attributeKvRepository.findAllKeysByDeviceProfileId(tenantId.getId(), deviceProfileId.getId()) - .stream().map(this::getKey).collect(Collectors.toList()); + .stream().map(id -> keyDictionaryDao.getKey(id)).collect(Collectors.toList()); } else { return attributeKvRepository.findAllKeysByTenantId(tenantId.getId()) - .stream().map(this::getKey).collect(Collectors.toList()); + .stream().map(id -> keyDictionaryDao.getKey(id)).collect(Collectors.toList()); } } @@ -162,13 +165,13 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl public List findAllKeysByEntityIds(TenantId tenantId, List entityIds) { return attributeKvRepository .findAllKeysByEntityIds(entityIds.stream().map(EntityId::getId).collect(Collectors.toList())) - .stream().map(this::getKey).collect(Collectors.toList()); + .stream().map(id -> keyDictionaryDao.getKey(id)).collect(Collectors.toList()); } @Override public ListenableFuture save(TenantId tenantId, EntityId entityId, AttributeScope attributeScope, AttributeKvEntry attribute) { AttributeKvEntity entity = new AttributeKvEntity(); - entity.setId(new AttributeKvCompositeKey(entityId.getId(), attributeScope.getId(), getOrSaveKeyId(attribute.getKey()))); + entity.setId(new AttributeKvCompositeKey(entityId.getId(), attributeScope.getId(), keyDictionaryDao.getOrSaveKeyId(attribute.getKey()))); entity.setLastUpdateTs(attribute.getLastUpdateTs()); entity.setStrValue(attribute.getStrValue().orElse(null)); entity.setDoubleValue(attribute.getDoubleValue().orElse(null)); @@ -187,7 +190,7 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl List> futuresList = new ArrayList<>(keys.size()); for (String key : keys) { futuresList.add(service.submit(() -> { - attributeKvRepository.delete(entityId.getId(), attributeScope.getId(), getOrSaveKeyId(key)); + attributeKvRepository.delete(entityId.getId(), attributeScope.getId(), keyDictionaryDao.getOrSaveKeyId(key)); return key; })); } @@ -200,9 +203,4 @@ public class JpaAttributeDao extends JpaAbstractDaoListeningExecutorService impl attributeType, attributeKey); } - - private String getKey(Integer attributeKey) { - Optional byKeyId = keyDictionaryRepository.findByKeyId(attributeKey); - return byKeyId.map(KeyDictionaryEntry::getKey).orElse(null); - } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java index 991043db88..c023b2ede3 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java @@ -32,6 +32,7 @@ import org.thingsboard.server.common.data.kv.ReadTsKvQueryResult; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.dao.DaoUtil; +import org.thingsboard.server.dao.dictionary.KeyDictionaryDao; import org.thingsboard.server.dao.model.sql.AbstractTsKvEntity; import org.thingsboard.server.dao.model.sqlts.ts.TsKvEntity; import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams; @@ -61,6 +62,9 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq @Autowired private StatsFactory statsFactory; + @Autowired + private KeyDictionaryDao keyDictionaryDao; + @PostConstruct protected void init() { TbSqlBlockingQueueParams tsParams = TbSqlBlockingQueueParams.builder() @@ -93,7 +97,7 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq return service.submit(() -> { tsKvRepository.delete( entityId.getId(), - getOrSaveKeyId(query.getKey()), + keyDictionaryDao.getOrSaveKeyId(query.getKey()), query.getStartTs(), query.getEndTs()); return null; @@ -132,7 +136,7 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq } private ReadTsKvQueryResult findAllAsyncWithLimit(EntityId entityId, ReadTsKvQuery query) { - Integer keyId = getOrSaveKeyId(query.getKey()); + Integer keyId = keyDictionaryDao.getOrSaveKeyId(query.getKey()); List tsKvEntities = tsKvRepository.findAllWithLimit( entityId.getId(), keyId, @@ -160,7 +164,7 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq } protected TsKvEntity switchAggregation(EntityId entityId, String key, long startTs, long endTs, Aggregation aggregation) { - var keyId = getOrSaveKeyId(key); + var keyId = keyDictionaryDao.getOrSaveKeyId(key); switch (aggregation) { case AVG: return tsKvRepository.findAvg(entityId.getId(), keyId, startTs, endTs); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java index 404855edf4..b7e871e3f5 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/SqlTimeseriesLatestDao.java @@ -37,6 +37,7 @@ import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.kv.TsKvLatestRemovingResult; import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.dao.DaoUtil; +import org.thingsboard.server.dao.dictionary.KeyDictionaryDao; import org.thingsboard.server.dao.model.sql.AbstractTsKvEntity; import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestCompositeKey; import org.thingsboard.server.dao.model.sqlts.latest.TsKvLatestEntity; @@ -103,6 +104,9 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme @Autowired private StatsFactory statsFactory; + @Autowired + private KeyDictionaryDao keyDictionaryDao; + @PostConstruct protected void init() { TbSqlBlockingQueueParams tsLatestParams = TbSqlBlockingQueueParams.builder() @@ -209,7 +213,7 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme TsKvLatestCompositeKey compositeKey = new TsKvLatestCompositeKey( entityId.getId(), - getOrSaveKeyId(key)); + keyDictionaryDao.getOrSaveKeyId(key)); Optional entry = tsKvLatestRepository.findById(compositeKey); if (entry.isPresent()) { TsKvLatestEntity tsKvLatestEntity = entry.get(); @@ -232,7 +236,7 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme if (ts >= query.getStartTs() && ts < query.getEndTs()) { TsKvLatestEntity latestEntity = new TsKvLatestEntity(); latestEntity.setEntityId(entityId.getId()); - latestEntity.setKey(getOrSaveKeyId(query.getKey())); + latestEntity.setKey(keyDictionaryDao.getOrSaveKeyId(query.getKey())); removedLatestFuture = service.submit(() -> { tsKvLatestRepository.delete(latestEntity); return true; @@ -259,7 +263,7 @@ public class SqlTimeseriesLatestDao extends BaseAbstractSqlTimeseriesDao impleme TsKvLatestEntity latestEntity = new TsKvLatestEntity(); latestEntity.setEntityId(entityId.getId()); latestEntity.setTs(tsKvEntry.getTs()); - latestEntity.setKey(getOrSaveKeyId(tsKvEntry.getKey())); + latestEntity.setKey(keyDictionaryDao.getOrSaveKeyId(tsKvEntry.getKey())); latestEntity.setStrValue(tsKvEntry.getStrValue().orElse(null)); latestEntity.setDoubleValue(tsKvEntry.getDoubleValue().orElse(null)); latestEntity.setLongValue(tsKvEntry.getLongValue().orElse(null)); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/dictionary/JpaKeyDictionaryDao.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/dictionary/JpaKeyDictionaryDao.java new file mode 100644 index 0000000000..3c2a4cd228 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/dictionary/JpaKeyDictionaryDao.java @@ -0,0 +1,92 @@ +/** + * Copyright © 2016-2023 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 lombok.extern.slf4j.Slf4j; +import org.hibernate.exception.ConstraintViolationException; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.dao.DataIntegrityViolationException; +import org.springframework.stereotype.Component; +import org.thingsboard.server.dao.dictionary.KeyDictionaryDao; +import org.thingsboard.server.dao.model.sqlts.dictionary.KeyDictionaryCompositeKey; +import org.thingsboard.server.dao.model.sqlts.dictionary.KeyDictionaryEntry; +import org.thingsboard.server.dao.sql.JpaAbstractDaoListeningExecutorService; +import org.thingsboard.server.dao.util.SqlDao; + +import java.util.Optional; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.locks.ReentrantLock; + +@Component +@Slf4j +@SqlDao +public class JpaKeyDictionaryDao extends JpaAbstractDaoListeningExecutorService implements KeyDictionaryDao { + + private final ConcurrentMap keyDictionaryMap = new ConcurrentHashMap<>(); + protected static final ReentrantLock creationLock = new ReentrantLock(); + + @Autowired + private KeyDictionaryRepository keyDictionaryRepository; + + @Override + public Integer getOrSaveKeyId(String strKey) { + Integer keyId = keyDictionaryMap.get(strKey); + if (keyId == null) { + Optional tsKvDictionaryOptional; + tsKvDictionaryOptional = keyDictionaryRepository.findById(new KeyDictionaryCompositeKey(strKey)); + if (tsKvDictionaryOptional.isEmpty()) { + creationLock.lock(); + try { + keyId = keyDictionaryMap.get(strKey); + if (keyId != null) { + return keyId; + } + tsKvDictionaryOptional = keyDictionaryRepository.findById(new KeyDictionaryCompositeKey(strKey)); + if (tsKvDictionaryOptional.isEmpty()) { + KeyDictionaryEntry keyDictionaryEntry = new KeyDictionaryEntry(); + keyDictionaryEntry.setKey(strKey); + try { + KeyDictionaryEntry saved = keyDictionaryRepository.save(keyDictionaryEntry); + keyDictionaryMap.put(saved.getKey(), saved.getKeyId()); + keyId = saved.getKeyId(); + } catch (DataIntegrityViolationException | ConstraintViolationException e) { + tsKvDictionaryOptional = keyDictionaryRepository.findById(new KeyDictionaryCompositeKey(strKey)); + KeyDictionaryEntry dictionary = tsKvDictionaryOptional.orElseThrow(() -> new RuntimeException("Failed to get KeyDictionaryEntry entity from DB!")); + keyDictionaryMap.put(dictionary.getKey(), dictionary.getKeyId()); + keyId = dictionary.getKeyId(); + } + } else { + keyId = tsKvDictionaryOptional.get().getKeyId(); + } + } finally { + creationLock.unlock(); + } + } else { + keyId = tsKvDictionaryOptional.get().getKeyId(); + keyDictionaryMap.put(strKey, keyId); + } + } + return keyId; + } + + @Override + public String getKey(Integer attributeKey) { + Optional byKeyId = keyDictionaryRepository.findByKeyId(attributeKey); + return byKeyId.map(KeyDictionaryEntry::getKey).orElse(null); + } + +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/sql/JpaSqlTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/sql/JpaSqlTimeseriesDao.java index 15a7e99129..4b0d4d9428 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/sql/JpaSqlTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/sql/JpaSqlTimeseriesDao.java @@ -27,6 +27,7 @@ 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.TsKvEntry; +import org.thingsboard.server.dao.dictionary.KeyDictionaryDao; import org.thingsboard.server.dao.model.sqlts.ts.TsKvEntity; import org.thingsboard.server.dao.sqlts.AbstractChunkedAggregationTimeseriesDao; import org.thingsboard.server.dao.sqlts.insert.sql.SqlPartitioningRepository; @@ -59,6 +60,8 @@ public class JpaSqlTimeseriesDao extends AbstractChunkedAggregationTimeseriesDao @Autowired private SqlPartitioningRepository partitioningRepository; + @Autowired + private KeyDictionaryDao keyDictionaryDao; private SqlTsPartitionDate tsFormat; @@ -83,7 +86,7 @@ public class JpaSqlTimeseriesDao extends AbstractChunkedAggregationTimeseriesDao int dataPointDays = getDataPointDays(tsKvEntry, computeTtl(ttl)); savePartitionIfNotExist(tsKvEntry.getTs()); String strKey = tsKvEntry.getKey(); - Integer keyId = getOrSaveKeyId(strKey); + Integer keyId = keyDictionaryDao.getOrSaveKeyId(strKey); TsKvEntity entity = new TsKvEntity(); entity.setEntityId(entityId.getId()); entity.setTs(tsKvEntry.getTs()); diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleTimeseriesDao.java index 128073323c..d7a6b45423 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/timescale/TimescaleTimeseriesDao.java @@ -33,6 +33,7 @@ import org.thingsboard.server.common.data.kv.ReadTsKvQueryResult; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.stats.StatsFactory; import org.thingsboard.server.dao.DaoUtil; +import org.thingsboard.server.dao.dictionary.KeyDictionaryDao; import org.thingsboard.server.dao.model.sql.AbstractTsKvEntity; import org.thingsboard.server.dao.model.sqlts.timescale.ts.TimescaleTsKvEntity; import org.thingsboard.server.dao.sql.TbSqlBlockingQueueParams; @@ -69,6 +70,9 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements @Autowired protected InsertTsRepository insertRepository; + @Autowired + protected KeyDictionaryDao keyDictionaryDao; + protected TbSqlBlockingQueueWrapper tsQueue; @PostConstruct @@ -108,7 +112,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements public ListenableFuture save(TenantId tenantId, EntityId entityId, TsKvEntry tsKvEntry, long ttl) { int dataPointDays = getDataPointDays(tsKvEntry, computeTtl(ttl)); String strKey = tsKvEntry.getKey(); - Integer keyId = getOrSaveKeyId(strKey); + Integer keyId = keyDictionaryDao.getOrSaveKeyId(strKey); TimescaleTsKvEntity entity = new TimescaleTsKvEntity(); entity.setEntityId(entityId.getId()); entity.setTs(tsKvEntry.getTs()); @@ -130,7 +134,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements @Override public ListenableFuture remove(TenantId tenantId, EntityId entityId, DeleteTsKvQuery query) { String strKey = query.getKey(); - Integer keyId = getOrSaveKeyId(strKey); + Integer keyId = keyDictionaryDao.getOrSaveKeyId(strKey); return service.submit(() -> { tsKvRepository.delete( entityId.getId(), @@ -161,7 +165,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements private ReadTsKvQueryResult findAllAsyncWithLimit(EntityId entityId, ReadTsKvQuery query) { String strKey = query.getKey(); - Integer keyId = getOrSaveKeyId(strKey); + Integer keyId = keyDictionaryDao.getOrSaveKeyId(strKey); List timescaleTsKvEntities = tsKvRepository.findAllWithLimit( entityId.getId(), keyId, @@ -205,7 +209,7 @@ public class TimescaleTimeseriesDao extends AbstractSqlTimeseriesDao implements } private List switchAggregation(String key, long startTs, long endTs, long timeBucket, Aggregation aggregation, UUID entityId) { - Integer keyId = getOrSaveKeyId(key); + Integer keyId = keyDictionaryDao.getOrSaveKeyId(key); switch (aggregation) { case AVG: return aggregationRepository.findAvg(entityId, keyId, timeBucket, startTs, endTs);