From f9c65696c9a2b479e21063dc8c33206d114ba5ac Mon Sep 17 00:00:00 2001 From: Oleksandra Matviienko Date: Wed, 15 Apr 2026 14:52:18 +0200 Subject: [PATCH 1/4] Add REST equivalent of WebSocket AggHistoryCmd for entity data queries PE UI exports widget data via REST; without a REST aggregated-history endpoint the Entity Table rendered aggregated values while CSV export returned raw latest. New POST /api/entitiesQuery/find/aggHistory mirrors the WS command and populates aggLatest per entity. Caps page size at MAX_PAGE_SIZE=100, validates endTs >= startTs, and fills aggLatest with EMPTY ComparisonTsValue on per-entity fetch failure for a predictable response contract. --- .../controller/EntityQueryController.java | 32 +++ .../query/DefaultEntityQueryService.java | 74 ++++++ .../service/query/EntityQueryService.java | 3 + .../controller/EntityQueryControllerTest.java | 222 ++++++++++++++++++ 4 files changed, 331 insertions(+) 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..46eaa509e3 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; @@ -49,6 +50,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 +151,36 @@ 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.") + @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 IllegalArgumentException("endTs must be >= startTs"); + } + EntityDataPageLink pageLink = request.getQuery().getPageLink(); + if (pageLink != null && pageLink.getPageSize() > MAX_PAGE_SIZE) { + pageLink.setPageSize(MAX_PAGE_SIZE); + } + 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..e4596d0f13 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,10 +36,14 @@ 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.BaseReadTsKvQuery; +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; import org.thingsboard.server.common.data.query.AlarmDataQuery; +import org.thingsboard.server.common.data.query.ComparisonTsValue; import org.thingsboard.server.common.data.query.ComplexFilterPredicate; import org.thingsboard.server.common.data.query.DynamicValue; import org.thingsboard.server.common.data.query.EntityCountQuery; @@ -53,6 +57,7 @@ import org.thingsboard.server.common.data.query.FilterPredicateType; import org.thingsboard.server.common.data.query.KeyFilter; import org.thingsboard.server.common.data.query.KeyFilterPredicate; import org.thingsboard.server.common.data.query.SimpleKeyFilterPredicate; +import org.thingsboard.server.common.data.query.TsValue; import org.thingsboard.server.dao.alarm.AlarmService; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.entity.EntityService; @@ -63,10 +68,14 @@ 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.ReadTsKvQueryInfo; +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.Collection; import java.util.Collections; +import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -339,4 +348,69 @@ 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 = new HashMap<>(); + for (AggKey key : cmd.getKeys()) { + if (key.getPreviousValueOnly() == null || !key.getPreviousValueOnly()) { + var q = new BaseReadTsKvQuery(key.getKey(), cmd.getStartTs(), cmd.getEndTs(), + cmd.getEndTs() - cmd.getStartTs(), 1, key.getAgg()); + queries.put(q.getId(), new ReadTsKvQueryInfo(key, q, false)); + } + if (key.getPreviousStartTs() != null && key.getPreviousEndTs() != null && key.getPreviousEndTs() >= key.getPreviousStartTs()) { + var q = new BaseReadTsKvQuery(key.getKey(), key.getPreviousStartTs(), key.getPreviousEndTs(), + key.getPreviousEndTs() - key.getPreviousStartTs(), 1, key.getAgg()); + queries.put(q.getId(), new ReadTsKvQueryInfo(key, q, true)); + } + } + 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(); + if (queryResults != null) { + for (ReadTsKvQueryResult queryResult : queryResults) { + ReadTsKvQueryInfo info = queries.get(queryResult.getQueryId()); + ComparisonTsValue comparisonTsValue = entityData.getAggLatest() + .computeIfAbsent(info.getKey().getId(), k -> new ComparisonTsValue()); + if (info.isPrevious()) { + comparisonTsValue.setPrevious(queryResult.toTsValue(info.getQuery())); + } else { + comparisonTsValue.setCurrent(queryResult.toTsValue(info.getQuery())); + } + } + } + cmd.getKeys().forEach(key -> entityData.getAggLatest() + .putIfAbsent(key.getId(), new ComparisonTsValue(TsValue.EMPTY, TsValue.EMPTY))); + } catch (InterruptedException | ExecutionException e) { + log.warn("[{}] Failed to fetch aggregated historical data", entityData.getEntityId(), e); + cmd.getKeys().forEach(key -> entityData.getAggLatest() + .putIfAbsent(key.getId(), new ComparisonTsValue(TsValue.EMPTY, TsValue.EMPTY))); + } + }); + 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/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)); From 6f7d86f9206cf65b42ed04fa6e32bb8ff7ab41a3 Mon Sep 17 00:00:00 2001 From: Oleksandra Matviienko Date: Thu, 30 Apr 2026 14:23:10 +0200 Subject: [PATCH 2/4] Return 400 instead of 500 for invalid aggHistory time range --- .../thingsboard/server/controller/EntityQueryController.java | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) 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 46eaa509e3..49e7e6d1eb 100644 --- a/application/src/main/java/org/thingsboard/server/controller/EntityQueryController.java +++ b/application/src/main/java/org/thingsboard/server/controller/EntityQueryController.java @@ -32,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; @@ -165,7 +166,7 @@ public class EntityQueryController extends BaseController { checkNotNull(request.getAggHistoryCmd()); AggHistoryCmd cmd = request.getAggHistoryCmd(); if (cmd.getEndTs() < cmd.getStartTs()) { - throw new IllegalArgumentException("endTs must be >= startTs"); + throw new ThingsboardException("endTs must be >= startTs", ThingsboardErrorCode.BAD_REQUEST_PARAMS); } EntityDataPageLink pageLink = request.getQuery().getPageLink(); if (pageLink != null && pageLink.getPageSize() > MAX_PAGE_SIZE) { From fc8ddede169bc0664eb010e94efc08626cfacdd3 Mon Sep 17 00:00:00 2001 From: Oleksandra Matviienko Date: Tue, 19 May 2026 00:20:05 +0200 Subject: [PATCH 3/4] Remove pageSize cap from /find/aggHistory endpoint Aligns with sibling /api/entitiesQuery/find, which serves entity pages unbounded and is the endpoint paired with /aggHistory in the PE entity- table export. Capping aggHistory at 100 while /find returns up to 1000 caused entity and aggregated-value lists to fall out of sync on export. The expensive timeseries reads use tsService.findAllByQueries, the same path as the WebSocket AggHistoryCmd, which has no such cap. --- .../thingsboard/server/controller/EntityQueryController.java | 4 ---- 1 file changed, 4 deletions(-) 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 49e7e6d1eb..581279d25f 100644 --- a/application/src/main/java/org/thingsboard/server/controller/EntityQueryController.java +++ b/application/src/main/java/org/thingsboard/server/controller/EntityQueryController.java @@ -168,10 +168,6 @@ public class EntityQueryController extends BaseController { if (cmd.getEndTs() < cmd.getStartTs()) { throw new ThingsboardException("endTs must be >= startTs", ThingsboardErrorCode.BAD_REQUEST_PARAMS); } - EntityDataPageLink pageLink = request.getQuery().getPageLink(); - if (pageLink != null && pageLink.getPageSize() > MAX_PAGE_SIZE) { - pageLink.setPageSize(MAX_PAGE_SIZE); - } resolveQuery(request.getQuery()); return entityQueryService.findEntityDataAggHistoryByQuery(getCurrentUser(), request.getQuery(), cmd); } From 2696dec73838f70e096ba1dc6c3c2b60e0f5902f Mon Sep 17 00:00:00 2001 From: Oleksandra Matviienko Date: Wed, 10 Jun 2026 14:27:22 +0200 Subject: [PATCH 4/4] Share aggregated-history logic between WebSocket and REST entity-data paths Extract the query-building and aggLatest-population blocks into AggregationQueryUtils; the WebSocket AggHistoryCmd handler and the REST /find/aggHistory endpoint now call the shared methods. Document the per-entity fan-out in the endpoint API notes. --- .../controller/EntityQueryController.java | 3 +- .../query/DefaultEntityQueryService.java | 38 +------- .../subscription/AggregationQueryUtils.java | 90 +++++++++++++++++++ ...efaultTbEntityDataSubscriptionService.java | 32 +------ 4 files changed, 99 insertions(+), 64 deletions(-) create mode 100644 application/src/main/java/org/thingsboard/server/service/subscription/AggregationQueryUtils.java 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 581279d25f..e6aed81c4b 100644 --- a/application/src/main/java/org/thingsboard/server/controller/EntityQueryController.java +++ b/application/src/main/java/org/thingsboard/server/controller/EntityQueryController.java @@ -155,7 +155,8 @@ public class EntityQueryController extends BaseController { @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.") + "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( 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 e4596d0f13..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,14 +36,12 @@ 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.BaseReadTsKvQuery; 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; import org.thingsboard.server.common.data.query.AlarmDataQuery; -import org.thingsboard.server.common.data.query.ComparisonTsValue; import org.thingsboard.server.common.data.query.ComplexFilterPredicate; import org.thingsboard.server.common.data.query.DynamicValue; import org.thingsboard.server.common.data.query.EntityCountQuery; @@ -57,7 +55,6 @@ import org.thingsboard.server.common.data.query.FilterPredicateType; import org.thingsboard.server.common.data.query.KeyFilter; import org.thingsboard.server.common.data.query.KeyFilterPredicate; import org.thingsboard.server.common.data.query.SimpleKeyFilterPredicate; -import org.thingsboard.server.common.data.query.TsValue; import org.thingsboard.server.dao.alarm.AlarmService; import org.thingsboard.server.dao.attributes.AttributesService; import org.thingsboard.server.dao.entity.EntityService; @@ -68,14 +65,13 @@ 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 org.thingsboard.server.service.ws.telemetry.cmd.v2.AggKey; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; -import java.util.HashMap; import java.util.LinkedHashMap; import java.util.List; import java.util.Map; @@ -358,19 +354,7 @@ public class DefaultEntityQueryService implements EntityQueryService { return response; } TenantId tenantId = securityUser.getTenantId(); - Map queries = new HashMap<>(); - for (AggKey key : cmd.getKeys()) { - if (key.getPreviousValueOnly() == null || !key.getPreviousValueOnly()) { - var q = new BaseReadTsKvQuery(key.getKey(), cmd.getStartTs(), cmd.getEndTs(), - cmd.getEndTs() - cmd.getStartTs(), 1, key.getAgg()); - queries.put(q.getId(), new ReadTsKvQueryInfo(key, q, false)); - } - if (key.getPreviousStartTs() != null && key.getPreviousEndTs() != null && key.getPreviousEndTs() >= key.getPreviousStartTs()) { - var q = new BaseReadTsKvQuery(key.getKey(), key.getPreviousStartTs(), key.getPreviousEndTs(), - key.getPreviousEndTs() - key.getPreviousStartTs(), 1, key.getAgg()); - queries.put(q.getId(), new ReadTsKvQueryInfo(key, q, true)); - } - } + 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, @@ -381,24 +365,10 @@ public class DefaultEntityQueryService implements EntityQueryService { fetchResultMap.forEach((entityData, future) -> { try { List queryResults = future.get(); - if (queryResults != null) { - for (ReadTsKvQueryResult queryResult : queryResults) { - ReadTsKvQueryInfo info = queries.get(queryResult.getQueryId()); - ComparisonTsValue comparisonTsValue = entityData.getAggLatest() - .computeIfAbsent(info.getKey().getId(), k -> new ComparisonTsValue()); - if (info.isPrevious()) { - comparisonTsValue.setPrevious(queryResult.toTsValue(info.getQuery())); - } else { - comparisonTsValue.setCurrent(queryResult.toTsValue(info.getQuery())); - } - } - } - cmd.getKeys().forEach(key -> entityData.getAggLatest() - .putIfAbsent(key.getId(), new ComparisonTsValue(TsValue.EMPTY, TsValue.EMPTY))); + AggregationQueryUtils.populateAggLatest(entityData, queryResults, queries, cmd.getKeys(), null); } catch (InterruptedException | ExecutionException e) { log.warn("[{}] Failed to fetch aggregated historical data", entityData.getEntityId(), e); - cmd.getKeys().forEach(key -> entityData.getAggLatest() - .putIfAbsent(key.getId(), new ComparisonTsValue(TsValue.EMPTY, TsValue.EMPTY))); + AggregationQueryUtils.populateAggLatest(entityData, null, queries, cmd.getKeys(), null); } }); response.setResult(pageData); 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!"));