diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java index f2a6916751..b6e1193cfb 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java @@ -23,7 +23,6 @@ import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.apache.commons.lang3.math.NumberUtils; import org.springframework.stereotype.Service; import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg; @@ -31,21 +30,14 @@ import org.thingsboard.server.actors.calculatedField.MultipleTbCallback; import org.thingsboard.server.cluster.TbClusterService; import org.thingsboard.server.common.data.DataConstants; import org.thingsboard.server.common.data.EntityType; -import org.thingsboard.server.common.data.StringUtils; import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.OutputType; import org.thingsboard.server.common.data.id.CalculatedFieldId; 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.BaseAttributeKvEntry; import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; -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.ReadTsKvQuery; -import org.thingsboard.server.common.data.kv.StringDataEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; @@ -70,9 +62,6 @@ import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; -import org.thingsboard.server.service.cf.ctx.state.ScriptCalculatedFieldState; -import org.thingsboard.server.service.cf.ctx.state.SimpleCalculatedFieldState; -import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.TsRollingArgumentEntry; import java.util.ArrayList; @@ -80,12 +69,15 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Map.Entry; -import java.util.Optional; import java.util.UUID; import java.util.concurrent.ExecutionException; import java.util.stream.Collectors; import static org.thingsboard.server.common.data.DataConstants.SCOPE; +import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultAttributeEntry; +import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultTsKvEntry; +import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createStateByType; +import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformSingleValueArgument; import static org.thingsboard.server.utils.CalculatedFieldUtils.toProto; @TbRuleEngineComponent @@ -244,30 +236,17 @@ public class DefaultCalculatedFieldProcessingService implements CalculatedFieldP private ListenableFuture fetchKvEntry(TenantId tenantId, EntityId entityId, Argument argument) { return switch (argument.getRefEntityKey().getType()) { case TS_ROLLING -> fetchTsRolling(tenantId, entityId, argument); - case ATTRIBUTE -> transformSingleValueArgument( - Futures.transform( - attributesService.find(tenantId, entityId, argument.getRefEntityKey().getScope(), argument.getRefEntityKey().getKey()), - result -> result.or(() -> Optional.of(new BaseAttributeKvEntry(createDefaultKvEntry(argument), System.currentTimeMillis(), 0L))), - calculatedFieldCallbackExecutor) - ); - case TS_LATEST -> transformSingleValueArgument( - Futures.transform( - timeseriesService.findLatest(tenantId, entityId, argument.getRefEntityKey().getKey()), - result -> result.or(() -> Optional.of(new BasicTsKvEntry(System.currentTimeMillis(), createDefaultKvEntry(argument), 0L))), - calculatedFieldCallbackExecutor)); + case ATTRIBUTE -> Futures.transform( + attributesService.find(tenantId, entityId, argument.getRefEntityKey().getScope(), argument.getRefEntityKey().getKey()), + result -> transformSingleValueArgument(result.orElseGet(() -> createDefaultAttributeEntry(argument, System.currentTimeMillis()))), + calculatedFieldCallbackExecutor); + case TS_LATEST -> Futures.transform( + timeseriesService.findLatest(tenantId, entityId, argument.getRefEntityKey().getKey()), + result -> transformSingleValueArgument(result.orElseGet(() -> createDefaultTsKvEntry(argument, System.currentTimeMillis()))), + calculatedFieldCallbackExecutor); }; } - private ListenableFuture transformSingleValueArgument(ListenableFuture> kvEntryFuture) { - return Futures.transform(kvEntryFuture, kvEntry -> { - if (kvEntry.isPresent() && kvEntry.get().getValue() != null) { - return ArgumentEntry.createSingleValueArgument(kvEntry.get()); - } else { - return new SingleValueArgumentEntry(); - } - }, calculatedFieldCallbackExecutor); - } - private ListenableFuture fetchTsRolling(TenantId tenantId, EntityId entityId, Argument argument) { long currentTime = System.currentTimeMillis(); long timeWindow = argument.getTimeWindow() == 0 ? System.currentTimeMillis() : argument.getTimeWindow(); @@ -282,28 +261,6 @@ public class DefaultCalculatedFieldProcessingService implements CalculatedFieldP return Futures.transform(tsRollingFuture, tsRolling -> tsRolling == null ? new TsRollingArgumentEntry(limit, timeWindow) : ArgumentEntry.createTsRollingArgument(tsRolling, limit, timeWindow), calculatedFieldCallbackExecutor); } - private KvEntry createDefaultKvEntry(Argument argument) { - String key = argument.getRefEntityKey().getKey(); - String defaultValue = argument.getDefaultValue(); - if (StringUtils.isBlank(defaultValue)) { - return new StringDataEntry(key, null); - } - if (NumberUtils.isParsable(defaultValue)) { - return new DoubleDataEntry(key, Double.parseDouble(defaultValue)); - } - if ("true".equalsIgnoreCase(defaultValue) || "false".equalsIgnoreCase(defaultValue)) { - return new BooleanDataEntry(key, Boolean.parseBoolean(defaultValue)); - } - return new StringDataEntry(key, defaultValue); - } - - private CalculatedFieldState createStateByType(CalculatedFieldCtx ctx) { - return switch (ctx.getCfType()) { - case SIMPLE -> new SimpleCalculatedFieldState(ctx.getArgNames()); - case SCRIPT -> new ScriptCalculatedFieldState(ctx.getArgNames()); - }; - } - private static class TbCallbackWrapper implements TbQueueCallback { private final TbCallback callback; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java index 94a18256a3..14370aa68c 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java @@ -37,15 +37,13 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { protected Map arguments; protected boolean sizeExceedsLimit; - protected long latestTimestamp = DEFAULT_LAST_UPDATE_TS; - public BaseCalculatedFieldState(List requiredArguments) { this.requiredArguments = requiredArguments; this.arguments = new HashMap<>(); } public BaseCalculatedFieldState() { - this(new ArrayList<>(), new HashMap<>(), false, DEFAULT_LAST_UPDATE_TS); + this(new ArrayList<>(), new HashMap<>(), false); } @Override @@ -75,7 +73,6 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { if (entryUpdated) { stateUpdated = true; - updateLastUpdateTimestamp(newEntry); } } @@ -111,15 +108,29 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState { protected abstract void validateNewEntry(ArgumentEntry newEntry); - private void updateLastUpdateTimestamp(ArgumentEntry entry) { - long newTs = this.latestTimestamp; - if (entry instanceof SingleValueArgumentEntry singleValueArgumentEntry) { - newTs = singleValueArgumentEntry.getTs(); - } else if (entry instanceof TsRollingArgumentEntry tsRollingArgumentEntry) { - Map.Entry lastEntry = tsRollingArgumentEntry.getTsRecords().lastEntry(); - newTs = (lastEntry != null) ? lastEntry.getKey() : DEFAULT_LAST_UPDATE_TS; + public long getLatestTimestamp() { + long latestTs = DEFAULT_LAST_UPDATE_TS; + + boolean allDefault = arguments.values().stream().allMatch(entry -> { + if (entry instanceof SingleValueArgumentEntry single) { + return single.isDefaultValue(); + } + return false; + }); + + for (ArgumentEntry entry : arguments.values()) { + if (entry instanceof SingleValueArgumentEntry single) { + if (allDefault) { + latestTs = Math.max(latestTs, single.getTs()); + } else if (!single.isDefaultValue()) { + latestTs = Math.max(latestTs, single.getTs()); + } + } else if (entry instanceof TsRollingArgumentEntry rolling) { + latestTs = Math.max(latestTs, rolling.getLatestTs()); + } } - this.latestTimestamp = Math.max(this.latestTimestamp, newTs); + + return latestTs; } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java index 0997fd6cbb..2f9a7de940 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/SingleValueArgumentEntry.java @@ -19,7 +19,6 @@ import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.core.type.TypeReference; import lombok.AllArgsConstructor; import lombok.Data; -import lombok.NoArgsConstructor; import org.thingsboard.common.util.JacksonUtil; import org.thingsboard.script.api.tbel.TbelCfArg; import org.thingsboard.script.api.tbel.TbelCfSingleValueArg; @@ -120,7 +119,7 @@ public class SingleValueArgumentEntry implements ArgumentEntry { @Override public boolean updateEntry(ArgumentEntry entry) { if (entry instanceof SingleValueArgumentEntry singleValueEntry) { - if (singleValueEntry.getTs() <= this.ts) { + if (singleValueEntry.getTs() < this.ts) { return false; } @@ -136,4 +135,9 @@ public class SingleValueArgumentEntry implements ArgumentEntry { } return false; } + + public boolean isDefaultValue() { + return DEFAULT_VERSION.equals(this.version); + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java index b5a680a072..e01d8b7369 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/TsRollingArgumentEntry.java @@ -31,6 +31,8 @@ import java.util.List; import java.util.Map; import java.util.TreeMap; +import static org.thingsboard.server.service.cf.ctx.state.BaseCalculatedFieldState.DEFAULT_LAST_UPDATE_TS; + @Data @NoArgsConstructor @AllArgsConstructor @@ -83,6 +85,11 @@ public class TsRollingArgumentEntry implements ArgumentEntry { return tsRecords; } + public long getLatestTs() { + var lastEntry = tsRecords.lastEntry(); + return (lastEntry != null) ? lastEntry.getKey() : DEFAULT_LAST_UPDATE_TS; + } + @Override public TbelCfArg toTbelCfArg() { List values = new ArrayList<>(tsRecords.size()); diff --git a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java new file mode 100644 index 0000000000..1296df6e87 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java @@ -0,0 +1,75 @@ +/** + * Copyright © 2016-2025 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.utils; + +import lombok.NonNull; +import org.apache.commons.lang3.math.NumberUtils; +import org.thingsboard.server.common.data.StringUtils; +import org.thingsboard.server.common.data.cf.configuration.Argument; +import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry; +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.StringDataEntry; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry; +import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx; +import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState; +import org.thingsboard.server.service.cf.ctx.state.ScriptCalculatedFieldState; +import org.thingsboard.server.service.cf.ctx.state.SimpleCalculatedFieldState; +import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry; + +import static org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry.DEFAULT_VERSION; + +public class CalculatedFieldArgumentUtils { + + public static ArgumentEntry transformSingleValueArgument(@NonNull KvEntry kvEntry) { + return kvEntry.getValue() != null ? ArgumentEntry.createSingleValueArgument(kvEntry) : new SingleValueArgumentEntry(); + } + + public static TsKvEntry createDefaultTsKvEntry(Argument argument, long ts) { + return new BasicTsKvEntry(ts, createDefaultKvEntry(argument), DEFAULT_VERSION); + } + + public static AttributeKvEntry createDefaultAttributeEntry(Argument argument, long ts) { + return new BaseAttributeKvEntry(createDefaultKvEntry(argument), ts, DEFAULT_VERSION); + } + + private static KvEntry createDefaultKvEntry(Argument argument) { + String key = argument.getRefEntityKey().getKey(); + String defaultValue = argument.getDefaultValue(); + if (StringUtils.isBlank(defaultValue)) { + return new StringDataEntry(key, null); + } + if (NumberUtils.isParsable(defaultValue)) { + return new DoubleDataEntry(key, Double.parseDouble(defaultValue)); + } + if ("true".equalsIgnoreCase(defaultValue) || "false".equalsIgnoreCase(defaultValue)) { + return new BooleanDataEntry(key, Boolean.parseBoolean(defaultValue)); + } + return new StringDataEntry(key, defaultValue); + } + + public static CalculatedFieldState createStateByType(CalculatedFieldCtx ctx) { + return switch (ctx.getCfType()) { + case SIMPLE -> new SimpleCalculatedFieldState(ctx.getArgNames()); + case SCRIPT -> new ScriptCalculatedFieldState(ctx.getArgNames()); + }; + } + +} diff --git a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java index b500f95d45..c83e8c5ee3 100644 --- a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java @@ -45,6 +45,7 @@ import java.util.concurrent.TimeUnit; import static org.assertj.core.api.Assertions.assertThat; import static org.awaitility.Awaitility.await; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; @DaoSqlTest public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTest { @@ -570,6 +571,7 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes @Test public void testScriptCalculatedFieldWhenUsedLatestTsInScript() throws Exception { Device testDevice = createDevice("Test device", "1234567890"); + long ts = System.currentTimeMillis() - 300000L; doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode(String.format("{\"ts\": %s, \"values\": {\"temperature\":30}}", ts))); @@ -606,6 +608,91 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes }); } + @Test + public void testSimpleCalculatedFieldWhenUseLatestTsIsTrueAndDefaultArguments() throws Exception { + Device testDevice = createDevice("Test device", "1234567890"); + + CalculatedField calculatedField = new CalculatedField(); + calculatedField.setEntityId(testDevice.getId()); + calculatedField.setType(CalculatedFieldType.SIMPLE); + calculatedField.setName("a + b + c"); + calculatedField.setDebugSettings(DebugSettings.all()); + calculatedField.setConfigurationVersion(1); + + SimpleCalculatedFieldConfiguration config = new SimpleCalculatedFieldConfiguration(); + + Argument argument1 = new Argument(); + ReferencedEntityKey refEntityKey1 = new ReferencedEntityKey("a", ArgumentType.TS_LATEST, null); + argument1.setRefEntityKey(refEntityKey1); + argument1.setDefaultValue("100"); + Argument argument2 = new Argument(); + ReferencedEntityKey refEntityKey2 = new ReferencedEntityKey("b", ArgumentType.TS_LATEST, null); + argument2.setRefEntityKey(refEntityKey2); + argument2.setDefaultValue("200"); + Argument argument3 = new Argument(); + ReferencedEntityKey refEntityKey3 = new ReferencedEntityKey("c", ArgumentType.TS_LATEST, null); + argument3.setRefEntityKey(refEntityKey3); + argument3.setDefaultValue("300"); + config.setArguments(Map.of("a", argument1, "b", argument2, "c", argument3)); + config.setExpression("a + b + c"); + + Output output = new Output(); + output.setName("d"); + output.setType(OutputType.TIME_SERIES); + output.setDecimalsByDefault(0); + config.setOutput(output); + + config.setUseLatestTs(true); + + calculatedField.setConfiguration(config); + + CalculatedField savedCalculatedField = doPost("/api/calculatedField", calculatedField, CalculatedField.class); + + await().alias("create CF -> perform initial calculation with default arguments").atMost(TIMEOUT, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + ObjectNode d = getLatestTelemetry(testDevice.getId(), "d"); + assertThat(d).isNotNull(); + assertThat(d.get("d").get(0).get("value").asText()).isEqualTo("600"); + }); + + doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"a\":10}")); + + await().alias("update telemetry -> save result with ts of 'a' argument").atMost(TIMEOUT, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + ObjectNode keys = getLatestTelemetry(testDevice.getId(), "d", "a"); + assertThat(keys).isNotNull(); + String aTs = keys.get("a").get(0).get("ts").asText(); + assertThat(keys.get("d").get(0).get("ts").asText()).isEqualTo(aTs); + assertThat(keys.get("d").get(0).get("value").asText()).isEqualTo("510"); + }); + + doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"b\":20}")); + doPost("/api/plugins/telemetry/DEVICE/" + testDevice.getUuidId() + "/timeseries/" + DataConstants.SERVER_SCOPE, JacksonUtil.toJsonNode("{\"c\":30}")); + + await().alias("update telemetry -> save result with latest ts of updated arguments").atMost(TIMEOUT, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + ObjectNode keys = getLatestTelemetry(testDevice.getId(), "d"); + assertThat(keys).isNotNull(); + assertThat(keys.get("d").get(0).get("value").asText()).isEqualTo("60"); + }); + + String latestTs = getLatestTelemetry(testDevice.getId(), "d").get("d").get(0).get("ts").asText(); + + doDelete("/api/plugins/telemetry/DEVICE/" + testDevice.getId() + "/timeseries/delete?keys=b&deleteAllDataForKeys=true").andExpect(status().isOk()); + + await().alias("delete telemetry -> save result with previous latest ts and default argument").atMost(TIMEOUT, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + ObjectNode keys = getLatestTelemetry(testDevice.getId(), "d"); + assertThat(keys).isNotNull(); + assertThat(keys.get("d").get(0).get("ts").asText()).isEqualTo(latestTs); + assertThat(keys.get("d").get(0).get("value").asText()).isEqualTo("240"); + }); + } + @Test public void testSimpleCalculatedFieldWhenCtxBecameUninitialized() throws Exception { Device testDevice = createDevice("Test device", "1234567890"); diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sqlts/dictionary/KeyDictionaryCompositeKey.java b/dao/src/main/java/org/thingsboard/server/dao/model/sqlts/dictionary/KeyDictionaryCompositeKey.java index 4f3285b9bf..00e49ea703 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/sqlts/dictionary/KeyDictionaryCompositeKey.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sqlts/dictionary/KeyDictionaryCompositeKey.java @@ -25,7 +25,7 @@ import java.io.Serializable; @Data @NoArgsConstructor @AllArgsConstructor -public class KeyDictionaryCompositeKey implements Serializable{ +public class KeyDictionaryCompositeKey implements Serializable { @Transient private static final long serialVersionUID = -4089175869616037523L; diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sqlts/dictionary/KeyDictionaryEntry.java b/dao/src/main/java/org/thingsboard/server/dao/model/sqlts/dictionary/KeyDictionaryEntry.java index a95c7a2bc6..8365e8facc 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/sqlts/dictionary/KeyDictionaryEntry.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sqlts/dictionary/KeyDictionaryEntry.java @@ -36,8 +36,7 @@ public final class KeyDictionaryEntry { @Column(name = KEY_COLUMN) private String key; - @Column(name = KEY_ID_COLUMN, unique = true, columnDefinition = "int") - @Generated - private int keyId; + @Column(name = KEY_ID_COLUMN, unique = true, columnDefinition = "int", insertable = false, updatable = false) + private Integer keyId; -} \ No newline at end of file +} 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 index c14c069f23..46bc10010e 100644 --- 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 @@ -17,8 +17,6 @@ package org.thingsboard.server.dao.sqlts.dictionary; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; -import org.hibernate.exception.ConstraintViolationException; -import org.springframework.dao.DataIntegrityViolationException; import org.springframework.stereotype.Component; import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; @@ -48,43 +46,34 @@ public class JpaKeyDictionaryDao extends JpaAbstractDaoListeningExecutorService @Transactional(propagation = Propagation.NOT_SUPPORTED) @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); + Integer cached = keyDictionaryMap.get(strKey); + if (cached != null) { + return cached; + } + var compositeKey = new KeyDictionaryCompositeKey(strKey); + Optional existingId = keyDictionaryRepository.findById(compositeKey).map(KeyDictionaryEntry::getKeyId); + if (existingId.isPresent()) { + return cacheAndReturn(strKey, existingId.get()); + } + creationLock.lock(); + try { + Integer fromCache = keyDictionaryMap.get(strKey); + if (fromCache != null) { + return fromCache; } + Integer keyId = keyDictionaryRepository.upsertAndGetKeyId(strKey); + if (keyId != null) { + return cacheAndReturn(strKey, keyId); + } + log.warn("upsertAndGetKeyId returned: [{}] for key: [{}], falling back to findById", keyId, strKey); + keyId = keyDictionaryRepository.findById(compositeKey) + .map(KeyDictionaryEntry::getKeyId) + .orElseThrow(() -> new IllegalStateException( + "Failed to resolve keyId for string key: " + strKey + " after fallback.")); + return cacheAndReturn(strKey, keyId); + } finally { + creationLock.unlock(); } - return keyId; } @Override @@ -98,4 +87,9 @@ public class JpaKeyDictionaryDao extends JpaAbstractDaoListeningExecutorService return DaoUtil.pageToPageData(keyDictionaryRepository.findAll(DaoUtil.toPageable(pageLink))); } + private Integer cacheAndReturn(String key, Integer keyId) { + keyDictionaryMap.put(key, keyId); + return keyId; + } + } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sqlts/dictionary/KeyDictionaryRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sqlts/dictionary/KeyDictionaryRepository.java index d264cd9966..e836cedb19 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sqlts/dictionary/KeyDictionaryRepository.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sqlts/dictionary/KeyDictionaryRepository.java @@ -19,6 +19,7 @@ import org.springframework.data.domain.Page; import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Query; +import org.springframework.data.repository.query.Param; import org.thingsboard.server.dao.model.sqlts.dictionary.KeyDictionaryCompositeKey; import org.thingsboard.server.dao.model.sqlts.dictionary.KeyDictionaryEntry; @@ -31,4 +32,7 @@ public interface KeyDictionaryRepository extends JpaRepository findAll(Pageable pageable); -} \ No newline at end of file + @Query(value = "INSERT INTO key_dictionary (key) VALUES (:key) ON CONFLICT (key) DO UPDATE SET key = EXCLUDED.key RETURNING key_id", nativeQuery = true) + Integer upsertAndGetKeyId(@Param("key") String key); + +} diff --git a/dao/src/test/java/org/thingsboard/server/dao/sqlts/dictionary/KeyDictionaryDaoTest.java b/dao/src/test/java/org/thingsboard/server/dao/sqlts/dictionary/KeyDictionaryDaoTest.java new file mode 100644 index 0000000000..11c14c99f4 --- /dev/null +++ b/dao/src/test/java/org/thingsboard/server/dao/sqlts/dictionary/KeyDictionaryDaoTest.java @@ -0,0 +1,111 @@ +/** + * Copyright © 2016-2025 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.junit.Test; +import org.springframework.beans.factory.annotation.Autowired; +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.service.AbstractServiceTest; +import org.thingsboard.server.dao.service.DaoSqlTest; + +import java.util.Arrays; +import java.util.Optional; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; + +@DaoSqlTest +public class KeyDictionaryDaoTest extends AbstractServiceTest { + + @Autowired + private KeyDictionaryDao keyDictionaryDao; + + @Autowired + private KeyDictionaryRepository keyDictionaryRepository; + + private static final String KEY = "testKeyDictionaryDaoTestKey"; + + @Test + public void testGetOrSaveKeyId_concurrent() throws Exception { + int threads = 8; + ExecutorService executor = Executors.newFixedThreadPool(threads); + + CountDownLatch allReady = new CountDownLatch(threads); + CountDownLatch start = new CountDownLatch(1); + CountDownLatch allDone = new CountDownLatch(threads); + + Integer[] keyIds = new Integer[threads]; + + try { + for (int i = 0; i < threads; i++) { + final int idx = i; + executor.submit(() -> { + allReady.countDown(); + try { + // wait until all threads are ready + start.await(); + // concurrent call + Integer id = keyDictionaryDao.getOrSaveKeyId(KEY); + keyIds[idx] = id; + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + allDone.countDown(); + } + }); + } + + // ensure all threads are queued + allReady.await(5, TimeUnit.SECONDS); + // fire the start gun + start.countDown(); + // wait for all to finish + allDone.await(10, TimeUnit.SECONDS); + } finally { + executor.shutdownNow(); + } + + // basic sanity + for (int i = 0; i < threads; i++) { + assertThat(keyIds[i]) + .as("keyId[%s]", i) + .isNotNull() + .isGreaterThan(0); + } + + // all threads must see the same keyId + int first = keyIds[0]; + assertThat(first).isGreaterThan(0); + assertThat(Arrays.stream(keyIds).distinct().count()) + .as("all threads should get the same keyId") + .isEqualTo(1); + + // DB must have exactly one row for this key and the same id + KeyDictionaryCompositeKey id = new KeyDictionaryCompositeKey(KEY); + Optional entry = keyDictionaryRepository.findById(id); + + assertThat(entry.isPresent()).isTrue(); + assertThat(entry.get().getKeyId()).isEqualTo(first); + + keyDictionaryRepository.deleteById(id); + } + +}