tsai
/
thingsboard
mirror of https://github.com/thingsboard/thingsboard.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
Browse Source

cassandra partition - implemented configurable period support for always_exist_in_reading mode

pull/7629/head
Sergey Matvienko 4 years ago
parent
55192e5c8f
commit
293a7ab078
7 changed files with 531 additions and 35 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Split View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 21
      dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java
  2. 93
      dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoPartitioningDaysAlwaysExistsTest.java
  3. 93
      dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoPartitioningHoursAlwaysExistsTest.java
  4. 40
      dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoPartitioningIndefiniteAlwaysExistsTest.java
  5. 93
      dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoPartitioningMinutesAlwaysExistsTest.java
  6. 134
      dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoPartitioningMonthsAlwaysExistsTest.java
  7. 92
      dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoPartitioningYearsAlwaysExistsTest.java

21
dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java
View File

@ -60,7 +60,6 @@ import java.time.LocalDateTime;
import java.time.ZoneOffset;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
@ -83,8 +82,6 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD
protected static final int MIN_AGGREGATION_STEP_MS = 1000;
public static final String ASC_ORDER = "ASC";
public static final long SECONDS_IN_DAY = TimeUnit.DAYS.toSeconds(1);
static final long DAYS_32_MS = TimeUnit.DAYS.toMillis(32);
protected static final List<Long> FIXED_PARTITION = List.of(0L);
private CassandraTsPartitionsCache cassandraTsPartitionsCache;
@ -110,6 +107,7 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD
private boolean setNullValuesEnabled;
private NoSqlTsPartitionDate tsFormat;
private long partitionDurationPlusOneMs;
private PreparedStatement partitionInsertStmt;
private PreparedStatement partitionInsertTtlStmt;
@ -134,6 +132,9 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD
Optional<NoSqlTsPartitionDate> partition = NoSqlTsPartitionDate.parse(partitioning);
if (partition.isPresent()) {
tsFormat = partition.get();
if (!isFixedPartitioning()) {
partitionDurationPlusOneMs = partitionToMs(tsFormat) + 1;
}
if (!isFixedPartitioning() && partitionsCacheSize > 0) {
cassandraTsPartitionsCache = new CassandraTsPartitionsCache(partitionsCacheSize);
}
@ -143,6 +144,18 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD
}
}
long partitionToMs(final NoSqlTsPartitionDate partition) {
switch (partition) {
case MINUTES: return TimeUnit.MINUTES.toMillis(1);
case HOURS: return TimeUnit.HOURS.toMillis(1);
case DAYS: return TimeUnit.DAYS.toMillis(1);
case MONTHS: return TimeUnit.DAYS.toMillis(31); //the longest month
case YEARS: return TimeUnit.DAYS.toMillis(366); // leap year
}
log.warn("Can not convert partition to milliseconds. There are no mapping [{}] for partitioning [{}]", partition, partitioning);
throw new RuntimeException("Failed to convert partition to ms: " + partitioning + "!");
}
@PreDestroy
public void stop() {
super.stopExecutor();
@ -439,7 +452,7 @@ public class CassandraBaseTimeseriesDao extends AbstractCassandraBaseTimeseriesD
partitions.add(minPartition);
long currentPartition = minPartition;
while (maxPartition > (currentPartition = toPartitionTs(currentPartition + DAYS_32_MS))){
while (maxPartition > (currentPartition = toPartitionTs(currentPartition + partitionDurationPlusOneMs))){
partitions.add(currentPartition);
}

93
dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoPartitioningDaysAlwaysExistsTest.java
View File

@ -0,0 +1,93 @@
/**
* 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.timeseries;
import lombok.extern.slf4j.Slf4j;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Answers;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit4.SpringRunner;
import org.thingsboard.server.dao.cassandra.CassandraCluster;
import org.thingsboard.server.dao.nosql.CassandraBufferedRateReadExecutor;
import org.thingsboard.server.dao.nosql.CassandraBufferedRateWriteExecutor;
import java.text.ParseException;
import java.util.List;
import static org.apache.commons.lang3.time.DateFormatUtils.ISO_DATETIME_TIME_ZONE_FORMAT;
import static org.assertj.core.api.Assertions.assertThat;
@RunWith(SpringRunner.class)
@SpringBootTest(classes = CassandraBaseTimeseriesDao.class)
@TestPropertySource(properties = {
"database.ts.type=cassandra",
"cassandra.query.ts_key_value_partitioning=DAYS",
"cassandra.query.ts_key_value_partitioning_always_exist_in_reading=true",
"cassandra.query.ts_key_value_partitions_max_cache_size=100000",
"cassandra.query.ts_key_value_partitions_cache_stats_enabled=true",
"cassandra.query.ts_key_value_partitions_cache_stats_interval=60",
"cassandra.query.ts_key_value_ttl=0",
"cassandra.query.set_null_values_enabled=false",
})
@Slf4j
public class CassandraBaseTimeseriesDaoPartitioningDaysAlwaysExistsTest {
@Autowired
CassandraBaseTimeseriesDao tsDao;
@MockBean(answer = Answers.RETURNS_MOCKS)
@Qualifier("CassandraCluster")
CassandraCluster cassandraCluster;
@MockBean
CassandraBufferedRateReadExecutor cassandraBufferedRateReadExecutor;
@MockBean
CassandraBufferedRateWriteExecutor cassandraBufferedRateWriteExecutor;
@Test
public void testToPartitionsDays() throws ParseException {
assertThat(tsDao.getPartitioning()).isEqualTo("DAYS");
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-02T00:00:00Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-02T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-03T00:00:01Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-03T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-31T23:59:59Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-31T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2023-12-31T23:59:59Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2023-12-31T00:00:00Z").getTime());
}
@Ignore //TODO make test for Days
@Test
public void testCalculatePartitionsDays() throws ParseException {
}
}

93
dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoPartitioningHoursAlwaysExistsTest.java
View File

@ -0,0 +1,93 @@
/**
* 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.timeseries;
import lombok.extern.slf4j.Slf4j;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Answers;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit4.SpringRunner;
import org.thingsboard.server.dao.cassandra.CassandraCluster;
import org.thingsboard.server.dao.nosql.CassandraBufferedRateReadExecutor;
import org.thingsboard.server.dao.nosql.CassandraBufferedRateWriteExecutor;
import java.text.ParseException;
import java.util.List;
import static org.apache.commons.lang3.time.DateFormatUtils.ISO_DATETIME_TIME_ZONE_FORMAT;
import static org.assertj.core.api.Assertions.assertThat;
@RunWith(SpringRunner.class)
@SpringBootTest(classes = CassandraBaseTimeseriesDao.class)
@TestPropertySource(properties = {
"database.ts.type=cassandra",
"cassandra.query.ts_key_value_partitioning=HOURS",
"cassandra.query.ts_key_value_partitioning_always_exist_in_reading=true",
"cassandra.query.ts_key_value_partitions_max_cache_size=100000",
"cassandra.query.ts_key_value_partitions_cache_stats_enabled=true",
"cassandra.query.ts_key_value_partitions_cache_stats_interval=60",
"cassandra.query.ts_key_value_ttl=0",
"cassandra.query.set_null_values_enabled=false",
})
@Slf4j
public class CassandraBaseTimeseriesDaoPartitioningHoursAlwaysExistsTest {
@Autowired
CassandraBaseTimeseriesDao tsDao;
@MockBean(answer = Answers.RETURNS_MOCKS)
@Qualifier("CassandraCluster")
CassandraCluster cassandraCluster;
@MockBean
CassandraBufferedRateReadExecutor cassandraBufferedRateReadExecutor;
@MockBean
CassandraBufferedRateWriteExecutor cassandraBufferedRateWriteExecutor;
@Test
public void testToPartitionsHours() throws ParseException {
assertThat(tsDao.getPartitioning()).isEqualTo("HOURS");
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-02T01:00:00Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-02T01:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-03T02:00:01Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-03T02:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-31T23:59:59Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-31T23:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2023-12-31T23:59:59Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2023-12-31T23:00:00Z").getTime());
}
@Ignore //TODO make test for Hours
@Test
public void testCalculatePartitionsHours() throws ParseException {
}
}

40
dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoTest.java → dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoPartitioningIndefiniteAlwaysExistsTest.java
View File

@ -16,6 +16,7 @@
package org.thingsboard.server.dao.timeseries;
import lombok.extern.slf4j.Slf4j;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Answers;
@ -39,7 +40,7 @@ import static org.assertj.core.api.Assertions.assertThat;
@SpringBootTest(classes = CassandraBaseTimeseriesDao.class)
@TestPropertySource(properties = {
"database.ts.type=cassandra",
"cassandra.query.ts_key_value_partitioning=MONTHS",
"cassandra.query.ts_key_value_partitioning=INDEFINITE",
"cassandra.query.ts_key_value_partitioning_always_exist_in_reading=true",
"cassandra.query.ts_key_value_partitions_max_cache_size=100000",
"cassandra.query.ts_key_value_partitions_cache_stats_enabled=true",
@ -48,7 +49,7 @@ import static org.assertj.core.api.Assertions.assertThat;
"cassandra.query.set_null_values_enabled=false",
})
@Slf4j
public class CassandraBaseTimeseriesDaoTest {
public class CassandraBaseTimeseriesDaoPartitioningIndefiniteAlwaysExistsTest {
@Autowired
CassandraBaseTimeseriesDao tsDao;
@ -63,38 +64,15 @@ public class CassandraBaseTimeseriesDaoTest {
CassandraBufferedRateWriteExecutor cassandraBufferedRateWriteExecutor;
@Test
public void testToPartitionsMonths() throws ParseException {
assertThat(tsDao.getPartitioning()).isEqualTo("MONTHS");
assertThat(tsDao.toPartitionTs(ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime())).isEqualTo(1640995200000L);
assertThat(tsDao.toPartitionTs(ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-01T00:00:00Z").getTime())).isEqualTo(1651363200000L);
assertThat(tsDao.toPartitionTs(ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-01T00:00:01Z").getTime())).isEqualTo(1651363200000L);
assertThat(tsDao.toPartitionTs(ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-31T23:59:59Z").getTime())).isEqualTo(1651363200000L);
assertThat(tsDao.toPartitionTs(ISO_DATETIME_TIME_ZONE_FORMAT.parse("2023-12-31T23:59:59Z").getTime())).isEqualTo(1701388800000L);
public void testToPartitionsIndefinite() throws ParseException {
assertThat(tsDao.getPartitioning()).isEqualTo("INDEFINITE");
assertThat(tsDao.toPartitionTs(ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime())).isEqualTo(0L);
}
@Test
public void testCalculatePartitions() throws ParseException {
long startTs = tsDao.toPartitionTs(ISO_DATETIME_TIME_ZONE_FORMAT.parse("2019-12-12T00:00:00Z").getTime());
long nextTs = tsDao.toPartitionTs(ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-01-31T23:59:59Z").getTime());
long leapTs = tsDao.toPartitionTs(ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-02-29T23:59:59Z").getTime());
long endTs = tsDao.toPartitionTs(ISO_DATETIME_TIME_ZONE_FORMAT.parse("2021-01-31T23:59:59Z").getTime());
log.warn("startTs {}, nextTs {}, leapTs {}, endTs {}", startTs, nextTs, leapTs, endTs);
assertThat(tsDao.calculatePartitions(0, 0)).isEqualTo(List.of(0L));
assertThat(tsDao.calculatePartitions(0, 1)).isEqualTo(List.of(0L, 1L));
assertThat(tsDao.calculatePartitions(startTs, startTs)).isEqualTo(List.of(1575158400000L));
assertThat(tsDao.calculatePartitions(startTs, nextTs)).isEqualTo(List.of(1575158400000L, 1577836800000L));
assertThat(tsDao.calculatePartitions(startTs, leapTs)).isEqualTo(List.of(1575158400000L, 1577836800000L, 1580515200000L));
assertThat(tsDao.calculatePartitions(startTs, endTs)).hasSize(14);
assertThat(tsDao.calculatePartitions(startTs, endTs)).isEqualTo(List.of(
1575158400000L,
1577836800000L, 1580515200000L, 1583020800000L,
1585699200000L, 1588291200000L, 1590969600000L,
1593561600000L, 1596240000000L, 1598918400000L,
1601510400000L, 1604188800000L, 1606780800000L,
1609459200000L));
@Test
public void testCalculatePartitionsIndefinite() throws ParseException {
//Indefinite partitioning should never call tsDao.calculatePartitions()
}
}

93
dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoPartitioningMinutesAlwaysExistsTest.java
View File

@ -0,0 +1,93 @@
/**
* 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.timeseries;
import lombok.extern.slf4j.Slf4j;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Answers;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit4.SpringRunner;
import org.thingsboard.server.dao.cassandra.CassandraCluster;
import org.thingsboard.server.dao.nosql.CassandraBufferedRateReadExecutor;
import org.thingsboard.server.dao.nosql.CassandraBufferedRateWriteExecutor;
import java.text.ParseException;
import java.util.List;
import static org.apache.commons.lang3.time.DateFormatUtils.ISO_DATETIME_TIME_ZONE_FORMAT;
import static org.assertj.core.api.Assertions.assertThat;
@RunWith(SpringRunner.class)
@SpringBootTest(classes = CassandraBaseTimeseriesDao.class)
@TestPropertySource(properties = {
"database.ts.type=cassandra",
"cassandra.query.ts_key_value_partitioning=MINUTES",
"cassandra.query.ts_key_value_partitioning_always_exist_in_reading=true",
"cassandra.query.ts_key_value_partitions_max_cache_size=100000",
"cassandra.query.ts_key_value_partitions_cache_stats_enabled=true",
"cassandra.query.ts_key_value_partitions_cache_stats_interval=60",
"cassandra.query.ts_key_value_ttl=0",
"cassandra.query.set_null_values_enabled=false",
})
@Slf4j
public class CassandraBaseTimeseriesDaoPartitioningMinutesAlwaysExistsTest {
@Autowired
CassandraBaseTimeseriesDao tsDao;
@MockBean(answer = Answers.RETURNS_MOCKS)
@Qualifier("CassandraCluster")
CassandraCluster cassandraCluster;
@MockBean
CassandraBufferedRateReadExecutor cassandraBufferedRateReadExecutor;
@MockBean
CassandraBufferedRateWriteExecutor cassandraBufferedRateWriteExecutor;
@Test
public void testToPartitionsMinutes() throws ParseException {
assertThat(tsDao.getPartitioning()).isEqualTo("MINUTES");
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-02T00:01:00Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-02T00:01:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-03T00:02:01Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-03T00:02:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-31T23:59:59Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-31T23:59:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2023-12-31T23:59:59Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2023-12-31T23:59:00Z").getTime());
}
@Ignore //TODO make test for Minutes
@Test
public void testCalculatePartitionsMinutes() throws ParseException {
}
}

134
dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoPartitioningMonthsAlwaysExistsTest.java
View File

@ -0,0 +1,134 @@
/**
* 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.timeseries;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Answers;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit4.SpringRunner;
import org.thingsboard.server.dao.cassandra.CassandraCluster;
import org.thingsboard.server.dao.nosql.CassandraBufferedRateReadExecutor;
import org.thingsboard.server.dao.nosql.CassandraBufferedRateWriteExecutor;
import java.text.ParseException;
import java.util.List;
import static org.apache.commons.lang3.time.DateFormatUtils.ISO_DATETIME_TIME_ZONE_FORMAT;
import static org.assertj.core.api.Assertions.assertThat;
@RunWith(SpringRunner.class)
@SpringBootTest(classes = CassandraBaseTimeseriesDao.class)
@TestPropertySource(properties = {
"database.ts.type=cassandra",
"cassandra.query.ts_key_value_partitioning=MONTHS",
"cassandra.query.ts_key_value_partitioning_always_exist_in_reading=true",
"cassandra.query.ts_key_value_partitions_max_cache_size=100000",
"cassandra.query.ts_key_value_partitions_cache_stats_enabled=true",
"cassandra.query.ts_key_value_partitions_cache_stats_interval=60",
"cassandra.query.ts_key_value_ttl=0",
"cassandra.query.set_null_values_enabled=false",
})
@Slf4j
public class CassandraBaseTimeseriesDaoPartitioningMonthsAlwaysExistsTest {
@Autowired
CassandraBaseTimeseriesDao tsDao;
@MockBean(answer = Answers.RETURNS_MOCKS)
@Qualifier("CassandraCluster")
CassandraCluster cassandraCluster;
@MockBean
CassandraBufferedRateReadExecutor cassandraBufferedRateReadExecutor;
@MockBean
CassandraBufferedRateWriteExecutor cassandraBufferedRateWriteExecutor;
@Test
public void testToPartitionsMonths() throws ParseException {
assertThat(tsDao.getPartitioning()).isEqualTo("MONTHS");
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime())).isEqualTo(1640995200000L).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-01T00:00:00Z").getTime())).isEqualTo(1651363200000L).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-01T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-01T00:00:01Z").getTime())).isEqualTo(1651363200000L).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-01T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-31T23:59:59Z").getTime())).isEqualTo(1651363200000L).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-01T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2023-12-31T23:59:59Z").getTime())).isEqualTo(1701388800000L).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2023-12-01T00:00:00Z").getTime());
}
@Test
public void testCalculatePartitionsMonths() throws ParseException {
long startTs = tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2019-12-12T00:00:00Z").getTime());
long nextTs = tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-01-31T23:59:59Z").getTime());
long leapTs = tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-02-29T23:59:59Z").getTime());
long endTs = tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2021-01-31T23:59:59Z").getTime());
log.info("startTs {}, nextTs {}, leapTs {}, endTs {}", startTs, nextTs, leapTs, endTs);
assertThat(tsDao.calculatePartitions(0, 0)).isEqualTo(List.of(0L));
assertThat(tsDao.calculatePartitions(0, 1)).isEqualTo(List.of(0L, 1L));
assertThat(tsDao.calculatePartitions(startTs, startTs)).isEqualTo(List.of(1575158400000L)).isEqualTo(List.of(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2019-12-01T00:00:00Z").getTime()));
assertThat(tsDao.calculatePartitions(startTs, nextTs)).isEqualTo(List.of(1575158400000L, 1577836800000L)).isEqualTo(List.of(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2019-12-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-01-01T00:00:00Z").getTime()));
assertThat(tsDao.calculatePartitions(startTs, leapTs)).isEqualTo(List.of(1575158400000L, 1577836800000L, 1580515200000L)).isEqualTo(List.of(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2019-12-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-01-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-02-01T00:00:00Z").getTime()));
assertThat(tsDao.calculatePartitions(startTs, endTs)).hasSize(14).isEqualTo(List.of(
1575158400000L,
1577836800000L, 1580515200000L, 1583020800000L,
1585699200000L, 1588291200000L, 1590969600000L,
1593561600000L, 1596240000000L, 1598918400000L,
1601510400000L, 1604188800000L, 1606780800000L,
1609459200000L)).isEqualTo(List.of(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2019-12-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-01-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-02-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-03-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-04-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-05-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-06-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-07-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-08-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-09-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-10-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-11-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2020-12-01T00:00:00Z").getTime(),
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2021-01-01T00:00:00Z").getTime()));
}
}

92
dao/src/test/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDaoPartitioningYearsAlwaysExistsTest.java
View File

@ -0,0 +1,92 @@
/**
* 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.timeseries;
import lombok.extern.slf4j.Slf4j;
import org.junit.Ignore;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.Answers;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.mock.mockito.MockBean;
import org.springframework.test.context.TestPropertySource;
import org.springframework.test.context.junit4.SpringRunner;
import org.thingsboard.server.dao.cassandra.CassandraCluster;
import org.thingsboard.server.dao.nosql.CassandraBufferedRateReadExecutor;
import org.thingsboard.server.dao.nosql.CassandraBufferedRateWriteExecutor;
import java.text.ParseException;
import java.util.List;
import static org.apache.commons.lang3.time.DateFormatUtils.ISO_DATETIME_TIME_ZONE_FORMAT;
import static org.assertj.core.api.Assertions.assertThat;
@RunWith(SpringRunner.class)
@SpringBootTest(classes = CassandraBaseTimeseriesDao.class)
@TestPropertySource(properties = {
"database.ts.type=cassandra",
"cassandra.query.ts_key_value_partitioning=YEARS",
"cassandra.query.ts_key_value_partitioning_always_exist_in_reading=true",
"cassandra.query.ts_key_value_partitions_max_cache_size=100000",
"cassandra.query.ts_key_value_partitions_cache_stats_enabled=true",
"cassandra.query.ts_key_value_partitions_cache_stats_interval=60",
"cassandra.query.ts_key_value_ttl=0",
"cassandra.query.set_null_values_enabled=false",
})
@Slf4j
public class CassandraBaseTimeseriesDaoPartitioningYearsAlwaysExistsTest {
@Autowired
CassandraBaseTimeseriesDao tsDao;
@MockBean(answer = Answers.RETURNS_MOCKS)
@Qualifier("CassandraCluster")
CassandraCluster cassandraCluster;
@MockBean
CassandraBufferedRateReadExecutor cassandraBufferedRateReadExecutor;
@MockBean
CassandraBufferedRateWriteExecutor cassandraBufferedRateWriteExecutor;
@Test
public void testToPartitionsYears() throws ParseException {
assertThat(tsDao.getPartitioning()).isEqualTo("YEARS");
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-01T00:00:00Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-01T00:00:01Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-05-31T23:59:59Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2022-01-01T00:00:00Z").getTime());
assertThat(tsDao.toPartitionTs(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2023-12-31T23:59:59Z").getTime())).isEqualTo(
ISO_DATETIME_TIME_ZONE_FORMAT.parse("2023-01-01T00:00:00Z").getTime());
}
@Ignore //TODO make test for Years
@Test
public void testCalculatePartitionsYears() throws ParseException {
}
}
Write Preview
Loading…
Cancel
Save