added new protos for states

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

@@ -71,11 +71,13 @@ import org.thingsboard.server.common.util.ProtoUtils;
 import org.thingsboard.server.dao.attributes.AttributesService;
 import org.thingsboard.server.dao.cf.CalculatedFieldService;
 import org.thingsboard.server.dao.timeseries.TimeseriesService;
-import org.thingsboard.server.gen.transport.TransportProtos;
 import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto;
 import org.thingsboard.server.gen.transport.TransportProtos.AttributeValueProto;
+import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto;
 import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityUpdateMsgProto;
+import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldIdProto;
 import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto;
+import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldLinkedTelemetryMsgProto.Builder;
 import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldTelemetryMsgProto;
 import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleEvent;
 import org.thingsboard.server.gen.transport.TransportProtos.ComponentLifecycleMsgProto;
@@ -84,7 +86,6 @@ import org.thingsboard.server.gen.transport.TransportProtos.ToCalculatedFieldNot
 import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
 import org.thingsboard.server.queue.TbQueueCallback;
 import org.thingsboard.server.queue.TbQueueMsgMetadata;
-import org.thingsboard.server.queue.discovery.HashPartitionService;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtx;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService;
@@ -383,7 +384,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
                         log.info("Initializing state for all entities in profile: tenantId=[{}], profileId=[{}]", tenantId, entityId);
                         Map<String, Argument> commonArguments = calculatedFieldCtx.getArguments().entrySet().stream()
                                 .filter(entry -> entry.getValue().getRefEntityId() != null)
-                                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
+                                .collect(Collectors.toMap(Entry::getKey, Entry::getValue));
                         fetchArguments(tenantId, entityId, commonArguments, commonArgs -> {
                             calculatedFieldCache.getEntitiesByProfile(tenantId, entityId).forEach(targetEntityId -> {
                                 initializeStateForEntity(calculatedFieldCtx, targetEntityId, commonArgs, callback);
@@ -551,7 +552,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
     private void executeTelemetryUpdate(CalculatedFieldCtx cfCtx, EntityId entityId, List<CalculatedFieldId> previousCalculatedFieldIds, Map<String, KvEntry> updatedTelemetry) {
         log.info("Received telemetry update msg: tenantId=[{}], entityId=[{}], calculatedFieldId=[{}]", cfCtx.getTenantId(), entityId, cfCtx.getCfId());
         Map<String, ArgumentEntry> argumentValues = updatedTelemetry.entrySet().stream()
-                .collect(Collectors.toMap(Map.Entry::getKey, entry -> ArgumentEntry.createSingleValueArgument(entry.getValue())));
+                .collect(Collectors.toMap(Entry::getKey, entry -> ArgumentEntry.createSingleValueArgument(entry.getValue())));
 
 //        updateOrInitializeState(cfCtx, entityId, argumentValues, previousCalculatedFieldIds);
     }
@@ -677,7 +678,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
             if (broadcast) {
                 broadcasts.add(link);
             } else {
-                TopicPartitionInfo tpi = partitionService.resolve(HashPartitionService.CALCULATED_FIELD_QUEUE_KEY, link.entityId());
+                TopicPartitionInfo tpi = partitionService.resolve(CALCULATED_FIELD_QUEUE_KEY, link.entityId());
                 unicasts.computeIfAbsent(tpi, k -> new ArrayList<>()).add(link);
             }
         }
@@ -710,7 +711,7 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
     }
 
     private CalculatedFieldLinkedTelemetryMsgProto buildLinkedTelemetryMsgProto(CalculatedFieldTelemetryMsgProto telemetryProto, List<CalculatedFieldEntityCtxId> links) {
-        TransportProtos.CalculatedFieldLinkedTelemetryMsgProto.Builder builder = TransportProtos.CalculatedFieldLinkedTelemetryMsgProto.newBuilder();
+        Builder builder = CalculatedFieldLinkedTelemetryMsgProto.newBuilder();
         builder.setMsg(telemetryProto);
         for (CalculatedFieldEntityCtxId link : links) {
             builder.addLinks(toProto(link));
@@ -719,8 +720,10 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
     }
 
     //TODO: IM: move to utils;
-    private TransportProtos.CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) {
-        return TransportProtos.CalculatedFieldEntityCtxIdProto.newBuilder()
+    private CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) {
+        return CalculatedFieldEntityCtxIdProto.newBuilder()
+                .setTenantIdMSB(ctxId.tenantId().getId().getMostSignificantBits())
+                .setTenantIdLSB(ctxId.tenantId().getId().getLeastSignificantBits())
                 .setCalculatedFieldIdMSB(ctxId.cfId().getId().getMostSignificantBits())
                 .setCalculatedFieldIdLSB(ctxId.cfId().getId().getLeastSignificantBits())
                 .setEntityType(ctxId.entityId().getEntityType().name())
@@ -890,8 +893,8 @@ public class DefaultCalculatedFieldExecutionService extends AbstractPartitionBas
         return telemetryMsg;
     }
 
-    private TransportProtos.CalculatedFieldIdProto toProto(CalculatedFieldId cfId) {
-        return TransportProtos.CalculatedFieldIdProto.newBuilder()
+    private CalculatedFieldIdProto toProto(CalculatedFieldId cfId) {
+        return CalculatedFieldIdProto.newBuilder()
                 .setCalculatedFieldIdMSB(cfId.getId().getMostSignificantBits())
                 .setCalculatedFieldIdLSB(cfId.getId().getLeastSignificantBits())
                 .build();

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

@@ -22,6 +22,8 @@ import org.rocksdb.RocksIterator;
 import org.rocksdb.WriteOptions;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Service;
+import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto;
+import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
 import org.thingsboard.server.utils.RocksDBConfig;
 
 import java.nio.charset.StandardCharsets;
@@ -49,6 +51,14 @@ public class RocksDBService {
         }
     }
 
+    public void put(CalculatedFieldEntityCtxIdProto key, CalculatedFieldStateProto value) {
+        try {
+            db.put(writeOptions, key.toByteArray(), value.toByteArray());
+        } catch (RocksDBException e) {
+            log.error("Failed to store data to RocksDB", e);
+        }
+    }
+
     public void delete(String key) {
         try {
             db.delete(writeOptions, key.getBytes(StandardCharsets.UTF_8));
@@ -67,18 +77,20 @@ public class RocksDBService {
         }
     }
 
-    public Map<String, String> getAll() {
-        Map<String, String> map = new HashMap<>();
+    public Map<CalculatedFieldEntityCtxIdProto, CalculatedFieldStateProto> getAll() {
+        Map<CalculatedFieldEntityCtxIdProto, CalculatedFieldStateProto> results = new HashMap<>();
         try (RocksIterator iterator = db.newIterator()) {
             for (iterator.seekToFirst(); iterator.isValid(); iterator.next()) {
-                String key = new String(iterator.key(), StandardCharsets.UTF_8);
-                String value = new String(iterator.value(), StandardCharsets.UTF_8);
-                map.put(key, value);
+                try {
+                    CalculatedFieldEntityCtxIdProto key = CalculatedFieldEntityCtxIdProto.parseFrom(iterator.key());
+                    CalculatedFieldStateProto value = CalculatedFieldStateProto.parseFrom(iterator.value());
+                    results.put(key, value);
+                } catch (Exception e) {
+                    log.error("Failed to retrieve data from RocksDB", e);
+                }
             }
-        } catch (Exception e) {
-            log.error("Failed to retrieve data from RocksDB", e);
         }
-        return map;
+        return results;
     }
 
 }

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

@@ -24,8 +24,6 @@ public interface CalculatedFieldStateService {
 
     Map<CalculatedFieldEntityCtxId, CalculatedFieldState> restoreStates();
 
-    CalculatedFieldState restoreState(CalculatedFieldEntityCtxId ctxId);
-
     void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback);
 
     void removeState(CalculatedFieldEntityCtxId ctxId);

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

@@ -19,14 +19,28 @@ import lombok.RequiredArgsConstructor;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.stereotype.Service;
 import org.thingsboard.common.util.JacksonUtil;
+import org.thingsboard.server.common.data.StringUtils;
+import org.thingsboard.server.common.data.cf.CalculatedFieldType;
+import org.thingsboard.server.common.data.id.CalculatedFieldId;
+import org.thingsboard.server.common.data.id.EntityId;
+import org.thingsboard.server.common.data.id.EntityIdFactory;
+import org.thingsboard.server.common.data.id.TenantId;
+import org.thingsboard.server.common.data.kv.BasicKvEntry;
+import org.thingsboard.server.common.data.kv.KvEntry;
 import org.thingsboard.server.common.msg.queue.TbCallback;
+import org.thingsboard.server.common.util.ProtoUtils;
+import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldEntityCtxIdProto;
+import org.thingsboard.server.gen.transport.TransportProtos.CalculatedFieldStateProto;
+import org.thingsboard.server.gen.transport.TransportProtos.RollingArgumentProto;
+import org.thingsboard.server.gen.transport.TransportProtos.SingleValueArgumentProto;
+import org.thingsboard.server.gen.transport.TransportProtos.SingleValueProto;
 import org.thingsboard.server.service.cf.RocksDBService;
-import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtx;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId;
 import org.thingsboard.server.service.cf.ctx.CalculatedFieldStateService;
 
 import java.util.Map;
-import java.util.Optional;
+import java.util.TreeMap;
+import java.util.UUID;
 import java.util.stream.Collectors;
 
 @Service
@@ -40,21 +54,14 @@ public class RocksDBStateService implements CalculatedFieldStateService {
     public Map<CalculatedFieldEntityCtxId, CalculatedFieldState> restoreStates() {
         return rocksDBService.getAll().entrySet().stream()
                 .collect(Collectors.toMap(
-                        entry -> JacksonUtil.fromString(entry.getKey(), CalculatedFieldEntityCtxId.class),
-                        entry -> JacksonUtil.fromString(entry.getValue(), CalculatedFieldState.class)
+                        entry -> fromProto(entry.getKey()),
+                        entry -> fromProto(entry.getValue())
                 ));
     }
 
     @Override
-    public CalculatedFieldState restoreState(CalculatedFieldEntityCtxId ctxId) {
-        return Optional.ofNullable(rocksDBService.get(JacksonUtil.writeValueAsString(ctxId)))
-                .map(storedState -> JacksonUtil.fromString(storedState, CalculatedFieldState.class))
-                .orElse(null);
-    }
-
-    @Override
-    public void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback){
-        rocksDBService.put(JacksonUtil.writeValueAsString(stateId), JacksonUtil.writeValueAsString(state));
+    public void persistState(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state, TbCallback callback) {
+        rocksDBService.put(toProto(stateId), toProto(stateId, state));
         callback.onSuccess();
     }
 
@@ -63,4 +70,120 @@ public class RocksDBStateService implements CalculatedFieldStateService {
         rocksDBService.delete(JacksonUtil.writeValueAsString(ctxId));
     }
 
+    private CalculatedFieldEntityCtxIdProto toProto(CalculatedFieldEntityCtxId ctxId) {
+        return CalculatedFieldEntityCtxIdProto.newBuilder()
+                .setTenantIdMSB(ctxId.tenantId().getId().getMostSignificantBits())
+                .setTenantIdLSB(ctxId.tenantId().getId().getLeastSignificantBits())
+                .setCalculatedFieldIdMSB(ctxId.cfId().getId().getMostSignificantBits())
+                .setCalculatedFieldIdLSB(ctxId.cfId().getId().getLeastSignificantBits())
+                .setEntityType(ctxId.entityId().getEntityType().name())
+                .setEntityIdMSB(ctxId.entityId().getId().getMostSignificantBits())
+                .setEntityIdLSB(ctxId.entityId().getId().getLeastSignificantBits())
+                .build();
+    }
+
+    private CalculatedFieldEntityCtxId fromProto(CalculatedFieldEntityCtxIdProto ctxIdProto) {
+        TenantId tenantId = TenantId.fromUUID(new UUID(ctxIdProto.getTenantIdMSB(), ctxIdProto.getTenantIdLSB()));
+        EntityId entityId = EntityIdFactory.getByTypeAndUuid(ctxIdProto.getEntityType(), new UUID(ctxIdProto.getEntityIdMSB(), ctxIdProto.getEntityIdLSB()));
+        CalculatedFieldId calculatedFieldId = new CalculatedFieldId(new UUID(ctxIdProto.getCalculatedFieldIdMSB(), ctxIdProto.getCalculatedFieldIdLSB()));
+        return new CalculatedFieldEntityCtxId(tenantId, calculatedFieldId, entityId);
+    }
+
+    private CalculatedFieldStateProto toProto(CalculatedFieldEntityCtxId stateId, CalculatedFieldState state) {
+        CalculatedFieldStateProto.Builder builder = CalculatedFieldStateProto.newBuilder()
+                .setId(toProto(stateId))
+                .setType(state.getType().name())
+                .addAllRequiredArguments(state.getRequiredArguments());
+
+        state.getArguments().forEach((argName, argEntry) -> {
+            if (argEntry instanceof SingleValueArgumentEntry singleValueArgumentEntry) {
+                builder.addSingleValueArguments(toSingleValueArgumentProto(argName, singleValueArgumentEntry));
+            } else if (argEntry instanceof TsRollingArgumentEntry rollingArgumentEntry) {
+                builder.addRollingValueArguments(toRollingArgumentProto(argName, rollingArgumentEntry));
+            }
+        });
+
+        return builder.build();
+    }
+
+    private SingleValueArgumentProto toSingleValueArgumentProto(String argName, SingleValueArgumentEntry entry) {
+        SingleValueProto.Builder singleValueProtoBuilder = SingleValueProto.newBuilder()
+                .setTs(entry.getTs());
+
+        if (entry.getVersion() != null) {
+            singleValueProtoBuilder.setVersion(entry.getVersion());
+        }
+
+        KvEntry value = entry.getValue();
+        if (value != null) {
+            singleValueProtoBuilder.setHasV(true)
+                    .setValue(ProtoUtils.toKeyValueProto(value));
+        }
+
+        return SingleValueArgumentProto.newBuilder()
+                .setArgName(argName)
+                .setValue(singleValueProtoBuilder.build())
+                .build();
+    }
+
+    private RollingArgumentProto toRollingArgumentProto(String argName, TsRollingArgumentEntry entry) {
+        RollingArgumentProto.Builder rollingArgumentProtoBuilder = RollingArgumentProto.newBuilder()
+                .setArgName(argName);
+
+        entry.getTsRecords().forEach((ts, value) -> {
+            SingleValueProto.Builder singleValueProtoBuilder = SingleValueProto.newBuilder()
+                    .setTs(ts);
+
+            if (value != null) {
+                singleValueProtoBuilder.setHasV(true)
+                        .setValue(ProtoUtils.toKeyValueProto(value));
+            }
+
+            rollingArgumentProtoBuilder.addValues(singleValueProtoBuilder.build());
+        });
+
+        return rollingArgumentProtoBuilder.build();
+    }
+
+    private CalculatedFieldState fromProto(CalculatedFieldStateProto proto) {
+        if (StringUtils.isEmpty(proto.getType())) {
+            return null;
+        }
+
+        CalculatedFieldType type = CalculatedFieldType.valueOf(proto.getType());
+
+        CalculatedFieldState state = switch (type) {
+            case SIMPLE -> new SimpleCalculatedFieldState(proto.getRequiredArgumentsList());
+            case SCRIPT -> new ScriptCalculatedFieldState(proto.getRequiredArgumentsList());
+        };
+
+        proto.getSingleValueArgumentsList().forEach(argProto ->
+                state.getArguments().put(argProto.getArgName(), fromSingleValueArgumentProto(argProto)));
+
+        if (CalculatedFieldType.SCRIPT.equals(type)) {
+            proto.getRollingValueArgumentsList().forEach(argProto ->
+                    state.getArguments().put(argProto.getArgName(), fromRollingArgumentProto(argProto)));
+        }
+
+        return state;
+    }
+
+    private SingleValueArgumentEntry fromSingleValueArgumentProto(SingleValueArgumentProto proto) {
+        SingleValueProto valueProto = proto.getValue();
+        BasicKvEntry value = valueProto.getHasV() ? ProtoUtils.fromProto(valueProto.getValue()) : null;
+
+        return new SingleValueArgumentEntry(valueProto.getTs(), value, valueProto.getVersion());
+    }
+
+    private TsRollingArgumentEntry fromRollingArgumentProto(RollingArgumentProto proto) {
+        TreeMap<Long, BasicKvEntry> tsRecords = new TreeMap<>();
+
+        proto.getValuesList().forEach(singleValueProto -> {
+            BasicKvEntry value = singleValueProto.getHasV() ? ProtoUtils.fromProto(singleValueProto.getValue()) : null;
+            tsRecords.put(singleValueProto.getTs(), value);
+        });
+
+        return new TsRollingArgumentEntry(tsRecords);
+    }
+
 }

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

@@ -24,6 +24,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.cf.configuration.Argument;
 import org.thingsboard.server.common.data.cf.configuration.Output;
+import org.thingsboard.server.common.data.kv.BasicKvEntry;
 import org.thingsboard.server.service.cf.CalculatedFieldResult;
 
 import java.util.List;
@@ -53,7 +54,7 @@ public class ScriptCalculatedFieldState extends BaseCalculatedFieldState {
         arguments.forEach((key, argumentEntry) -> {
             if (argumentEntry instanceof TsRollingArgumentEntry tsRollingEntry) {
                 Argument argument = ctx.getArguments().get(key);
-                TreeMap<Long, Object> tsRecords = tsRollingEntry.getTsRecords();
+                TreeMap<Long, BasicKvEntry> tsRecords = tsRollingEntry.getTsRecords();
                 if (tsRecords.size() > argument.getLimit()) {
                     tsRecords.pollFirstEntry();
                 }

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

@@ -21,6 +21,7 @@ import lombok.Data;
 import lombok.NoArgsConstructor;
 import org.thingsboard.server.common.data.cf.CalculatedFieldType;
 import org.thingsboard.server.common.data.cf.configuration.Output;
+import org.thingsboard.server.common.data.kv.BasicKvEntry;
 import org.thingsboard.server.service.cf.CalculatedFieldResult;
 
 import java.util.List;
@@ -52,7 +53,8 @@ public class SimpleCalculatedFieldState extends BaseCalculatedFieldState {
 
         for (Map.Entry<String, ArgumentEntry> entry : this.arguments.entrySet()) {
             try {
-                expr.setVariable(entry.getKey(), Double.parseDouble(entry.getValue().getValue().toString()));
+                BasicKvEntry kvEntry = ((SingleValueArgumentEntry) entry.getValue()).getValue();
+                expr.setVariable(entry.getKey(), Double.parseDouble(kvEntry.getValueAsString()));
             } catch (NumberFormatException e) {
                 throw new IllegalArgumentException("Argument '" + entry.getKey() + "' is not a number.");
             }

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

@@ -19,6 +19,7 @@ import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BasicKvEntry;
 import org.thingsboard.server.common.data.kv.KvEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
 import org.thingsboard.server.common.util.ProtoUtils;
@@ -33,19 +34,19 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
     public static final ArgumentEntry EMPTY = new SingleValueArgumentEntry(0);
 
     private long ts;
-    private Object value;
+    private BasicKvEntry value;
     private Long version;
 
     public SingleValueArgumentEntry(TsKvProto entry) {
         this.ts = entry.getTs();
         this.version = entry.getVersion();
-        this.value = ProtoUtils.fromProto(entry).getValue();
+        this.value = ProtoUtils.fromProto(entry.getKv());
     }
 
     public SingleValueArgumentEntry(AttributeValueProto entry) {
         this.ts = entry.getLastUpdateTs();
         this.version = entry.getVersion();
-        this.value = ProtoUtils.fromProto(entry).getValue();
+        this.value = ProtoUtils.basicKvEntryFromProto(entry);
     }
 
     public SingleValueArgumentEntry(KvEntry entry) {
@@ -56,7 +57,7 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
             this.ts = attributeKvEntry.getLastUpdateTs();
             this.version = attributeKvEntry.getVersion();
         }
-        this.value = entry.getValue();
+        this.value = ProtoUtils.basicKvEntryFromKvEntry(entry);
     }
 
     /**
@@ -72,11 +73,6 @@ public class SingleValueArgumentEntry implements ArgumentEntry {
         return ArgumentEntryType.SINGLE_VALUE;
     }
 
-    @Override
-    public Object getValue() {
-        return value;
-    }
-
     @Override
     public ArgumentEntry copy() {
         return new SingleValueArgumentEntry(this.ts, this.value, this.version);

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

@@ -21,11 +21,15 @@ import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.math.NumberUtils;
+import org.thingsboard.server.common.data.kv.BasicKvEntry;
+import org.thingsboard.server.common.data.kv.KvEntry;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
+import org.thingsboard.server.common.util.ProtoUtils;
 
 import java.util.List;
 import java.util.Map;
 import java.util.TreeMap;
+import java.util.stream.Collectors;
 
 @Data
 @NoArgsConstructor
@@ -37,10 +41,10 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
 
     private static final int MAX_ROLLING_ARGUMENT_ENTRY_SIZE = 1000;
 
-    private TreeMap<Long, Object> tsRecords = new TreeMap<>();
+    private TreeMap<Long, BasicKvEntry> tsRecords = new TreeMap<>();
 
     public TsRollingArgumentEntry(List<TsKvEntry> kvEntries) {
-        kvEntries.forEach(tsKvEntry -> addTsRecord(tsKvEntry.getTs(), tsKvEntry.getValue()));
+        kvEntries.forEach(tsKvEntry -> addTsRecord(tsKvEntry.getTs(), tsKvEntry));
     }
 
     /**
@@ -58,7 +62,15 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
     @JsonIgnore
     @Override
     public Object getValue() {
-        return tsRecords;
+        return tsRecords.entrySet()
+                .stream()
+                .collect(Collectors.toMap(
+                        Map.Entry::getKey,
+                        entry -> entry.getValue().getValue(),
+                        (oldValue, newValue) -> oldValue,
+                        TreeMap::new
+                ));
+
     }
 
     @Override
@@ -79,7 +91,7 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
 
     private boolean updateTsRollingEntry(TsRollingArgumentEntry tsRollingEntry) {
         boolean updated = false;
-        for (Map.Entry<Long, Object> tsRecordEntry : tsRollingEntry.getTsRecords().entrySet()) {
+        for (Map.Entry<Long, BasicKvEntry> tsRecordEntry : tsRollingEntry.getTsRecords().entrySet()) {
             updated |= addTsRecordIfAbsent(tsRecordEntry.getKey(), tsRecordEntry.getValue());
         }
         return updated;
@@ -89,7 +101,7 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
         return addTsRecordIfAbsent(singleValueEntry.getTs(), singleValueEntry.getValue());
     }
 
-    private boolean addTsRecordIfAbsent(Long ts, Object value) {
+    private boolean addTsRecordIfAbsent(Long ts, KvEntry value) {
         if (!tsRecords.containsKey(ts)) {
             addTsRecord(ts, value);
             return true;
@@ -97,9 +109,9 @@ public class TsRollingArgumentEntry implements ArgumentEntry {
         return false;
     }
 
-    private void addTsRecord(Long ts, Object value) {
-        if (NumberUtils.isParsable(value.toString())) {
-            tsRecords.put(ts, value);
+    private void addTsRecord(Long ts, KvEntry value) {
+        if (NumberUtils.isParsable(value.getValue().toString())) {
+            tsRecords.put(ts, ProtoUtils.basicKvEntryFromKvEntry(value));
             if (tsRecords.size() > MAX_ROLLING_ARGUMENT_ENTRY_SIZE) {
                 tsRecords.pollFirstEntry();
             }

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

@@ -17,6 +17,9 @@ package org.thingsboard.server.service.cf.ctx.state;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.thingsboard.server.common.data.kv.BasicKvEntry;
+import org.thingsboard.server.common.data.kv.DoubleDataEntry;
+import org.thingsboard.server.common.data.kv.StringDataEntry;
 
 import java.util.Map;
 import java.util.TreeMap;
@@ -32,10 +35,10 @@ public class TsRollingArgumentEntryTest {
 
     @BeforeEach
     void setUp() {
-        TreeMap<Long, Object> values = new TreeMap<>();
-        values.put(ts - 40, 10);
-        values.put(ts - 30, 12);
-        values.put(ts - 20, 17);
+        TreeMap<Long, BasicKvEntry> values = new TreeMap<>();
+        values.put(ts - 40, new DoubleDataEntry("key", 10.0));
+        values.put(ts - 30, new DoubleDataEntry("key", 12.0));
+        values.put(ts - 20, new DoubleDataEntry("key", 17.0));
 
         entry = new TsRollingArgumentEntry(values);
     }
@@ -47,7 +50,7 @@ public class TsRollingArgumentEntryTest {
 
     @Test
     void testUpdateEntryWhenSingleValueEntryPassed() {
-        SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 10, 23, 123L);
+        SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 10, new DoubleDataEntry("key", 23.0), 123L);
 
         assertThat(entry.updateEntry(newEntry)).isTrue();
         assertThat(entry.getTsRecords()).hasSize(4);
@@ -56,7 +59,7 @@ public class TsRollingArgumentEntryTest {
 
     @Test
     void testUpdateEntryWhenSingleValueEntryWithTheSameTsPassed() {
-        SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 20, 23, 123L);
+        SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 20, new DoubleDataEntry("key", 23.0), 123L);
 
         assertThat(entry.updateEntry(newEntry)).isFalse();
     }
@@ -64,10 +67,10 @@ public class TsRollingArgumentEntryTest {
     @Test
     void testUpdateEntryWhenRollingEntryPassed() {
         TsRollingArgumentEntry newEntry = new TsRollingArgumentEntry();
-        TreeMap<Long, Object> values = new TreeMap<>();
-        values.put(ts - 20, 16);
-        values.put(ts - 10, 7);
-        values.put(ts - 5, 1);
+        TreeMap<Long, BasicKvEntry> values = new TreeMap<>();
+        values.put(ts - 20, new DoubleDataEntry("key", 16.0));
+        values.put(ts - 10, new DoubleDataEntry("key", 7.0));
+        values.put(ts - 5, new DoubleDataEntry("key", 1.0));
         newEntry.setTsRecords(values);
 
         assertThat(entry.updateEntry(newEntry)).isTrue();
@@ -83,7 +86,7 @@ public class TsRollingArgumentEntryTest {
 
     @Test
     void testUpdateEntryWhenValueIsNotNumber() {
-        SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 10, "string", 123L);
+        SingleValueArgumentEntry newEntry = new SingleValueArgumentEntry(ts - 10, new StringDataEntry("key", "string"), 123L);
 
         assertThatThrownBy(() -> entry.updateEntry(newEntry))
                 .isInstanceOf(IllegalArgumentException.class)

common/proto/src/main/java/org/thingsboard/server/common/util/ProtoUtils.java

@@ -58,6 +58,7 @@ import org.thingsboard.server.common.data.id.TenantProfileId;
 import org.thingsboard.server.common.data.kv.AttributeKey;
 import org.thingsboard.server.common.data.kv.AttributeKvEntry;
 import org.thingsboard.server.common.data.kv.BaseAttributeKvEntry;
+import org.thingsboard.server.common.data.kv.BasicKvEntry;
 import org.thingsboard.server.common.data.kv.BasicTsKvEntry;
 import org.thingsboard.server.common.data.kv.BooleanDataEntry;
 import org.thingsboard.server.common.data.kv.DoubleDataEntry;
@@ -90,7 +91,7 @@ import org.thingsboard.server.common.msg.rule.engine.DeviceDeleteMsg;
 import org.thingsboard.server.common.msg.rule.engine.DeviceEdgeUpdateMsg;
 import org.thingsboard.server.common.msg.rule.engine.DeviceNameOrTypeUpdateMsg;
 import org.thingsboard.server.gen.transport.TransportProtos;
-import org.thingsboard.server.gen.transport.TransportProtos.TsKvProto;
+import org.thingsboard.server.gen.transport.TransportProtos.KeyValueProto;
 
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -630,6 +631,42 @@ public class ProtoUtils {
         return new BaseAttributeKvEntry(entry, proto.getLastUpdateTs(), proto.hasVersion() ? proto.getVersion() : null);
     }
 
+    public static BasicKvEntry basicKvEntryFromProto(TransportProtos.AttributeValueProto proto) {
+        boolean hasValue = proto.getHasV();
+        String key = proto.getKey();
+        return switch (proto.getType()) {
+            case BOOLEAN_V -> new BooleanDataEntry(key, hasValue ? proto.getBoolV() : null);
+            case LONG_V -> new LongDataEntry(key, hasValue ? proto.getLongV() : null);
+            case DOUBLE_V -> new DoubleDataEntry(key, hasValue ? proto.getDoubleV() : null);
+            case STRING_V -> new StringDataEntry(key, hasValue ? proto.getStringV() : null);
+            case JSON_V -> new JsonDataEntry(key, hasValue ? proto.getJsonV() : null);
+            default -> null;
+        };
+    }
+
+    public static BasicKvEntry fromProto(KeyValueProto proto) {
+        String key = proto.getKey();
+        return switch (proto.getType()) {
+            case BOOLEAN_V -> new BooleanDataEntry(key, proto.getBoolV());
+            case LONG_V -> new LongDataEntry(key, proto.getLongV());
+            case DOUBLE_V -> new DoubleDataEntry(key, proto.getDoubleV());
+            case STRING_V -> new StringDataEntry(key, proto.getStringV());
+            case JSON_V -> new JsonDataEntry(key, proto.getJsonV());
+            default -> null;
+        };
+    }
+
+    public static BasicKvEntry basicKvEntryFromKvEntry(KvEntry kvEntry) {
+        String key = kvEntry.getKey();
+        return switch (kvEntry.getDataType()) {
+            case BOOLEAN -> new BooleanDataEntry(key, kvEntry.getBooleanValue().orElse(null));
+            case LONG -> new LongDataEntry(key, kvEntry.getLongValue().orElse(null));
+            case DOUBLE -> new DoubleDataEntry(key, kvEntry.getDoubleValue().orElse(null));
+            case STRING -> new StringDataEntry(key, kvEntry.getStrValue().orElse(null));
+            case JSON -> new JsonDataEntry(key, kvEntry.getJsonValue().orElse(null));
+        };
+    }
+
     public static TsKvEntry fromProto(TransportProtos.TsKvProto proto) {
         TransportProtos.KeyValueProto kvProto = proto.getKv();
         String key = kvProto.getKey();

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

@@ -810,11 +810,13 @@ message CalculatedFieldLinkedTelemetryMsgProto {
 }
 
 message CalculatedFieldEntityCtxIdProto {
-  int64 calculatedFieldIdMSB = 1;
-  int64 calculatedFieldIdLSB = 2;
-  string entityType = 3;
-  int64 entityIdMSB = 4;
-  int64 entityIdLSB = 5;
+  int64 tenantIdMSB = 1;
+  int64 tenantIdLSB = 2;
+  int64 calculatedFieldIdMSB = 3;
+  int64 calculatedFieldIdLSB = 4;
+  string entityType = 5;
+  int64 entityIdMSB = 6;
+  int64 entityIdLSB = 7;
 }
 
 message CalculatedFieldIdProto {
@@ -825,13 +827,8 @@ message CalculatedFieldIdProto {
 message SingleValueProto {
   int64 ts = 1;
   int64 version = 2;
-  KeyValueType type = 3;
   bool has_v = 4;
-  bool bool_v = 5;
-  int64 long_v = 6;
-  double double_v = 7;
-  string string_v = 8;
-  string json_v = 9;
+  KeyValueProto value = 5;
 }
 
 message SingleValueArgumentProto {
@@ -847,8 +844,10 @@ message RollingArgumentProto {
 message CalculatedFieldStateProto {
   CalculatedFieldEntityCtxIdProto id = 1;
   // int32 version = 2;
-  repeated SingleValueArgumentProto singleValueArguments = 3;
-  repeated RollingArgumentProto rollingValueArguments = 4;
+  string type = 3;
+  repeated string requiredArguments = 4;
+  repeated SingleValueArgumentProto singleValueArguments = 5;
+  repeated RollingArgumentProto rollingValueArguments = 6;
 }
 
 //Used to report session state to tb-Service and persist this state in the cache on the tb-Service level.