From 3ad0b4b74de092e34fb375aa11a99e952dcfa414 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Thu, 8 Jan 2026 16:57:05 +0200 Subject: [PATCH] fixed relation update handling, handle max related entities limit --- ...alculatedFieldManagerMessageProcessor.java | 37 +++++-- ...tractCalculatedFieldProcessingService.java | 8 +- .../cf/ctx/state/CalculatedFieldCtx.java | 8 ++ .../RelatedEntitiesArgumentEntry.java | 12 +++ .../propagation/PropagationArgumentEntry.java | 12 ++- ...ntitiesAggregationCalculatedFieldTest.java | 101 ++++++++++++++++-- .../dao/relation/BaseRelationService.java | 4 +- 7 files changed, 161 insertions(+), 21 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 4e6711b150..f06c0f113a 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 @@ -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 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)) { 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 fed1e82e96..8284e81f70 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 @@ -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); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java index 6500bfb892..f74ec7514d 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/CalculatedFieldCtx.java +++ b/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; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesArgumentEntry.java index 1939318e7a..c1c8d38316 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/RelatedEntitiesArgumentEntry.java +++ b/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(); diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationArgumentEntry.java index 7b8b393371..c3b5951da6 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/propagation/PropagationArgumentEntry.java +++ b/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 updatedIds) { + private boolean checkAdded(Collection 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<>(); 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 10a7b46282..4c708bce56 100644 --- a/application/src/test/java/org/thingsboard/server/cf/RelatedEntitiesAggregationCalculatedFieldTest.java +++ b/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 arguments = new HashMap<>(); + Argument argument = new Argument(); + argument.setRefEntityKey(new ReferencedEntityKey("active", ArgumentType.TS_LATEST, null)); + argument.setDefaultValue("true"); + arguments.put("active", argument); + + Map 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, 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 0fd914df7c..b390b732d5 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 @@ -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 relations = relationFilter != null ? filterRelations(entityRelations, relationFilter) : entityRelations; + List 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()); }