From b22b556ceca937e5ff0517d3f90e84d55f8847d2 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Fri, 7 Nov 2025 09:44:30 +0200 Subject: [PATCH 1/3] fixed relation update event and added checks for entity type in relation for CFs processing --- ...alculatedFieldManagerMessageProcessor.java | 41 +++++++++++-- ...tractCalculatedFieldProcessingService.java | 9 ++- .../DefaultCalculatedFieldQueueService.java | 2 +- .../entitiy/EntityStateSourcingListener.java | 13 +++-- ...ntitiesAggregationCalculatedFieldTest.java | 58 +++++++++++++++++++ .../server/dao/relation/RelationService.java | 2 + .../common/data/cf/CalculatedField.java | 4 ++ .../dao/relation/BaseRelationService.java | 10 ++++ 8 files changed, 126 insertions(+), 13 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java index 8f632647c5..a86ddde2a1 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java @@ -27,9 +27,11 @@ import org.thingsboard.server.actors.service.DefaultActorService; import org.thingsboard.server.actors.shared.AbstractContextAwareMsgProcessor; import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.DataConstants; +import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.ProfileEntityIdInfo; import org.thingsboard.server.common.data.alarm.Alarm; +import org.thingsboard.server.common.data.asset.AssetProfile; import org.thingsboard.server.common.data.audit.ActionType; import org.thingsboard.server.common.data.cf.CalculatedField; import org.thingsboard.server.common.data.cf.CalculatedFieldLink; @@ -74,6 +76,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.Set; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.ScheduledFuture; @@ -292,17 +295,33 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) { + // 2 = 1 for entity processing + 1 for relation processing + MultipleTbCallback multiCallback = new MultipleTbCallback(2, callback); + deleteEntityRelations(msg, multiCallback); switch (msg.getEntityId().getEntityType()) { case DEVICE, ASSET -> entityProfileCache.removeEntityId(msg.getEntityId()); case CUSTOMER -> ownerEntities.remove(msg.getEntityId()); } ownerEntities.values().forEach(entities -> entities.remove(msg.getEntityId())); - if (isMyPartition(msg.getEntityId(), callback)) { + if (isMyPartition(msg.getEntityId(), multiCallback)) { log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId()); - getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); + getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), multiCallback)); } } + private void deleteEntityRelations(ComponentLifecycleMsg msg, TbCallback callback) { + List entityRelations = relationService.findEntityRelations(tenantId, msg.getEntityId()); + if (entityRelations.isEmpty()) { + callback.onSuccess(); + } + entityRelations.forEach(entityRelation -> { + Function> deleteAction = + relatedId -> (entityId, ctx, cb) -> deleteRelatedEntity(entityId, relatedId, ctx, cb); + + processRelationIfSupported(entityRelation, callback, deleteAction); + }); + } + private void onRelationChangedEvent(ComponentLifecycleMsg msg, TbCallback callback) { Function> relationAction = switch (msg.getEvent()) { case RELATION_UPDATED -> relatedId -> (entityId, ctx, cb) -> initRelatedEntity(entityId, relatedId, ctx, cb); @@ -316,16 +335,26 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } EntityRelation entityRelation = JacksonUtil.treeToValue(msg.getInfo(), EntityRelation.class); + processRelationIfSupported(entityRelation, callback, relationAction); + } + + private void processRelationIfSupported(EntityRelation entityRelation, + TbCallback callback, + Function> relationAction) { EntityId toId = entityRelation.getTo(); EntityId fromId = entityRelation.getFrom(); String relationType = entityRelation.getType(); + if (!(CalculatedField.isSupportedRefEntity(toId) || CalculatedField.isSupportedRefEntity(fromId))) { + callback.onSuccess(); + return; + } + MultipleTbCallback callbackForToAndFrom = new MultipleTbCallback(2, callback); processRelationByDirection(EntitySearchDirection.TO, relationType, toId, callbackForToAndFrom, relationAction.apply(fromId)); processRelationByDirection(EntitySearchDirection.FROM, relationType, fromId, callbackForToAndFrom, relationAction.apply(toId)); } - private void processRelationByDirection(EntitySearchDirection direction, String relationType, EntityId mainId, @@ -339,7 +368,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware List matchingCfs = cfsByEntityIdAndProfile.stream() .filter(cf -> { - if (cf.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration config ) { + if (cf.getCalculatedField().getConfiguration() instanceof RelatedEntitiesAggregationCalculatedFieldConfiguration config) { RelationPathLevel relation = config.getRelation(); return direction.equals(relation.direction()) && relationType.equals(relation.relationType()); } else { @@ -717,8 +746,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware private EntityId getProfileId(TenantId tenantId, EntityId entityId) { return switch (entityId.getEntityType()) { - case ASSET -> assetProfileCache.get(tenantId, (AssetId) entityId).getId(); - case DEVICE -> deviceProfileCache.get(tenantId, (DeviceId) entityId).getId(); + case ASSET -> Optional.ofNullable(assetProfileCache.get(tenantId, (AssetId) entityId)).map(AssetProfile::getId).orElse(null); + case DEVICE -> Optional.ofNullable(deviceProfileCache.get(tenantId, (DeviceId) entityId)).map(DeviceProfile::getId).orElse(null); default -> null; }; } 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 945792ebcd..f20df4159d 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 @@ -24,6 +24,7 @@ import jakarta.annotation.PreDestroy; import lombok.Data; import lombok.extern.slf4j.Slf4j; import org.thingsboard.common.util.ThingsBoardExecutors; +import org.thingsboard.server.common.data.cf.CalculatedField; 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.RelationPathQueryDynamicSourceConfiguration; @@ -194,8 +195,14 @@ public abstract class AbstractCalculatedFieldProcessingService { return switch (relation.direction()) { case FROM -> relations.stream() .map(EntityRelation::getTo) + .filter(CalculatedField::isSupportedRefEntity) .toList(); - case TO -> relations.isEmpty() ? List.of() : List.of(relations.get(0).getFrom()); + case TO -> relations.stream() + .map(EntityRelation::getFrom) + .filter(CalculatedField::isSupportedRefEntity) + .findFirst() + .map(List::of) + .orElseGet(Collections::emptyList); }; }, calculatedFieldCallbackExecutor); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java index c7e369a862..8f5d857d74 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java @@ -162,7 +162,7 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS } private boolean checkEntityForCalculatedFields(TenantId tenantId, EntityId entityId, Predicate filter, Predicate linkedEntityFilter, Predicate dynamicSourceFilter, Predicate relatedEntityFilter) { - if (!CalculatedField.SUPPORTED_REFERENCED_ENTITIES.contains(entityId.getEntityType())) { + if (!CalculatedField.isSupportedRefEntity(entityId)) { return false; } diff --git a/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java b/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java index 48a318f981..9a6a96155f 100644 --- a/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java +++ b/application/src/main/java/org/thingsboard/server/service/entitiy/EntityStateSourcingListener.java @@ -29,7 +29,6 @@ import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.Device; import org.thingsboard.server.common.data.DeviceProfile; import org.thingsboard.server.common.data.EntityType; -import org.thingsboard.server.common.data.ObjectType; import org.thingsboard.server.common.data.TbResource; import org.thingsboard.server.common.data.TbResourceInfo; import org.thingsboard.server.common.data.Tenant; @@ -49,6 +48,7 @@ import org.thingsboard.server.common.data.job.Job; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.notification.NotificationRequest; import org.thingsboard.server.common.data.plugin.ComponentLifecycleEvent; +import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.rule.RuleChainType; import org.thingsboard.server.common.data.security.DeviceCredentials; @@ -274,10 +274,13 @@ public class EntityStateSourcingListener { @TransactionalEventListener(fallbackExecution = true) public void handleEvent(RelationActionEvent relationEvent) { - if (relationEvent.getActionType() == ActionType.RELATION_ADD_OR_UPDATE) { - tbClusterService.onRelationUpdated(relationEvent.getTenantId(), relationEvent.getRelation(), TbQueueCallback.EMPTY); - } else if (relationEvent.getActionType() == ActionType.RELATION_DELETED) { - tbClusterService.onRelationDeleted(relationEvent.getTenantId(), relationEvent.getRelation(), TbQueueCallback.EMPTY); + EntityRelation relation = relationEvent.getRelation(); + if (CalculatedField.isSupportedRefEntity(relation.getFrom()) && CalculatedField.isSupportedRefEntity(relation.getTo())) { + if (relationEvent.getActionType() == ActionType.RELATION_ADD_OR_UPDATE) { + tbClusterService.onRelationUpdated(relationEvent.getTenantId(), relation, TbQueueCallback.EMPTY); + } else if (relationEvent.getActionType() == ActionType.RELATION_DELETED) { + tbClusterService.onRelationDeleted(relationEvent.getTenantId(), relation, TbQueueCallback.EMPTY); + } } } diff --git a/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java b/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java index a7b67baf3f..f4eb2b5167 100644 --- a/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java +++ b/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java @@ -51,6 +51,7 @@ import org.thingsboard.server.common.data.relation.EntityRelation; import org.thingsboard.server.common.data.relation.EntitySearchDirection; import org.thingsboard.server.common.data.relation.RelationPathLevel; import org.thingsboard.server.common.data.relation.RelationTypeGroup; +import org.thingsboard.server.common.data.rule.RuleChain; import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.controller.AbstractControllerTest; import org.thingsboard.server.dao.service.DaoSqlTest; @@ -299,6 +300,45 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr }); } + @Test + public void testCreateCfAndRelationToRuleChain_checkAggregation() throws Exception { + Asset asset2 = createAsset("Asset 2", assetProfile.getId()); + Device device3 = createDevice("Device 3", "1234567890333"); + postTelemetry(device3.getId(), "{\"occupied\":true}"); + + RuleChain ruleChain = new RuleChain(); + ruleChain.setName("RuleChain"); + ruleChain = doPost("/api/ruleChain", ruleChain, RuleChain.class); + postTelemetry(ruleChain.getId(), "{\"occupied\":true}"); + + createEntityRelation(asset2.getId(), device3.getId(), "Contains"); + createEntityRelation(asset2.getId(), ruleChain.getId(), "Contains"); + + createOccupancyCF(asset2.getId()); + + await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset2.getId(), Map.of( + "freeSpaces", "0", + "occupiedSpaces", "1", + "totalSpaces", "1" + )); + }); + + postTelemetry(ruleChain.getId(), "{\"occupied\":true}"); + + await().alias("update telemetry on rule chain and no aggregation performed").atMost(deduplicationInterval, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset2.getId(), Map.of( + "freeSpaces", "0", + "occupiedSpaces", "1", + "totalSpaces", "1" + )); + }); + } + @Test public void testDeleteCf_checkNoAggregation() throws Exception { CalculatedField cf = createOccupancyCF(asset.getId()); @@ -466,6 +506,24 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr }); } + @Test + public void testDeleteEntityByRelation_checkAggregation() throws Exception { + createOccupancyCF(asset.getId()); + checkInitialCalculation(); + + doDelete("/api/device/" + device1.getId()).andExpect(status().isOk()); + + await().alias("create relation and perform aggregation").atMost(deduplicationInterval, TimeUnit.SECONDS) + .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS) + .untilAsserted(() -> { + verifyTelemetry(asset.getId(), Map.of( + "freeSpaces", "1", + "occupiedSpaces", "0", + "totalSpaces", "1" + )); + }); + } + @Test public void testUpdateRelationPath_checkAggregation() throws Exception { CalculatedField cf = createOccupancyCF(asset.getId()); diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java index 214dc5247e..12b788435f 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java @@ -88,6 +88,8 @@ public interface RelationService { List findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery); + List findEntityRelations(TenantId tenantId, EntityId entityId); + // TODO: This method may be useful for some validations in the future // ListenableFuture checkRecursiveRelation(EntityId from, EntityId to); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java index 0d2543d4c2..8c5adefcf8 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/CalculatedField.java @@ -65,6 +65,10 @@ public class CalculatedField extends BaseData implements HasN EntityType.DEVICE, EntityType.ASSET, EntityType.CUSTOMER, EntityType.TENANT )); + public static boolean isSupportedRefEntity(EntityId entity) { + return SUPPORTED_REFERENCED_ENTITIES.contains(entity.getEntityType()); + } + private TenantId tenantId; private EntityId entityId; diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java index ff0d09ef2c..3fc5a9d5e7 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java @@ -541,6 +541,16 @@ public class BaseRelationService implements RelationService { return relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit); } + @Override + public List findEntityRelations(TenantId tenantId, EntityId entityId) { + List relations = new ArrayList<>(); + for (RelationTypeGroup relationTypeGroup : RelationTypeGroup.values()) { + relations.addAll(findByFrom(tenantId, entityId, relationTypeGroup)); + relations.addAll(findByTo(tenantId, entityId, relationTypeGroup)); + } + return relations; + } + private void validate(EntityRelationPathQuery relationPathQuery) { validateId((UUIDBased) relationPathQuery.rootEntityId(), id -> "Invalid root entity id: " + id); List levels = relationPathQuery.levels(); From 87e7a14bdb8f50100f79e06d9eede4259848260c Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Fri, 7 Nov 2025 14:14:44 +0200 Subject: [PATCH 2/3] added check for supported entities --- ...alculatedFieldManagerMessageProcessor.java | 26 ++------------ ...tractCalculatedFieldProcessingService.java | 9 ++--- .../server/dao/relation/RelationService.java | 5 +-- .../dao/relation/BaseRelationService.java | 36 ++++++++++++------- 4 files changed, 33 insertions(+), 43 deletions(-) diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java index a86ddde2a1..ec4598947a 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java @@ -295,33 +295,17 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } private void onEntityDeleted(ComponentLifecycleMsg msg, TbCallback callback) { - // 2 = 1 for entity processing + 1 for relation processing - MultipleTbCallback multiCallback = new MultipleTbCallback(2, callback); - deleteEntityRelations(msg, multiCallback); switch (msg.getEntityId().getEntityType()) { case DEVICE, ASSET -> entityProfileCache.removeEntityId(msg.getEntityId()); case CUSTOMER -> ownerEntities.remove(msg.getEntityId()); } ownerEntities.values().forEach(entities -> entities.remove(msg.getEntityId())); - if (isMyPartition(msg.getEntityId(), multiCallback)) { + if (isMyPartition(msg.getEntityId(), callback)) { log.debug("Pushing entity lifecycle msg to specific actor [{}]", msg.getEntityId()); - getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), multiCallback)); + getOrCreateActor(msg.getEntityId()).tell(new CalculatedFieldEntityDeleteMsg(tenantId, msg.getEntityId(), callback)); } } - private void deleteEntityRelations(ComponentLifecycleMsg msg, TbCallback callback) { - List entityRelations = relationService.findEntityRelations(tenantId, msg.getEntityId()); - if (entityRelations.isEmpty()) { - callback.onSuccess(); - } - entityRelations.forEach(entityRelation -> { - Function> deleteAction = - relatedId -> (entityId, ctx, cb) -> deleteRelatedEntity(entityId, relatedId, ctx, cb); - - processRelationIfSupported(entityRelation, callback, deleteAction); - }); - } - private void onRelationChangedEvent(ComponentLifecycleMsg msg, TbCallback callback) { Function> relationAction = switch (msg.getEvent()) { case RELATION_UPDATED -> relatedId -> (entityId, ctx, cb) -> initRelatedEntity(entityId, relatedId, ctx, cb); @@ -335,12 +319,6 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware } EntityRelation entityRelation = JacksonUtil.treeToValue(msg.getInfo(), EntityRelation.class); - processRelationIfSupported(entityRelation, callback, relationAction); - } - - private void processRelationIfSupported(EntityRelation entityRelation, - TbCallback callback, - Function> relationAction) { EntityId toId = entityRelation.getTo(); EntityId fromId = entityRelation.getFrom(); String relationType = entityRelation.getType(); 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 f20df4159d..c7bbf83066 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 @@ -57,6 +57,7 @@ import java.util.Map; import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutionException; +import java.util.function.Predicate; import java.util.stream.Collectors; import static org.thingsboard.server.common.data.cf.CalculatedFieldType.PROPAGATION; @@ -185,7 +186,8 @@ public abstract class AbstractCalculatedFieldProcessingService { } private ListenableFuture> resolveRelatedEntities(TenantId tenantId, EntityId entityId, RelationPathLevel relation) { - ListenableFuture> relationsFut = relationService.findByRelationPathQueryAsync(tenantId, new EntityRelationPathQuery(entityId, List.of(relation))); + Predicate filter = entityRelation -> CalculatedField.isSupportedRefEntity(entityRelation.getFrom()) && CalculatedField.isSupportedRefEntity(entityRelation.getTo()); + ListenableFuture> relationsFut = relationService.findFilteredRelationsByPathQueryAsync(tenantId, new EntityRelationPathQuery(entityId, List.of(relation)), filter); return Futures.transform(relationsFut, relations -> { if (relations == null) { @@ -195,11 +197,9 @@ public abstract class AbstractCalculatedFieldProcessingService { return switch (relation.direction()) { case FROM -> relations.stream() .map(EntityRelation::getTo) - .filter(CalculatedField::isSupportedRefEntity) .toList(); case TO -> relations.stream() .map(EntityRelation::getFrom) - .filter(CalculatedField::isSupportedRefEntity) .findFirst() .map(List::of) .orElseGet(Collections::emptyList); @@ -224,7 +224,8 @@ public abstract class AbstractCalculatedFieldProcessingService { case CURRENT_OWNER -> Futures.immediateFuture(List.of(resolveOwnerArgument(tenantId, entityId))); case RELATION_PATH_QUERY -> { var configuration = (RelationPathQueryDynamicSourceConfiguration) refDynamicSourceConfiguration; - yield Futures.transform(relationService.findByRelationPathQueryAsync(tenantId, configuration.toRelationPathQuery(entityId)), + Predicate filter = entityRelation -> CalculatedField.isSupportedRefEntity(entityRelation.getFrom()) && CalculatedField.isSupportedRefEntity(entityRelation.getTo()); + yield Futures.transform(relationService.findFilteredRelationsByPathQueryAsync(tenantId, configuration.toRelationPathQuery(entityId), filter), configuration::resolveEntityIds, calculatedFieldCallbackExecutor); } }; diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java index 12b788435f..aaa52f99a7 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java @@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.relation.RelationTypeGroup; import org.thingsboard.server.common.data.rule.RuleChainType; import java.util.List; +import java.util.function.Predicate; /** * Created by ashvayka on 27.04.17. @@ -86,9 +87,9 @@ public interface RelationService { ListenableFuture> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery); - List findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery); + ListenableFuture> findFilteredRelationsByPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery, Predicate relationFilter); - List findEntityRelations(TenantId tenantId, EntityId entityId); + List findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery); // TODO: This method may be useful for some validations in the future // ListenableFuture checkRecursiveRelation(EntityId from, EntityId to); diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java index 3fc5a9d5e7..dbde6efd96 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java @@ -71,6 +71,7 @@ import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.function.BiConsumer; +import java.util.function.Predicate; import static org.thingsboard.server.dao.service.Validator.validateId; import static org.thingsboard.server.dao.service.Validator.validatePositiveNumber; @@ -507,6 +508,11 @@ public class BaseRelationService implements RelationService { @Override public ListenableFuture> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery) { + return findFilteredRelationsByPathQueryAsync(tenantId, relationPathQuery, null); + } + + @Override + public ListenableFuture> findFilteredRelationsByPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery, Predicate relationFilter) { log.trace("Executing findByRelationPathQuery, tenantId [{}], relationPathQuery {}", tenantId, relationPathQuery); validateId(tenantId, id -> "Invalid tenant id: " + id); validate(relationPathQuery); @@ -518,10 +524,24 @@ public class BaseRelationService implements RelationService { case FROM -> findByFromAndTypeAsync(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON); case TO -> findByToAndTypeAsync(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON); }; - return Futures.transform(relationsFuture, entityRelations -> entityRelations.size() > limit ? - entityRelations.subList(0, limit) : entityRelations, MoreExecutors.directExecutor()); + return Futures.transform(relationsFuture, entityRelations -> { + if (entityRelations == null || entityRelations.isEmpty()) { + return Collections.emptyList(); + } + List relations = relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations; + return relations.size() > limit ? entityRelations.subList(0, limit) : entityRelations; + }, MoreExecutors.directExecutor()); } - return executor.submit(() -> relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit)); + return executor.submit(() -> { + List entityRelations = relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit); + return relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations; + }); + } + + private List filterRelations(List entityRelations, Predicate relationFilter) { + return entityRelations.stream() + .filter(relationFilter) + .toList(); } @Override @@ -541,16 +561,6 @@ public class BaseRelationService implements RelationService { return relationDao.findByRelationPathQuery(tenantId, relationPathQuery, limit); } - @Override - public List findEntityRelations(TenantId tenantId, EntityId entityId) { - List relations = new ArrayList<>(); - for (RelationTypeGroup relationTypeGroup : RelationTypeGroup.values()) { - relations.addAll(findByFrom(tenantId, entityId, relationTypeGroup)); - relations.addAll(findByTo(tenantId, entityId, relationTypeGroup)); - } - return relations; - } - private void validate(EntityRelationPathQuery relationPathQuery) { validateId((UUIDBased) relationPathQuery.rootEntityId(), id -> "Invalid root entity id: " + id); List levels = relationPathQuery.levels(); From da61794ec5e523f8294f5f483e65b0bc00693e05 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Fri, 7 Nov 2025 14:20:48 +0200 Subject: [PATCH 3/3] minor relations fix --- .../thingsboard/server/dao/relation/BaseRelationService.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java index dbde6efd96..ffb800c7f7 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java @@ -529,7 +529,7 @@ public class BaseRelationService implements RelationService { return Collections.emptyList(); } List relations = relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations; - return relations.size() > limit ? entityRelations.subList(0, limit) : entityRelations; + return relations.size() > limit ? relations.subList(0, limit) : relations; }, MoreExecutors.directExecutor()); } return executor.submit(() -> {