added state to proto and fixed deduplication logic

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

@@ -106,21 +106,22 @@ public class LatestValuesAggregationCalculatedFieldState extends BaseCalculatedF
 
     @Override
     public ListenableFuture<CalculatedFieldResult> performCalculation(Map<String, ArgumentEntry> updatedArgs, CalculatedFieldCtx ctx) throws Exception {
-        boolean shouldRecalculate = updatedArgs == null || updatedArgs.isEmpty();
-        if (!shouldRecalculate() && !shouldRecalculate) {
+        boolean cfUpdated = updatedArgs != null && updatedArgs.isEmpty();
+        if (shouldRecalculate() || cfUpdated) {
+            Output output = ctx.getOutput();
+            ObjectNode aggResult = aggregateMetrics(output);
+            lastMetricsEvalTs = System.currentTimeMillis();
+            ctx.scheduleReevaluation(deduplicationInterval, actorCtx);
+            return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder()
+                    .type(output.getType())
+                    .scope(output.getScope())
+                    .result(createResultJson(ctx.isUseLatestTs(), aggResult))
+                    .build());
+        } else {
             return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder()
                     .result(null)
                     .build());
         }
-        Output output = ctx.getOutput();
-        ObjectNode aggResult = aggregateMetrics(output);
-        lastMetricsEvalTs = System.currentTimeMillis();
-        ctx.scheduleReevaluation(deduplicationInterval, actorCtx);
-        return Futures.immediateFuture(TelemetryCalculatedFieldResult.builder()
-                .type(output.getType())
-                .scope(output.getScope())
-                .result(createResultJson(ctx.isUseLatestTs(), aggResult))
-                .build());
     }
 
     private boolean shouldRecalculate() {

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

@@ -1,60 +0,0 @@
-{
-  "type": "LATEST_VALUES_AGGREGATION",
-  "name": "Occupied spaces",
-  "debugSettings": {
-    "failuresEnabled": true,
-    "allEnabled": true,
-    "allEnabledUntil": 1769907492297
-  },
-  "entityId": {
-    "entityType": "ASSET",
-    "id": "f8ad0800-a9a6-11f0-bbe6-459b63b420fe"
-  },
-  "configuration": {
-    "type": "LATEST_VALUES_AGGREGATION",
-    "relation": {
-      "direction": "FROM",
-      "relationType": "Contains"
-    },
-    "arguments": {
-      "oc": {
-        "refEntityKey": {
-          "key": "occupied",
-          "type": "TS_LATEST"
-        },
-        "defaultValue": "false"
-      }
-    },
-    "deduplicationIntervalMillis": 20000,
-    "metrics": {
-      "totalSpaces": {
-        "function": "COUNT",
-        "input": {
-          "type": "function",
-          "function" : "return 1;"
-        }
-      },
-      "occupiedSpaces": {
-        "function": "COUNT",
-        "filter": "return oc == true",
-        "input": {
-          "type": "key",
-          "key" : "oc"
-        }
-      },
-      "freeSpaces": {
-        "function": "COUNT",
-        "filter": "return oc == false",
-        "input": {
-          "type": "key",
-          "key" : "oc"
-        }
-      }
-    },
-    "output": {
-      "type": "TIME_SERIES",
-      "decimals": 2
-    },
-    "useLatestTsFromInputs": "true"
-  }
-}

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

@@ -35,6 +35,7 @@ import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdPro
 import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.GeofencingArgumentProto;
 import org.thingsboard.server.gen.transport.TransportProtos.GeofencingZoneProto;
+import org.thingsboard.server.gen.transport.TransportProtos.LatestValuesAggregationStateProto;
 import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TsDoubleValProto;
 import org.thingsboard.server.gen.transport.TransportProtos.TsRollingArgumentProto;
@@ -96,13 +97,11 @@ public class CalculatedFieldUtils {
                 .setId(toProto(stateId))
                 .setType(state.getType().name());
 
-        if (state instanceof LatestValuesAggregationCalculatedFieldState aggState) {
-            builder.setLastArgsUpdateTs(aggState.getLastArgsRefreshTs());
-        }
+        LatestValuesAggregationStateProto.Builder aggBuilder = LatestValuesAggregationStateProto.newBuilder();
         state.getArguments().forEach((argName, argEntry) -> {
             if (argEntry instanceof AggArgumentEntry aggArgumentEntry) {
                 aggArgumentEntry.getAggInputs()
-                        .forEach((entityId, entry) -> builder.addAggArguments(toAggSingleArgumentProto(argName, entityId, entry)));
+                        .forEach((entityId, entry) -> aggBuilder.addAggArguments(toAggSingleArgumentProto(argName, entityId, entry)));
             } else if (argEntry instanceof SingleValueArgumentEntry singleValueArgumentEntry) {
                 builder.addSingleValueArguments(toSingleValueArgumentProto(argName, singleValueArgumentEntry));
             } else if (argEntry instanceof TsRollingArgumentEntry rollingArgumentEntry) {
@@ -120,6 +119,10 @@ public class CalculatedFieldUtils {
                 alarmStateProto.setClearRuleState(toAlarmRuleStateProto(alarmState.getClearRuleState()));
             }
         }
+        if (state instanceof LatestValuesAggregationCalculatedFieldState aggState) {
+            aggBuilder.setLastArgsUpdateTs(aggState.getLastArgsRefreshTs());
+            builder.setLatestValuesAggregationState(aggBuilder.build());
+        }
         return builder.build();
     }
 
@@ -237,15 +240,16 @@ public class CalculatedFieldUtils {
             }
             case LATEST_VALUES_AGGREGATION -> {
                 LatestValuesAggregationCalculatedFieldState aggState = (LatestValuesAggregationCalculatedFieldState) state;
+                LatestValuesAggregationStateProto aggregationStateProto = proto.getLatestValuesAggregationState();
                 Map<String, Map<EntityId, ArgumentEntry>> arguments = new HashMap<>();
-                proto.getAggArgumentsList().forEach(argProto -> {
+                aggregationStateProto.getAggArgumentsList().forEach(argProto -> {
                     AggSingleEntityArgumentEntry entry = fromAggSingleValueArgumentProto(argProto);
                     arguments.computeIfAbsent(argProto.getValue().getArgName(), name -> new HashMap<>()).put(entry.getEntityId(), entry);
                 });
                 arguments.forEach((argName, entityInputs) -> {
                     aggState.getArguments().put(argName, new AggArgumentEntry(entityInputs, false));
                 });
-                aggState.setLastArgsRefreshTs(proto.getLastArgsUpdateTs());
+                aggState.setLastArgsRefreshTs(aggregationStateProto.getLastArgsUpdateTs());
             }
         }
 

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

@@ -489,10 +489,16 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
         configuration.setMetrics(Map.of("maxTemperature", aggMetric));
         saveCalculatedField(cf);
 
+        await().alias("update metrics and perform aggregation").atMost(deduplicationInterval / 2, TimeUnit.MILLISECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    verifyTelemetry(asset.getId(), Map.of("maxTemperature", "24"));
+                });
+
         postTelemetry(device1.getId(), "{\"temperature\":101.3}");
         postTelemetry(device2.getId(), "{\"temperature\":25.8}");
 
-        await().alias("update metrics and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS)
+        await().alias("update telemetry and perform aggregation").atMost(deduplicationInterval, TimeUnit.MILLISECONDS)
                 .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
                 .untilAsserted(() -> {
                     verifyTelemetry(asset.getId(), Map.of("maxTemperature", "26"));
@@ -544,9 +550,15 @@ public class LatestValuesAggregationCalculatedFieldTest extends AbstractControll
         configuration.setDeduplicationIntervalMillis(2 * deduplicationInterval);
         saveCalculatedField(cf);
 
+        await().alias("update deduplication interval and perform aggregation").atMost(deduplicationInterval / 2, TimeUnit.MILLISECONDS)
+                .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
+                .untilAsserted(() -> {
+                    verifyTelemetry(asset.getId(), Map.of("avgTemperature", "24"));
+                });
+
         postTelemetry(device2.getId(), "{\"temperature\":32.1}");
 
-        await().alias("update deduplication interval and perform aggregation").atMost(2 * deduplicationInterval, TimeUnit.MILLISECONDS)
+        await().alias("update telemetry and perform aggregation").atMost(2 * deduplicationInterval, TimeUnit.MILLISECONDS)
                 .pollInterval(POLL_INTERVAL, TimeUnit.SECONDS)
                 .untilAsserted(() -> {
                     verifyTelemetry(asset.getId(), Map.of("avgTemperature", "28"));

common/proto/src/main/proto/queue.proto

@@ -920,6 +920,11 @@ message AggSingleArgumentEntryProto {
   SingleValueArgumentProto value = 2;
 }
 
+message LatestValuesAggregationStateProto {
+  int64 lastArgsUpdateTs = 1;
+  repeated AggSingleArgumentEntryProto aggArguments = 2;
+}
+
 message CalculatedFieldStateProto {
   CalculatedFieldEntityCtxIdProto id = 1;
   string type = 2;
@@ -927,8 +932,7 @@ message CalculatedFieldStateProto {
   repeated TsRollingArgumentProto rollingValueArguments = 4;
   repeated GeofencingArgumentProto geofencingArguments = 5;
   AlarmStateProto alarmState = 6;
-  repeated AggSingleArgumentEntryProto aggArguments = 7;
-  int64 lastArgsUpdateTs = 8;
+  LatestValuesAggregationStateProto latestValuesAggregationState = 7;
 }
 
 //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level.