From 6a307041b699f4d2a5d1c27e8755b8538f0fec31 Mon Sep 17 00:00:00 2001 From: IrynaMatveieva Date: Fri, 17 Oct 2025 15:55:44 +0300 Subject: [PATCH] added tests for entry --- ...CalculatedFieldEntityMessageProcessor.java | 4 +- ...tractCalculatedFieldProcessingService.java | 4 +- .../DefaultCalculatedFieldQueueService.java | 24 ++-- .../ctx/state/BaseCalculatedFieldState.java | 3 +- .../state/aggregation/AggArgumentEntry.java | 7 +- .../AggSingleEntityArgumentEntry.java | 30 +++-- .../cf/ctx/state/AggArgumentEntryTest.java | 112 ++++++++++++++++++ .../AggSingleEntityArgumentEntryTest.java | 101 ++++++++++++++++ .../server/dao/relation/RelationService.java | 2 + .../data/cf/configuration/Argument.java | 4 + ...gregationCalculatedFieldConfiguration.java | 13 ++ .../dao/relation/BaseRelationService.java | 15 +++ 12 files changed, 288 insertions(+), 31 deletions(-) create mode 100644 application/src/test/java/org/thingsboard/server/service/cf/ctx/state/AggArgumentEntryTest.java create mode 100644 application/src/test/java/org/thingsboard/server/service/cf/ctx/state/AggSingleEntityArgumentEntryTest.java diff --git a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java index ce29dc06e0..58113deb08 100644 --- a/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java +++ b/application/src/main/java/org/thingsboard/server/actors/calculatedField/CalculatedFieldEntityMessageProcessor.java @@ -67,7 +67,6 @@ import java.util.HashSet; import java.util.LinkedList; import java.util.List; import java.util.Map; -import java.util.Map.Entry; import java.util.Set; import java.util.UUID; import java.util.concurrent.TimeUnit; @@ -686,9 +685,8 @@ public class CalculatedFieldEntityMessageProcessor extends AbstractContextAwareM Map.Entry::getKey, argEntry -> new AggSingleEntityArgumentEntry(entityId, argEntry.getValue()) )); - } else { - fetchedArgs.values().forEach(arg -> arg.setForceResetPrevious(true)); } + fetchedArgs.values().forEach(arg -> arg.setForceResetPrevious(true)); return fetchedArgs; } 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 94695bcc8c..7f47dba2ed 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 @@ -332,7 +332,7 @@ public abstract class AbstractCalculatedFieldProcessingService { var attributeOptFuture = attributesService.find(tenantId, entityId, argument.getRefEntityKey().getScope(), argument.getRefEntityKey().getKey()); return Futures.transform(attributeOptFuture, attrOpt -> { log.debug("[{}][{}] Fetched attribute for key {}: {}", tenantId, entityId, argument.getRefEntityKey(), attrOpt); - AttributeKvEntry attributeKvEntry = attrOpt.orElseGet(() -> new BaseAttributeKvEntry(createDefaultKvEntry(argument), defaultLastUpdateTs, 0L)); + AttributeKvEntry attributeKvEntry = attrOpt.orElseGet(() -> new BaseAttributeKvEntry(createDefaultKvEntry(argument), defaultLastUpdateTs, SingleValueArgumentEntry.DEFAULT_VERSION)); return transformAggSingleArgument(entityId, Optional.of(attributeKvEntry)); }, calculatedFieldCallbackExecutor); } @@ -344,7 +344,7 @@ public abstract class AbstractCalculatedFieldProcessingService { timeseriesService.findLatest(tenantId, entityId, key), result -> { log.debug("[{}][{}] Fetched latest timeseries {}: {}", tenantId, entityId, key, result); - Optional tsKvEntry = result.or(() -> Optional.of(new BasicTsKvEntry(defaultTs, createDefaultKvEntry(argument), 0L))); + Optional tsKvEntry = result.or(() -> Optional.of(new BasicTsKvEntry(defaultTs, createDefaultKvEntry(argument), SingleValueArgumentEntry.DEFAULT_VERSION))); return transformAggSingleArgument(entityId, tsKvEntry); }, calculatedFieldCallbackExecutor); } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java index 37c7b2ef6c..147882c3c2 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/DefaultCalculatedFieldQueueService.java @@ -37,8 +37,9 @@ import org.thingsboard.server.common.data.kv.TimeseriesSaveResult; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.msg.TbMsgType; import org.thingsboard.server.common.data.relation.EntityRelation; +import org.thingsboard.server.common.data.relation.EntityRelationPathQuery; +import org.thingsboard.server.common.data.relation.EntitySearchDirection; import org.thingsboard.server.common.data.relation.RelationPathLevel; -import org.thingsboard.server.common.data.relation.RelationTypeGroup; import org.thingsboard.server.common.util.ProtoUtils; import org.thingsboard.server.dao.relation.RelationService; import org.thingsboard.server.gen.transport.TransportProtos.AttributeScopeProto; @@ -191,20 +192,15 @@ public class DefaultCalculatedFieldQueueService implements CalculatedFieldQueueS for (CalculatedFieldCtx cfCtx : cfCtxs) { if (cfCtx.getCalculatedField().getConfiguration() instanceof LatestValuesAggregationCalculatedFieldConfiguration aggConfig) { RelationPathLevel relation = aggConfig.getRelation(); - switch (relation.direction()) { - case FROM -> { - List byToAndType = relationService.findByToAndType(tenantId, entityId, relation.relationType(), RelationTypeGroup.COMMON); - if (!byToAndType.isEmpty()) { - return true; - } - } - case TO -> { - List byFromAndType = relationService.findByFromAndType(tenantId, entityId, relation.relationType(), RelationTypeGroup.COMMON); - if (!byFromAndType.isEmpty()) { - return true; - } - } + EntitySearchDirection inverseDirection = switch (relation.direction()) { + case FROM -> EntitySearchDirection.TO; + case TO -> EntitySearchDirection.FROM; }; + RelationPathLevel inverseRelation = new RelationPathLevel(inverseDirection, relation.relationType()); + List byRelationPathQuery = relationService.findByRelationPathQuery(tenantId, new EntityRelationPathQuery(entityId, List.of(inverseRelation))); + if (!byRelationPathQuery.isEmpty()) { + return true; + } } } diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java index 2d853fd3fd..ee10857e6f 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/BaseCalculatedFieldState.java @@ -22,6 +22,7 @@ import org.thingsboard.server.actors.TbActorRef; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.msg.queue.TopicPartitionInfo; import org.thingsboard.server.service.cf.ctx.CalculatedFieldEntityCtxId; +import org.thingsboard.server.service.cf.ctx.state.aggregation.AggSingleEntityArgumentEntry; import org.thingsboard.server.utils.CalculatedFieldUtils; import java.io.Closeable; @@ -73,7 +74,7 @@ public abstract class BaseCalculatedFieldState implements CalculatedFieldState, ArgumentEntry existingEntry = arguments.get(key); boolean entryUpdated; - if (existingEntry == null || newEntry.isForceResetPrevious()) { + if (existingEntry == null || !(newEntry instanceof AggSingleEntityArgumentEntry) && newEntry.isForceResetPrevious()) { validateNewEntry(key, newEntry); arguments.put(key, newEntry); entryUpdated = true; diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggArgumentEntry.java index 7e3a8623e4..e002aece2c 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggArgumentEntry.java @@ -52,7 +52,12 @@ public class AggArgumentEntry implements ArgumentEntry { if (aggSingleEntityArgumentEntry.isDeleted()) { aggInputs.remove(aggSingleEntityArgumentEntry.getEntityId()); } else { - aggInputs.put(aggSingleEntityArgumentEntry.getEntityId(), aggSingleEntityArgumentEntry); + ArgumentEntry argumentEntry = aggInputs.get(aggSingleEntityArgumentEntry.getEntityId()); + if (argumentEntry != null) { + argumentEntry.updateEntry(aggSingleEntityArgumentEntry); + } else { + aggInputs.put(aggSingleEntityArgumentEntry.getEntityId(), aggSingleEntityArgumentEntry); + } } return true; } else { diff --git a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggSingleEntityArgumentEntry.java b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggSingleEntityArgumentEntry.java index 32ce77311d..6430fe3f1c 100644 --- a/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggSingleEntityArgumentEntry.java +++ b/application/src/main/java/org/thingsboard/server/service/cf/ctx/state/aggregation/AggSingleEntityArgumentEntry.java @@ -62,25 +62,35 @@ public class AggSingleEntityArgumentEntry extends SingleValueArgumentEntry { @Override public boolean updateEntry(ArgumentEntry entry) { - if (entry instanceof AggSingleEntityArgumentEntry singleValueEntry) { - if (singleValueEntry.getTs() <= ts) { - return false; + if (entry instanceof AggSingleEntityArgumentEntry aggSingleEntityEntry) { + if (aggSingleEntityEntry.isForceResetPrevious()) { + return applyNewEntry(aggSingleEntityEntry); } - Long newVersion = singleValueEntry.getVersion(); + if (aggSingleEntityEntry.getTs() < this.ts) { + if (!isDefaultValue()) { + return false; + } + } + + Long newVersion = aggSingleEntityEntry.getVersion(); if (newVersion == null || this.version == null || newVersion > this.version) { - this.ts = singleValueEntry.getTs(); - this.version = newVersion; - this.kvEntryValue = singleValueEntry.getKvEntryValue(); - this.entityId = singleValueEntry.getEntityId(); - return true; + return applyNewEntry(aggSingleEntityEntry); } } else { - throw new IllegalArgumentException("Unsupported argument entry type for single value argument entry: " + entry.getType()); + throw new IllegalArgumentException("Unsupported argument entry type for aggregation single entity argument entry: " + entry.getType()); } return false; } + private boolean applyNewEntry(AggSingleEntityArgumentEntry entry) { + this.ts = entry.getTs(); + this.version = entry.getVersion(); + this.kvEntryValue = entry.getKvEntryValue(); + this.entityId = entry.getEntityId(); + return true; + } + @Override public ArgumentEntryType getType() { return ArgumentEntryType.AGGREGATE_LATEST_SINGLE; diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/AggArgumentEntryTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/AggArgumentEntryTest.java new file mode 100644 index 0000000000..c1c2d85c55 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/AggArgumentEntryTest.java @@ -0,0 +1,112 @@ +/** + * Copyright © 2016-2025 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +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.id.DeviceId; +import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.kv.BasicTsKvEntry; +import org.thingsboard.server.common.data.kv.LongDataEntry; +import org.thingsboard.server.service.cf.ctx.state.aggregation.AggArgumentEntry; +import org.thingsboard.server.service.cf.ctx.state.aggregation.AggSingleEntityArgumentEntry; + +import java.util.HashMap; +import java.util.Map; +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +public class AggArgumentEntryTest { + + private AggArgumentEntry entry; + + private final DeviceId device1 = new DeviceId(UUID.fromString("1984e5f4-9ff0-4187-84ae-e4438bba4c8a")); + private final DeviceId device2 = new DeviceId(UUID.fromString("937fc062-1a9d-438f-aa22-55a93fc908b7")); + + private final long ts = System.currentTimeMillis(); + + @BeforeEach + void setUp() { + Map aggInputs = new HashMap<>(); + aggInputs.put(device1, new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 100, new LongDataEntry("key", 12L), 1L))); + aggInputs.put(device2, new AggSingleEntityArgumentEntry(device2, new BasicTsKvEntry(ts - 150, new LongDataEntry("key", 16L), 6L))); + + entry = new AggArgumentEntry(aggInputs, false); + } + + @Test + void testUpdateEntryWhenNotAggEntryPassed() { + assertThatThrownBy(() -> entry.updateEntry(new TsRollingArgumentEntry(5, 30000L))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Unsupported argument entry type for aggregation argument entry: " + ArgumentEntryType.TS_ROLLING); + } + + @Test + void testUpdateEntryWhenAggArgumentEntryPasser() { + DeviceId device3 = new DeviceId(UUID.randomUUID()); + DeviceId device4 = new DeviceId(UUID.randomUUID()); + + AggArgumentEntry aggArgumentEntry = new AggArgumentEntry(Map.of( + device3, new AggSingleEntityArgumentEntry(device3, new BasicTsKvEntry(ts - 50, new LongDataEntry("key", 16L), 13L)), + device4, new AggSingleEntityArgumentEntry(device4, new BasicTsKvEntry(ts - 60, new LongDataEntry("key", 23L), 7L)) + ), false); + + assertThat(entry.updateEntry(aggArgumentEntry)).isTrue(); + + Map aggInputs = entry.getAggInputs(); + assertThat(aggInputs.size()).isEqualTo(4); + assertThat(aggInputs.get(device3)).isEqualTo(aggArgumentEntry.getAggInputs().get(device3)); + assertThat(aggInputs.get(device4)).isEqualTo(aggArgumentEntry.getAggInputs().get(device4)); + } + + @Test + void testUpdateEntryWhenAggSingleEntityArgumentEntryPassedAndNoEntriesById() { + DeviceId device3 = new DeviceId(UUID.randomUUID()); + + AggSingleEntityArgumentEntry singleEntityArgumentEntry = new AggSingleEntityArgumentEntry(device3, new BasicTsKvEntry(ts - 50, new LongDataEntry("key", 18L), 10L)); + + assertThat(entry.updateEntry(singleEntityArgumentEntry)).isTrue(); + + Map aggInputs = entry.getAggInputs(); + assertThat(aggInputs.size()).isEqualTo(3); + assertThat(aggInputs.get(device3)).isEqualTo(singleEntityArgumentEntry); + } + + @Test + void testUpdateEntryWhenAggSingleEntityArgumentEntryPassedAndEntryByIdExist() { + AggSingleEntityArgumentEntry singleEntityArgumentEntry = new AggSingleEntityArgumentEntry(device2, new BasicTsKvEntry(ts - 50, new LongDataEntry("key", 18L), 10L)); + + assertThat(entry.updateEntry(singleEntityArgumentEntry)).isTrue(); + + Map aggInputs = entry.getAggInputs(); + assertThat(aggInputs.size()).isEqualTo(2); + assertThat(aggInputs.get(device2)).isEqualTo(singleEntityArgumentEntry); + } + + @Test + void testUpdateEntryWhenDeletedAggSingleEntityArgumentEntryPassed() { + AggSingleEntityArgumentEntry singleEntityArgumentEntry = new AggSingleEntityArgumentEntry(device2, true); + + assertThat(entry.updateEntry(singleEntityArgumentEntry)).isTrue(); + + Map aggInputs = entry.getAggInputs(); + assertThat(aggInputs.size()).isEqualTo(1); + assertThat(aggInputs.get(device2)).isNull(); + } + +} diff --git a/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/AggSingleEntityArgumentEntryTest.java b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/AggSingleEntityArgumentEntryTest.java new file mode 100644 index 0000000000..0401bb0156 --- /dev/null +++ b/application/src/test/java/org/thingsboard/server/service/cf/ctx/state/AggSingleEntityArgumentEntryTest.java @@ -0,0 +1,101 @@ +/** + * Copyright © 2016-2025 The Thingsboard Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +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.id.DeviceId; +import org.thingsboard.server.common.data.kv.BasicTsKvEntry; +import org.thingsboard.server.common.data.kv.LongDataEntry; +import org.thingsboard.server.service.cf.ctx.state.aggregation.AggSingleEntityArgumentEntry; + +import java.util.UUID; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +public class AggSingleEntityArgumentEntryTest { + + private AggSingleEntityArgumentEntry entry; + + private final DeviceId device1 = new DeviceId(UUID.fromString("1984e5f4-9ff0-4187-84ae-e4438bba4c8a")); + + private final long ts = System.currentTimeMillis(); + + @BeforeEach + void setUp() { + entry = new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 100, new LongDataEntry("key", 12L), 22L)); + } + + @Test + void testUpdateEntryWhenNotAggEntryPassed() { + assertThatThrownBy(() -> entry.updateEntry(new TsRollingArgumentEntry(5, 30000L))) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Unsupported argument entry type for aggregation single entity argument entry: " + ArgumentEntryType.TS_ROLLING); + } + + @Test + void testUpdateEntryWhenResetPrevious() { + AggSingleEntityArgumentEntry singleEntityArgumentEntry = new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 50, new LongDataEntry("key", 18L), 100L)); + singleEntityArgumentEntry.setForceResetPrevious(true); + + assertThat(entry.updateEntry(singleEntityArgumentEntry)).isTrue(); + assertThat(entry.getTs()).isEqualTo(singleEntityArgumentEntry.getTs()); + assertThat(entry.getKvEntryValue()).isEqualTo(singleEntityArgumentEntry.getKvEntryValue()); + assertThat(entry.getVersion()).isEqualTo(singleEntityArgumentEntry.getVersion()); + } + + + @Test + void testUpdateEntryWithTheSameTsAndVersion() { + assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 100, new LongDataEntry("key", 19L), 22L)))).isFalse(); + } + + @Test + void testUpdateEntryWithTheSameTsAndDifferentVersion() { + assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 100, new LongDataEntry("key", 134L), 23L)))).isTrue(); + } + + @Test + void testUpdateEntryWhenNewVersionIsNull() { + assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 40, new LongDataEntry("key", 56L), null)))).isTrue(); + assertThat(entry.getValue()).isEqualTo(56L); + assertThat(entry.getVersion()).isNull(); + } + + @Test + void testUpdateEntryWhenNewVersionIsGreaterThanCurrent() { + assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 40, new LongDataEntry("key", 76L), 23L)))).isTrue(); + assertThat(entry.getValue()).isEqualTo(76L); + assertThat(entry.getVersion()).isEqualTo(23); + } + + @Test + void testUpdateEntryWhenNewVersionIsLessThanCurrent() { + assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 40, new LongDataEntry("key", 11L), 20L)))).isFalse(); + } + + @Test + void testUpdateEntryWhenValueWasNotChanged() { + assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 40, new LongDataEntry("key", 18L), 45L)))).isTrue(); + } + + @Test + void testUpdateEntryWithOldTs() { + assertThat(entry.updateEntry(new AggSingleEntityArgumentEntry(device1, new BasicTsKvEntry(ts - 150, new LongDataEntry("key", 155L), 45L)))).isFalse(); + } + +} diff --git a/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java b/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java index a0bc9a72e6..214dc5247e 100644 --- a/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java +++ b/common/dao-api/src/main/java/org/thingsboard/server/dao/relation/RelationService.java @@ -86,6 +86,8 @@ public interface RelationService { ListenableFuture> findByRelationPathQueryAsync(TenantId tenantId, EntityRelationPathQuery relationPathQuery); + List findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery); + // TODO: This method may be useful for some validations in the future // ListenableFuture checkRecursiveRelation(EntityId from, EntityId to); diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/Argument.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/Argument.java index 4de4551e73..8d4a831cf9 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/Argument.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/Argument.java @@ -45,4 +45,8 @@ public class Argument { return hasDynamicSource() && refDynamicSourceConfiguration.getType() == CFArgumentDynamicSourceType.CURRENT_OWNER; } + public boolean hasTsRollingArgument() { + return ArgumentType.TS_ROLLING.equals(refEntityKey.getType()); + } + } diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/LatestValuesAggregationCalculatedFieldConfiguration.java b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/LatestValuesAggregationCalculatedFieldConfiguration.java index 43a3360ce6..721930e676 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/LatestValuesAggregationCalculatedFieldConfiguration.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/cf/configuration/aggregation/LatestValuesAggregationCalculatedFieldConfiguration.java @@ -41,6 +41,19 @@ public class LatestValuesAggregationCalculatedFieldConfiguration implements Argu @Override public void validate() { + if (relation == null) { + throw new IllegalArgumentException("Relation must be specified!"); + } + relation.validate(); + if (arguments.containsKey("ctx")) { + throw new IllegalArgumentException("Argument name 'ctx' is reserved and cannot be used."); + } + if (arguments.values().stream().anyMatch(Argument::hasTsRollingArgument)) { + throw new IllegalArgumentException("Calculated field with type: '" + getType() + "' doesn't support TS_ROLLING arguments."); + } + if (metrics.isEmpty()) { + throw new IllegalArgumentException("Latest value aggregation calculated field must have at least one metric."); + } } } 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 18d1806fe4..ec203e5309 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 @@ -514,6 +514,21 @@ public class BaseRelationService implements RelationService { return executor.submit(() -> relationDao.findByRelationPathQuery(tenantId, relationPathQuery)); } + @Override + public List findByRelationPathQuery(TenantId tenantId, EntityRelationPathQuery relationPathQuery) { + log.trace("Executing findByRelationPathQuery, tenantId [{}], relationPathQuery {}", tenantId, relationPathQuery); + validateId(tenantId, id -> "Invalid tenant id: " + id); + validate(relationPathQuery); + if (relationPathQuery.levels().size() == 1) { + RelationPathLevel relationPathLevel = relationPathQuery.levels().get(0); + return switch (relationPathLevel.direction()) { + case FROM -> findByFromAndType(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON); + case TO -> findByToAndType(tenantId, relationPathQuery.rootEntityId(), relationPathLevel.relationType(), RelationTypeGroup.COMMON); + }; + } + return relationDao.findByRelationPathQuery(tenantId, relationPathQuery); + } + private void validate(EntityRelationPathQuery relationPathQuery) { validateId((UUIDBased) relationPathQuery.rootEntityId(), id -> "Invalid root entity id: " + id); List levels = relationPathQuery.levels();