minor refactoring

application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java

@@ -69,9 +69,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.UUID;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
 
 import static org.thingsboard.server.utils.CalculatedFieldArgumentUtils.createStateByType;
@@ -264,7 +262,7 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM
 
     @SneakyThrows
     private Map<String, ArgumentEntry> fetchAggArguments(CalculatedFieldCtx ctx, EntityId entityId) {
-        ListenableFuture<Map<String, ArgumentEntry>> argumentsFuture = cfService.fetchAggArguments(ctx, entityId);
+        ListenableFuture<Map<String, ArgumentEntry>> argumentsFuture = cfService.fetchAggEntityArguments(ctx, entityId);
         // Ugly but necessary. We do not expect to often fetch data from DB. Only once per <Entity, CalculatedField> pair lifetime.
         // This call happens while processing the CF pack from the queue consumer. So the timeout should be relatively low.
         // Alternatively, we can fetch the state outside the actor system and push separate command to create this actor,

application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldManagerMessageProcessor.java

@@ -24,8 +24,8 @@ import org.thingsboard.server.actors.TbCalculatedFieldEntityActorId;
 import org.thingsboard.server.actors.calculatedField.EntityInitCalculatedFieldMsg.StateAction;
 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.AttributeScope;
+import org.thingsboard.server.common.data.Customer;
 import org.thingsboard.server.common.data.DataConstants;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.ProfileEntityIdInfo;
@@ -255,7 +255,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
         MultipleTbCallback callbackFor2 = new MultipleTbCallback(2, callback);
 
         // process aggregation cfs(in any)
-        List<CalculatedFieldCtx> cfsRelatedToEntity = getCalculatedFieldsRelatedToEntity(entityId, profileId);
+        List<CalculatedFieldCtx> cfsRelatedToEntity = getCfsWithRelationToEntity(entityId, profileId);
         if (!cfsRelatedToEntity.isEmpty()) {
             MultipleTbCallback multiCallback = new MultipleTbCallback(cfsRelatedToEntity.size(), callbackFor2);
             cfsRelatedToEntity.forEach(ctx -> {
@@ -289,8 +289,8 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
             MultipleTbCallback callbackFor2 = new MultipleTbCallback(2, callback);
 
             // process aggregation cfs(in any)
-            List<CalculatedFieldCtx> oldCfsRelatedToEntity = getCalculatedFieldsRelatedToEntity(msg.getEntityId(), msg.getOldProfileId());
-            List<CalculatedFieldCtx> newCfsRelatedToEntity = getCalculatedFieldsRelatedToEntity(msg.getEntityId(), msg.getProfileId());
+            List<CalculatedFieldCtx> oldCfsRelatedToEntity = getCfsWithRelationToEntity(msg.getEntityId(), msg.getOldProfileId());
+            List<CalculatedFieldCtx> newCfsRelatedToEntity = getCfsWithRelationToEntity(msg.getEntityId(), msg.getProfileId());
             var fieldsWithRelatedEntityCount = oldCfsRelatedToEntity.size() + newCfsRelatedToEntity.size();
             if (fieldsWithRelatedEntityCount > 0) {
                 MultipleTbCallback multiCallback = new MultipleTbCallback(fieldsWithRelatedEntityCount, callbackFor2);
@@ -335,7 +335,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
             }
             ownerEntities.values().forEach(entities -> entities.remove(msg.getEntityId()));
 
-            getCalculatedFieldsRelatedToEntity(msg.getEntityId(), msg.getProfileId()).forEach(ctx -> {
+            getCfsWithRelationToEntity(msg.getEntityId(), msg.getProfileId()).forEach(ctx -> {
                 applyToTargetCfEntityActors(ctx, callback, (id, cb) -> deleteRelatedEntity(id, msg.getEntityId(), cb));
             });
             if (isMyPartition(msg.getEntityId(), callback)) {
@@ -559,6 +559,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
     private void onCfDeleted(ComponentLifecycleMsg msg, TbCallback callback) {
         var cfId = new CalculatedFieldId(msg.getEntityId().getId());
         var cfCtx = calculatedFields.remove(cfId); // fixme wtf? why isn't ctx closed properly?
+        cfTriggers.remove(cfId);
         if (cfCtx == null) {
             log.debug("[{}] CF was already deleted [{}]", tenantId, cfId);
             callback.onSuccess();
@@ -613,76 +614,55 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
 
     private List<CalculatedFieldEntityCtxId> filterAggregationCfs(CalculatedFieldTelemetryMsg msg) {
         EntityId entityId = msg.getEntityId();
-
-        List<CalculatedFieldCtx> aggregationCalculatedFields = cfTriggers.entrySet().stream()
+        return cfTriggers.entrySet().stream()
                 .filter(entry -> aggMatches(entry.getValue(), msg.getProto()))
                 .map(Entry::getKey)
                 .map(calculatedFields::get)
                 .filter(Objects::nonNull)
+                .flatMap(cf -> findRelationsForCf(entityId, cf).stream())
                 .toList();
-
-        List<CalculatedFieldEntityCtxId> filteredByRelationCfs = new ArrayList<>();
-        for (CalculatedFieldCtx cf : aggregationCalculatedFields) {
-            EntityId cfEntityId = cf.getEntityId();
-            if (cf.getCalculatedField().getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration aggConfig) {
-                RelationPathLevel relation = aggConfig.getSource().getRelation();
-                EntityId cfEntityProfileId = isProfileEntity(cfEntityId.getEntityType())
-                        ? cfEntityId
-                        : getProfileId(tenantId, cfEntityId);
-                EntityId targetEntity = switch (relation.direction()) {
-                    case FROM ->
-                            relationService.findByToAndTypeAndEntityProfile(tenantId, entityId, relation.relationType(), cfEntityProfileId).getFrom();
-                    case TO ->
-                            relationService.findByFromAndTypeAndEntityProfile(tenantId, entityId, relation.relationType(), cfEntityProfileId).get(0).getTo();
-                };
-                if (targetEntity != null) {
-                    filteredByRelationCfs.add(new CalculatedFieldEntityCtxId(tenantId, cf.getCfId(), targetEntity));
-                }
-            }
-        }
-        return filteredByRelationCfs;
     }
 
-    private List<CalculatedFieldCtx> getCalculatedFieldsRelatedToEntity(EntityId entityId, EntityId profileId) {
-        List<CalculatedFieldCtx> aggCFsUsedProfile = cfTriggers.entrySet().stream()
+    private List<CalculatedFieldCtx> getCfsWithRelationToEntity(EntityId entityId, EntityId profileId) {
+        return cfTriggers.entrySet().stream()
                 .filter(entry -> entry.getValue().matchesProfile(profileId))
                 .map(Entry::getKey)
                 .map(calculatedFields::get)
                 .filter(Objects::nonNull)
+                .filter(cf -> !findRelationsForCf(entityId, cf).isEmpty())
                 .toList();
-
-        List<CalculatedFieldCtx> filteredByRelationCfs = new ArrayList<>();
-        for (CalculatedFieldCtx cf : aggCFsUsedProfile) {
-            CalculatedFieldEntityCtxId calculatedFieldEntityCtxId = filterCfByRelationWithEntity(entityId, cf);
-            if (calculatedFieldEntityCtxId != null) {
-                filteredByRelationCfs.add(cf);
-            }
-        }
-        return filteredByRelationCfs;
     }
 
-    private CalculatedFieldEntityCtxId filterCfByRelationWithEntity(EntityId entityId, CalculatedFieldCtx cf) {
-        EntityId cfEntityId = cf.getEntityId();
-        if (cf.getCalculatedField().getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration aggConfig) {
-            RelationPathLevel relation = aggConfig.getSource().getRelation();
-            EntityId cfEntityProfileId = isProfileEntity(cfEntityId.getEntityType())
+    private List<CalculatedFieldEntityCtxId> findRelationsForCf(EntityId entityId, CalculatedFieldCtx cf) {
+        List<CalculatedFieldEntityCtxId> result = new ArrayList<>();
+        if (cf.getCalculatedField().getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration configuration) {
+            AggSource source = configuration.getSource();
+            RelationPathLevel relation = source.getRelation();
+            EntityId cfEntityId = cf.getEntityId();
+            EntityId targetProfileId = isProfileEntity(cfEntityId.getEntityType())
                     ? cfEntityId
                     : getProfileId(tenantId, cfEntityId);
-            EntityId targetEntity = switch (relation.direction()) {
+            switch (relation.direction()) {
                 case FROM -> {
-                    EntityRelation entityRelation = relationService.findByToAndTypeAndEntityProfile(tenantId, entityId, relation.relationType(), cfEntityProfileId);
-                    yield entityRelation == null ? null : entityRelation.getFrom();
+                    List<EntityRelation> relationsByTo = relationService.findByToAndTypeAndEntityProfile(tenantId, entityId, relation.relationType(), targetProfileId);
+                    if (relationsByTo != null && !relationsByTo.isEmpty()) {
+                        EntityRelation entityRelation = relationsByTo.get(0); // only one supported
+                        result.add(new CalculatedFieldEntityCtxId(tenantId, cf.getCfId(), entityRelation.getFrom()));
+                    }
                 }
                 case TO -> {
-                    EntityRelation entityRelation = relationService.findByFromAndTypeAndEntityProfile(tenantId, entityId, relation.relationType(), cfEntityProfileId).get(0);
-                    yield entityRelation == null ? null : entityRelation.getTo();
+                    List<EntityRelation> relationsByFrom = relationService.findByFromAndTypeAndEntityProfile(tenantId, entityId, relation.relationType(), targetProfileId);
+                    if (relationsByFrom != null && !relationsByFrom.isEmpty()) {
+                        for (EntityRelation entityRelation : relationsByFrom) {
+                            if (entityRelation.getTo().equals(cf.getEntityId())) {
+                                result.add(new CalculatedFieldEntityCtxId(tenantId, cf.getCfId(), entityRelation.getTo()));
+                            }
+                        }
+                    }
                 }
-            };
-            if (targetEntity != null) {
-                return new CalculatedFieldEntityCtxId(tenantId, cf.getCfId(), targetEntity);
             }
         }
-        return null;
+        return result;
     }
 
     private boolean aggMatches(CfAggTrigger cfAggTrigger, CalculatedFieldTelemetryMsgProto proto) {

application/src/main/java/org/thingsboard/server/service/cf/AbstractCalculatedFieldProcessingService.java

@@ -26,7 +26,6 @@ import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.common.util.ThingsBoardExecutors;
 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.CalculatedFieldConfiguration;
 import org.thingsboard.server.common.data.cf.configuration.RelationPathQueryDynamicSourceConfiguration;
 import org.thingsboard.server.common.data.cf.configuration.aggregation.AggSource;
 import org.thingsboard.server.common.data.cf.configuration.aggregation.LatestValuesAggregationCalculatedFieldConfiguration;
@@ -40,6 +39,7 @@ import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
 import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.data.relation.EntityRelation;
+import org.thingsboard.server.common.data.relation.ProfileEntityRelationPathQuery;
 import org.thingsboard.server.common.data.relation.RelationPathLevel;
 import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration;
 import org.thingsboard.server.dao.attributes.AttributesService;
@@ -51,10 +51,11 @@ import org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx;
 import org.thingsboard.server.service.cf.ctx.state.SingleValueArgumentEntry;
 import org.thingsboard.server.service.cf.ctx.state.aggregation.AggSingleArgumentEntry;
 
-import java.util.Collection;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
@@ -105,7 +106,7 @@ public abstract class AbstractCalculatedFieldProcessingService {
                 }
                 yield futures;
             }
-            case LATEST_VALUES_AGGREGATION -> fetchAggregationArgumentFutures(ctx, entityId);
+            case LATEST_VALUES_AGGREGATION -> fetchAggArguments(ctx, entityId, ts);
         };
         return Futures.whenAllComplete(argFutures.values())
                 .call(() -> resolveArgumentFutures(argFutures),
@@ -122,17 +123,32 @@ public abstract class AbstractCalculatedFieldProcessingService {
         return resolveOwnerArgument(tenantId, entityId);
     }
 
-    private List<EntityId> resolveRelatedEntities(TenantId tenantId, EntityId entityId, AggSource aggSource) {
+    private ListenableFuture<List<EntityId>> resolveRelatedEntities(TenantId tenantId, EntityId entityId, AggSource aggSource) {
         RelationPathLevel relation = aggSource.getRelation();
-        return switch (relation.direction()) {
-            case FROM -> aggSource.getEntityProfiles().stream()
-                    .map(profile -> relationService.findByFromAndTypeAndEntityProfile(tenantId, entityId, relation.relationType(), profile))
-                    .flatMap(Collection::stream)
-                    .map(EntityRelation::getTo)
+
+        List<ListenableFuture<List<EntityRelation>>> relationListsFut = new ArrayList<>();
+        if (aggSource.getEntityProfiles().isEmpty()) {
+            relationListsFut.add(relationService.findByProfileEntityRelationPathQueryAsync(tenantId, new ProfileEntityRelationPathQuery(entityId, relation, null)));
+        } else {
+            aggSource.getEntityProfiles().forEach(profile -> relationListsFut.add(relationService.findByProfileEntityRelationPathQueryAsync(tenantId, new ProfileEntityRelationPathQuery(entityId, relation, profile))));
+        }
+
+        return Futures.transform(Futures.allAsList(relationListsFut), relationLists -> {
+            if (relationLists == null) {
+                return new ArrayList<>();
+            }
+            List<EntityRelation> allRelations = relationLists.stream()
+                    .filter(Objects::nonNull)
+                    .flatMap(List::stream)
                     .toList();
-            case TO ->
-                    aggSource.getEntityProfiles().stream().map(profile -> relationService.findByToAndTypeAndEntityProfile(tenantId, entityId, relation.relationType(), profile).getFrom()).toList();
-        };
+
+            return switch (relation.direction()) {
+                        case FROM -> allRelations.stream()
+                            .map(EntityRelation::getTo)
+                            .toList();
+                        case TO -> allRelations.isEmpty() ? List.of() : List.of(allRelations.get(0).getFrom());
+                    };
+            }, calculatedFieldCallbackExecutor);
     }
 
     protected Map<String, ArgumentEntry> resolveArgumentFutures(Map<String, ListenableFuture<ArgumentEntry>> argFutures) {
@@ -174,6 +190,34 @@ public abstract class AbstractCalculatedFieldProcessingService {
         return argFutures;
     }
 
+    protected Map<String, ListenableFuture<ArgumentEntry>> fetchAggArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) {
+        LatestValuesAggregationCalculatedFieldConfiguration aggConfig = (LatestValuesAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration();
+
+        ListenableFuture<List<EntityId>> relatedEntities = resolveRelatedEntities(ctx.getTenantId(), entityId, aggConfig.getSource());
+
+        Map<String, ListenableFuture<ArgumentEntry>> futures = new HashMap<>();
+        aggConfig.getInputs().forEach((key, refKey) -> {
+            Argument argument = new Argument();
+            argument.setRefEntityKey(refKey);
+            futures.put(key, Futures.transformAsync(relatedEntities, entityIds -> fetchAggArgumentEntry(ctx.getTenantId(), entityIds, argument, System.currentTimeMillis()), MoreExecutors.directExecutor()));
+        });
+        return futures;
+    }
+
+    protected ListenableFuture<Map<String, ArgumentEntry>> fetchEntityAggArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) {
+        LatestValuesAggregationCalculatedFieldConfiguration aggConfig = (LatestValuesAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration();
+        Map<String, ListenableFuture<ArgumentEntry>> futures = new HashMap<>();
+        aggConfig.getInputs().forEach((key, refKey) -> {
+            Argument argument = new Argument();
+            argument.setRefEntityKey(refKey);
+            ListenableFuture<ArgumentEntry> argEntryFut = fetchSingleAggArgumentEntry(ctx.getTenantId(), entityId, argument, ts);
+            futures.put(key, argEntryFut);
+        });
+        return Futures.whenAllComplete(futures.values())
+                .call(() -> resolveArgumentFutures(futures),
+                        MoreExecutors.directExecutor());
+    }
+
     private ListenableFuture<List<EntityId>> resolveGeofencingEntityIds(TenantId tenantId, EntityId entityId, Map.Entry<String, Argument> entry) {
         Argument value = entry.getValue();
         if (value.getRefEntityId() != null) {
@@ -197,50 +241,6 @@ public abstract class AbstractCalculatedFieldProcessingService {
         return ownerService.getOwner(tenantId, entityId);
     }
 
-    private Map<String, ListenableFuture<ArgumentEntry>> fetchAggregationArgumentFutures(CalculatedFieldCtx ctx, EntityId entityId) {
-        LatestValuesAggregationCalculatedFieldConfiguration aggConfig = (LatestValuesAggregationCalculatedFieldConfiguration) ctx.getCalculatedField().getConfiguration();
-
-        List<EntityId> entityIds = resolveRelatedEntities(ctx.getTenantId(), entityId, aggConfig.getSource());
-
-        Map<String, ListenableFuture<ArgumentEntry>> futures = new HashMap<>();
-        aggConfig.getInputs().forEach((key, refKey) -> {
-            Argument argument = new Argument();
-            argument.setRefEntityKey(refKey);
-            futures.put(key, fetchAggArgumentEntry(ctx.getTenantId(), entityIds, argument, System.currentTimeMillis()));
-        });
-        return futures;
-    }
-
-    public ListenableFuture<ArgumentEntry> fetchAggArgumentEntry(TenantId tenantId, List<EntityId> aggEntities, Argument argument, long startTs) {
-        List<ListenableFuture<Map.Entry<EntityId, ? extends ArgumentEntry>>> futures = aggEntities.stream()
-                .map(entityId -> fetchSingleAggArgumentEntry(tenantId, entityId, argument, startTs))
-                .toList();
-
-        ListenableFuture<List<Map.Entry<EntityId, ? extends ArgumentEntry>>> allFutures = Futures.allAsList(futures);
-
-        return Futures.transform(allFutures,
-                entries -> ArgumentEntry.createAggArgument(
-                        entries.stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
-                ),
-                MoreExecutors.directExecutor());
-    }
-
-    protected ListenableFuture<Map<String, ArgumentEntry>> fetchAggArguments(CalculatedFieldCtx ctx, EntityId entityId, long ts) {
-        CalculatedFieldConfiguration configuration = ctx.getCalculatedField().getConfiguration();
-        LatestValuesAggregationCalculatedFieldConfiguration aggConfig = (LatestValuesAggregationCalculatedFieldConfiguration) configuration;
-        Map<String, ListenableFuture<ArgumentEntry>> futures = new HashMap<>();
-        aggConfig.getInputs().forEach((key, refKey) -> {
-            Argument argument = new Argument();
-            argument.setRefEntityKey(refKey);
-
-            ListenableFuture<ArgumentEntry> argumentEntryListenableFuture = fetchAggArgumentEntry(ctx.getTenantId(), List.of(entityId), argument, System.currentTimeMillis());
-            futures.put(key, argumentEntryListenableFuture);
-        });
-        return Futures.whenAllComplete(futures.values())
-                .call(() -> resolveArgumentFutures(futures),
-                        MoreExecutors.directExecutor());
-    }
-
     private ListenableFuture<ArgumentEntry> fetchGeofencingKvEntry(TenantId tenantId, List<EntityId> geofencingEntities, Argument argument) {
         if (argument.getRefEntityKey().getType() != ArgumentType.ATTRIBUTE) {
             throw new IllegalStateException("Unsupported argument key type: " + argument.getRefEntityKey().getType());
@@ -265,6 +265,22 @@ public abstract class AbstractCalculatedFieldProcessingService {
                 .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))), MoreExecutors.directExecutor());
     }
 
+    public ListenableFuture<ArgumentEntry> fetchAggArgumentEntry(TenantId tenantId, List<EntityId> aggEntities, Argument argument, long startTs) {List<ListenableFuture<Map.Entry<EntityId, ArgumentEntry>>> futures = aggEntities.stream()
+                .map(entityId -> {
+                    ListenableFuture<ArgumentEntry> singleAggEntryFut = fetchSingleAggArgumentEntry(tenantId, entityId, argument, startTs);
+                    return Futures.transform(singleAggEntryFut, singleAggEntry -> Map.entry(entityId, singleAggEntry), MoreExecutors.directExecutor());
+                })
+                .toList();
+
+        ListenableFuture<List<Map.Entry<EntityId, ? extends ArgumentEntry>>> allFutures = Futures.allAsList(futures);
+
+        return Futures.transform(allFutures,
+                entries -> ArgumentEntry.createAggArgument(
+                        entries.stream().collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
+                ),
+                MoreExecutors.directExecutor());
+    }
+
     protected ListenableFuture<ArgumentEntry> fetchArgumentValue(TenantId tenantId, EntityId entityId, Argument argument, long startTs) {
         return switch (argument.getRefEntityKey().getType()) {
             case TS_ROLLING -> fetchTsRolling(tenantId, entityId, argument, startTs);
@@ -309,7 +325,15 @@ public abstract class AbstractCalculatedFieldProcessingService {
                         }, calculatedFieldCallbackExecutor));
     }
 
-    protected ListenableFuture<Map.Entry<EntityId, ? extends ArgumentEntry>> fetchSingleAggArgumentEntry(TenantId tenantId, EntityId entityId, Argument argument, long startTs) {
+    private ReadTsKvQuery buildTsRollingQuery(TenantId tenantId, Argument argument, long startTs, long endTs) {
+        long maxDataPoints = apiLimitService.getLimit(
+                tenantId, DefaultTenantProfileConfiguration::getMaxDataPointsPerRollingArg);
+        int argumentLimit = argument.getLimit();
+        int limit = argumentLimit == 0 || argumentLimit > maxDataPoints ? (int) maxDataPoints : argumentLimit;
+        return new BaseReadTsKvQuery(argument.getRefEntityKey().getKey(), startTs, endTs, 0, limit, Aggregation.NONE);
+    }
+
+    private ListenableFuture<ArgumentEntry> 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);
@@ -317,36 +341,26 @@ public abstract class AbstractCalculatedFieldProcessingService {
         };
     }
 
-    private ListenableFuture<Map.Entry<EntityId, ? extends ArgumentEntry>> fetchAttributeAggEntry(TenantId tenantId, EntityId entityId, Argument argument, long defaultLastUpdateTs) {
+    private ListenableFuture<ArgumentEntry> 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, 0L));
-            AggSingleArgumentEntry entry = new AggSingleArgumentEntry(entityId, attributeKvEntry);
-            return Map.entry(entityId, entry);
+            return new AggSingleArgumentEntry(entityId, attributeKvEntry);
         }, calculatedFieldCallbackExecutor);
     }
 
-    protected ListenableFuture<Map.Entry<EntityId, ? extends ArgumentEntry>> fetchTsLatestAggEntry(TenantId tenantId, EntityId entityId, Argument argument, long defaultTs) {
+    private ListenableFuture<ArgumentEntry> 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> tsKvEntry = result.or(() -> Optional.of(new BasicTsKvEntry(System.currentTimeMillis(), createDefaultKvEntry(argument), 0L)));
-                    AggSingleArgumentEntry entry = new AggSingleArgumentEntry(entityId, tsKvEntry.get());
-                    return Map.entry(entityId, entry);
+                    Optional<TsKvEntry> tsKvEntry = result.or(() -> Optional.of(new BasicTsKvEntry(defaultTs, createDefaultKvEntry(argument), 0L)));
+                    return new AggSingleArgumentEntry(entityId, tsKvEntry.get());
                 }, calculatedFieldCallbackExecutor);
     }
 
-    private ReadTsKvQuery buildTsRollingQuery(TenantId tenantId, Argument argument, long startTs, long endTs) {
-        long maxDataPoints = apiLimitService.getLimit(
-                tenantId, DefaultTenantProfileConfiguration::getMaxDataPointsPerRollingArg);
-        int argumentLimit = argument.getLimit();
-        int limit = argumentLimit == 0 || argumentLimit > maxDataPoints ? (int) maxDataPoints : argumentLimit;
-        return new BaseReadTsKvQuery(argument.getRefEntityKey().getKey(), startTs, endTs, 0, limit, Aggregation.NONE);
-    }
-
 }

application/src/main/java/org/thingsboard/server/service/cf/CalculatedFieldProcessingService.java

@@ -33,7 +33,7 @@ public interface CalculatedFieldProcessingService {
 
     ListenableFuture<Map<String, ArgumentEntry>> fetchArguments(CalculatedFieldCtx ctx, EntityId entityId);
 
-    ListenableFuture<Map<String, ArgumentEntry>> fetchAggArguments(CalculatedFieldCtx ctx, EntityId entityId);
+    ListenableFuture<Map<String, ArgumentEntry>> fetchAggEntityArguments(CalculatedFieldCtx ctx, EntityId entityId);
 
     Map<String, ArgumentEntry> fetchDynamicArgsFromDb(CalculatedFieldCtx ctx, EntityId entityId);
 

application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldProcessingService.java

@@ -91,8 +91,8 @@ public class DefaultCalculatedFieldProcessingService extends AbstractCalculatedF
     }
 
     @Override
-    public ListenableFuture<Map<String, ArgumentEntry>> fetchAggArguments(CalculatedFieldCtx ctx, EntityId entityId) {
-        return super.fetchAggArguments(ctx, entityId, System.currentTimeMillis());
+    public ListenableFuture<Map<String, ArgumentEntry>> fetchAggEntityArguments(CalculatedFieldCtx ctx, EntityId entityId) {
+        return super.fetchEntityAggArguments(ctx, entityId, System.currentTimeMillis());
     }
 
     @Override

application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java

@@ -27,6 +27,7 @@ import org.thingsboard.server.cluster.TbClusterService;
 import org.thingsboard.server.common.data.EntityType;
 import org.thingsboard.server.common.data.cf.CalculatedField;
 import org.thingsboard.server.common.data.cf.CalculatedFieldLink;
+import org.thingsboard.server.common.data.cf.configuration.aggregation.AggSource;
 import org.thingsboard.server.common.data.cf.configuration.aggregation.CfAggTrigger;
 import org.thingsboard.server.common.data.cf.configuration.aggregation.LatestValuesAggregationCalculatedFieldConfiguration;
 import org.thingsboard.server.common.data.id.CalculatedFieldId;
@@ -39,6 +40,7 @@ import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.data.msg.TbMsgType;
 import org.thingsboard.server.common.data.relation.EntityRelation;
 import org.thingsboard.server.common.data.relation.RelationPathLevel;
+import org.thingsboard.server.common.data.relation.RelationTypeGroup;
 import org.thingsboard.server.common.util.ProtoUtils;
 import org.thingsboard.server.dao.relation.RelationService;
 import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto;
@@ -191,19 +193,27 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS
         for (CalculatedFieldCtx cfCtx : cfCtxs) {
             EntityId cfEntityId = cfCtx.getEntityId();
             if (cfCtx.getCalculatedField().getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration aggConfig) {
-                RelationPathLevel relation = aggConfig.getSource().getRelation();
+                AggSource source = aggConfig.getSource();
+                RelationPathLevel relation = source.getRelation();
                 EntityId cfEntityProfileId = isProfileEntity(cfEntityId.getEntityType())
                         ? cfEntityId
                         : calculatedFieldCache.getProfileId(tenantId, cfEntityId);
-                EntityRelation entityRelation = switch (relation.direction()) {
-                    case FROM ->
-                            relationService.findByToAndTypeAndEntityProfile(tenantId, entityId, relation.relationType(), cfEntityProfileId);
-                    case TO ->
-                            relationService.findByFromAndTypeAndEntityProfile(tenantId, entityId, relation.relationType(), cfEntityProfileId).get(0);
+                switch (relation.direction()) {
+                    case FROM -> {
+                        List<EntityRelation> byToAndType = relationService.findByToAndType(tenantId, entityId, relation.relationType(), RelationTypeGroup.COMMON);
+//                        List<EntityRelation> byTo = relationService.findByToAndTypeAndEntityProfile(tenantId, entityId, relation.relationType(), cfEntityProfileId);
+                        if (!byToAndType.isEmpty()) {
+                            return true;
+                        }
+                    }
+                    case TO -> {
+                        List<EntityRelation> byFromAndType = relationService.findByFromAndType(tenantId, entityId, relation.relationType(), RelationTypeGroup.COMMON);
+//                        List<EntityRelation> byFrom = relationService.findByFromAndTypeAndEntityProfile(tenantId, entityId, relation.relationType(), cfEntityProfileId);
+                        if (!byFromAndType.isEmpty()) {
+                            return true;
+                        }
+                    }
                 };
-                if (entityRelation != null) {
-                    return true;
-                }
             }
         }
 

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggArgumentEntry.java

@@ -18,6 +18,7 @@ package org.thingsboard.server.service.cf.ctx.state.aggregation;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import org.thingsboard.script.api.tbel.TbelCfArg;
+import org.thingsboard.script.api.tbel.TbelCfLatestValuesAggregation;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.service.cf.ctx.state.ArgumentEntry;
 import org.thingsboard.server.service.cf.ctx.state.ArgumentEntryType;
@@ -66,7 +67,7 @@ public class AggArgumentEntry implements ArgumentEntry {
 
     @Override
     public TbelCfArg toTbelCfArg() {
-        return null;
+        return new TbelCfLatestValuesAggregation(aggInputs.values());
     }
 
 }

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/agg.json

@@ -7,8 +7,8 @@
     "allEnabledUntil": 1769907492297
   },
   "entityId": {
-    "entityType": "ASSET",
-    "id": "cc830710-a4cf-11f0-87cb-2d6683c4fccf"
+    "entityType": "ASSET_PROFILE",
+    "id": "bb8ddd40-a8bc-11f0-869b-e9d81fa6eaf1"
   },
   "configuration": {
     "type": "LATEST_VALUES_AGGREGATION",

application/src/main/java/org/thingsboard/server/service/queue/processing/AbstractConsumerService.java

@@ -184,7 +184,6 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
             }
         } else if (EntityType.DEVICE_PROFILE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             deviceProfileCache.evict(tenantId, new DeviceProfileId(componentLifecycleMsg.getEntityId().getId()));
-            actorContext.getRelationService().evictRelationsByProfile(tenantId, componentLifecycleMsg.getEntityId());
         } else if (EntityType.DEVICE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             deviceProfileCache.evict(tenantId, new DeviceId(componentLifecycleMsg.getEntityId().getId()));
             if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.CREATED)) {
@@ -200,7 +199,6 @@ public abstract class AbstractConsumerService<N extends com.google.protobuf.Gene
             }
         } else if (EntityType.ASSET_PROFILE.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             assetProfileCache.evict(tenantId, new AssetProfileId(componentLifecycleMsg.getEntityId().getId()));
-            actorContext.getRelationService().evictRelationsByProfile(tenantId, componentLifecycleMsg.getEntityId());
         } else if (EntityType.ASSET.equals(componentLifecycleMsg.getEntityId().getEntityType())) {
             assetProfileCache.evict(tenantId, new AssetId(componentLifecycleMsg.getEntityId().getId()));
             if (componentLifecycleMsg.getEvent().equals(ComponentLifecycleEvent.CREATED)) {

application/src/test/java/org/thingsboard/server/cf/LatestValuesAggregationCalculatedFieldTest.java

@@ -28,6 +28,7 @@ import org.thingsboard.server.common.data.asset.AssetProfile;
 import org.thingsboard.server.common.data.cf.CalculatedField;
 import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.cf.configuration.ArgumentType;
+import org.thingsboard.server.common.data.cf.configuration.CalculatedFieldConfiguration;
 import org.thingsboard.server.common.data.cf.configuration.Output;
 import org.thingsboard.server.common.data.cf.configuration.OutputType;
 import org.thingsboard.server.common.data.cf.configuration.ReferencedEntityKey;
@@ -52,6 +53,7 @@ import org.thingsboard.server.common.data.security.Authority;
 import org.thingsboard.server.controller.AbstractControllerTest;
 import org.thingsboard.server.dao.service.DaoSqlTest;
 
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -110,7 +112,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
         createEntityRelation(asset.getId(), device1.getId(), "Contains");
         createEntityRelation(asset.getId(), device2.getId(), "Contains");
 
-        calculatedField = createOccupancyCF(asset.getId(), List.of(deviceProfile.getId()));
+        calculatedField = createOccupancyCF("Occupied spaces", asset.getId(), List.of(deviceProfile.getId()));
 
         checkInitialCalculation();
     }
@@ -200,6 +202,49 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
                 });
     }
 
+    @Test
+    public void testCfOnProfile_checkMetricsCalculation() throws Exception {
+        Asset asset2 = createAsset("Asset 2", assetProfile.getId());
+        Device device3 = createDevice("Device 3", deviceProfile.getId(), "1234567890333");
+        postTelemetry(device3.getId(), "{\"occupied\":false}");
+        Device device4 = createDevice("Device 4", deviceProfile.getId(), "1234567890444");
+        postTelemetry(device4.getId(), "{\"occupied\":false}");
+
+        createEntityRelation(asset2.getId(), device3.getId(), "Contains");
+        createEntityRelation(asset2.getId(), device4.getId(), "Contains");
+
+        CalculatedField calculatedField2 = createOccupancyCF("Occupied spaces 2", assetProfile.getId(), List.of(deviceProfile.getId()));
+
+        await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    ObjectNode occupancy = getLatestTelemetry(asset.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces");
+                    assertThat(occupancy).isNotNull();
+                    assertThat(occupancy.get("freeSpaces").get(0).get("value").asText()).isEqualTo("1");
+                    assertThat(occupancy.get("occupiedSpaces").get(0).get("value").asText()).isEqualTo("1");
+                    assertThat(occupancy.get("totalSpaces").get(0).get("value").asText()).isEqualTo("2");
+
+                    ObjectNode occupancy2 = getLatestTelemetry(asset2.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces");
+                    assertThat(occupancy2).isNotNull();
+                    assertThat(occupancy2.get("freeSpaces").get(0).get("value").asText()).isEqualTo("2");
+                    assertThat(occupancy2.get("occupiedSpaces").get(0).get("value").asText()).isEqualTo("0");
+                    assertThat(occupancy2.get("totalSpaces").get(0).get("value").asText()).isEqualTo("2");
+                });
+
+        postTelemetry(device3.getId(), "{\"occupied\":true}");
+
+        await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.MILLISECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    ObjectNode occupancy2 = getLatestTelemetry(asset2.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces");
+                    assertThat(occupancy2).isNotNull();
+                    assertThat(occupancy2.get("freeSpaces").get(0).get("value").asText()).isEqualTo("1");
+                    assertThat(occupancy2.get("occupiedSpaces").get(0).get("value").asText()).isEqualTo("1");
+                    assertThat(occupancy2.get("totalSpaces").get(0).get("value").asText()).isEqualTo("2");
+                });
+    }
+
+
     @Test
     public void testDeleteRelation_checkMetricsCalculation() throws Exception {
         deleteEntityRelation(new EntityRelation(asset.getId(), device1.getId(), "Contains", RelationTypeGroup.COMMON));
@@ -215,6 +260,28 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
                 });
     }
 
+//    @Test
+//    public void testCfWithoutTargetProfileSpecified_checkMetricsCalculation() throws Exception {
+//        Device device3 = createDevice("Device 3", "1234567890333");
+//        postTelemetry(device3.getId(), "{\"occupied\":true}");
+//        createEntityRelation(asset.getId(), device3.getId(), "Contains");
+//
+//        var configuration = (LatestValuesAggregationCalculatedFieldConfiguration) calculatedField.getConfiguration();
+//        configuration.getSource().setEntityProfiles(Collections.emptyList());
+//        calculatedField.setConfiguration(configuration);
+//        saveCalculatedField(calculatedField);
+//
+//        await().alias("update cf and perform aggregation for 3 devices").atMost(deduplicationInterval, TimeUnit.MILLISECONDS)
+//                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+//                .untilAsserted(() -> {
+//                    ObjectNode occupancy = getLatestTelemetry(asset.getId(), "freeSpaces", "occupiedSpaces", "totalSpaces");
+//                    assertThat(occupancy).isNotNull();
+//                    assertThat(occupancy.get("freeSpaces").get(0).get("value").asText()).isEqualTo("1");
+//                    assertThat(occupancy.get("occupiedSpaces").get(0).get("value").asText()).isEqualTo("2");
+//                    assertThat(occupancy.get("totalSpaces").get(0).get("value").asText()).isEqualTo("3");
+//                });
+//    }
+
     private void checkInitialCalculation() {
         await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS)
                 .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
@@ -229,7 +296,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
         assertThat(occupancy.get("totalSpaces").get(0).get("value").asText()).isEqualTo("2");
     }
 
-    private CalculatedField createOccupancyCF(EntityId entityId, List<EntityId> profiles) {
+    private CalculatedField createOccupancyCF(String name, EntityId entityId, List<EntityId> profiles) {
         Map<String, AggMetric> aggMetrics = new HashMap<>();
 
         AggMetric freeSpaces = new AggMetric();
@@ -252,7 +319,7 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
         Output output = new Output();
         output.setType(OutputType.TIME_SERIES);
 
-        return createAggCf("Occupied spaces", entityId,
+        return createAggCf(name, entityId,
                 buildSource(EntitySearchDirection.FROM, "Contains", profiles),
                 Map.of("oc", new ReferencedEntityKey("occupied", ArgumentType.TS_LATEST, null)),
                 aggMetrics,

common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java

@@ -22,6 +22,7 @@ import org.thingsboard.server.common.data.relation.EntityRelation;
 import org.thingsboard.server.common.data.relation.EntityRelationInfo;
 import org.thingsboard.server.common.data.relation.EntityRelationPathQuery;
 import org.thingsboard.server.common.data.relation.EntityRelationsQuery;
+import org.thingsboard.server.common.data.relation.ProfileEntityRelationPathQuery;
 import org.thingsboard.server.common.data.relation.RelationTypeGroup;
 import org.thingsboard.server.common.data.rule.RuleChainType;
 
@@ -86,11 +87,17 @@ public interface RelationService {
 
     ListenableFuture<List<EntityRelation>> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery);
 
+    ListenableFuture<List<EntityRelation>> findByProfileEntityRelationPathQueryAsync(TenantId tenantId, ProfileEntityRelationPathQuery relationPathQuery);
+
+    List<EntityRelation> findByProfileEntityRelationPathQuery(TenantId tenantId, ProfileEntityRelationPathQuery relationPathQuery);
+
+    ListenableFuture<List<EntityRelation>> findByFromAndTypeAndEntityProfileAsync(TenantId tenantId, EntityId from, String relationType, EntityId targetProfileId);
+
     List<EntityRelation> findByFromAndTypeAndEntityProfile(TenantId tenantId, EntityId from, String relationType, EntityId profileId);
 
-    EntityRelation findByToAndTypeAndEntityProfile(TenantId tenantId, EntityId to, String relationType, EntityId profileId);
+    ListenableFuture<List<EntityRelation>> findByToAndTypeAndEntityProfileAsync(TenantId tenantId, EntityId to, String relationType, EntityId targetProfileId);
 
-    void evictRelationsByProfile(TenantId tenantId, EntityId profileId);
+    List<EntityRelation> findByToAndTypeAndEntityProfile(TenantId tenantId, EntityId to, String relationType, EntityId profileId);
 
     void evictRelationsByEntityAndProfile(TenantId tenantId, EntityId entityId, EntityId profileId);
 

common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/CfAggTrigger.java

@@ -42,7 +42,7 @@ public class CfAggTrigger {
     }
 
     public boolean matchesProfile(EntityId profileId) {
-        return entityProfiles.contains(profileId);
+        return entityProfiles.isEmpty() || entityProfiles.contains(profileId);
     }
 
     public boolean matchesTimeSeries(List<TsKvEntry> telemetry) {

common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/LatestValuesAggregationCalculatedFieldConfiguration.java

@@ -45,7 +45,7 @@ public class LatestValuesAggregationCalculatedFieldConfiguration implements Calc
     public CfAggTrigger buildTrigger() {
         return CfAggTrigger.builder()
                 .inputs(List.copyOf(inputs.values()))
-                .entityProfiles(source.getEntityProfiles())
+                .entityProfiles(List.copyOf(source.getEntityProfiles()))
                 .build();
     }
 

common/data/src/main/java/org/thingsboard/server/common/data/relation/ProfileEntityRelationPathQuery.java

@@ -0,0 +1,21 @@
+/**
+ * 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.common.data.relation;
+
+import org.thingsboard.server.common.data.id.EntityId;
+
+public record ProfileEntityRelationPathQuery(EntityId rootEntityId, RelationPathLevel level, EntityId targetEntityProfileId) {
+}

common/script/script-api/src/main/java/org/thingsboard/script/api/tbel/TbelCfLatestValuesAggregation.java

@@ -39,6 +39,6 @@ public class TbelCfLatestValuesAggregation implements TbelCfArg {
 
     @Override
     public long memorySize() {
-        return 32;
+        return OBJ_SIZE;
     }
 }

dao/src/main/java/org/thingsboard/server/dao/relation/BaseRelationService.java

@@ -45,6 +45,7 @@ import org.thingsboard.server.common.data.relation.EntityRelationInfo;
 import org.thingsboard.server.common.data.relation.EntityRelationPathQuery;
 import org.thingsboard.server.common.data.relation.EntityRelationsQuery;
 import org.thingsboard.server.common.data.relation.EntitySearchDirection;
+import org.thingsboard.server.common.data.relation.ProfileEntityRelationPathQuery;
 import org.thingsboard.server.common.data.relation.RelationEntityTypeFilter;
 import org.thingsboard.server.common.data.relation.RelationPathLevel;
 import org.thingsboard.server.common.data.relation.RelationTypeGroup;
@@ -515,32 +516,92 @@ public class BaseRelationService implements RelationService {
     }
 
     @Override
-    public List<EntityRelation> findByFromAndTypeAndEntityProfile(TenantId tenantId, EntityId from, String relationType, EntityId profileId) {
-        RelationCacheKey cacheKey = RelationCacheKey.builder().from(from).type(relationType).typeGroup(RelationTypeGroup.COMMON).direction(EntitySearchDirection.FROM).entityProfile(profileId).build();
-        return cache.getAndPutInTransaction(cacheKey,
-                () -> relationDao.findByFromAndTypeAndProfile(tenantId, from, relationType, RelationTypeGroup.COMMON, profileId),
-                RelationCacheValue::getRelations,
-                relations -> RelationCacheValue.builder().relations(relations).build(), false);
+    public ListenableFuture<List<EntityRelation>> findByProfileEntityRelationPathQueryAsync(TenantId tenantId, ProfileEntityRelationPathQuery relationPathQuery) {
+        log.trace("Executing findByProfileEntityRelationPathQueryAsync, tenantId [{}], relationPathQuery {}", tenantId, relationPathQuery);
+        validateId(tenantId, id -> "Invalid tenant id: " + id);
+        validate(relationPathQuery);
+        RelationPathLevel relationPathLevel = relationPathQuery.level();
+        return switch (relationPathLevel.direction()) {
+            case FROM -> findByFromAndTypeAndEntityProfileAsync(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), relationPathQuery.targetEntityProfileId());
+            case TO -> findByToAndTypeAndEntityProfileAsync(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), relationPathQuery.targetEntityProfileId());
+        };
     }
 
     @Override
-    public EntityRelation findByToAndTypeAndEntityProfile(TenantId tenantId, EntityId to, String relationType, EntityId profileId) {
-        RelationCacheKey cacheKey = RelationCacheKey.builder().to(to).type(relationType).typeGroup(RelationTypeGroup.COMMON).direction(EntitySearchDirection.TO).entityProfile(profileId).build();
-        return cache.getAndPutInTransaction(cacheKey,
-                () -> relationDao.findByToAndTypeAndProfile(tenantId, to, relationType, RelationTypeGroup.COMMON, profileId),
-                RelationCacheValue::getRelation,
-                relation -> RelationCacheValue.builder().relation(relation).build(), false);
+    public List<EntityRelation> findByProfileEntityRelationPathQuery(TenantId tenantId, ProfileEntityRelationPathQuery relationPathQuery) {
+        log.trace("Executing findByProfileEntityRelationPathQuery, tenantId [{}], relationPathQuery {}", tenantId, relationPathQuery);
+        validateId(tenantId, id -> "Invalid tenant id: " + id);
+        validate(relationPathQuery);
+        return relationDao.findByProfileEntityRelationPathQuery(tenantId, relationPathQuery);
+//        RelationPathLevel relationPathLevel = relationPathQuery.level();
+//        return switch (relationPathLevel.direction()) {
+//            case FROM -> findByFromAndTypeAndEntityProfile(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), relationPathQuery.targetEntityProfileId());
+//            case TO -> findByToAndTypeAndEntityProfile(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), relationPathQuery.targetEntityProfileId());
+//        };
+    }
+
+    @Override
+    public ListenableFuture<List<EntityRelation>> findByFromAndTypeAndEntityProfileAsync(TenantId tenantId, EntityId from, String relationType, EntityId targetProfileId) {
+        log.trace("Executing findByFromAndTypeAndEntityProfileAsync [{}][{}][{}]", from, relationType, targetProfileId);
+        validate(from);
+        validateType(relationType);
+        if (targetProfileId == null) {
+            return findByFromAndTypeAsync(tenantId, from, relationType, RelationTypeGroup.COMMON);
+        }
+        return executor.submit(() -> findByFromAndTypeAndEntityProfile(tenantId, from, relationType, targetProfileId));
+    }
+
+    @Override
+    public List<EntityRelation> findByFromAndTypeAndEntityProfile(TenantId tenantId, EntityId from, String relationType, EntityId targetProfileId) {
+        if (targetProfileId == null) {
+            return findByFromAndType(tenantId, from, relationType, RelationTypeGroup.COMMON);
+        }
+//        RelationCacheKey cacheKey = RelationCacheKey.builder().from(from).type(relationType).typeGroup(RelationTypeGroup.COMMON).direction(EntitySearchDirection.FROM).entityProfile(targetProfileId).build();
+//        return cache.getAndPutInTransaction(cacheKey,
+//                () -> relationDao.findByFromAndTypeAndProfile(tenantId, from, relationType, RelationTypeGroup.COMMON, targetProfileId),
+//                RelationCacheValue::getRelations,
+//                relations -> RelationCacheValue.builder().relations(relations).build(), false);
+
+        return relationDao.findByFromAndTypeAndProfile(tenantId, from, relationType, RelationTypeGroup.COMMON, targetProfileId);
     }
 
     @Override
-    public void evictRelationsByProfile(TenantId tenantId, EntityId profileId) {
-        RelationCacheKey key = RelationCacheKey.builder().entityProfile(profileId).build();
-        cache.evict(List.of(key));
-        log.debug("Processed evict relations by key: {}", key);
+    public ListenableFuture<List<EntityRelation>> findByToAndTypeAndEntityProfileAsync(TenantId tenantId, EntityId to, String relationType, EntityId targetProfileId) {
+        log.trace("Executing findByToAndTypeAndEntityProfileAsync [{}][{}][{}]", to, relationType, targetProfileId);
+        validate(to);
+        validateType(relationType);
+        if (targetProfileId == null) {
+            return findByToAndTypeAsync(tenantId, to, relationType, RelationTypeGroup.COMMON);
+        }
+        return executor.submit(() -> findByToAndTypeAndEntityProfile(tenantId, to, relationType, targetProfileId));
+    }
+
+    @Override
+    public List<EntityRelation> findByToAndTypeAndEntityProfile(TenantId tenantId, EntityId to, String relationType, EntityId targetProfileId) {
+        if (targetProfileId == null) {
+            return findByFromAndType(tenantId, to, relationType, RelationTypeGroup.COMMON);
+        }
+//        RelationCacheKey cacheKey = RelationCacheKey.builder().to(to).type(relationType).typeGroup(RelationTypeGroup.COMMON).direction(EntitySearchDirection.TO).entityProfile(targetProfileId).build();
+//        return cache.getAndPutInTransaction(cacheKey,
+//                () -> relationDao.findByToAndTypeAndProfile(tenantId, to, relationType, RelationTypeGroup.COMMON, targetProfileId),
+//                RelationCacheValue::getRelations,
+//                relations -> RelationCacheValue.builder().relations(relations).build(), false);
+
+        return relationDao.findByToAndTypeAndProfile(tenantId, to, relationType, RelationTypeGroup.COMMON, targetProfileId);
     }
 
     @Override
     public void evictRelationsByEntityAndProfile(TenantId tenantId, EntityId entityId, EntityId profileId) {
+
+//        List<RelationCacheKey> keys = new ArrayList<>(5);
+//        keys.add(new RelationCacheKey(entityId, null, event.getType(), event.getTypeGroup()));
+//        keys.add(new RelationCacheKey(event.getFrom(), null, event.getType(), event.getTypeGroup(), EntitySearchDirection.FROM));
+//        keys.add(new RelationCacheKey(event.getFrom(), null, null, event.getTypeGroup(), EntitySearchDirection.FROM));
+//        keys.add(new RelationCacheKey(null, event.getTo(), event.getType(), event.getTypeGroup(), EntitySearchDirection.TO));
+//        keys.add(new RelationCacheKey(null, event.getTo(), null, event.getTypeGroup(), EntitySearchDirection.TO));
+//        cache.evict(keys);
+//        log.debug("Processed evict event: {}", event);
+
         List<RelationCacheKey> keys = new ArrayList<>(2);
         keys.add(RelationCacheKey.builder().from(entityId).entityProfile(profileId).build());
         keys.add(RelationCacheKey.builder().to(entityId).entityProfile(profileId).build());
@@ -548,6 +609,11 @@ public class BaseRelationService implements RelationService {
         log.debug("Processed evict relations by keys: {}", keys);
     }
 
+    private void validate(ProfileEntityRelationPathQuery relationPathQuery) {
+        validateId((UUIDBased) relationPathQuery.rootEntityId(), id -> "Invalid root entity id: " + id);
+        relationPathQuery.level().validate();
+    }
+
     private void validate(EntityRelationPathQuery relationPathQuery) {
         validateId((UUIDBased) relationPathQuery.rootEntityId(), id -> "Invalid root entity id: " + id);
         List<RelationPathLevel> levels = relationPathQuery.levels();

dao/src/main/java/org/thingsboard/server/dao/relation/RelationDao.java

@@ -20,6 +20,7 @@ import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.relation.EntityRelation;
 import org.thingsboard.server.common.data.relation.EntityRelationPathQuery;
+import org.thingsboard.server.common.data.relation.ProfileEntityRelationPathQuery;
 import org.thingsboard.server.common.data.relation.RelationTypeGroup;
 import org.thingsboard.server.common.data.rule.RuleChainType;
 
@@ -40,7 +41,7 @@ public interface RelationDao {
 
     List<EntityRelation> findAllByTo(TenantId tenantId, EntityId to, RelationTypeGroup typeGroup);
 
-    EntityRelation findByToAndTypeAndProfile(TenantId tenantId, EntityId to, String relationType, RelationTypeGroup typeGroup, EntityId profileId);
+    List<EntityRelation> findByToAndTypeAndProfile(TenantId tenantId, EntityId to, String relationType, RelationTypeGroup typeGroup, EntityId profileId);
 
     List<EntityRelation> findAllByTo(TenantId tenantId, EntityId to);
 
@@ -78,4 +79,6 @@ public interface RelationDao {
 
     List<EntityRelation> findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery);
 
+    List<EntityRelation> findByProfileEntityRelationPathQuery(TenantId tenantId, ProfileEntityRelationPathQuery query);
+
 }

dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java

@@ -27,6 +27,7 @@ import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.relation.EntityRelation;
 import org.thingsboard.server.common.data.relation.EntityRelationPathQuery;
 import org.thingsboard.server.common.data.relation.EntitySearchDirection;
+import org.thingsboard.server.common.data.relation.ProfileEntityRelationPathQuery;
 import org.thingsboard.server.common.data.relation.RelationPathLevel;
 import org.thingsboard.server.common.data.relation.RelationTypeGroup;
 import org.thingsboard.server.common.data.rule.RuleChainType;
@@ -41,9 +42,12 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.UUID;
 import java.util.stream.Collectors;
 
+import static org.thingsboard.server.dao.model.ModelConstants.ASSET_TABLE_NAME;
+import static org.thingsboard.server.dao.model.ModelConstants.DEVICE_TABLE_NAME;
 import static org.thingsboard.server.dao.model.ModelConstants.RELATION_FROM_ID_PROPERTY;
 import static org.thingsboard.server.dao.model.ModelConstants.RELATION_FROM_TYPE_PROPERTY;
 import static org.thingsboard.server.dao.model.ModelConstants.RELATION_TABLE_NAME;
@@ -118,8 +122,14 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple
     }
 
     @Override
-    public EntityRelation findByToAndTypeAndProfile(TenantId tenantId, EntityId to, String relationType, RelationTypeGroup typeGroup, EntityId profileId) {
-        return DaoUtil.getData(relationRepository.findByToAndProfile(to.getId(), to.getEntityType().name(), typeGroup.name(), relationType, profileId.getId()));
+    public List<EntityRelation> findByToAndTypeAndProfile(TenantId tenantId, EntityId to, String relationType, RelationTypeGroup typeGroup, EntityId profileId) {
+        return DaoUtil.convertDataList(
+                relationRepository.findByToAndProfile(
+                        to.getId(),
+                        to.getEntityType().name(),
+                        typeGroup.name(),
+                        relationType,
+                        profileId.getId()));
     }
 
     @Override
@@ -402,4 +412,92 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple
         return sb.toString();
     }
 
+    @Override
+    public List<EntityRelation> findByProfileEntityRelationPathQuery(TenantId tenantId, ProfileEntityRelationPathQuery query) {
+        String sql = buildProfileEntityRelationPathSql(query);
+        Object[] params = buildProfileEntityRelationPathParams(query);
+
+        log.trace("[{}] profile entity relation path query: {}", tenantId, sql);
+
+        return jdbcTemplate.queryForList(sql, params).stream()
+                .map(row -> {
+                    var entityRelation = new EntityRelation();
+                    var fromId = (UUID) row.get(RELATION_FROM_ID_PROPERTY);
+                    var fromType = (String) row.get(RELATION_FROM_TYPE_PROPERTY);
+                    var toId = (UUID) row.get(RELATION_TO_ID_PROPERTY);
+                    var toType = (String) row.get(RELATION_TO_TYPE_PROPERTY);
+                    var grp = (String) row.get(RELATION_TYPE_GROUP_PROPERTY);
+                    var type = (String) row.get(RELATION_TYPE_PROPERTY);
+                    var version = (Long) row.get(VERSION_COLUMN);
+
+                    entityRelation.setFrom(EntityIdFactory.getByTypeAndUuid(fromType, fromId));
+                    entityRelation.setTo(EntityIdFactory.getByTypeAndUuid(toType, toId));
+                    entityRelation.setType(type);
+                    entityRelation.setTypeGroup(RelationTypeGroup.valueOf(grp));
+                    entityRelation.setVersion(version);
+                    return entityRelation;
+                })
+                .collect(Collectors.toList());
+    }
+
+    private Object[] buildProfileEntityRelationPathParams(ProfileEntityRelationPathQuery query) {
+        final List<Object> params = new ArrayList<>();
+
+        params.add(query.rootEntityId().getId());
+        params.add(query.rootEntityId().getEntityType().name());
+
+        params.add(query.level().relationType());
+
+        if (query.targetEntityProfileId() != null) {
+            params.add(query.targetEntityProfileId().getId());
+            params.add(query.targetEntityProfileId().getId());
+        }
+
+        return params.toArray();
+    }
+
+    private static String buildProfileEntityRelationPathSql(ProfileEntityRelationPathQuery query) {
+        EntitySearchDirection direction = query.level().direction();
+
+        StringBuilder sb = new StringBuilder();
+
+        sb.append("\n")
+                .append("SELECT r.from_id, r.from_type, r.to_id, r.to_type,\n")
+                .append("       r.relation_type_group, r.relation_type, r.version\n")
+                .append("FROM ").append(RELATION_TABLE_NAME).append(" r\n");
+
+        sb.append("JOIN ").append(DEVICE_TABLE_NAME).append(" d ON ");
+        if (EntitySearchDirection.FROM == direction) {
+            sb.append("r.to_id = d.id AND r.to_type = 'DEVICE'").append("\n");
+        } else {
+            sb.append("r.from_id = d.id AND r.from_type = 'DEVICE'").append("\n");
+        }
+
+        sb.append("JOIN ").append(ASSET_TABLE_NAME).append(" a ON ");
+        if (EntitySearchDirection.FROM == direction) {
+            sb.append("r.to_id = a.id AND r.to_type = 'ASSET'").append("\n");
+        } else {
+            sb.append("r.from_id = a.id AND r.from_type = 'ASSET'").append("\n");
+        }
+
+        if (EntitySearchDirection.FROM == direction) {
+            sb.append("WHERE r.from_id = ?").append("\n")
+                    .append("AND r.from_type = ?").append("\n");
+        } else {
+            sb.append("WHERE r.to_id = ?").append("\n")
+                    .append("AND r.to_type = ?").append("\n");
+        }
+
+        sb.append("AND r.relation_type = ?").append("\n")
+                .append("AND r.relation_type_group = '").append(RelationTypeGroup.COMMON).append("'\n");
+
+        if (query.targetEntityProfileId() != null) {
+            sb.append("AND ((d.device_profile_id = ?) OR (a.asset_profile_id = ?))").append("\n");
+        }
+
+        sb.append("AND (d.id IS NOT NULL OR a.id IS NOT NULL)");
+
+        return sb.toString();
+    }
+
 }

dao/src/main/java/org/thingsboard/server/dao/sql/relation/RelationRepository.java

@@ -126,9 +126,8 @@ public interface RelationRepository
                   AND r.relation_type_group = :relationTypeGroup
                   AND ((d.device_profile_id = :profileId) OR (a.asset_profile_id = :profileId))
                   AND (d.id IS NOT NULL OR a.id IS NOT NULL)
-                LIMIT 1
             """, nativeQuery = true)
-    Optional<RelationEntity> findByToAndProfile(@Param("toId") UUID toId,
+    List<RelationEntity> findByToAndProfile(@Param("toId") UUID toId,
                                                 @Param("toType") String toType,
                                                 @Param("relationTypeGroup") String relationTypeGroup,
                                                 @Param("relationType") String relationType,