From 76606f63ba1036b5577dd9acdd60a12c8b0b5a23 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Mon, 3 Nov 2025 12:14:44 +0200 Subject: [PATCH] added upgrade script --- .../main/data/upgrade/basic/schema_update.sql | 19 +++++ ...tractCalculatedFieldProcessingService.java | 85 ++++++++++++++++--- ...faultCalculatedFieldProcessingService.java | 2 +- .../cf/CalculatedFieldIntegrationTest.java | 2 +- ...AttributeSkipRuleEngineOutputStrategy.java | 4 + ...imeSeriesSkipRuleEngineOutputStrategy.java | 5 ++ 6 files changed, 101 insertions(+), 16 deletions(-) diff --git a/application/src/main/data/upgrade/basic/schema_update.sql b/application/src/main/data/upgrade/basic/schema_update.sql index fe79fce3a2..cc4fbaad14 100644 --- a/application/src/main/data/upgrade/basic/schema_update.sql +++ b/application/src/main/data/upgrade/basic/schema_update.sql @@ -69,3 +69,22 @@ ALTER TABLE calculated_field DROP CONSTRAINT IF EXISTS calculated_field_unq_key; ALTER TABLE calculated_field ADD CONSTRAINT calculated_field_unq_key UNIQUE (entity_id, type, name); -- CALCULATED FIELD UNIQUE CONSTRAINT UPDATE END + +-- CALCULATED FIELD OUTPUT STRATEGY UPGRADE START + +UPDATE calculated_field +SET configuration = jsonb_set( + configuration::jsonb, + '{output}', + (configuration::jsonb -> 'output') + || jsonb_build_object( + 'strategy', + jsonb_build_object( + 'type', 'PUSH_TO_RULE_ENGINE' + ) + ), + false + ) +WHERE (configuration::jsonb -> 'output' -> 'strategy') IS NULL; + +-- CALCULATED FIELD OUTPUT STRATEGY UPGRADE END diff --git a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java index 455093a116..f9c5d6db78 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java @@ -27,12 +27,15 @@ import jakarta.annotation.PostConstruct; import jakarta.annotation.PreDestroy; import lombok.Data; import lombok.extern.slf4j.Slf4j; +import org.thingsboard.common.util.DonAsynchron; import org.thingsboard.common.util.ThingsBoardExecutors; import org.thingsboard.rule.engine.api.AttributesSaveRequest; +import org.thingsboard.rule.engine.api.AttributesSaveRequest.Strategy; import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.adaptor.JsonConverter; import org.thingsboard.server.common.data.cf.configuration.Argument; import org.thingsboard.server.common.data.cf.configuration.ArgumentType; +import org.thingsboard.server.common.data.cf.configuration.AttributeSkipRuleEngineOutputStrategy; import org.thingsboard.server.common.data.cf.configuration.OutputType; import org.thingsboard.server.common.data.cf.configuration.RelationPathQueryDynamicSourceConfiguration; import org.thingsboard.server.common.data.cf.configuration.TimeSeriesSkipRuleEngineOutputStrategy; @@ -71,6 +74,7 @@ import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutionException; +import java.util.function.Function; import java.util.stream.Collectors; import static org.thingsboard.server.common.data.cf.CalculatedFieldType.PROPAGATION; @@ -344,8 +348,8 @@ public abstract class AbstractCalculatedFieldProcessingService { SettableFuture future = SettableFuture.create(); switch (type) { - case ATTRIBUTES -> saveAttributes(tenantId, entityId, jsonResult, cfIds, future); - case TIME_SERIES -> saveTimeSeries(tenantId, entityId, jsonResult, ((TimeSeriesSkipRuleEngineOutputStrategy) cfResult.getOutputStrategy()).getTtl(), cfIds, System.currentTimeMillis(), TimeseriesSaveRequest.Strategy.PROCESS_ALL, future); + case ATTRIBUTES -> saveAttributes(tenantId, entityId, cfResult, cfIds, future); + case TIME_SERIES -> saveTimeSeries(tenantId, entityId, cfResult, cfIds, System.currentTimeMillis(), future); } if (log.isTraceEnabled()) { @@ -365,20 +369,54 @@ public abstract class AbstractCalculatedFieldProcessingService { } } - private void saveAttributes(TenantId tenantId, EntityId entityId, JsonElement jsonResult, List cfIds, SettableFuture future) { + private void saveAttributes(TenantId tenantId, EntityId entityId, TelemetryCalculatedFieldResult cfResult, List cfIds, SettableFuture future) { + if (!(cfResult.getOutputStrategy() instanceof AttributeSkipRuleEngineOutputStrategy outputStrategy)) { + return; + } + JsonElement jsonResult = JsonParser.parseString(Objects.requireNonNull(cfResult.stringValue())); + + AttributesSaveRequest.Strategy strategy = new Strategy(outputStrategy.isSaveAttribute(), outputStrategy.isSendWsUpdate(), outputStrategy.isProcessCfs()); List attributeKvEntries = JsonConverter.convertToAttributes(jsonResult); - tsSubService.saveAttributesInternal(AttributesSaveRequest.builder() - .tenantId(tenantId) - .entityId(entityId) - .entries(attributeKvEntries) - .strategy(AttributesSaveRequest.Strategy.PROCESS_ALL) - .previousCalculatedFieldIds(cfIds) - .future(future) - .build() - ); + + if (!outputStrategy.isUpdateAttributesOnlyOnValueChange()) { + tsSubService.saveAttributesInternal(AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(entityId) + .entries(attributeKvEntries) + .strategy(strategy) + .previousCalculatedFieldIds(cfIds) + .future(future) + .build() + ); + return; + } + + List keys = attributeKvEntries.stream().map(KvEntry::getKey).collect(Collectors.toList()); + + ListenableFuture> findFuture = attributesService.find(tenantId, entityId, cfResult.getScope(), keys); + + DonAsynchron.withCallback(findFuture, + existingAttributes -> { + List attributesChanged = filterChangedAttr(existingAttributes, attributeKvEntries); + tsSubService.saveAttributesInternal(AttributesSaveRequest.builder() + .tenantId(tenantId) + .entityId(entityId) + .entries(attributesChanged) + .strategy(strategy) + .previousCalculatedFieldIds(cfIds) + .future(future) + .build() + ); + }, + future::setException, + MoreExecutors.directExecutor()); } - private void saveTimeSeries(TenantId tenantId, EntityId entityId, JsonElement jsonResult, Long ttl, List cfIds, long ts, TimeseriesSaveRequest.Strategy strategy, SettableFuture future) { + private void saveTimeSeries(TenantId tenantId, EntityId entityId, TelemetryCalculatedFieldResult cfResult, List cfIds, long ts, SettableFuture future) { + if (!(cfResult.getOutputStrategy() instanceof TimeSeriesSkipRuleEngineOutputStrategy outputStrategy)) { + return; + } + JsonElement jsonResult = JsonParser.parseString(Objects.requireNonNull(cfResult.stringValue())); Map> tsKvMap = JsonConverter.convertToTelemetry(jsonResult, ts); List tsEntries = new ArrayList<>(); for (Map.Entry> tsKvEntry : tsKvMap.entrySet()) { @@ -386,15 +424,34 @@ public abstract class AbstractCalculatedFieldProcessingService { tsEntries.add(new BasicTsKvEntry(tsKvEntry.getKey(), kvEntry)); } } + TimeseriesSaveRequest.Strategy strategy = new TimeseriesSaveRequest.Strategy(outputStrategy.isSaveTimeSeries(), outputStrategy.isSaveLatest(), outputStrategy.isSendWsUpdate(), outputStrategy.isProcessCfs()); tsSubService.saveTimeseriesInternal(TimeseriesSaveRequest.builder() .tenantId(tenantId) .entityId(entityId) .entries(tsEntries) - .ttl(ttl) + .ttl(outputStrategy.getTtl()) .strategy(strategy) .previousCalculatedFieldIds(cfIds) .future(future) .build()); } + private List filterChangedAttr(List existingAttributes, List newAttributes) { + if (existingAttributes == null || existingAttributes.isEmpty()) { + return newAttributes; + } + + Map currentAttrMap = existingAttributes.stream() + .collect(Collectors.toMap(AttributeKvEntry::getKey, Function.identity(), (existing, replacement) -> existing)); + + return newAttributes.stream() + .filter(item -> { + AttributeKvEntry cacheAttr = currentAttrMap.get(item.getKey()); + return cacheAttr == null + || !Objects.equals(item.getValue(), cacheAttr.getValue()) //JSON and String can be equals by value, but different by type + || !Objects.equals(item.getDataType(), cacheAttr.getDataType()); + }) + .collect(Collectors.toList()); + } + } 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 3cec856746..a6058c7dcb 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 @@ -17,7 +17,7 @@ package org.thingsboard.server.service.cf; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; -import org.apache.logging.log4j.util.TriConsumer; +import org.apache.commons.lang3.function.TriConsumer; import org.springframework.stereotype.Service; import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg; import org.thingsboard.server.actors.calculatedField.MultipleTbCallback; 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 874d5c8cd6..6a76a2fb99 100644 --- a/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/CalculatedFieldIntegrationTest.java @@ -1235,7 +1235,7 @@ public class CalculatedFieldIntegrationTest extends CalculatedFieldControllerTes output.setName("fahrenheitTemp"); output.setType(OutputType.TIME_SERIES); output.setDecimalsByDefault(1); - output.setStrategy(new TimeSeriesSkipRuleEngineOutputStrategy(1000L)); + output.setStrategy(new TimeSeriesSkipRuleEngineOutputStrategy(1000L, true, true, true, true)); config.setOutput(output); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AttributeSkipRuleEngineOutputStrategy.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AttributeSkipRuleEngineOutputStrategy.java index b796042de3..5d30eb36f9 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AttributeSkipRuleEngineOutputStrategy.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/AttributeSkipRuleEngineOutputStrategy.java @@ -26,4 +26,8 @@ public class AttributeSkipRuleEngineOutputStrategy extends SkipRuleEngineOutputS private boolean updateAttributesOnlyOnValueChange; + private boolean saveAttribute; + private boolean sendWsUpdate; + private boolean processCfs; + } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/TimeSeriesSkipRuleEngineOutputStrategy.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/TimeSeriesSkipRuleEngineOutputStrategy.java index 27cc561035..3165fc5eac 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/TimeSeriesSkipRuleEngineOutputStrategy.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/TimeSeriesSkipRuleEngineOutputStrategy.java @@ -26,4 +26,9 @@ public class TimeSeriesSkipRuleEngineOutputStrategy extends SkipRuleEngineOutput private long ttl; + private boolean saveTimeSeries; + private boolean saveLatest; + private boolean sendWsUpdate; + private boolean processCfs; + }