From 7bf7e0f994e4d69c5ca2f1f8e7464021c9310f48 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Tue, 21 Oct 2025 15:33:02 +0300 Subject: [PATCH] removed unnecessary fetch arguments methods --- ...tractCalculatedFieldProcessingService.java | 77 ++++++------------- ...faultCalculatedFieldProcessingService.java | 3 - .../service/cf/ctx/state/ArgumentEntry.java | 8 +- .../utils/CalculatedFieldArgumentUtils.java | 9 --- 4 files changed, 27 insertions(+), 70 deletions(-) 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 0c0f2f23e9..cfd7c9af3e 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 @@ -64,7 +64,6 @@ import static org.thingsboard.server.common.data.cf.configuration.geofencing.Ent import static org.thingsboard.server.common.data.cf.configuration.geofencing.EntityCoordinates.ENTITY_ID_LONGITUDE_ARGUMENT_KEY; import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultAttributeEntry; import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createDefaultKvEntry; -import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformAggSingleArgument; import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.transformSingleValueArgument; @Data @@ -98,7 +97,7 @@ public abstract class AbstractCalculatedFieldProcessingService { Map> argFutures = switch (ctx.getCfType()) { case GEOFENCING -> fetchGeofencingCalculatedFieldArguments(ctx, entityId, false, ts); case SIMPLE, SCRIPT, ALARM, PROPAGATION -> getBaseCalculatedFieldArguments(ctx, entityId, ts); - case RELATED_ENTITIES_AGGREGATION -> fetchAggArguments(ctx, entityId, ts); + case RELATED_ENTITIES_AGGREGATION -> fetchRelatedEntitiesAggArguments(ctx, entityId, ts); }; if (ctx.getCfType() == PROPAGATION) { argFutures.put(PROPAGATION_CONFIG_ARGUMENT, fetchPropagationCalculatedFieldArgument(ctx, entityId)); @@ -128,23 +127,6 @@ public abstract class AbstractCalculatedFieldProcessingService { return resolveOwnerArgument(tenantId, entityId); } - private ListenableFuture> resolveRelatedEntities(TenantId tenantId, EntityId entityId, RelationPathLevel relation) { - ListenableFuture> relationsFut = relationService.findByRelationPathQueryAsync(tenantId, new EntityRelationPathQuery(entityId, List.of(relation))); - - return Futures.transform(relationsFut, relations -> { - if (relations == null) { - return new ArrayList<>(); - } - - return switch (relation.direction()) { - case FROM -> relations.stream() - .map(EntityRelation::getTo) - .toList(); - case TO -> relations.isEmpty() ? List.of() : List.of(relations.get(0).getFrom()); - }; - }, calculatedFieldCallbackExecutor); - } - protected Map resolveArgumentFutures(Map> argFutures) { return argFutures.entrySet().stream() .collect(Collectors.toMap( @@ -189,7 +171,7 @@ public abstract class AbstractCalculatedFieldProcessingService { return argFutures; } - protected Map> fetchAggArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) { + protected Map> fetchRelatedEntitiesAggArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) { RelatedEntitiesAggregationCalculatedFieldConfiguration aggConfig = (RelatedEntitiesAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration(); ListenableFuture> relatedEntitiesFut = resolveRelatedEntities(ctx.getTenantId(), entityId, aggConfig.getRelation()); @@ -197,10 +179,27 @@ public abstract class AbstractCalculatedFieldProcessingService { return aggConfig.getArguments().entrySet().stream() .collect(Collectors.toMap( Map.Entry::getKey, - entry -> Futures.transformAsync(relatedEntitiesFut, relatedEntities -> fetchAggArgumentEntry(ctx.getTenantId(), relatedEntities, entry.getValue(), ts), MoreExecutors.directExecutor()) + entry -> Futures.transformAsync(relatedEntitiesFut, relatedEntities -> fetchRelatedEntitiesArgumentEntry(ctx.getTenantId(), relatedEntities, entry.getValue(), ts), MoreExecutors.directExecutor()) )); } + private ListenableFuture> resolveRelatedEntities(TenantId tenantId, EntityId entityId, RelationPathLevel relation) { + ListenableFuture> relationsFut = relationService.findByRelationPathQueryAsync(tenantId, new EntityRelationPathQuery(entityId, List.of(relation))); + + return Futures.transform(relationsFut, relations -> { + if (relations == null) { + return new ArrayList<>(); + } + + return switch (relation.direction()) { + case FROM -> relations.stream() + .map(EntityRelation::getTo) + .toList(); + case TO -> relations.isEmpty() ? List.of() : List.of(relations.get(0).getFrom()); + }; + }, calculatedFieldCallbackExecutor); + } + private ListenableFuture> resolveGeofencingEntityIds(TenantId tenantId, EntityId entityId, Map.Entry entry) { Argument value = entry.getValue(); if (value.getRefEntityId() != null) { @@ -252,11 +251,11 @@ public abstract class AbstractCalculatedFieldProcessingService { .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))), MoreExecutors.directExecutor()); } - public ListenableFuture fetchAggArgumentEntry(TenantId tenantId, List aggEntities, Argument argument, long startTs) { + public ListenableFuture fetchRelatedEntitiesArgumentEntry(TenantId tenantId, List aggEntities, Argument argument, long startTs) { List>> futures = aggEntities.stream() .map(entityId -> { - ListenableFuture singleAggEntryFut = fetchSingleAggArgumentEntry(tenantId, entityId, argument, startTs); - return Futures.transform(singleAggEntryFut, singleAggEntry -> Map.entry(entityId, singleAggEntry), MoreExecutors.directExecutor()); + ListenableFuture argumentEntryFut = fetchArgumentValue(tenantId, entityId, argument, startTs); + return Futures.transform(argumentEntryFut, argumentEntry -> Map.entry(entityId, ArgumentEntry.createAggSingleArgument(entityId, argumentEntry)), MoreExecutors.directExecutor()); }) .toList(); @@ -321,34 +320,4 @@ public abstract class AbstractCalculatedFieldProcessingService { return new BaseReadTsKvQuery(argument.getRefEntityKey().getKey(), startTs, endTs, 0, limit, Aggregation.NONE); } - private ListenableFuture fetchSingleAggArgumentEntry(TenantId tenantId, EntityId entityId, Argument argument, long startTs) { - return switch (argument.getRefEntityKey().getType()) { - case TS_ROLLING -> throw new IllegalStateException("TS_ROLLING is not supported for aggregation"); - case ATTRIBUTE -> fetchAttributeAggEntry(tenantId, entityId, argument, startTs); - case TS_LATEST -> fetchTsLatestAggEntry(tenantId, entityId, argument, startTs); - }; - } - - private ListenableFuture fetchAttributeAggEntry(TenantId tenantId, EntityId entityId, Argument argument, long defaultLastUpdateTs) { - log.trace("[{}][{}] Fetching attribute for key {}", tenantId, entityId, argument.getRefEntityKey()); - var attributeOptFuture = attributesService.find(tenantId, entityId, argument.getRefEntityKey().getScope(), argument.getRefEntityKey().getKey()); - return Futures.transform(attributeOptFuture, attrOpt -> { - log.debug("[{}][{}] Fetched attribute for key {}: {}", tenantId, entityId, argument.getRefEntityKey(), attrOpt); - AttributeKvEntry attributeKvEntry = attrOpt.orElseGet(() -> new BaseAttributeKvEntry(createDefaultKvEntry(argument), defaultLastUpdateTs, SingleValueArgumentEntry.DEFAULT_VERSION)); - return transformAggSingleArgument(entityId, Optional.of(attributeKvEntry)); - }, calculatedFieldCallbackExecutor); - } - - private ListenableFuture fetchTsLatestAggEntry(TenantId tenantId, EntityId entityId, Argument argument, long defaultTs) { - String key = argument.getRefEntityKey().getKey(); - log.trace("[{}][{}] Fetching latest timeseries {}", tenantId, entityId, key); - return Futures.transform( - timeseriesService.findLatest(tenantId, entityId, key), - result -> { - log.debug("[{}][{}] Fetched latest timeseries {}: {}", tenantId, entityId, key, result); - Optional tsKvEntry = result.or(() -> Optional.of(new BasicTsKvEntry(defaultTs, createDefaultKvEntry(argument), SingleValueArgumentEntry.DEFAULT_VERSION))); - return transformAggSingleArgument(entityId, tsKvEntry); - }, calculatedFieldCallbackExecutor); - } - } 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 d7957dce9b..52393d0ffe 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 @@ -15,9 +15,7 @@ */ package org.thingsboard.server.service.cf; -import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; -import com.google.common.util.concurrent.MoreExecutors; import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import org.thingsboard.server.actors.calculatedField.CalculatedFieldTelemetryMsg; @@ -56,7 +54,6 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.UUID; -import java.util.stream.Collectors; import static org.thingsboard.server.common.data.cf.configuration.PropagationCalculatedFieldConfiguration.PROPAGATION_CONFIG_ARGUMENT; import static org.thingsboard.server.utils.CalculatedFieldUtils.toProto; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java index 5bb16292be..863d6f5b50 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/ArgumentEntry.java @@ -22,8 +22,8 @@ import org.thingsboard.script.api.tbel.TbelCfArg; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.kv.KvEntry; import org.thingsboard.server.common.data.kv.TsKvEntry; -import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.aggregation.AggSingleEntityArgumentEntry; +import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.propagation.PropagationArgumentEntry; @@ -40,7 +40,7 @@ import java.util.Map; @JsonSubTypes.Type(value = TsRollingArgumentEntry.class, name = "TS_ROLLING"), @JsonSubTypes.Type(value = GeofencingArgumentEntry.class, name = "GEOFENCING"), @JsonSubTypes.Type(value = PropagationArgumentEntry.class, name = "PROPAGATION"), - @JsonSubTypes.Type(value = RelatedEntitiesArgumentEntry.class, name = "AGGREGATE_LATEST"), + @JsonSubTypes.Type(value = RelatedEntitiesArgumentEntry.class, name = "RELATED_ENTITIES"), @JsonSubTypes.Type(value = AggSingleEntityArgumentEntry.class, name = "AGGREGATE_LATEST_SINGLE") }) public interface ArgumentEntry { @@ -80,8 +80,8 @@ public interface ArgumentEntry { return new RelatedEntitiesArgumentEntry(entityIdkvEntryMap, false); } - static ArgumentEntry createAggSingleArgument(EntityId entityId, KvEntry kvEntry) { - return new AggSingleEntityArgumentEntry(entityId, kvEntry); + static ArgumentEntry createAggSingleArgument(EntityId entityId, ArgumentEntry argumentEntry) { + return new AggSingleEntityArgumentEntry(entityId, argumentEntry); } } diff --git a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java index 239ffedbc7..0c0d401688 100644 --- a/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java +++ b/application/src/main/java/org/thingsboard/server/utils/CalculatedFieldArgumentUtils.java @@ -34,7 +34,6 @@ 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.aggregation.AggSingleEntityArgumentEntry; import org.thingsboard.server.service.cf.ctx.state.aggregation.RelatedEntitiesAggregationCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.alarm.AlarmCalculatedFieldState; import org.thingsboard.server.service.cf.ctx.state.geofencing.GeofencingCalculatedFieldState; @@ -56,14 +55,6 @@ public class CalculatedFieldArgumentUtils { } } - public static ArgumentEntry transformAggSingleArgument(EntityId entityId, Optional kvEntry) { - if (kvEntry.isPresent() && kvEntry.get().getValue() != null) { - return ArgumentEntry.createAggSingleArgument(entityId, kvEntry.get()); - } else { - return new AggSingleEntityArgumentEntry(); - } - } - public static KvEntry createDefaultKvEntry(Argument argument) { String key = argument.getRefEntityKey().getKey(); String defaultValue = argument.getDefaultValue();