fixed relation update handling, handle max related entities limit

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

@@ -262,17 +262,35 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
     }
 
     private void onTenantProfileUpdated(ComponentLifecycleMsg msg, TbCallback callback) {
+        checkCfIntervalForUpdate();
+
+        long maxRelatedEntitiesPerCfArgument = systemContext.getApiLimitService().getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxRelatedEntitiesToReturnPerCfArgument);
+        Set<CalculatedFieldCtx> cfsToReinit = new HashSet<>();
+        Stream.concat(
+                calculatedFields.values().stream(),
+                entityIdCalculatedFields.values().stream().flatMap(Collection::stream)
+        ).forEach(ctx -> {
+            if (ctx.hasRelatedEntities() && ctx.getMaxRelatedEntitiesPerCfArgument() != maxRelatedEntitiesPerCfArgument) {
+                cfsToReinit.add(ctx);
+            }
+            ctx.setTenantProfileProperties();
+        });
+
+        if (!cfsToReinit.isEmpty()) {
+            MultipleTbCallback cfsReinitCallback = new MultipleTbCallback(cfsToReinit.size(), callback);
+            cfsToReinit.forEach(ctx -> applyToTargetCfEntityActors(ctx, cfsReinitCallback, (id, cb) -> initCfForEntity(id, ctx, StateAction.REINIT, cb)));
+        } else {
+            callback.onSuccess();
+        }
+    }
+
+    private void checkCfIntervalForUpdate() {
         long updatedCfCheckInterval = systemContext.getApiLimitService().getLimit(tenantId, DefaultTenantProfileConfiguration::getCfReevaluationCheckInterval);
         if (cfCheckInterval != updatedCfCheckInterval) {
             cfCheckInterval = updatedCfCheckInterval;
             cancelReevaluationTask();
             scheduleCfsReevaluation();
         }
-        Stream.concat(
-                calculatedFields.values().stream(),
-                entityIdCalculatedFields.values().stream().flatMap(Collection::stream)
-        ).forEach(CalculatedFieldCtx::setTenantProfileProperties);
-        callback.onSuccess();
     }
 
     private void onEntityCreated(ComponentLifecycleMsg msg, TbCallback callback) {
@@ -383,10 +401,7 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
                 .toList();
 
         MultipleTbCallback directionCallback = new MultipleTbCallback(matchingCfs.size(), parentCallback);
-
-        matchingCfs.forEach(ctx ->
-                applyToTargetCfEntityActors(ctx, directionCallback, (entityId, cb) -> relationAction.accept(entityId, ctx, cb))
-        );
+        matchingCfs.forEach(ctx -> relationAction.accept(mainId, ctx, directionCallback));
     }
 
     private void onCfCreated(ComponentLifecycleMsg msg, TbCallback callback) throws CalculatedFieldException {
@@ -581,6 +596,10 @@ public class CalculatedFieldManagerMessageProcessor extends AbstractContextAware
             if (byRelationPathQuery != null && !byRelationPathQuery.isEmpty()) {
                 switch (relation.direction()) {
                     case FROM -> {
+                        if (byRelationPathQuery.size() > 1) {
+                            throw new IllegalStateException("More than one relation found with direction 'TO' " +
+                                    "for relation type '" + relation.relationType() + "'. Found: " + byRelationPathQuery.size());
+                        }
                         EntityRelation entityRelation = byRelationPathQuery.get(0); // only one supported
                         EntityId relatedId = entityRelation.getFrom();
                         if (matchesCfEntity.test(relatedId)) {

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

@@ -250,11 +250,17 @@ public abstract class AbstractCalculatedFieldProcessingService {
                 case FROM -> relations.stream()
                         .map(EntityRelation::getTo)
                         .toList();
-                case TO -> relations.stream()
+                case TO -> {
+                    if (relations.size() > 1) {
+                        throw new IllegalStateException("More than one relation found with direction 'TO' " +
+                                        "for relation type '" + relation.relationType() + "'. Found: " + relations.size());
+                    }
+                    yield relations.stream()
                         .map(EntityRelation::getFrom)
                         .findFirst()
                         .map(List::of)
                         .orElseGet(Collections::emptyList);
+                }
             };
         }, calculatedFieldCallbackExecutor);
     }

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

@@ -129,6 +129,7 @@ public class CalculatedFieldCtx implements Closeable {
     private long scheduledUpdateIntervalMillis;
     private long cfCheckReevaluationIntervalMillis;
     private long alarmReevaluationIntervalMillis;
+    private long maxRelatedEntitiesPerCfArgument;
 
     private Argument propagationArgument;
     private boolean applyExpressionForResolvedArguments;
@@ -307,6 +308,7 @@ public class CalculatedFieldCtx implements Closeable {
         this.intermediateAggregationIntervalMillis = TimeUnit.SECONDS.toMillis(apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getIntermediateAggregationIntervalInSecForCF));
         this.cfCheckReevaluationIntervalMillis = TimeUnit.SECONDS.toMillis(apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getCfReevaluationCheckInterval));
         this.alarmReevaluationIntervalMillis = TimeUnit.SECONDS.toMillis(apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getAlarmsReevaluationInterval));
+        this.maxRelatedEntitiesPerCfArgument = apiLimitService.getLimit(tenantId, DefaultTenantProfileConfiguration::getMaxRelatedEntitiesToReturnPerCfArgument);
     }
 
     public double evaluateSimpleExpression(Expression expression, CalculatedFieldState state) {
@@ -756,6 +758,12 @@ public class CalculatedFieldCtx implements Closeable {
         return scheduledUpdateIntervalMillis == DISABLED_INTERVAL_VALUE;
     }
 
+    public boolean hasRelatedEntities() {
+        return CalculatedFieldType.GEOFENCING == cfType
+                || CalculatedFieldType.PROPAGATION == cfType
+                || CalculatedFieldType.RELATED_ENTITIES_AGGREGATION == cfType;
+    }
+
     public boolean shouldFetchRelatedEntities(CalculatedFieldState state) {
         if (!cfHasRelationPathQuerySource) {
             return false;

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

@@ -66,10 +66,12 @@ public class RelatedEntitiesArgumentEntry implements ArgumentEntry, HasLatestTs
     @Override
     public boolean updateEntry(ArgumentEntry entry, CalculatedFieldCtx ctx) {
         if (entry instanceof RelatedEntitiesArgumentEntry relatedEntitiesArgumentEntry) {
+            checkMaxRelatedEntitiesPerArgument(ctx);
             entityInputs.putAll(relatedEntitiesArgumentEntry.entityInputs);
             return true;
         } else if (entry instanceof SingleValueArgumentEntry singleValueArgumentEntry) {
             if (entry.isForceResetPrevious()) {
+                checkMaxRelatedEntitiesPerArgument(ctx);
                 entityInputs.put(singleValueArgumentEntry.getEntityId(), singleValueArgumentEntry);
                 return true;
             }
@@ -77,6 +79,7 @@ public class RelatedEntitiesArgumentEntry implements ArgumentEntry, HasLatestTs
             if (argumentEntry != null) {
                 argumentEntry.updateEntry(singleValueArgumentEntry, ctx);
             } else {
+                checkMaxRelatedEntitiesPerArgument(ctx);
                 entityInputs.put(singleValueArgumentEntry.getEntityId(), singleValueArgumentEntry);
             }
             return true;
@@ -85,6 +88,15 @@ public class RelatedEntitiesArgumentEntry implements ArgumentEntry, HasLatestTs
         }
     }
 
+    private void checkMaxRelatedEntitiesPerArgument(CalculatedFieldCtx ctx) {
+        if (entityInputs.size() >= ctx.getMaxRelatedEntitiesPerCfArgument()) {
+            throw new IllegalArgumentException(
+                    "Exceeded the maximum allowed related entities per argument '"
+                            + ctx.getMaxRelatedEntitiesPerCfArgument() + "'. Increase the limit in the tenant profile configuration."
+            );
+        }
+    }
+
     @Override
     public boolean isEmpty() {
         return entityInputs.isEmpty();

application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationArgumentEntry.java

@@ -65,7 +65,7 @@ public class PropagationArgumentEntry implements ArgumentEntry {
             throw new IllegalArgumentException("Unsupported argument entry type for propagation argument entry: " + entry.getType());
         }
         if (updated.getAdded() != null) {
-            return checkAdded(updated.getAdded());
+            return checkAdded(updated.getAdded(), ctx);
         }
         if (updated.getRemoved() != null) {
             return entityIds.remove(updated.getRemoved());
@@ -80,7 +80,7 @@ public class PropagationArgumentEntry implements ArgumentEntry {
                 return true;
             }
             boolean retained = entityIds.retainAll(dbEntityIds);
-            boolean added = checkAdded(dbEntityIds);
+            boolean added = checkAdded(dbEntityIds, ctx);
             return retained || added;
         }
         if (updated.isEmpty()) {
@@ -91,8 +91,14 @@ public class PropagationArgumentEntry implements ArgumentEntry {
         return true;
     }
 
-    private boolean checkAdded(Collection<EntityId> updatedIds) {
+    private boolean checkAdded(Collection<EntityId> updatedIds, CalculatedFieldCtx ctx) {
         for (EntityId id : updatedIds) {
+            if (entityIds.size() >= ctx.getMaxRelatedEntitiesPerCfArgument()) {
+                throw new IllegalArgumentException(
+                        "Exceeded the maximum allowed related entities per argument '"
+                                + ctx.getMaxRelatedEntitiesPerCfArgument() + "'. Increase the limit in the tenant profile configuration."
+                );
+            }
             if (entityIds.add(id)) {
                 if (added == null) {
                     added = new ArrayList<>();

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

@@ -470,21 +470,47 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
 
     @Test
     public void testCreateRelation_checkAggregation() throws Exception {
-        createOccupancyCF(asset.getId());
-        checkInitialCalculation();
+        Asset asset2 = createAsset("Asset 2", assetProfile.getId());
+        Device device3 = createDevice("Device 3", "1234567890333");
+        Device device4 = createDevice("Device 4", "1234567890444");
 
-        Device device3 = createDevice("Device 3", deviceProfile.getId(), "1234567890333");
+        createEntityRelation(asset2.getId(), device3.getId(), "Contains");
+        createEntityRelation(asset2.getId(), device4.getId(), "Contains");
 
-        postTelemetry(device3.getId(), "{\"occupied\":true}");
+        createOccupancyCF(assetProfile.getId());
 
-        createEntityRelation(asset.getId(), device3.getId(), "Contains");
+        await().alias("create CF and perform initial aggregation").atMost(TIMEOUT, TimeUnit.SECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    verifyTelemetry(asset.getId(), Map.of(
+                            "freeSpaces", "1",
+                            "occupiedSpaces", "1",
+                            "totalSpaces", "2"
+                    ));
 
-        await().alias("create relation and perform aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS)
+                    verifyTelemetry(asset2.getId(), Map.of(
+                            "freeSpaces", "2",
+                            "occupiedSpaces", "0",
+                            "totalSpaces", "2"
+                    ));
+                });
+
+        Device device5 = createDevice("Device 5", "1234567890555");
+        createEntityRelation(asset2.getId(), device5.getId(), "Contains");
+
+        await().alias("create relation and perform aggregation on asset 2")
+                .atMost(TIMEOUT, TimeUnit.SECONDS)
                 .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
                 .untilAsserted(() -> {
                     verifyTelemetry(asset.getId(), Map.of(
                             "freeSpaces", "1",
-                            "occupiedSpaces", "2",
+                            "occupiedSpaces", "1",
+                            "totalSpaces", "2"
+                    ));
+
+                    verifyTelemetry(asset2.getId(), Map.of(
+                            "freeSpaces", "3",
+                            "occupiedSpaces", "0",
                             "totalSpaces", "3"
                     ));
                 });
@@ -721,6 +747,43 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
                 });
     }
 
+    @Test
+    public void testUpdateMaxRelatedEntitiesPerArgument_checkAggregation() throws Exception {
+        loginSysAdmin();
+
+        updateDefaultTenantProfileConfig(tenantProfileConfig -> {
+            tenantProfileConfig.setMaxRelatedEntitiesToReturnPerCfArgument(1);
+        });
+
+        login("tenant@thingsboard.org", "testPassword");
+
+        createCountCF(asset.getId());
+
+        await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    ObjectNode numberOfDevices = getLatestTelemetry(asset.getId(), "numberOfDevices");
+                    assertThat(numberOfDevices).isNotNull();
+                    assertThat(numberOfDevices.get("numberOfDevices").get(0).get("value").asText()).isEqualTo("1");
+                });
+
+        loginSysAdmin();
+
+        updateDefaultTenantProfileConfig(tenantProfileConfig -> {
+            tenantProfileConfig.setMaxRelatedEntitiesToReturnPerCfArgument(10);
+        });
+
+        login("tenant@thingsboard.org", "testPassword");
+
+        await().alias("update max related entities per argument and perform initial aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    ObjectNode numberOfDevices = getLatestTelemetry(asset.getId(), "numberOfDevices");
+                    assertThat(numberOfDevices).isNotNull();
+                    assertThat(numberOfDevices.get("numberOfDevices").get(0).get("value").asText()).isEqualTo("2");
+                });
+    }
+
     private void checkInitialCalculation() {
         await().alias("create CF and perform initial aggregation").atMost(deduplicationInterval * 2, TimeUnit.SECONDS)
                 .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
@@ -832,6 +895,30 @@ public class RelatedEntitiesAggregationCalculatedFieldTest extends AbstractContr
                 output);
     }
 
+    private CalculatedField createCountCF(EntityId entityId) {
+        Map<String, Argument> arguments = new HashMap<>();
+        Argument argument = new Argument();
+        argument.setRefEntityKey(new ReferencedEntityKey("active", ArgumentType.TS_LATEST, null));
+        argument.setDefaultValue("true");
+        arguments.put("active", argument);
+
+        Map<String, AggMetric> aggMetrics = new HashMap<>();
+
+        AggMetric avgMetric = new AggMetric();
+        avgMetric.setFunction(AggFunction.COUNT);
+        avgMetric.setInput(new AggKeyInput("active"));
+        aggMetrics.put("numberOfDevices", avgMetric);
+
+        TimeSeriesOutput output = new TimeSeriesOutput();
+        output.setDecimalsByDefault(0);
+
+        return createAggCf("Number of devices", entityId,
+                new RelationPathLevel(EntitySearchDirection.FROM, "Contains"),
+                arguments,
+                aggMetrics,
+                output);
+    }
+
     private CalculatedField createAggCf(String name,
                                         EntityId entityId,
                                         RelationPathLevel relation,

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

@@ -60,6 +60,7 @@ import org.thingsboard.server.dao.usagerecord.ApiLimitService;
 
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Comparator;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -517,7 +518,8 @@ class BaseRelationService implements RelationService {
                 if (entityRelations == null || entityRelations.isEmpty()) {
                     return Collections.emptyList();
                 }
-                List<EntityRelation> relations = relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations;
+                List<EntityRelation> relations = new ArrayList<>(relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations);
+                relations.sort(Comparator.comparing(r -> r.getFrom().getId()));
                 return relations.size() > limit ? relations.subList(0, limit) : relations;
             }, directExecutor());
         }