Merge pull request #14565 from irynamatveieva/aggregation-cf/improvements

Related entities aggregation calculated field improvements

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

@@ -110,21 +110,26 @@ public interface CalculatedFieldState extends Closeable {
         private static final String MISSING_PROPAGATION_TARGETS_ERROR = "No entities found via 'Propagation path to related entities'. " +
                                                                         "Verify the configured relation type and direction.";
         private static final String MISSING_PROPAGATION_TARGETS_AND_ARGUMENTS_ERROR = MISSING_PROPAGATION_TARGETS_ERROR + " Missing arguments to propagate: ";
+        public static final String MISSING_AGGREGATION_ENTITIES_ERROR = "No entities found via 'Aggregation path to related entities'. " +
+                                                                        "Verify the configured relation type and direction.";
         public static final ReadinessStatus READY = new ReadinessStatus(true, null);
 
+        public static ReadinessStatus notReady(String errorMsg) {
+            return new ReadinessStatus(false, errorMsg);
+        }
+
         public static ReadinessStatus from(List<String> emptyOrMissingArguments) {
             if (CollectionsUtil.isEmpty(emptyOrMissingArguments)) {
                 return ReadinessStatus.READY;
             }
             boolean propagationCtxIsEmpty = emptyOrMissingArguments.remove(PROPAGATION_CONFIG_ARGUMENT);
             if (!propagationCtxIsEmpty) {
-                return new ReadinessStatus(false, MISSING_REQUIRED_ARGUMENTS_ERROR + String.join(", ", emptyOrMissingArguments));
+                return notReady(MISSING_REQUIRED_ARGUMENTS_ERROR + String.join(", ", emptyOrMissingArguments));
             }
             if (emptyOrMissingArguments.isEmpty()) {
-                return new ReadinessStatus(false, MISSING_PROPAGATION_TARGETS_ERROR);
+                return notReady(MISSING_PROPAGATION_TARGETS_ERROR);
             }
-            return new ReadinessStatus(false, MISSING_PROPAGATION_TARGETS_AND_ARGUMENTS_ERROR +
-                                              String.join(", ", emptyOrMissingArguments));
+            return notReady(MISSING_PROPAGATION_TARGETS_AND_ARGUMENTS_ERROR + String.join(", ", emptyOrMissingArguments));
         }
     }
 

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

@@ -53,6 +53,7 @@ import java.util.stream.Collectors;
 
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.thingsboard.server.service.cf.ctx.state.CalculatedFieldCtx.DISABLED_INTERVAL_VALUE;
+import static org.thingsboard.server.service.cf.ctx.state.CalculatedFieldState.ReadinessStatus.MISSING_AGGREGATION_ENTITIES_ERROR;
 
 @Slf4j
 public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculatedFieldState implements ScheduledRefreshSupported {
@@ -177,6 +178,7 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat
         });
         lastMetricsEvalTs = DEFAULT_LAST_UPDATE_TS;
         lastArgsRefreshTs = System.currentTimeMillis();
+        readinessStatus = checkReadiness();
     }
 
     public void scheduleReevaluation() {
@@ -288,4 +290,18 @@ public class RelatedEntitiesAggregationCalculatedFieldState extends BaseCalculat
 
     record EntityArgument(EntityInfo entity, JsonNode entityArguments) {}
 
+    @Override
+    protected ReadinessStatus checkReadiness() {
+        if (arguments == null) {
+            return ReadinessStatus.notReady(MISSING_AGGREGATION_ENTITIES_ERROR);
+        }
+        for (String requiredArgumentKey : requiredArguments) {
+            ArgumentEntry argumentEntry = arguments.get(requiredArgumentKey);
+            if (argumentEntry == null || argumentEntry.isEmpty()) {
+                return ReadinessStatus.notReady(MISSING_AGGREGATION_ENTITIES_ERROR);
+            }
+        }
+        return ReadinessStatus.READY;
+    }
+
 }

application/src/main/java/org/thingsboard/server/utils/CalculatedFieldUtils.java

@@ -113,8 +113,12 @@ public class CalculatedFieldUtils {
                 case PROPAGATION -> builder.addAllPropagationEntityIds(toPropagationEntityIdsProto((PropagationArgumentEntry) argEntry));
                 case RELATED_ENTITIES -> {
                     RelatedEntitiesArgumentEntry relatedEntitiesArgumentEntry = (RelatedEntitiesArgumentEntry) argEntry;
-                    relatedEntitiesArgumentEntry.getEntityInputs()
-                            .forEach((entityId, entry) -> builder.addSingleValueArguments(toSingleValueArgumentProto(argName, (SingleValueArgumentEntry) entry)));
+                    Map<EntityId, ArgumentEntry> entityInputs = relatedEntitiesArgumentEntry.getEntityInputs();
+                    if (entityInputs.isEmpty()) {
+                        builder.addSingleValueArguments(SingleValueArgumentProto.newBuilder().setArgName(argName).build());
+                    } else {
+                        entityInputs.forEach((entityId, entry) -> builder.addSingleValueArguments(toSingleValueArgumentProto(argName, (SingleValueArgumentEntry) entry)));
+                    }
                 }
                 case ENTITY_AGGREGATION -> {
                     EntityAggregationArgumentEntry entityAggregationArgumentEntry = (EntityAggregationArgumentEntry) argEntry;
@@ -241,7 +245,10 @@ public class CalculatedFieldUtils {
             Map<String, Map<EntityId, ArgumentEntry>> arguments = new HashMap<>();
             proto.getSingleValueArgumentsList().forEach(argProto -> {
                 SingleValueArgumentEntry entry = fromSingleValueArgumentProto(argProto);
-                arguments.computeIfAbsent(argProto.getArgName(), name -> new HashMap<>()).put(entry.getEntityId(), entry);
+                Map<EntityId, ArgumentEntry> entityInputs = arguments.computeIfAbsent(argProto.getArgName(), name -> new HashMap<>());
+                if (entry.getEntityId() != null) {
+                    entityInputs.put(entry.getEntityId(), entry);
+                }
             });
             arguments.forEach((argName, entityInputs) -> {
                 relatedEntitiesAggState.getArguments().put(argName, new RelatedEntitiesArgumentEntry(entityInputs, false));

application/src/test/java/org/thingsboard/server/service/cf/ctx/state/RelatedEntitiesAggregationCalculatedFieldStateTest.java

@@ -113,8 +113,10 @@ public class RelatedEntitiesAggregationCalculatedFieldStateTest {
     }
 
     @Test
-    void testIsReadyReturnFalseWhenNoArgumentsSet() {
+    void testIsReadyWhenNoRelatedEntities() {
         assertThat(state.isReady()).isFalse();
+        assertThat(state.getReadinessStatus().errorMsg())
+                .isEqualTo("No entities found via 'Aggregation path to related entities'. Verify the configured relation type and direction.");
     }
 
     @Test