Merge branch 'aggregationTsDao_fix' of github.com:van-vanich/thingsboard into develop/3.4

dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDao.java

@@ -26,6 +26,7 @@ import org.springframework.data.domain.Sort;
 import org.thingsboard.server.common.data.id.EntityId;
 import org.thingsboard.server.common.data.id.TenantId;
 import org.thingsboard.server.common.data.kv.Aggregation;
+import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.DeleteTsKvQuery;
 import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
 import org.thingsboard.server.common.data.kv.TsKvEntry;
@@ -122,14 +123,18 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq
         if (query.getAggregation() == Aggregation.NONE) {
             return findAllAsyncWithLimit(entityId, query);
         } else {
-            long stepTs = query.getStartTs();
             List<ListenableFuture<Optional<TsKvEntry>>> futures = new ArrayList<>();
-            while (stepTs < query.getEndTs()) {
-                long startTs = stepTs;
-                long endTs = stepTs + query.getInterval();
+            long endPeriod = query.getEndTs();
+            long startPeriod = query.getStartTs();
+            long step = query.getInterval();
+            while (startPeriod <= endPeriod) {
+                long startTs = startPeriod;
+                long endTs = Math.min(startPeriod + step, endPeriod + 1);
                 long ts = startTs + (endTs - startTs) / 2;
-                futures.add(findAndAggregateAsync(entityId, query.getKey(), startTs, endTs, ts, query.getAggregation()));
-                stepTs = endTs;
+                ReadTsKvQuery subQuery = new BaseReadTsKvQuery(query.getKey(), startTs, endTs, ts, 1, query.getAggregation(), query.getOrder());
+                ListenableFuture<Optional<TsKvEntry>> aggregateTsKvEntry = findAndAggregateAsync(entityId, subQuery.getKey(), startTs, endTs, ts,  query.getAggregation());
+                futures.add(aggregateTsKvEntry);
+                startPeriod = endTs;
             }
             return getTskvEntriesFuture(Futures.allAsList(futures));
         }
@@ -148,7 +153,7 @@ public abstract class AbstractChunkedAggregationTimeseriesDao extends AbstractSq
         return Futures.immediateFuture(DaoUtil.convertDataList(tsKvEntities));
     }
 
-    private ListenableFuture<Optional<TsKvEntry>> findAndAggregateAsync(EntityId entityId, String key, long startTs, long endTs, long ts, Aggregation aggregation) {
+    ListenableFuture<Optional<TsKvEntry>> findAndAggregateAsync(EntityId entityId, String key, long startTs, long endTs, long ts, Aggregation aggregation) {
         List<CompletableFuture<TsKvEntity>> entitiesFutures = new ArrayList<>();
         switchAggregation(entityId, key, startTs, endTs, aggregation, entitiesFutures);
         return Futures.transform(setFutures(entitiesFutures), entity -> {

dao/src/main/java/org/thingsboard/server/dao/sqlts/AbstractSqlTimeseriesDao.java

@@ -91,7 +91,7 @@ public abstract class AbstractSqlTimeseriesDao extends BaseAbstractSqlTimeseries
                 .stream()
                 .map(query -> findAllAsync(tenantId, entityId, query))
                 .collect(Collectors.toList());
-        return Futures.transform(Futures.allAsList(futures), new Function<List<List<TsKvEntry>>, List<TsKvEntry>>() {
+        return Futures.transform(Futures.allAsList(futures), new Function<>() {
             @Nullable
             @Override
             public List<TsKvEntry> apply(@Nullable List<List<TsKvEntry>> results) {

dao/src/main/java/org/thingsboard/server/dao/sqlts/AggregationTimeseriesDao.java

@@ -26,4 +26,4 @@ import java.util.List;
 public interface AggregationTimeseriesDao {
 
     ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, ReadTsKvQuery query);
-}
+}

dao/src/main/java/org/thingsboard/server/dao/sqlts/BaseAbstractSqlTimeseriesDao.java

@@ -39,9 +39,7 @@ import java.util.stream.Collectors;
 public abstract class BaseAbstractSqlTimeseriesDao extends JpaAbstractDaoListeningExecutorService {
 
     private final ConcurrentMap<String, Integer> tsKvDictionaryMap = new ConcurrentHashMap<>();
-
     protected static final ReentrantLock tsCreationLock = new ReentrantLock();
-
     @Autowired
     protected TsKvDictionaryRepository dictionaryRepository;
 
@@ -96,4 +94,5 @@ public abstract class BaseAbstractSqlTimeseriesDao extends JpaAbstractDaoListeni
             }
         }, service);
     }
+
 }

dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java

@@ -141,7 +141,7 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD
     @Override
     public ListenableFuture<List<TsKvEntry>> findAllAsync(TenantId tenantId, EntityId entityId, List<ReadTsKvQuery> queries) {
         List<ListenableFuture<List<TsKvEntry>>> futures = queries.stream().map(query -> findAllAsync(tenantId, entityId, query)).collect(Collectors.toList());
-        return Futures.transform(Futures.allAsList(futures), new Function<List<List<TsKvEntry>>, List<TsKvEntry>>() {
+        return Futures.transform(Futures.allAsList(futures), new Function<>() {
             @Nullable
             @Override
             public List<TsKvEntry> apply(@Nullable List<List<TsKvEntry>> results) {
@@ -270,18 +270,20 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD
         if (query.getAggregation() == Aggregation.NONE) {
             return findAllAsyncWithLimit(tenantId, entityId, query);
         } else {
+            long startPeriod = query.getStartTs();
+            long endPeriod = query.getEndTs();
             long step = Math.max(query.getInterval(), MIN_AGGREGATION_STEP_MS);
-            long stepTs = query.getStartTs();
             List<ListenableFuture<Optional<TsKvEntry>>> futures = new ArrayList<>();
-            while (stepTs < query.getEndTs()) {
-                long startTs = stepTs;
-                long endTs = stepTs + step;
-                ReadTsKvQuery subQuery = new BaseReadTsKvQuery(query.getKey(), startTs, endTs, step, 1, query.getAggregation(), query.getOrder());
+            while (startPeriod <= endPeriod) {
+                long startTs = startPeriod;
+                long endTs = Math.min(startPeriod + step, endPeriod + 1);
+                long ts = endTs - startTs;
+                ReadTsKvQuery subQuery = new BaseReadTsKvQuery(query.getKey(), startTs, endTs, ts, 1, query.getAggregation(), query.getOrder());
                 futures.add(findAndAggregateAsync(tenantId, entityId, subQuery, toPartitionTs(startTs), toPartitionTs(endTs)));
-                stepTs = endTs;
+                startPeriod = endTs;
             }
             ListenableFuture<List<Optional<TsKvEntry>>> future = Futures.allAsList(futures);
-            return Futures.transform(future, new Function<List<Optional<TsKvEntry>>, List<TsKvEntry>>() {
+            return Futures.transform(future, new Function<>() {
                 @Nullable
                 @Override
                 public List<TsKvEntry> apply(@Nullable List<Optional<TsKvEntry>> input) {

dao/src/test/java/org/thingsboard/server/dao/service/timeseries/BaseTimeseriesServiceTest.java

@@ -184,6 +184,140 @@ public abstract class BaseTimeseriesServiceTest extends AbstractServiceTest {
         Assert.assertEquals(toTsEntry(TS - 1, stringKvEntry), entries.get(2));
     }
 
+    @Test
+    public void testFindByQuery_whenPeriodEqualsOneMilisecondPeriod() throws Exception {
+        DeviceId deviceId = new DeviceId(Uuids.timeBased());
+        saveEntries(deviceId, TS - 1L);
+        saveEntries(deviceId, TS);
+        saveEntries(deviceId, TS + 1L);
+
+        List<ReadTsKvQuery> queries = List.of(new BaseReadTsKvQuery(LONG_KEY, TS, TS, 1, 1, Aggregation.COUNT, DESC_ORDER));
+
+        List<TsKvEntry> entries = tsService.findAll(tenantId, deviceId, queries).get();
+        Assert.assertEquals(1, entries.size());
+        Assert.assertEquals(toTsEntry(TS, new LongDataEntry(LONG_KEY, 1L)), entries.get(0));
+
+        EntityView entityView = saveAndCreateEntityView(deviceId, List.of(LONG_KEY));
+
+        entries = tsService.findAll(tenantId, entityView.getId(), queries).get();
+        Assert.assertEquals(1, entries.size());
+        Assert.assertEquals(toTsEntry(TS, new LongDataEntry(LONG_KEY, 1L)), entries.get(0));
+    }
+
+    @Test
+    public void testFindByQuery_whenPeriodEqualsInterval() throws Exception {
+        DeviceId deviceId = new DeviceId(Uuids.timeBased());
+        saveEntries(deviceId, TS - 1L);
+        for (long i = TS; i <= TS + 100L; i += 10L) {
+            saveEntries(deviceId, i);
+        }
+        saveEntries(deviceId, TS + 100L + 1L);
+
+        List<ReadTsKvQuery> queries = List.of(new BaseReadTsKvQuery(LONG_KEY, TS, TS + 100, 101, 1, Aggregation.COUNT, DESC_ORDER));
+
+        List<TsKvEntry> entries = tsService.findAll(tenantId, deviceId, queries).get();
+        Assert.assertEquals(1, entries.size());
+        Assert.assertEquals(toTsEntry(TS + 50, new LongDataEntry(LONG_KEY, 11L)), entries.get(0));
+
+        EntityView entityView = saveAndCreateEntityView(deviceId, List.of(LONG_KEY));
+
+        entries = tsService.findAll(tenantId, entityView.getId(), queries).get();
+        Assert.assertEquals(1, entries.size());
+        Assert.assertEquals(toTsEntry(TS + 50, new LongDataEntry(LONG_KEY, 11L)), entries.get(0));
+    }
+
+    @Test
+    public void testFindByQuery_whenPeriodHaveTwoIntervalWithEqualsLength() throws Exception {
+        DeviceId deviceId = new DeviceId(Uuids.timeBased());
+        saveEntries(deviceId, TS - 1L);
+        for (long i = TS; i <= TS + 100000L; i += 10000L) {
+            saveEntries(deviceId, i);
+        }
+        saveEntries(deviceId, TS + 100000L + 1L);
+
+        List<ReadTsKvQuery> queries = List.of(new BaseReadTsKvQuery(LONG_KEY, TS, TS + 99999, 50000, 1, Aggregation.COUNT, DESC_ORDER));
+
+        List<TsKvEntry> entries = tsService.findAll(tenantId, deviceId, queries).get();
+        Assert.assertEquals(2, entries.size());
+        Assert.assertEquals(toTsEntry(TS + 25000, new LongDataEntry(LONG_KEY, 5L)), entries.get(0));
+        Assert.assertEquals(toTsEntry(TS + 75000, new LongDataEntry(LONG_KEY, 5L)), entries.get(1));
+
+        EntityView entityView = saveAndCreateEntityView(deviceId, List.of(LONG_KEY));
+
+        entries = tsService.findAll(tenantId, entityView.getId(), queries).get();
+        Assert.assertEquals(2, entries.size());
+        Assert.assertEquals(toTsEntry(TS + 25000, new LongDataEntry(LONG_KEY, 5L)), entries.get(0));
+        Assert.assertEquals(toTsEntry(TS + 75000, new LongDataEntry(LONG_KEY, 5L)), entries.get(1));
+    }
+
+    @Test
+    public void testFindByQuery_whenPeriodHaveTwoInterval_whereSecondShorterThanFirst() throws Exception {
+        DeviceId deviceId = new DeviceId(Uuids.timeBased());
+        saveEntries(deviceId, TS - 1L);
+        for (long i = TS; i <= TS + 80000L; i += 10000L) {
+            saveEntries(deviceId, i);
+        }
+        saveEntries(deviceId, TS + 80000L + 1L);
+
+        List<ReadTsKvQuery> queries = List.of(new BaseReadTsKvQuery(LONG_KEY, TS, TS + 80000, 50000, 1, Aggregation.COUNT, DESC_ORDER));
+
+        List<TsKvEntry> entries = tsService.findAll(tenantId, deviceId, queries).get();
+        Assert.assertEquals(2, entries.size());
+        Assert.assertEquals(toTsEntry(TS + 25000, new LongDataEntry(LONG_KEY, 5L)), entries.get(0));
+        Assert.assertEquals(toTsEntry(TS + 65000, new LongDataEntry(LONG_KEY, 4L)), entries.get(1));
+
+        EntityView entityView = saveAndCreateEntityView(deviceId, List.of(LONG_KEY));
+
+        entries = tsService.findAll(tenantId, entityView.getId(), queries).get();
+        Assert.assertEquals(2, entries.size());
+        Assert.assertEquals(toTsEntry(TS + 25000, new LongDataEntry(LONG_KEY, 5L)), entries.get(0));
+        Assert.assertEquals(toTsEntry(TS + 65000, new LongDataEntry(LONG_KEY, 4L)), entries.get(1));
+    }
+
+    @Test
+    public void testFindByQuery_whenPeriodHaveTwoIntervalWithEqualsLength_whereNotAllEntriesInRange() throws Exception {
+        DeviceId deviceId = new DeviceId(Uuids.timeBased());
+        for (long i = TS - 1L; i <= TS + 100000L + 1L; i += 10000) {
+            saveEntries(deviceId, i);
+        }
+
+        List<ReadTsKvQuery> queries = List.of(new BaseReadTsKvQuery(LONG_KEY, TS, TS + 99999, 50000, 1, Aggregation.COUNT, DESC_ORDER));
+
+        List<TsKvEntry> entries = tsService.findAll(tenantId, deviceId, queries).get();
+        Assert.assertEquals(2, entries.size());
+        Assert.assertEquals(toTsEntry(TS + 25000, new LongDataEntry(LONG_KEY, 5L)), entries.get(0));
+        Assert.assertEquals(toTsEntry(TS + 75000, new LongDataEntry(LONG_KEY, 5L)), entries.get(1));
+
+        EntityView entityView = saveAndCreateEntityView(deviceId, List.of(LONG_KEY));
+
+        entries = tsService.findAll(tenantId, entityView.getId(), queries).get();
+        Assert.assertEquals(2, entries.size());
+        Assert.assertEquals(toTsEntry(TS + 25000, new LongDataEntry(LONG_KEY, 5L)), entries.get(0));
+        Assert.assertEquals(toTsEntry(TS + 75000, new LongDataEntry(LONG_KEY, 5L)), entries.get(1));
+    }
+
+    @Test
+    public void testFindByQuery_whenPeriodHaveTwoInterval_whereSecondShorterThanFirst_andNotAllEntriesInRange() throws Exception {
+        DeviceId deviceId = new DeviceId(Uuids.timeBased());
+        for (long i = TS - 1L; i <= TS + 100000L + 1L; i += 10000L) {
+            saveEntries(deviceId, i);
+        }
+
+        List<ReadTsKvQuery> queries = List.of(new BaseReadTsKvQuery(LONG_KEY, TS, TS + 80000, 50000, 1, Aggregation.COUNT, DESC_ORDER));
+
+        List<TsKvEntry> entries = tsService.findAll(tenantId, deviceId, queries).get();
+        Assert.assertEquals(2, entries.size());
+        Assert.assertEquals(toTsEntry(TS + 25000, new LongDataEntry(LONG_KEY, 5L)), entries.get(0));
+        Assert.assertEquals(toTsEntry(TS + 65000, new LongDataEntry(LONG_KEY, 3L)), entries.get(1));
+
+        EntityView entityView = saveAndCreateEntityView(deviceId, List.of(LONG_KEY));
+
+        entries = tsService.findAll(tenantId, entityView.getId(), queries).get();
+        Assert.assertEquals(2, entries.size());
+        Assert.assertEquals(toTsEntry(TS + 25000, new LongDataEntry(LONG_KEY, 5L)), entries.get(0));
+        Assert.assertEquals(toTsEntry(TS + 65000, new LongDataEntry(LONG_KEY, 3L)), entries.get(1));
+    }
+
     @Test
     public void testFindByQueryDescOrder() throws Exception {
         DeviceId deviceId = new DeviceId(Uuids.timeBased());

dao/src/test/java/org/thingsboard/server/dao/sqlts/AbstractChunkedAggregationTimeseriesDaoTest.java

@@ -0,0 +1,156 @@
+/**
+ * Copyright © 2016-2022 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.dao.sqlts;
+
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.junit.Before;
+import org.junit.Test;
+import org.thingsboard.server.common.data.kv.BaseReadTsKvQuery;
+import org.thingsboard.server.common.data.kv.ReadTsKvQuery;
+import org.thingsboard.server.common.data.kv.TsKvEntry;
+
+import java.util.Optional;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.BDDMockito.willCallRealMethod;
+import static org.mockito.BDDMockito.willReturn;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.thingsboard.server.common.data.id.TenantId.SYS_TENANT_ID;
+import static org.thingsboard.server.common.data.kv.Aggregation.COUNT;
+
+public class AbstractChunkedAggregationTimeseriesDaoTest {
+
+    final int LIMIT = 1;
+    final String TEMP = "temp";
+    final String DESC = "DESC";
+    AbstractChunkedAggregationTimeseriesDao tsDao;
+
+    @Before
+    public void setUp() throws Exception {
+        tsDao = spy(AbstractChunkedAggregationTimeseriesDao.class);
+        ListenableFuture<Optional<TsKvEntry>> optionalListenableFuture = Futures.immediateFuture(Optional.of(mock(TsKvEntry.class)));
+        willReturn(optionalListenableFuture).given(tsDao).findAndAggregateAsync(any(), anyString(), anyLong(), anyLong(), anyLong(), any());
+        willReturn(Futures.immediateFuture(mock(TsKvEntry.class))).given(tsDao).getTskvEntriesFuture(any());
+    }
+
+    @Test
+    public void givenIntervalNotMultiplePeriod_whenAggregateCount_thenLastIntervalShorterThanOthersAndEqualsEndTs() {
+        ReadTsKvQuery query = new BaseReadTsKvQuery(TEMP, 1, 3000, 2000, LIMIT, COUNT, DESC);
+        ReadTsKvQuery subQueryFirst = new BaseReadTsKvQuery(TEMP, 1, 2001, 1001, LIMIT, COUNT, DESC);
+        ReadTsKvQuery subQuerySecond = new BaseReadTsKvQuery(TEMP, 2001, 3001, 2501, LIMIT, COUNT, DESC);
+        tsDao.findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        verify(tsDao, times(2)).findAndAggregateAsync(any(), any(), anyLong(), anyLong(), anyLong(), any());
+        verify(tsDao, times(1)).findAndAggregateAsync(SYS_TENANT_ID, subQueryFirst.getKey(), 1, 2001, getTsForReadTsKvQuery(1, 2001), COUNT);
+        verify(tsDao, times(1)).findAndAggregateAsync(SYS_TENANT_ID, subQuerySecond.getKey(), 2001, 3000 + 1, getTsForReadTsKvQuery(2001, 3001), COUNT);
+    }
+
+    @Test
+    public void givenIntervalNotMultiplePeriod_whenAggregateCount_thenIntervalEqualsPeriod() {
+        ReadTsKvQuery query = new BaseReadTsKvQuery(TEMP, 1, 3000, 3000, LIMIT, COUNT, DESC);
+        ReadTsKvQuery subQueryFirst = new BaseReadTsKvQuery(TEMP, 1, 3001, 1501, LIMIT, COUNT, DESC);
+        willCallRealMethod().given(tsDao).findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        assertThat(tsDao.findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query)).isNotNull();
+        verify(tsDao, times(1)).findAndAggregateAsync(any(), any(), anyLong(), anyLong(), anyLong(), any());
+        verify(tsDao, times(1)).findAndAggregateAsync(SYS_TENANT_ID, subQueryFirst.getKey(), 1, 3000 + 1, getTsForReadTsKvQuery(1, 3001), COUNT);
+    }
+
+    @Test
+    public void givenIntervalNotMultiplePeriod_whenAggregateCount_thenIntervalEqualsPeriodMinusOne() {
+        ReadTsKvQuery query = new BaseReadTsKvQuery(TEMP, 1, 3000, 2999, LIMIT, COUNT, DESC);
+        ReadTsKvQuery subQueryFirst = new BaseReadTsKvQuery(TEMP, 1, 3000, 1500, LIMIT, COUNT, DESC);
+        ReadTsKvQuery subQuerySecond = new BaseReadTsKvQuery(TEMP, 3000, 3001, 3000, LIMIT, COUNT, DESC);
+        willCallRealMethod().given(tsDao).findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        tsDao.findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        verify(tsDao, times(2)).findAndAggregateAsync(any(), any(), anyLong(), anyLong(), anyLong(), any());
+        verify(tsDao, times(1)).findAndAggregateAsync(SYS_TENANT_ID, subQueryFirst.getKey(), 1, 3000, getTsForReadTsKvQuery(1, 3000), COUNT);
+        verify(tsDao, times(1)).findAndAggregateAsync(SYS_TENANT_ID, subQuerySecond.getKey(), 3000, 3001, getTsForReadTsKvQuery(3000, 3001), COUNT);
+
+    }
+
+    @Test
+    public void givenIntervalNotMultiplePeriod_whenAggregateCount_thenIntervalEqualsPeriodPlusOne() {
+        ReadTsKvQuery query = new BaseReadTsKvQuery(TEMP, 1, 3000, 3001, LIMIT, COUNT, DESC);
+        ReadTsKvQuery subQueryFirst = new BaseReadTsKvQuery(TEMP, 1, 3001, 1501, LIMIT, COUNT, DESC);
+        willCallRealMethod().given(tsDao).findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        tsDao.findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        verify(tsDao, times(1)).findAndAggregateAsync(any(), any(), anyLong(), anyLong(), anyLong(), any());
+        verify(tsDao, times(1)).findAndAggregateAsync(SYS_TENANT_ID, subQueryFirst.getKey(), 1, 3001, getTsForReadTsKvQuery(1, 3001), COUNT);
+    }
+
+    @Test
+    public void givenIntervalNotMultiplePeriod_whenAggregateCount_thenIntervalEqualsOneMillisecondAndStartTsIsZero() {
+        ReadTsKvQuery query = new BaseReadTsKvQuery(TEMP, 0, 0, 1, LIMIT, COUNT, DESC);
+        ReadTsKvQuery subQueryFirst = new BaseReadTsKvQuery(TEMP, 0, 1, 0, LIMIT, COUNT, DESC);
+        willCallRealMethod().given(tsDao).findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        tsDao.findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        verify(tsDao, times(1)).findAndAggregateAsync(any(), any(), anyLong(), anyLong(), anyLong(), any());
+        verify(tsDao, times(1)).findAndAggregateAsync(SYS_TENANT_ID, subQueryFirst.getKey(), 0, 1, getTsForReadTsKvQuery(0, 1), COUNT);
+    }
+
+    @Test
+    public void givenIntervalNotMultiplePeriod_whenAggregateCount_thenIntervalEqualsOneMillisecondAndStartTsIsOne() {
+        ReadTsKvQuery query = new BaseReadTsKvQuery(TEMP, 1, 1, 1, LIMIT, COUNT, DESC);
+        ReadTsKvQuery subQuery = new BaseReadTsKvQuery(TEMP, 1, 2, 1, LIMIT, COUNT, DESC);
+        willCallRealMethod().given(tsDao).findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        tsDao.findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        verify(tsDao, times(1)).findAndAggregateAsync(any(), any(), anyLong(), anyLong(), anyLong(), any());
+        verify(tsDao, times(1)).findAndAggregateAsync(SYS_TENANT_ID, subQuery.getKey(), 1, 2, getTsForReadTsKvQuery(1, 2), COUNT);
+    }
+
+    @Test
+    public void givenIntervalNotMultiplePeriod_whenAggregateCount_thenIntervalEqualsOneMillisecondAndStartTsIsIntegerMax() {
+        ReadTsKvQuery query = new BaseReadTsKvQuery(TEMP, Integer.MAX_VALUE, Integer.MAX_VALUE, 1, LIMIT, COUNT, DESC);
+        ReadTsKvQuery subQueryFirst = new BaseReadTsKvQuery(TEMP, Integer.MAX_VALUE, Integer.MAX_VALUE + 1L, Integer.MAX_VALUE, LIMIT, COUNT, DESC);
+        willCallRealMethod().given(tsDao).findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        tsDao.findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        verify(tsDao, times(1)).findAndAggregateAsync(any(), any(), anyLong(), anyLong(), anyLong(), any());
+        verify(tsDao, times(1)).findAndAggregateAsync(SYS_TENANT_ID, subQueryFirst.getKey(), Integer.MAX_VALUE, 1L + Integer.MAX_VALUE, getTsForReadTsKvQuery(Integer.MAX_VALUE, 1L + Integer.MAX_VALUE), COUNT);
+    }
+
+    @Test
+    public void givenIntervalNotMultiplePeriod_whenAggregateCount_thenIntervalEqualsBigNumber() {
+        ReadTsKvQuery query = new BaseReadTsKvQuery(TEMP, 1, 3000, Integer.MAX_VALUE, LIMIT, COUNT, DESC);
+        ReadTsKvQuery subQueryFirst = new BaseReadTsKvQuery(TEMP, 1, 3001, 1501, LIMIT, COUNT, DESC);
+        willCallRealMethod().given(tsDao).findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        tsDao.findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        verify(tsDao, times(1)).findAndAggregateAsync(any(), any(), anyLong(), anyLong(), anyLong(), any());
+        verify(tsDao, times(1)).findAndAggregateAsync(SYS_TENANT_ID, subQueryFirst.getKey(), 1, 3001, getTsForReadTsKvQuery(1, 3001), COUNT);
+    }
+
+    @Test
+    public void givenIntervalNotMultiplePeriod_whenAggregateCount_thenCountIntervalEqualsPeriodSize() {
+        ReadTsKvQuery query = new BaseReadTsKvQuery(TEMP, 1, 3000, 3, LIMIT, COUNT, DESC);
+        willCallRealMethod().given(tsDao).findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        tsDao.findAllAsync(SYS_TENANT_ID, SYS_TENANT_ID, query);
+        verify(tsDao, times(1000)).findAndAggregateAsync(any(), any(), anyLong(), anyLong(), anyLong(), any());
+        for (long i = 1; i <= 3000; i += 3) {
+            ReadTsKvQuery querySub = new BaseReadTsKvQuery(TEMP, i, i + 3, i + (i + 3 - i) / 2, LIMIT, COUNT, DESC);
+            verify(tsDao, times(1)).findAndAggregateAsync(SYS_TENANT_ID, querySub.getKey(), i, i + 3, getTsForReadTsKvQuery(i, i + 3), COUNT);
+        }
+    }
+
+    long getTsForReadTsKvQuery(long startTs, long endTs) {
+        return startTs + (endTs - startTs) / 2L;
+    }
+
+}