From d6d2de6d9d4557ff29b18edb0d5f7b8fdb746ecc Mon Sep 17 00:00:00 2001 From: Volodymyr Babak Date: Wed, 14 Jun 2017 19:27:35 +0300 Subject: [PATCH] Added ts kv latest and ts kv jps dao support --- .../server/common/data/kv/TsKvQuery.java | 2 - .../dao/model/sql/AttributeKvEntity.java | 4 + .../dao/model/sql/RelationCompositeKey.java | 8 ++ .../server/dao/model/sql/RelationEntity.java | 89 +------------------ .../dao/model/sql/TsKvCompositeKey.java | 40 +++++++++ .../server/dao/model/sql/TsKvEntity.java | 69 ++++++++++++++ .../dao/model/sql/TsKvLatestCompositeKey.java | 39 ++++++++ .../dao/model/sql/TsKvLatestEntity.java | 68 ++++++++++++++ .../dao/sql/timeseries/JpaTimeseriesDao.java | 68 +++++++++++--- .../sql/timeseries/TsKvLatestRepository.java | 30 +++++++ .../dao/sql/timeseries/TsKvRepository.java | 25 ++++++ .../dao/timeseries/BaseTimeseriesService.java | 12 ++- .../CassandraBaseTimeseriesDao.java | 17 ++-- .../server/dao/timeseries/TimeseriesDao.java | 6 +- dao/src/main/resources/postgres/schema.sql | 50 ++++++++--- 15 files changed, 398 insertions(+), 129 deletions(-) create mode 100644 dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvCompositeKey.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvEntity.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvLatestCompositeKey.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvLatestEntity.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvLatestRepository.java create mode 100644 dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvRepository.java diff --git a/common/data/src/main/java/org/thingsboard/server/common/data/kv/TsKvQuery.java b/common/data/src/main/java/org/thingsboard/server/common/data/kv/TsKvQuery.java index 8d60f525f4..8ede698241 100644 --- a/common/data/src/main/java/org/thingsboard/server/common/data/kv/TsKvQuery.java +++ b/common/data/src/main/java/org/thingsboard/server/common/data/kv/TsKvQuery.java @@ -15,8 +15,6 @@ */ package org.thingsboard.server.common.data.kv; -import java.util.Optional; - public interface TsKvQuery { String getKey(); diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/AttributeKvEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/AttributeKvEntity.java index 2ea58afcf2..dd3f28362f 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/sql/AttributeKvEntity.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/AttributeKvEntity.java @@ -16,6 +16,8 @@ package org.thingsboard.server.dao.model.sql; import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.ToString; import org.thingsboard.server.common.data.kv.*; import org.thingsboard.server.dao.model.ToData; @@ -29,6 +31,8 @@ import static org.thingsboard.server.dao.model.ModelConstants.*; @Entity @Table(name = "attribute_kv") @IdClass(AttributeKvCompositeKey.class) +@EqualsAndHashCode +@ToString public class AttributeKvEntity implements ToData, Serializable { @Id diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/RelationCompositeKey.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/RelationCompositeKey.java index 3e13ff5e6f..f08f873b61 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/sql/RelationCompositeKey.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/RelationCompositeKey.java @@ -17,15 +17,23 @@ package org.thingsboard.server.dao.model.sql; import lombok.AllArgsConstructor; import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.ToString; import org.thingsboard.server.common.data.relation.EntityRelation; +import javax.persistence.Transient; import java.io.Serializable; import java.util.UUID; @AllArgsConstructor @Data +@EqualsAndHashCode +@ToString public class RelationCompositeKey implements Serializable { + @Transient + private static final long serialVersionUID = -4089175869616037592L; + private UUID fromId; private String fromType; private UUID toId; diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/RelationEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/RelationEntity.java index 1b15aa1589..f8aa72cb8a 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/model/sql/RelationEntity.java +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/RelationEntity.java @@ -17,6 +17,8 @@ package org.thingsboard.server.dao.model.sql; import com.fasterxml.jackson.databind.JsonNode; import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.ToString; import org.hibernate.annotations.Type; import org.thingsboard.server.common.data.id.EntityIdFactory; import org.thingsboard.server.common.data.relation.EntityRelation; @@ -32,11 +34,10 @@ import static org.thingsboard.server.dao.model.ModelConstants.*; @Entity @Table(name = RELATION_COLUMN_FAMILY_NAME) @IdClass(RelationCompositeKey.class) +@EqualsAndHashCode +@ToString public final class RelationEntity implements ToData { - @Transient - private static final long serialVersionUID = -4089175869616037592L; - @Id @Column(name = RELATION_FROM_ID_PROPERTY) private UUID fromId; @@ -83,88 +84,6 @@ public final class RelationEntity implements ToData { this.additionalInfo = relation.getAdditionalInfo(); } - @Override - public int hashCode() { - final int prime = 31; - int result = 1; - result = prime * result + ((additionalInfo == null) ? 0 : additionalInfo.hashCode()); - result = prime * result + ((toId == null) ? 0 : toId.hashCode()); - result = prime * result + ((toType == null) ? 0 : toType.hashCode()); - result = prime * result + ((fromId == null) ? 0 : fromId.hashCode()); - result = prime * result + ((fromType == null) ? 0 : fromType.hashCode()); - result = prime * result + ((relationType == null) ? 0 : relationType.hashCode()); - result = prime * result + ((relationTypeGroup == null) ? 0 : relationTypeGroup.hashCode()); - return result; - } - - @Override - public boolean equals(Object obj) { - if (this == obj) - return true; - if (obj == null) - return false; - if (getClass() != obj.getClass()) - return false; - RelationEntity other = (RelationEntity) obj; - if (additionalInfo == null) { - if (other.additionalInfo != null) - return false; - } else if (!additionalInfo.equals(other.additionalInfo)) - return false; - if (toId == null) { - if (other.toId != null) - return false; - } else if (!toId.equals(other.toId)) - return false; - if (fromId == null) { - if (other.fromId != null) - return false; - } else if (!fromId.equals(other.fromId)) - return false; - if (toType == null) { - if (other.toType != null) - return false; - } else if (!toType.equals(other.toType)) - return false; - if (fromType == null) { - if (other.fromType != null) - return false; - } else if (!fromType.equals(other.fromType)) - return false; - if (relationType == null) { - if (other.relationType != null) - return false; - } else if (!relationType.equals(other.relationType)) - return false; - if (relationTypeGroup == null) { - if (other.relationTypeGroup != null) - return false; - } else if (!relationTypeGroup.equals(other.relationTypeGroup)) - return false; - return true; - } - - @Override - public String toString() { - StringBuilder builder = new StringBuilder(); - builder.append("AssetEntity [toId="); - builder.append(toId); - builder.append(", toType="); - builder.append(toType); - builder.append(", fromId="); - builder.append(fromId); - builder.append(", fromType="); - builder.append(fromType); - builder.append(", relationType="); - builder.append(relationType); - builder.append(", relationTypeGroup="); - builder.append(relationTypeGroup); - builder.append(", additionalInfo="); - builder.append(additionalInfo); - builder.append("]"); - return builder.toString(); - } - @Override public EntityRelation toData() { EntityRelation relation = new EntityRelation(); diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvCompositeKey.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvCompositeKey.java new file mode 100644 index 0000000000..037c2dff6f --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvCompositeKey.java @@ -0,0 +1,40 @@ +/** + * Copyright © 2016-2017 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.model.sql; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.ToString; + +import javax.persistence.Transient; +import java.io.Serializable; +import java.util.UUID; + +@Data +@AllArgsConstructor +@EqualsAndHashCode +@ToString +public class TsKvCompositeKey implements Serializable{ + + @Transient + private static final long serialVersionUID = -4089175869616037523L; + + private String entityType; + private UUID entityId; + private String key; + private long ts; +} \ No newline at end of file diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvEntity.java new file mode 100644 index 0000000000..7d6901ddfa --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvEntity.java @@ -0,0 +1,69 @@ +/** + * Copyright © 2016-2017 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.model.sql; + +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.ToString; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.dao.model.ToData; + +import javax.persistence.*; +import java.util.UUID; + +import static org.thingsboard.server.dao.model.ModelConstants.*; + +@Data +@Entity +@Table(name = "ts_kv") +@IdClass(TsKvCompositeKey.class) +@EqualsAndHashCode +@ToString +public final class TsKvEntity implements ToData { + + @Id + @Column(name = ENTITY_TYPE_COLUMN) + private String entityType; + + @Id + @Column(name = ENTITY_ID_COLUMN) + private UUID entityId; + + @Id + @Column(name = KEY_COLUMN) + private String key; + + @Id + @Column(name = TS_COLUMN) + private long ts; + + @Column(name = BOOLEAN_VALUE_COLUMN) + private Boolean booleanValue; + + @Column(name = STRING_VALUE_COLUMN) + private String strValue; + + @Column(name = LONG_VALUE_COLUMN) + private Long longValue; + + @Column(name = DOUBLE_VALUE_COLUMN) + private Double doubleValue; + + @Override + public TsKvEntry toData() { + return null; + } +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvLatestCompositeKey.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvLatestCompositeKey.java new file mode 100644 index 0000000000..a68ddf7067 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvLatestCompositeKey.java @@ -0,0 +1,39 @@ +/** + * Copyright © 2016-2017 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.model.sql; + +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.ToString; + +import javax.persistence.Transient; +import java.io.Serializable; +import java.util.UUID; + +@Data +@AllArgsConstructor +@EqualsAndHashCode +@ToString +public class TsKvLatestCompositeKey implements Serializable{ + + @Transient + private static final long serialVersionUID = -4089175869616037523L; + + private String entityType; + private UUID entityId; + private String key; +} \ No newline at end of file diff --git a/dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvLatestEntity.java b/dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvLatestEntity.java new file mode 100644 index 0000000000..803b89a64e --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/model/sql/TsKvLatestEntity.java @@ -0,0 +1,68 @@ +/** + * Copyright © 2016-2017 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.model.sql; + +import lombok.Data; +import lombok.EqualsAndHashCode; +import lombok.ToString; +import org.thingsboard.server.common.data.kv.TsKvEntry; +import org.thingsboard.server.dao.model.ToData; + +import javax.persistence.*; +import java.util.UUID; + +import static org.thingsboard.server.dao.model.ModelConstants.*; + +@Data +@Entity +@Table(name = "ts_kv_latest") +@IdClass(TsKvLatestCompositeKey.class) +@EqualsAndHashCode +@ToString +public final class TsKvLatestEntity implements ToData { + + @Id + @Column(name = ENTITY_TYPE_COLUMN) + private String entityType; + + @Id + @Column(name = ENTITY_ID_COLUMN) + private UUID entityId; + + @Id + @Column(name = KEY_COLUMN) + private String key; + + @Column(name = TS_COLUMN) + private long ts; + + @Column(name = BOOLEAN_VALUE_COLUMN) + private Boolean booleanValue; + + @Column(name = STRING_VALUE_COLUMN) + private String strValue; + + @Column(name = LONG_VALUE_COLUMN) + private Long longValue; + + @Column(name = DOUBLE_VALUE_COLUMN) + private Double doubleValue; + + @Override + public TsKvEntry toData() { + return null; + } +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/JpaTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/JpaTimeseriesDao.java index 9e53cc6b99..f3371f23e4 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/JpaTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/JpaTimeseriesDao.java @@ -15,13 +15,20 @@ */ package org.thingsboard.server.dao.sql.timeseries; +import com.google.common.collect.Lists; import com.google.common.util.concurrent.ListenableFuture; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.stereotype.Component; import org.thingsboard.server.common.data.id.EntityId; import org.thingsboard.server.common.data.kv.TsKvEntry; import org.thingsboard.server.common.data.kv.TsKvQuery; +import org.thingsboard.server.dao.DaoUtil; +import org.thingsboard.server.dao.model.sql.TsKvEntity; +import org.thingsboard.server.dao.model.sql.TsKvLatestCompositeKey; +import org.thingsboard.server.dao.model.sql.TsKvLatestEntity; +import org.thingsboard.server.dao.sql.JpaAbstractDaoListeningExecutorService; import org.thingsboard.server.dao.timeseries.TimeseriesDao; import java.util.List; @@ -29,41 +36,80 @@ import java.util.List; @Component @Slf4j @ConditionalOnProperty(prefix = "sql", value = "enabled", havingValue = "true") -public class JpaTimeseriesDao implements TimeseriesDao { +public class JpaTimeseriesDao extends JpaAbstractDaoListeningExecutorService implements TimeseriesDao { - @Override - public long toPartitionTs(long ts) { - return 0; - } + @Autowired + private TsKvRepository tsKvRepository; + + @Autowired + private TsKvLatestRepository tsKvLatestRepository; @Override public ListenableFuture> findAllAsync(EntityId entityId, List queries) { + return null; + } + + private ListenableFuture> findAllAsync(EntityId entityId, TsKvQuery query) { return null; } @Override public ListenableFuture findLatest(EntityId entityId, String key) { - return null; + TsKvLatestCompositeKey compositeKey = + new TsKvLatestCompositeKey( + entityId.getEntityType().name(), + entityId.getId(), + key); + return service.submit(() -> + DaoUtil.getData(tsKvLatestRepository.findOne(compositeKey))); } @Override public ListenableFuture> findAllLatest(EntityId entityId) { - return null; + return service.submit(() -> + DaoUtil.convertDataList(Lists.newArrayList( + tsKvLatestRepository.findAllByEntityTypeAndEntityId( + entityId.getEntityType().name(), + entityId.getId())))); } @Override - public ListenableFuture save(EntityId entityId, long partition, TsKvEntry tsKvEntry, long ttl) { - return null; + public ListenableFuture save(EntityId entityId, TsKvEntry tsKvEntry, long ttl) { + TsKvEntity entity = new TsKvEntity(); + entity.setEntityType(entityId.getEntityType().name()); + entity.setEntityId(entityId.getId()); + entity.setTs(tsKvEntry.getTs()); + entity.setKey(tsKvEntry.getKey()); + entity.setStrValue(tsKvEntry.getStrValue().orElse(null)); + entity.setDoubleValue(tsKvEntry.getDoubleValue().orElse(null)); + entity.setLongValue(tsKvEntry.getLongValue().orElse(null)); + entity.setBooleanValue(tsKvEntry.getBooleanValue().orElse(null)); + return service.submit(() -> { + tsKvRepository.save(entity); + return null; + }); } @Override - public ListenableFuture savePartition(EntityId entityId, long partition, String key, long ttl) { + public ListenableFuture savePartition(EntityId entityId, long tsKvEntryTs, String key, long ttl) { return null; } @Override public ListenableFuture saveLatest(EntityId entityId, TsKvEntry tsKvEntry) { - return null; + TsKvLatestEntity latestEntity = new TsKvLatestEntity(); + latestEntity.setEntityType(entityId.getEntityType().name()); + latestEntity.setEntityId(entityId.getId()); + latestEntity.setTs(tsKvEntry.getTs()); + latestEntity.setKey(tsKvEntry.getKey()); + latestEntity.setStrValue(tsKvEntry.getStrValue().orElse(null)); + latestEntity.setDoubleValue(tsKvEntry.getDoubleValue().orElse(null)); + latestEntity.setLongValue(tsKvEntry.getLongValue().orElse(null)); + latestEntity.setBooleanValue(tsKvEntry.getBooleanValue().orElse(null)); + return service.submit(() -> { + tsKvLatestRepository.save(latestEntity); + return null; + }); } } diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvLatestRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvLatestRepository.java new file mode 100644 index 0000000000..863064908f --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvLatestRepository.java @@ -0,0 +1,30 @@ +/** + * Copyright © 2016-2017 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.sql.timeseries; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.data.repository.CrudRepository; +import org.thingsboard.server.dao.model.sql.TsKvLatestCompositeKey; +import org.thingsboard.server.dao.model.sql.TsKvLatestEntity; + +import java.util.List; +import java.util.UUID; + +@ConditionalOnProperty(prefix = "sql", value = "enabled", havingValue = "true") +public interface TsKvLatestRepository extends CrudRepository { + + List findAllByEntityTypeAndEntityId(String entityType, UUID entityId); +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvRepository.java b/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvRepository.java new file mode 100644 index 0000000000..6b087847f9 --- /dev/null +++ b/dao/src/main/java/org/thingsboard/server/dao/sql/timeseries/TsKvRepository.java @@ -0,0 +1,25 @@ +/** + * Copyright © 2016-2017 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.sql.timeseries; + +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.data.repository.CrudRepository; +import org.thingsboard.server.dao.model.sql.TsKvCompositeKey; +import org.thingsboard.server.dao.model.sql.TsKvEntity; + +@ConditionalOnProperty(prefix = "sql", value = "enabled", havingValue = "true") +public interface TsKvRepository extends CrudRepository { +} diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java index c4a71b3352..49f8a832df 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java +++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/BaseTimeseriesService.java @@ -72,9 +72,8 @@ public class BaseTimeseriesService implements TimeseriesService { if (tsKvEntry == null) { throw new IncorrectParameterException("Key value entry can't be null"); } - long partitionTs = timeseriesDao.toPartitionTs(tsKvEntry.getTs()); List> futures = Lists.newArrayListWithExpectedSize(INSERTS_PER_ENTRY); - saveAndRegisterFutures(futures, entityId, tsKvEntry, partitionTs, 0L); + saveAndRegisterFutures(futures, entityId, tsKvEntry, 0L); return Futures.allAsList(futures); } @@ -85,16 +84,15 @@ public class BaseTimeseriesService implements TimeseriesService { if (tsKvEntry == null) { throw new IncorrectParameterException("Key value entry can't be null"); } - long partitionTs = timeseriesDao.toPartitionTs(tsKvEntry.getTs()); - saveAndRegisterFutures(futures, entityId, tsKvEntry, partitionTs, ttl); + saveAndRegisterFutures(futures, entityId, tsKvEntry, ttl); } return Futures.allAsList(futures); } - private void saveAndRegisterFutures(List> futures, EntityId entityId, TsKvEntry tsKvEntry, long partitionTs, long ttl) { - futures.add(timeseriesDao.savePartition(entityId, partitionTs, tsKvEntry.getKey(), ttl)); + private void saveAndRegisterFutures(List> futures, EntityId entityId, TsKvEntry tsKvEntry, long ttl) { + futures.add(timeseriesDao.savePartition(entityId, tsKvEntry.getTs(), tsKvEntry.getKey(), ttl)); futures.add(timeseriesDao.saveLatest(entityId, tsKvEntry)); - futures.add(timeseriesDao.save(entityId, partitionTs, tsKvEntry, ttl)); + futures.add(timeseriesDao.save(entityId, tsKvEntry, ttl)); } private static void validate(EntityId entityId) { diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java index 21e01e6ad1..bb9a1b571d 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/CassandraBaseTimeseriesDao.java @@ -89,12 +89,6 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem super.stopExecutor(); } - @Override - public long toPartitionTs(long ts) { - LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(ts), ZoneOffset.UTC); - return tsFormat.truncatedTo(time).toInstant(ZoneOffset.UTC).toEpochMilli(); - } - @Override public ListenableFuture> findAllAsync(EntityId entityId, List queries) { List>> futures = queries.stream().map(query -> findAllAsync(entityId, query)).collect(Collectors.toList()); @@ -163,6 +157,11 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem return resultFuture; } + private long toPartitionTs(long ts) { + LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(ts), ZoneOffset.UTC); + return tsFormat.truncatedTo(time).toInstant(ZoneOffset.UTC).toEpochMilli(); + } + private void findAllAsyncSequentiallyWithLimit(final TsKvQueryCursor cursor, final SimpleListenableFuture> resultFuture) { if (cursor.isFull() || !cursor.hasNextPartition()) { resultFuture.set(cursor.getData()); @@ -259,7 +258,8 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem } @Override - public ListenableFuture save(EntityId entityId, long partition, TsKvEntry tsKvEntry, long ttl) { + public ListenableFuture save(EntityId entityId, TsKvEntry tsKvEntry, long ttl) { + long partition = toPartitionTs(tsKvEntry.getTs()); DataType type = tsKvEntry.getDataType(); BoundStatement stmt = (ttl == 0 ? getSaveStmt(type) : getSaveTtlStmt(type)).bind(); stmt.setString(0, entityId.getEntityType().name()) @@ -275,7 +275,8 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem } @Override - public ListenableFuture savePartition(EntityId entityId, long partition, String key, long ttl) { + public ListenableFuture savePartition(EntityId entityId, long tsKvEntryTs, String key, long ttl) { + long partition = toPartitionTs(tsKvEntryTs); log.debug("Saving partition {} for the entity [{}-{}] and key {}", partition, entityId.getEntityType(), entityId.getId(), key); BoundStatement stmt = (ttl == 0 ? getPartitionInsertStmt() : getPartitionInsertTtlStmt()).bind(); stmt = stmt.setString(0, entityId.getEntityType().name()) diff --git a/dao/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesDao.java b/dao/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesDao.java index 8f1b0031de..5d593b7fa8 100644 --- a/dao/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesDao.java +++ b/dao/src/main/java/org/thingsboard/server/dao/timeseries/TimeseriesDao.java @@ -27,17 +27,15 @@ import java.util.List; */ public interface TimeseriesDao { - long toPartitionTs(long ts); - ListenableFuture> findAllAsync(EntityId entityId, List queries); ListenableFuture findLatest(EntityId entityId, String key); ListenableFuture> findAllLatest(EntityId entityId); - ListenableFuture save(EntityId entityId, long partition, TsKvEntry tsKvEntry, long ttl); + ListenableFuture save(EntityId entityId, TsKvEntry tsKvEntry, long ttl); - ListenableFuture savePartition(EntityId entityId, long partition, String key, long ttl); + ListenableFuture savePartition(EntityId entityId, long tsKvEntryTs, String key, long ttl); ListenableFuture saveLatest(EntityId entityId, TsKvEntry tsKvEntry); } diff --git a/dao/src/main/resources/postgres/schema.sql b/dao/src/main/resources/postgres/schema.sql index e78199fdf0..6791e0e7dd 100644 --- a/dao/src/main/resources/postgres/schema.sql +++ b/dao/src/main/resources/postgres/schema.sql @@ -51,18 +51,6 @@ CREATE TABLE IF NOT EXISTS alarm ( ); ALTER TABLE alarm OWNER TO postgres; -CREATE TABLE IF NOT EXISTS relation ( - from_id uuid, - from_type character varying(255), - to_id uuid, - to_type character varying(255), - relation_type_group character varying(255), - relation_type character varying(255), - additional_info jsonb, - CONSTRAINT relation_unq_key UNIQUE (from_id, from_type, relation_type_group, relation_type, to_id, to_type) -); -ALTER TABLE relation OWNER TO postgres; - CREATE TABLE IF NOT EXISTS asset ( id uuid NOT NULL CONSTRAINT asset_pkey PRIMARY KEY, additional_info jsonb, @@ -172,6 +160,18 @@ CREATE TABLE IF NOT EXISTS plugin ( ); ALTER TABLE plugin OWNER TO postgres; +CREATE TABLE IF NOT EXISTS relation ( + from_id uuid, + from_type character varying(255), + to_id uuid, + to_type character varying(255), + relation_type_group character varying(255), + relation_type character varying(255), + additional_info jsonb, + CONSTRAINT relation_unq_key UNIQUE (from_id, from_type, relation_type_group, relation_type, to_id, to_type) +); +ALTER TABLE relation OWNER TO postgres; + CREATE TABLE IF NOT EXISTS rule ( id uuid NOT NULL CONSTRAINT rule_pkey PRIMARY KEY, action jsonb, @@ -217,6 +217,32 @@ CREATE TABLE IF NOT EXISTS tenant ( ); ALTER TABLE tenant OWNER TO postgres; +CREATE TABLE IF NOT EXISTS ts_kv ( + entity_type character varying(255) NOT NULL, + entity_id uuid NOT NULL, + key character varying(255) NOT NULL, + ts bigint NOT NULL, + bool_v boolean, + str_v character varying(255), + long_v bigint, + dbl_v double precision, + CONSTRAINT ts_kv_unq_key UNIQUE (entity_type, entity_id, key, ts) +); +ALTER TABLE ts_kv OWNER TO postgres; + +CREATE TABLE IF NOT EXISTS ts_kv_latest ( + entity_type character varying(255) NOT NULL, + entity_id uuid NOT NULL, + key character varying(255) NOT NULL, + ts bigint NOT NULL, + bool_v boolean, + str_v character varying(255), + long_v bigint, + dbl_v double precision, + CONSTRAINT ts_kv_latest_unq_key UNIQUE (entity_type, entity_id, key) +); +ALTER TABLE ts_kv OWNER TO postgres; + CREATE TABLE IF NOT EXISTS user_credentials ( id uuid NOT NULL CONSTRAINT user_credentials_pkey PRIMARY KEY, activate_token character varying(255) UNIQUE,