diff --git a/application/src/main/java/org/thingsboard/server/controller/EntityQueryController.java b/application/src/main/java/org/thingsboard/server/controller/EntityQueryController.java index 1eee81ddd3..e6aed81c4b 100644 --- a/application/src/main/java/org/thingsboard/server/controller/EntityQueryController.java +++ b/application/src/main/java/org/thingsboard/server/controller/EntityQueryController.java @@ -17,6 +17,7 @@ package org.thingsboard.server.controller; import io.swagger.v3.oas.annotations.Parameter; import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.ResponseEntity; import org.springframework.security.access.prepost.PreAuthorize; @@ -31,6 +32,7 @@ import org.springframework.web.bind.annotation.RestController; import org.springframework.web.context.request.async.DeferredResult; import org.thingsboard.server.common.data.edqs.EdqsState; import org.thingsboard.server.common.data.edqs.ToCoreEdqsRequest; +import org.thingsboard.server.common.data.exception.ThingsboardErrorCode; import org.thingsboard.server.common.data.exception.ThingsboardException; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.id.UserId; @@ -49,6 +51,7 @@ import org.thingsboard.server.config.annotations.ApiOperation; import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.query.EntityQueryService; import org.thingsboard.server.service.security.permission.Operation; +import org.thingsboard.server.service.ws.telemetry.cmd.v2.AggHistoryCmd; import static org.thingsboard.server.controller.ControllerConstants.ALARM_DATA_QUERY_DESCRIPTION; import static org.thingsboard.server.controller.ControllerConstants.ATTRIBUTES_SCOPE_DESCRIPTION; @@ -149,6 +152,33 @@ public class EntityQueryController extends BaseController { return entityQueryService.getKeysByQuery(getCurrentUser(), tenantId, query, isTimeseries, isAttributes, scope); } + @ApiOperation(value = "Find Aggregated Historical Entity Data by Query", + notes = "Runs the entity data query and, for each matched entity, fetches a single aggregated value per key over [startTs, endTs] " + + "with per-key aggregation function. Optional previousStartTs/previousEndTs per key add a comparison window. " + + "REST equivalent of the WebSocket AggHistoryCmd. The aggregated values are returned in the 'aggLatest' field of each EntityData. " + + "Note: a separate timeseries aggregation query is issued per matched entity, so a large pageSize fans out proportionally - callers should page responsibly.") + @PreAuthorize("hasAnyAuthority('SYS_ADMIN', 'TENANT_ADMIN', 'CUSTOMER_USER')") + @PostMapping("/entitiesQuery/find/aggHistory") + public DeferredResult> findEntityDataAggHistoryByQuery( + @Parameter(description = "A JSON value representing the entity data query and aggregated history command.") + @RequestBody EntityDataAggHistoryRequest request) throws ThingsboardException { + checkNotNull(request); + checkNotNull(request.getQuery()); + checkNotNull(request.getAggHistoryCmd()); + AggHistoryCmd cmd = request.getAggHistoryCmd(); + if (cmd.getEndTs() < cmd.getStartTs()) { + throw new ThingsboardException("endTs must be >= startTs", ThingsboardErrorCode.BAD_REQUEST_PARAMS); + } + resolveQuery(request.getQuery()); + return entityQueryService.findEntityDataAggHistoryByQuery(getCurrentUser(), request.getQuery(), cmd); + } + + @Data + public static class EntityDataAggHistoryRequest { + private EntityDataQuery query; + private AggHistoryCmd aggHistoryCmd; + } + @PreAuthorize("hasAnyAuthority('SYS_ADMIN')") @PostMapping("/edqs/system/request") public void processSystemEdqsRequest(@RequestBody ToCoreEdqsRequest request) { diff --git a/application/src/main/java/org/thingsboard/server/service/query/DefaultEntityQueryService.java b/application/src/main/java/org/thingsboard/server/service/query/DefaultEntityQueryService.java index 38efff3d39..4ea9bc7529 100644 --- a/application/src/main/java/org/thingsboard/server/service/query/DefaultEntityQueryService.java +++ b/application/src/main/java/org/thingsboard/server/service/query/DefaultEntityQueryService.java @@ -36,6 +36,8 @@ import org.thingsboard.server.common.data.EntityType; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.id.TenantId; import org.thingsboard.server.common.data.kv.AttributeKvEntry; +import org.thingsboard.server.common.data.kv.ReadTsKvQuery; +import org.thingsboard.server.common.data.kv.ReadTsKvQueryResult; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.query.AlarmCountQuery; import org.thingsboard.server.common.data.query.AlarmData; @@ -63,6 +65,9 @@ import org.thingsboard.server.queue.util.TbCoreComponent; import org.thingsboard.server.service.executors.DbCallbackExecutorService; import org.thingsboard.server.service.security.AccessValidator; import org.thingsboard.server.service.security.model.SecurityUser; +import org.thingsboard.server.service.subscription.AggregationQueryUtils; +import org.thingsboard.server.service.subscription.ReadTsKvQueryInfo; +import org.thingsboard.server.service.ws.telemetry.cmd.v2.AggHistoryCmd; import java.util.ArrayList; import java.util.Collection; @@ -339,4 +344,43 @@ public class DefaultEntityQueryService implements EntityQueryService { }, dbCallbackExecutor); } + @Override + public DeferredResult> findEntityDataAggHistoryByQuery(SecurityUser securityUser, EntityDataQuery query, AggHistoryCmd cmd) { + DeferredResult> response = new DeferredResult<>(); + PageData pageData = findEntityDataByQuery(securityUser, query); + List entityDataList = pageData.getData(); + if (entityDataList.isEmpty() || cmd.getKeys() == null || cmd.getKeys().isEmpty()) { + response.setResult(pageData); + return response; + } + TenantId tenantId = securityUser.getTenantId(); + Map queries = AggregationQueryUtils.buildAggHistoryQueries(cmd.getKeys(), cmd.getStartTs(), cmd.getEndTs()); + List queryList = queries.values().stream().map(ReadTsKvQueryInfo::getQuery).collect(Collectors.toList()); + Map>> fetchResultMap = new LinkedHashMap<>(); + entityDataList.forEach(entityData -> fetchResultMap.put(entityData, + timeseriesService.findAllByQueries(tenantId, entityData.getEntityId(), queryList))); + Futures.addCallback(Futures.allAsList(fetchResultMap.values()), new FutureCallback<>() { + @Override + public void onSuccess(@Nullable List> ignored) { + fetchResultMap.forEach((entityData, future) -> { + try { + List queryResults = future.get(); + AggregationQueryUtils.populateAggLatest(entityData, queryResults, queries, cmd.getKeys(), null); + } catch (InterruptedException | ExecutionException e) { + log.warn("[{}] Failed to fetch aggregated historical data", entityData.getEntityId(), e); + AggregationQueryUtils.populateAggLatest(entityData, null, queries, cmd.getKeys(), null); + } + }); + response.setResult(pageData); + } + + @Override + public void onFailure(Throwable t) { + log.error("Failed to fetch aggregated historical data", t); + response.setErrorResult(t); + } + }, dbCallbackExecutor); + return response; + } + } diff --git a/application/src/main/java/org/thingsboard/server/service/query/EntityQueryService.java b/application/src/main/java/org/thingsboard/server/service/query/EntityQueryService.java index 464fde4410..e133508064 100644 --- a/application/src/main/java/org/thingsboard/server/service/query/EntityQueryService.java +++ b/application/src/main/java/org/thingsboard/server/service/query/EntityQueryService.java @@ -26,6 +26,7 @@ import org.thingsboard.server.common.data.query.EntityCountQuery; import org.thingsboard.server.common.data.query.EntityData; import org.thingsboard.server.common.data.query.EntityDataQuery; import org.thingsboard.server.service.security.model.SecurityUser; +import org.thingsboard.server.service.ws.telemetry.cmd.v2.AggHistoryCmd; public interface EntityQueryService { @@ -40,4 +41,6 @@ public interface EntityQueryService { DeferredResult getKeysByQuery(SecurityUser securityUser, TenantId tenantId, EntityDataQuery query, boolean isTimeseries, boolean isAttributes, String attributesScope); + DeferredResult> findEntityDataAggHistoryByQuery(SecurityUser securityUser, EntityDataQuery query, AggHistoryCmd aggHistoryCmd); + } diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/AggregationQueryUtils.java b/application/src/main/java/org/thingsboard/server/service/subscription/AggregationQueryUtils.java new file mode 100644 index 0000000000..05d3f48fd5 --- /dev/null +++ b/application/src/main/java/org/thingsboard/server/service/subscription/AggregationQueryUtils.java @@ -0,0 +1,90 @@ +/** + * Copyright © 2016-2026 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.subscription; + +import org.thingsboard.server.common.data.kv.Aggregation; +import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery; +import org.thingsboard.server.common.data.kv.ReadTsKvQueryResult; +import org.thingsboard.server.common.data.query.ComparisonTsValue; +import org.thingsboard.server.common.data.query.EntityData; +import org.thingsboard.server.common.data.query.TsValue; +import org.thingsboard.server.service.ws.telemetry.cmd.v2.AggKey; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +/** + * Shared aggregated-history logic used by both the WebSocket {@code AggHistoryCmd} handler + * ({@link DefaultTbEntityDataSubscriptionService}) and the REST {@code /find/aggHistory} endpoint. + */ +public class AggregationQueryUtils { + + /** + * Builds one current (and, when requested, one previous) timeseries query per {@link AggKey}, keyed by query id. + * A key with {@code previousValueOnly} skips the current window; a previous window is added only when both bounds + * are present and {@code previousEndTs >= previousStartTs}. + */ + public static Map buildAggHistoryQueries(List keys, long startTs, long endTs) { + Map queries = new HashMap<>(); + for (AggKey key : keys) { + if (key.getPreviousValueOnly() == null || !key.getPreviousValueOnly()) { + var query = singleBucketQuery(key.getKey(), startTs, endTs, key.getAgg()); + queries.put(query.getId(), new ReadTsKvQueryInfo(key, query, false)); + } + if (key.getPreviousStartTs() != null && key.getPreviousEndTs() != null && key.getPreviousEndTs() >= key.getPreviousStartTs()) { + var query = singleBucketQuery(key.getKey(), key.getPreviousStartTs(), key.getPreviousEndTs(), key.getAgg()); + queries.put(query.getId(), new ReadTsKvQueryInfo(key, query, true)); + } + } + return queries; + } + + /** + * A single bucket spanning the whole {@code [startTs, endTs]} window (interval = the full window, limit = 1), + * yielding exactly one aggregated point. + */ + private static BaseReadTsKvQuery singleBucketQuery(String key, long startTs, long endTs, Aggregation agg) { + return new BaseReadTsKvQuery(key, startTs, endTs, endTs - startTs, 1, agg); + } + + /** + * Maps the per-entity query results into {@code entityData.aggLatest} (current/previous per key) and fills any key + * without data with {@link TsValue#EMPTY}. When {@code lastTsCollector} is non-null, the last entry ts of each + * current value is recorded into it (used by the WS path to set up follow-up subscriptions). + */ + public static void populateAggLatest(EntityData entityData, List queryResults, + Map queries, List keys, + Map lastTsCollector) { + if (queryResults != null) { + for (ReadTsKvQueryResult queryResult : queryResults) { + ReadTsKvQueryInfo queryInfo = queries.get(queryResult.getQueryId()); + ComparisonTsValue comparisonTsValue = entityData.getAggLatest() + .computeIfAbsent(queryInfo.getKey().getId(), agg -> new ComparisonTsValue()); + if (queryInfo.isPrevious()) { + comparisonTsValue.setPrevious(queryResult.toTsValue(queryInfo.getQuery())); + } else { + comparisonTsValue.setCurrent(queryResult.toTsValue(queryInfo.getQuery())); + if (lastTsCollector != null) { + lastTsCollector.put(queryInfo.getQuery().getKey(), queryResult.getLastEntryTs()); + } + } + } + } + keys.forEach(key -> entityData.getAggLatest().putIfAbsent(key.getId(), new ComparisonTsValue(TsValue.EMPTY, TsValue.EMPTY))); + } + +} diff --git a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java index fc8c5c3be2..d1e1413715 100644 --- a/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java +++ b/application/src/main/java/org/thingsboard/server/service/subscription/DefaultTbEntityDataSubscriptionService.java @@ -41,7 +41,6 @@ import org.thingsboard.server.common.data.kv.ReadTsKvQueryResult; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.query.AlarmDataQuery; -import org.thingsboard.server.common.data.query.ComparisonTsValue; import org.thingsboard.server.common.data.query.EntityData; import org.thingsboard.server.common.data.query.EntityDataQuery; import org.thingsboard.server.common.data.query.EntityKey; @@ -296,17 +295,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc } private ListenableFuture handleAggHistoryCmd(TbEntityDataSubCtx ctx, AggHistoryCmd cmd) { - ConcurrentMap queries = new ConcurrentHashMap<>(); - for (AggKey key : cmd.getKeys()) { - if (key.getPreviousValueOnly() == null || !key.getPreviousValueOnly()) { - var query = new BaseReadTsKvQuery(key.getKey(), cmd.getStartTs(), cmd.getEndTs(), cmd.getEndTs() - cmd.getStartTs(), 1, key.getAgg()); - queries.put(query.getId(), new ReadTsKvQueryInfo(key, query, false)); - } - if (key.getPreviousStartTs() != null && key.getPreviousEndTs() != null && key.getPreviousEndTs() >= key.getPreviousStartTs()) { - var query = new BaseReadTsKvQuery(key.getKey(), key.getPreviousStartTs(), key.getPreviousEndTs(), key.getPreviousEndTs() - key.getPreviousStartTs(), 1, key.getAgg()); - queries.put(query.getId(), new ReadTsKvQueryInfo(key, query, true)); - } - } + Map queries = AggregationQueryUtils.buildAggHistoryQueries(cmd.getKeys(), cmd.getStartTs(), cmd.getEndTs()); return handleAggCmd(ctx, cmd.getKeys(), queries, cmd.getStartTs(), cmd.getEndTs(), false); } @@ -319,7 +308,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc return handleAggCmd(ctx, cmd.getKeys(), queries, cmd.getStartTs(), cmd.getStartTs() + cmd.getTimeWindow(), true); } - private ListenableFuture handleAggCmd(TbEntityDataSubCtx ctx, List keys, ConcurrentMap queries, + private ListenableFuture handleAggCmd(TbEntityDataSubCtx ctx, List keys, Map queries, long startTs, long endTs, boolean subscribe) { Map>> fetchResultMap = new HashMap<>(); List entityDataList = ctx.getData().getData(); @@ -335,22 +324,7 @@ public class DefaultTbEntityDataSubscriptionService implements TbEntityDataSubsc lastTsEntityMap.put(entityData, lastTsMap); List queryResults = future.get(); - if (queryResults != null) { - for (ReadTsKvQueryResult queryResult : queryResults) { - ReadTsKvQueryInfo queryInfo = queries.get(queryResult.getQueryId()); - ComparisonTsValue comparisonTsValue = entityData.getAggLatest().computeIfAbsent(queryInfo.getKey().getId(), agg -> new ComparisonTsValue()); - if (queryInfo.isPrevious()) { - comparisonTsValue.setPrevious(queryResult.toTsValue(queryInfo.getQuery())); - } else { - comparisonTsValue.setCurrent(queryResult.toTsValue(queryInfo.getQuery())); - lastTsMap.put(queryInfo.getQuery().getKey(), queryResult.getLastEntryTs()); - } - } - } - // Populate with empty values if no data found. - keys.forEach(key -> { - entityData.getAggLatest().putIfAbsent(key.getId(), new ComparisonTsValue(TsValue.EMPTY, TsValue.EMPTY)); - }); + AggregationQueryUtils.populateAggLatest(entityData, queryResults, queries, keys, lastTsMap); } catch (InterruptedException | ExecutionException e) { log.warn("[{}][{}][{}] Failed to fetch historical data", ctx.getSessionId(), ctx.getCmdId(), entityData.getEntityId(), e); ctx.sendWsMsg(new EntityDataUpdate(ctx.getCmdId(), SubscriptionErrorCode.INTERNAL_ERROR.getCode(), "Failed to fetch historical data!")); diff --git a/application/src/test/java/org/thingsboard/server/controller/EntityQueryControllerTest.java b/application/src/test/java/org/thingsboard/server/controller/EntityQueryControllerTest.java index e557626ee7..4c983350ef 100644 --- a/application/src/test/java/org/thingsboard/server/controller/EntityQueryControllerTest.java +++ b/application/src/test/java/org/thingsboard/server/controller/EntityQueryControllerTest.java @@ -18,6 +18,8 @@ package org.thingsboard.server.controller; import com.fasterxml.jackson.core.type.TypeReference; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.node.ObjectNode; +import com.google.common.util.concurrent.FutureCallback; +import org.checkerframework.checker.nullness.qual.Nullable; import org.junit.After; import org.junit.Assert; import org.junit.Before; @@ -27,6 +29,7 @@ import org.springframework.test.context.TestPropertySource; import org.springframework.test.web.servlet.ResultActions; import org.testcontainers.shaded.org.apache.commons.lang3.RandomStringUtils; import org.thingsboard.common.util.JacksonUtil; +import org.thingsboard.rule.engine.api.TimeseriesSaveRequest; import org.thingsboard.server.common.data.AttributeScope; import org.thingsboard.server.common.data.Customer; import org.thingsboard.server.common.data.Dashboard; @@ -42,6 +45,10 @@ import org.thingsboard.server.common.data.alarm.AlarmSeverity; import org.thingsboard.server.common.data.asset.Asset; import org.thingsboard.server.common.data.id.DeviceId; import org.thingsboard.server.common.data.id.EntityId; +import org.thingsboard.server.common.data.kv.Aggregation; +import org.thingsboard.server.common.data.kv.BasicTsKvEntry; +import org.thingsboard.server.common.data.kv.LongDataEntry; +import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.page.PageData; import org.thingsboard.server.common.data.query.AlarmCountQuery; import org.thingsboard.server.common.data.query.AlarmData; @@ -75,14 +82,20 @@ import org.thingsboard.server.common.data.relation.RelationEntityTypeFilter; import org.thingsboard.server.common.data.security.Authority; import org.thingsboard.server.common.data.tenant.profile.DefaultTenantProfileConfiguration; import org.thingsboard.server.common.data.tenant.profile.TenantProfileData; +import org.thingsboard.server.common.data.query.ComparisonTsValue; import org.thingsboard.server.dao.queue.QueueStatsService; import org.thingsboard.server.dao.service.DaoSqlTest; +import org.thingsboard.server.service.telemetry.TelemetrySubscriptionService; +import org.thingsboard.server.service.ws.telemetry.cmd.v2.AggHistoryCmd; +import org.thingsboard.server.service.ws.telemetry.cmd.v2.AggKey; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import java.util.stream.Collectors; import static org.assertj.core.api.Assertions.assertThat; @@ -107,6 +120,9 @@ public class EntityQueryControllerTest extends AbstractControllerTest { @Autowired private QueueStatsService queueStatsService; + @Autowired + private TelemetrySubscriptionService tsService; + @Before public void beforeTest() throws Exception { loginSysAdmin(); @@ -1329,6 +1345,212 @@ public class EntityQueryControllerTest extends AbstractControllerTest { return numericFilter; } + @Test + public void testFindEntityDataAggHistoryByQuery() throws Exception { + Device device = new Device(); + device.setName("AggHistoryDevice"); + device.setType("default"); + device = doPost("/api/device", device, Device.class); + + long now = System.currentTimeMillis(); + long startTs = now - TimeUnit.MINUTES.toMillis(30); + long endTs = now; + long previousStartTs = now - TimeUnit.MINUTES.toMillis(60); + long previousEndTs = startTs - 1; + + List currentData = new ArrayList<>(); + for (int i = 0; i < 20; i++) { + long ts = startTs + i * TimeUnit.MINUTES.toMillis(1); + currentData.add(new BasicTsKvEntry(ts, new LongDataEntry("temperature", 30L))); + } + List previousData = new ArrayList<>(); + for (int i = 0; i < 10; i++) { + long ts = previousStartTs + i * TimeUnit.MINUTES.toMillis(1); + previousData.add(new BasicTsKvEntry(ts, new LongDataEntry("temperature", 30L))); + } + sendTelemetry(device, currentData); + sendTelemetry(device, previousData); + + DeviceTypeFilter filter = new DeviceTypeFilter(List.of("default"), "AggHistoryDevice"); + EntityDataPageLink pageLink = new EntityDataPageLink(10, 0, null, null); + EntityDataQuery query = new EntityDataQuery(filter, pageLink, Collections.emptyList(), Collections.emptyList(), null); + + AggKey key = new AggKey(); + key.setId(1); + key.setKey("temperature"); + key.setAgg(Aggregation.SUM); + key.setPreviousStartTs(previousStartTs); + key.setPreviousEndTs(previousEndTs); + + AggHistoryCmd cmd = new AggHistoryCmd(); + cmd.setKeys(List.of(key)); + cmd.setStartTs(startTs); + cmd.setEndTs(endTs); + + ObjectNode request = JacksonUtil.newObjectNode(); + request.set("query", JacksonUtil.valueToTree(query)); + request.set("aggHistoryCmd", JacksonUtil.valueToTree(cmd)); + + PageData result = doPostAsyncWithTypedResponse("/api/entitiesQuery/find/aggHistory", + request, new TypeReference<>() {}, status().isOk()); + + assertThat(result.getData()).hasSize(1); + EntityData entityData = result.getData().get(0); + assertThat(entityData.getEntityId()).isEqualTo(device.getId()); + ComparisonTsValue agg = entityData.getAggLatest().get(1); + assertThat(agg).isNotNull(); + assertThat(agg.getCurrent().getValue()).isEqualTo("600"); + assertThat(agg.getPrevious().getValue()).isEqualTo("300"); + } + + @Test + public void testFindEntityDataAggHistoryByQuery_previousValueOnly() throws Exception { + Device device = new Device(); + device.setName("AggPrevOnlyDevice"); + device.setType("default"); + device = doPost("/api/device", device, Device.class); + + long now = System.currentTimeMillis(); + long startTs = now - TimeUnit.MINUTES.toMillis(30); + long endTs = now; + long previousStartTs = now - TimeUnit.MINUTES.toMillis(60); + long previousEndTs = startTs - 1; + + List currentData = new ArrayList<>(); + for (int i = 0; i < 5; i++) { + currentData.add(new BasicTsKvEntry(startTs + i * TimeUnit.MINUTES.toMillis(1), + new LongDataEntry("temperature", 100L))); + } + List previousData = new ArrayList<>(); + for (int i = 0; i < 3; i++) { + previousData.add(new BasicTsKvEntry(previousStartTs + i * TimeUnit.MINUTES.toMillis(1), + new LongDataEntry("temperature", 10L))); + } + sendTelemetry(device, currentData); + sendTelemetry(device, previousData); + + AggKey key = new AggKey(); + key.setId(7); + key.setKey("temperature"); + key.setAgg(Aggregation.SUM); + key.setPreviousStartTs(previousStartTs); + key.setPreviousEndTs(previousEndTs); + key.setPreviousValueOnly(true); + + PageData result = postAggHistory(device, startTs, endTs, List.of(key)); + + ComparisonTsValue agg = result.getData().get(0).getAggLatest().get(7); + assertThat(agg.getPrevious().getValue()).isEqualTo("30"); + assertThat(agg.getCurrent()).isNull(); + } + + @Test + public void testFindEntityDataAggHistoryByQuery_noTelemetry() throws Exception { + Device device = new Device(); + device.setName("AggEmptyDevice"); + device.setType("default"); + device = doPost("/api/device", device, Device.class); + + long now = System.currentTimeMillis(); + AggKey key = new AggKey(); + key.setId(1); + key.setKey("temperature"); + key.setAgg(Aggregation.SUM); + + PageData result = postAggHistory(device, now - TimeUnit.HOURS.toMillis(1), now, List.of(key)); + + ComparisonTsValue agg = result.getData().get(0).getAggLatest().get(1); + assertThat(agg).isNotNull(); + assertThat(agg.getCurrent().getValue()).isEqualTo("0"); + assertThat(agg.getPrevious()).isNull(); + } + + @Test + public void testFindEntityDataAggHistoryByQuery_multipleKeysMixedAggregation() throws Exception { + Device device = new Device(); + device.setName("AggMultiKeyDevice"); + device.setType("default"); + device = doPost("/api/device", device, Device.class); + + long now = System.currentTimeMillis(); + long startTs = now - TimeUnit.MINUTES.toMillis(30); + long endTs = now; + + List data = new ArrayList<>(); + long[] values = {10L, 20L, 30L, 40L, 50L}; + for (int i = 0; i < values.length; i++) { + data.add(new BasicTsKvEntry(startTs + i * TimeUnit.MINUTES.toMillis(1), + new LongDataEntry("temperature", values[i]))); + } + sendTelemetry(device, data); + + AggKey minKey = new AggKey(); + minKey.setId(1); + minKey.setKey("temperature"); + minKey.setAgg(Aggregation.MIN); + + AggKey maxKey = new AggKey(); + maxKey.setId(2); + maxKey.setKey("temperature"); + maxKey.setAgg(Aggregation.MAX); + + AggKey avgKey = new AggKey(); + avgKey.setId(3); + avgKey.setKey("temperature"); + avgKey.setAgg(Aggregation.AVG); + + PageData result = postAggHistory(device, startTs, endTs, List.of(minKey, maxKey, avgKey)); + + var aggLatest = result.getData().get(0).getAggLatest(); + assertThat(aggLatest.get(1).getCurrent().getValue()).isEqualTo("10"); + assertThat(aggLatest.get(2).getCurrent().getValue()).isEqualTo("50"); + assertThat(Double.parseDouble(aggLatest.get(3).getCurrent().getValue())).isEqualTo(30.0); + } + + private PageData postAggHistory(Device device, long startTs, long endTs, List keys) throws Exception { + DeviceTypeFilter filter = new DeviceTypeFilter(List.of("default"), device.getName()); + EntityDataPageLink pageLink = new EntityDataPageLink(10, 0, null, null); + EntityDataQuery query = new EntityDataQuery(filter, pageLink, Collections.emptyList(), Collections.emptyList(), null); + + AggHistoryCmd cmd = new AggHistoryCmd(); + cmd.setKeys(keys); + cmd.setStartTs(startTs); + cmd.setEndTs(endTs); + + ObjectNode request = JacksonUtil.newObjectNode(); + request.set("query", JacksonUtil.valueToTree(query)); + request.set("aggHistoryCmd", JacksonUtil.valueToTree(cmd)); + + return doPostAsyncWithTypedResponse("/api/entitiesQuery/find/aggHistory", + request, new TypeReference<>() {}, status().isOk()); + } + + private void sendTelemetry(Device device, List tsData) throws InterruptedException { + CountDownLatch latch = new CountDownLatch(1); + AtomicReference error = new AtomicReference<>(); + tsService.saveTimeseries(TimeseriesSaveRequest.builder() + .tenantId(device.getTenantId()) + .entityId(device.getId()) + .entries(tsData) + .callback(new FutureCallback<>() { + @Override + public void onSuccess(@Nullable Void result) { + latch.countDown(); + } + + @Override + public void onFailure(Throwable t) { + error.set(t); + latch.countDown(); + } + }) + .build()); + assertThat(latch.await(TIMEOUT, TimeUnit.SECONDS)).isTrue(); + if (error.get() != null) { + throw new AssertionError("Failed to save telemetry", error.get()); + } + } + private KeyFilter buildStringKeyFilter(EntityKeyType entityKeyType, String name, StringFilterPredicate.StringOperation operation, String value) { KeyFilter nameFilter = new KeyFilter(); nameFilter.setKey(new EntityKey(entityKeyType, name));