Commit d6d2de6d9d4557ff29b18edb0d5f7b8fdb746ecc

Authored by Volodymyr Babak
1 parent 4486d07f

Added ts kv latest and ts kv jps dao support

... ... @@ -15,8 +15,6 @@
15 15 */
16 16 package org.thingsboard.server.common.data.kv;
17 17
18   -import java.util.Optional;
19   -
20 18 public interface TsKvQuery {
21 19
22 20 String getKey();
... ...
... ... @@ -16,6 +16,8 @@
16 16 package org.thingsboard.server.dao.model.sql;
17 17
18 18 import lombok.Data;
  19 +import lombok.EqualsAndHashCode;
  20 +import lombok.ToString;
19 21 import org.thingsboard.server.common.data.kv.*;
20 22 import org.thingsboard.server.dao.model.ToData;
21 23
... ... @@ -29,6 +31,8 @@ import static org.thingsboard.server.dao.model.ModelConstants.*;
29 31 @Entity
30 32 @Table(name = "attribute_kv")
31 33 @IdClass(AttributeKvCompositeKey.class)
  34 +@EqualsAndHashCode
  35 +@ToString
32 36 public class AttributeKvEntity implements ToData<AttributeKvEntry>, Serializable {
33 37
34 38 @Id
... ...
... ... @@ -17,15 +17,23 @@ package org.thingsboard.server.dao.model.sql;
17 17
18 18 import lombok.AllArgsConstructor;
19 19 import lombok.Data;
  20 +import lombok.EqualsAndHashCode;
  21 +import lombok.ToString;
20 22 import org.thingsboard.server.common.data.relation.EntityRelation;
21 23
  24 +import javax.persistence.Transient;
22 25 import java.io.Serializable;
23 26 import java.util.UUID;
24 27
25 28 @AllArgsConstructor
26 29 @Data
  30 +@EqualsAndHashCode
  31 +@ToString
27 32 public class RelationCompositeKey implements Serializable {
28 33
  34 + @Transient
  35 + private static final long serialVersionUID = -4089175869616037592L;
  36 +
29 37 private UUID fromId;
30 38 private String fromType;
31 39 private UUID toId;
... ...
... ... @@ -17,6 +17,8 @@ package org.thingsboard.server.dao.model.sql;
17 17
18 18 import com.fasterxml.jackson.databind.JsonNode;
19 19 import lombok.Data;
  20 +import lombok.EqualsAndHashCode;
  21 +import lombok.ToString;
20 22 import org.hibernate.annotations.Type;
21 23 import org.thingsboard.server.common.data.id.EntityIdFactory;
22 24 import org.thingsboard.server.common.data.relation.EntityRelation;
... ... @@ -32,11 +34,10 @@ import static org.thingsboard.server.dao.model.ModelConstants.*;
32 34 @Entity
33 35 @Table(name = RELATION_COLUMN_FAMILY_NAME)
34 36 @IdClass(RelationCompositeKey.class)
  37 +@EqualsAndHashCode
  38 +@ToString
35 39 public final class RelationEntity implements ToData<EntityRelation> {
36 40
37   - @Transient
38   - private static final long serialVersionUID = -4089175869616037592L;
39   -
40 41 @Id
41 42 @Column(name = RELATION_FROM_ID_PROPERTY)
42 43 private UUID fromId;
... ... @@ -84,88 +85,6 @@ public final class RelationEntity implements ToData<EntityRelation> {
84 85 }
85 86
86 87 @Override
87   - public int hashCode() {
88   - final int prime = 31;
89   - int result = 1;
90   - result = prime * result + ((additionalInfo == null) ? 0 : additionalInfo.hashCode());
91   - result = prime * result + ((toId == null) ? 0 : toId.hashCode());
92   - result = prime * result + ((toType == null) ? 0 : toType.hashCode());
93   - result = prime * result + ((fromId == null) ? 0 : fromId.hashCode());
94   - result = prime * result + ((fromType == null) ? 0 : fromType.hashCode());
95   - result = prime * result + ((relationType == null) ? 0 : relationType.hashCode());
96   - result = prime * result + ((relationTypeGroup == null) ? 0 : relationTypeGroup.hashCode());
97   - return result;
98   - }
99   -
100   - @Override
101   - public boolean equals(Object obj) {
102   - if (this == obj)
103   - return true;
104   - if (obj == null)
105   - return false;
106   - if (getClass() != obj.getClass())
107   - return false;
108   - RelationEntity other = (RelationEntity) obj;
109   - if (additionalInfo == null) {
110   - if (other.additionalInfo != null)
111   - return false;
112   - } else if (!additionalInfo.equals(other.additionalInfo))
113   - return false;
114   - if (toId == null) {
115   - if (other.toId != null)
116   - return false;
117   - } else if (!toId.equals(other.toId))
118   - return false;
119   - if (fromId == null) {
120   - if (other.fromId != null)
121   - return false;
122   - } else if (!fromId.equals(other.fromId))
123   - return false;
124   - if (toType == null) {
125   - if (other.toType != null)
126   - return false;
127   - } else if (!toType.equals(other.toType))
128   - return false;
129   - if (fromType == null) {
130   - if (other.fromType != null)
131   - return false;
132   - } else if (!fromType.equals(other.fromType))
133   - return false;
134   - if (relationType == null) {
135   - if (other.relationType != null)
136   - return false;
137   - } else if (!relationType.equals(other.relationType))
138   - return false;
139   - if (relationTypeGroup == null) {
140   - if (other.relationTypeGroup != null)
141   - return false;
142   - } else if (!relationTypeGroup.equals(other.relationTypeGroup))
143   - return false;
144   - return true;
145   - }
146   -
147   - @Override
148   - public String toString() {
149   - StringBuilder builder = new StringBuilder();
150   - builder.append("AssetEntity [toId=");
151   - builder.append(toId);
152   - builder.append(", toType=");
153   - builder.append(toType);
154   - builder.append(", fromId=");
155   - builder.append(fromId);
156   - builder.append(", fromType=");
157   - builder.append(fromType);
158   - builder.append(", relationType=");
159   - builder.append(relationType);
160   - builder.append(", relationTypeGroup=");
161   - builder.append(relationTypeGroup);
162   - builder.append(", additionalInfo=");
163   - builder.append(additionalInfo);
164   - builder.append("]");
165   - return builder.toString();
166   - }
167   -
168   - @Override
169 88 public EntityRelation toData() {
170 89 EntityRelation relation = new EntityRelation();
171 90 if (toId != null && toType != null) {
... ...
  1 +/**
  2 + * Copyright © 2016-2017 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.dao.model.sql;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import lombok.Data;
  20 +import lombok.EqualsAndHashCode;
  21 +import lombok.ToString;
  22 +
  23 +import javax.persistence.Transient;
  24 +import java.io.Serializable;
  25 +import java.util.UUID;
  26 +
  27 +@Data
  28 +@AllArgsConstructor
  29 +@EqualsAndHashCode
  30 +@ToString
  31 +public class TsKvCompositeKey implements Serializable{
  32 +
  33 + @Transient
  34 + private static final long serialVersionUID = -4089175869616037523L;
  35 +
  36 + private String entityType;
  37 + private UUID entityId;
  38 + private String key;
  39 + private long ts;
  40 +}
\ No newline at end of file
... ...
  1 +/**
  2 + * Copyright © 2016-2017 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.dao.model.sql;
  17 +
  18 +import lombok.Data;
  19 +import lombok.EqualsAndHashCode;
  20 +import lombok.ToString;
  21 +import org.thingsboard.server.common.data.kv.TsKvEntry;
  22 +import org.thingsboard.server.dao.model.ToData;
  23 +
  24 +import javax.persistence.*;
  25 +import java.util.UUID;
  26 +
  27 +import static org.thingsboard.server.dao.model.ModelConstants.*;
  28 +
  29 +@Data
  30 +@Entity
  31 +@Table(name = "ts_kv")
  32 +@IdClass(TsKvCompositeKey.class)
  33 +@EqualsAndHashCode
  34 +@ToString
  35 +public final class TsKvEntity implements ToData<TsKvEntry> {
  36 +
  37 + @Id
  38 + @Column(name = ENTITY_TYPE_COLUMN)
  39 + private String entityType;
  40 +
  41 + @Id
  42 + @Column(name = ENTITY_ID_COLUMN)
  43 + private UUID entityId;
  44 +
  45 + @Id
  46 + @Column(name = KEY_COLUMN)
  47 + private String key;
  48 +
  49 + @Id
  50 + @Column(name = TS_COLUMN)
  51 + private long ts;
  52 +
  53 + @Column(name = BOOLEAN_VALUE_COLUMN)
  54 + private Boolean booleanValue;
  55 +
  56 + @Column(name = STRING_VALUE_COLUMN)
  57 + private String strValue;
  58 +
  59 + @Column(name = LONG_VALUE_COLUMN)
  60 + private Long longValue;
  61 +
  62 + @Column(name = DOUBLE_VALUE_COLUMN)
  63 + private Double doubleValue;
  64 +
  65 + @Override
  66 + public TsKvEntry toData() {
  67 + return null;
  68 + }
  69 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2017 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.dao.model.sql;
  17 +
  18 +import lombok.AllArgsConstructor;
  19 +import lombok.Data;
  20 +import lombok.EqualsAndHashCode;
  21 +import lombok.ToString;
  22 +
  23 +import javax.persistence.Transient;
  24 +import java.io.Serializable;
  25 +import java.util.UUID;
  26 +
  27 +@Data
  28 +@AllArgsConstructor
  29 +@EqualsAndHashCode
  30 +@ToString
  31 +public class TsKvLatestCompositeKey implements Serializable{
  32 +
  33 + @Transient
  34 + private static final long serialVersionUID = -4089175869616037523L;
  35 +
  36 + private String entityType;
  37 + private UUID entityId;
  38 + private String key;
  39 +}
\ No newline at end of file
... ...
  1 +/**
  2 + * Copyright © 2016-2017 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.dao.model.sql;
  17 +
  18 +import lombok.Data;
  19 +import lombok.EqualsAndHashCode;
  20 +import lombok.ToString;
  21 +import org.thingsboard.server.common.data.kv.TsKvEntry;
  22 +import org.thingsboard.server.dao.model.ToData;
  23 +
  24 +import javax.persistence.*;
  25 +import java.util.UUID;
  26 +
  27 +import static org.thingsboard.server.dao.model.ModelConstants.*;
  28 +
  29 +@Data
  30 +@Entity
  31 +@Table(name = "ts_kv_latest")
  32 +@IdClass(TsKvLatestCompositeKey.class)
  33 +@EqualsAndHashCode
  34 +@ToString
  35 +public final class TsKvLatestEntity implements ToData<TsKvEntry> {
  36 +
  37 + @Id
  38 + @Column(name = ENTITY_TYPE_COLUMN)
  39 + private String entityType;
  40 +
  41 + @Id
  42 + @Column(name = ENTITY_ID_COLUMN)
  43 + private UUID entityId;
  44 +
  45 + @Id
  46 + @Column(name = KEY_COLUMN)
  47 + private String key;
  48 +
  49 + @Column(name = TS_COLUMN)
  50 + private long ts;
  51 +
  52 + @Column(name = BOOLEAN_VALUE_COLUMN)
  53 + private Boolean booleanValue;
  54 +
  55 + @Column(name = STRING_VALUE_COLUMN)
  56 + private String strValue;
  57 +
  58 + @Column(name = LONG_VALUE_COLUMN)
  59 + private Long longValue;
  60 +
  61 + @Column(name = DOUBLE_VALUE_COLUMN)
  62 + private Double doubleValue;
  63 +
  64 + @Override
  65 + public TsKvEntry toData() {
  66 + return null;
  67 + }
  68 +}
... ...
... ... @@ -15,13 +15,20 @@
15 15 */
16 16 package org.thingsboard.server.dao.sql.timeseries;
17 17
  18 +import com.google.common.collect.Lists;
18 19 import com.google.common.util.concurrent.ListenableFuture;
19 20 import lombok.extern.slf4j.Slf4j;
  21 +import org.springframework.beans.factory.annotation.Autowired;
20 22 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
21 23 import org.springframework.stereotype.Component;
22 24 import org.thingsboard.server.common.data.id.EntityId;
23 25 import org.thingsboard.server.common.data.kv.TsKvEntry;
24 26 import org.thingsboard.server.common.data.kv.TsKvQuery;
  27 +import org.thingsboard.server.dao.DaoUtil;
  28 +import org.thingsboard.server.dao.model.sql.TsKvEntity;
  29 +import org.thingsboard.server.dao.model.sql.TsKvLatestCompositeKey;
  30 +import org.thingsboard.server.dao.model.sql.TsKvLatestEntity;
  31 +import org.thingsboard.server.dao.sql.JpaAbstractDaoListeningExecutorService;
25 32 import org.thingsboard.server.dao.timeseries.TimeseriesDao;
26 33
27 34 import java.util.List;
... ... @@ -29,41 +36,80 @@ import java.util.List;
29 36 @Component
30 37 @Slf4j
31 38 @ConditionalOnProperty(prefix = "sql", value = "enabled", havingValue = "true")
32   -public class JpaTimeseriesDao implements TimeseriesDao {
  39 +public class JpaTimeseriesDao extends JpaAbstractDaoListeningExecutorService implements TimeseriesDao {
33 40
34   - @Override
35   - public long toPartitionTs(long ts) {
36   - return 0;
37   - }
  41 + @Autowired
  42 + private TsKvRepository tsKvRepository;
  43 +
  44 + @Autowired
  45 + private TsKvLatestRepository tsKvLatestRepository;
38 46
39 47 @Override
40 48 public ListenableFuture<List<TsKvEntry>> findAllAsync(EntityId entityId, List<TsKvQuery> queries) {
  49 + return null;
  50 + }
  51 +
  52 + private ListenableFuture<List<TsKvEntry>> findAllAsync(EntityId entityId, TsKvQuery query) {
41 53 return null;
42 54 }
43 55
44 56 @Override
45 57 public ListenableFuture<TsKvEntry> findLatest(EntityId entityId, String key) {
46   - return null;
  58 + TsKvLatestCompositeKey compositeKey =
  59 + new TsKvLatestCompositeKey(
  60 + entityId.getEntityType().name(),
  61 + entityId.getId(),
  62 + key);
  63 + return service.submit(() ->
  64 + DaoUtil.getData(tsKvLatestRepository.findOne(compositeKey)));
47 65 }
48 66
49 67 @Override
50 68 public ListenableFuture<List<TsKvEntry>> findAllLatest(EntityId entityId) {
51   - return null;
  69 + return service.submit(() ->
  70 + DaoUtil.convertDataList(Lists.newArrayList(
  71 + tsKvLatestRepository.findAllByEntityTypeAndEntityId(
  72 + entityId.getEntityType().name(),
  73 + entityId.getId()))));
52 74 }
53 75
54 76 @Override
55   - public ListenableFuture<Void> save(EntityId entityId, long partition, TsKvEntry tsKvEntry, long ttl) {
56   - return null;
  77 + public ListenableFuture<Void> save(EntityId entityId, TsKvEntry tsKvEntry, long ttl) {
  78 + TsKvEntity entity = new TsKvEntity();
  79 + entity.setEntityType(entityId.getEntityType().name());
  80 + entity.setEntityId(entityId.getId());
  81 + entity.setTs(tsKvEntry.getTs());
  82 + entity.setKey(tsKvEntry.getKey());
  83 + entity.setStrValue(tsKvEntry.getStrValue().orElse(null));
  84 + entity.setDoubleValue(tsKvEntry.getDoubleValue().orElse(null));
  85 + entity.setLongValue(tsKvEntry.getLongValue().orElse(null));
  86 + entity.setBooleanValue(tsKvEntry.getBooleanValue().orElse(null));
  87 + return service.submit(() -> {
  88 + tsKvRepository.save(entity);
  89 + return null;
  90 + });
57 91 }
58 92
59 93 @Override
60   - public ListenableFuture<Void> savePartition(EntityId entityId, long partition, String key, long ttl) {
  94 + public ListenableFuture<Void> savePartition(EntityId entityId, long tsKvEntryTs, String key, long ttl) {
61 95 return null;
62 96 }
63 97
64 98 @Override
65 99 public ListenableFuture<Void> saveLatest(EntityId entityId, TsKvEntry tsKvEntry) {
66   - return null;
  100 + TsKvLatestEntity latestEntity = new TsKvLatestEntity();
  101 + latestEntity.setEntityType(entityId.getEntityType().name());
  102 + latestEntity.setEntityId(entityId.getId());
  103 + latestEntity.setTs(tsKvEntry.getTs());
  104 + latestEntity.setKey(tsKvEntry.getKey());
  105 + latestEntity.setStrValue(tsKvEntry.getStrValue().orElse(null));
  106 + latestEntity.setDoubleValue(tsKvEntry.getDoubleValue().orElse(null));
  107 + latestEntity.setLongValue(tsKvEntry.getLongValue().orElse(null));
  108 + latestEntity.setBooleanValue(tsKvEntry.getBooleanValue().orElse(null));
  109 + return service.submit(() -> {
  110 + tsKvLatestRepository.save(latestEntity);
  111 + return null;
  112 + });
67 113 }
68 114
69 115 }
... ...
  1 +/**
  2 + * Copyright © 2016-2017 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.dao.sql.timeseries;
  17 +
  18 +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
  19 +import org.springframework.data.repository.CrudRepository;
  20 +import org.thingsboard.server.dao.model.sql.TsKvLatestCompositeKey;
  21 +import org.thingsboard.server.dao.model.sql.TsKvLatestEntity;
  22 +
  23 +import java.util.List;
  24 +import java.util.UUID;
  25 +
  26 +@ConditionalOnProperty(prefix = "sql", value = "enabled", havingValue = "true")
  27 +public interface TsKvLatestRepository extends CrudRepository<TsKvLatestEntity, TsKvLatestCompositeKey> {
  28 +
  29 + List<TsKvLatestEntity> findAllByEntityTypeAndEntityId(String entityType, UUID entityId);
  30 +}
... ...
  1 +/**
  2 + * Copyright © 2016-2017 The Thingsboard Authors
  3 + *
  4 + * Licensed under the Apache License, Version 2.0 (the "License");
  5 + * you may not use this file except in compliance with the License.
  6 + * You may obtain a copy of the License at
  7 + *
  8 + * http://www.apache.org/licenses/LICENSE-2.0
  9 + *
  10 + * Unless required by applicable law or agreed to in writing, software
  11 + * distributed under the License is distributed on an "AS IS" BASIS,
  12 + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 + * See the License for the specific language governing permissions and
  14 + * limitations under the License.
  15 + */
  16 +package org.thingsboard.server.dao.sql.timeseries;
  17 +
  18 +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
  19 +import org.springframework.data.repository.CrudRepository;
  20 +import org.thingsboard.server.dao.model.sql.TsKvCompositeKey;
  21 +import org.thingsboard.server.dao.model.sql.TsKvEntity;
  22 +
  23 +@ConditionalOnProperty(prefix = "sql", value = "enabled", havingValue = "true")
  24 +public interface TsKvRepository extends CrudRepository<TsKvEntity, TsKvCompositeKey> {
  25 +}
... ...
... ... @@ -72,9 +72,8 @@ public class BaseTimeseriesService implements TimeseriesService {
72 72 if (tsKvEntry == null) {
73 73 throw new IncorrectParameterException("Key value entry can't be null");
74 74 }
75   - long partitionTs = timeseriesDao.toPartitionTs(tsKvEntry.getTs());
76 75 List<ListenableFuture<Void>> futures = Lists.newArrayListWithExpectedSize(INSERTS_PER_ENTRY);
77   - saveAndRegisterFutures(futures, entityId, tsKvEntry, partitionTs, 0L);
  76 + saveAndRegisterFutures(futures, entityId, tsKvEntry, 0L);
78 77 return Futures.allAsList(futures);
79 78 }
80 79
... ... @@ -85,16 +84,15 @@ public class BaseTimeseriesService implements TimeseriesService {
85 84 if (tsKvEntry == null) {
86 85 throw new IncorrectParameterException("Key value entry can't be null");
87 86 }
88   - long partitionTs = timeseriesDao.toPartitionTs(tsKvEntry.getTs());
89   - saveAndRegisterFutures(futures, entityId, tsKvEntry, partitionTs, ttl);
  87 + saveAndRegisterFutures(futures, entityId, tsKvEntry, ttl);
90 88 }
91 89 return Futures.allAsList(futures);
92 90 }
93 91
94   - private void saveAndRegisterFutures(List<ListenableFuture<Void>> futures, EntityId entityId, TsKvEntry tsKvEntry, long partitionTs, long ttl) {
95   - futures.add(timeseriesDao.savePartition(entityId, partitionTs, tsKvEntry.getKey(), ttl));
  92 + private void saveAndRegisterFutures(List<ListenableFuture<Void>> futures, EntityId entityId, TsKvEntry tsKvEntry, long ttl) {
  93 + futures.add(timeseriesDao.savePartition(entityId, tsKvEntry.getTs(), tsKvEntry.getKey(), ttl));
96 94 futures.add(timeseriesDao.saveLatest(entityId, tsKvEntry));
97   - futures.add(timeseriesDao.save(entityId, partitionTs, tsKvEntry, ttl));
  95 + futures.add(timeseriesDao.save(entityId, tsKvEntry, ttl));
98 96 }
99 97
100 98 private static void validate(EntityId entityId) {
... ...
... ... @@ -90,12 +90,6 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
90 90 }
91 91
92 92 @Override
93   - public long toPartitionTs(long ts) {
94   - LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(ts), ZoneOffset.UTC);
95   - return tsFormat.truncatedTo(time).toInstant(ZoneOffset.UTC).toEpochMilli();
96   - }
97   -
98   - @Override
99 93 public ListenableFuture<List<TsKvEntry>> findAllAsync(EntityId entityId, List<TsKvQuery> queries) {
100 94 List<ListenableFuture<List<TsKvEntry>>> futures = queries.stream().map(query -> findAllAsync(entityId, query)).collect(Collectors.toList());
101 95 return Futures.transform(Futures.allAsList(futures), new Function<List<List<TsKvEntry>>, List<TsKvEntry>>() {
... ... @@ -163,6 +157,11 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
163 157 return resultFuture;
164 158 }
165 159
  160 + private long toPartitionTs(long ts) {
  161 + LocalDateTime time = LocalDateTime.ofInstant(Instant.ofEpochMilli(ts), ZoneOffset.UTC);
  162 + return tsFormat.truncatedTo(time).toInstant(ZoneOffset.UTC).toEpochMilli();
  163 + }
  164 +
166 165 private void findAllAsyncSequentiallyWithLimit(final TsKvQueryCursor cursor, final SimpleListenableFuture<List<TsKvEntry>> resultFuture) {
167 166 if (cursor.isFull() || !cursor.hasNextPartition()) {
168 167 resultFuture.set(cursor.getData());
... ... @@ -259,7 +258,8 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
259 258 }
260 259
261 260 @Override
262   - public ListenableFuture<Void> save(EntityId entityId, long partition, TsKvEntry tsKvEntry, long ttl) {
  261 + public ListenableFuture<Void> save(EntityId entityId, TsKvEntry tsKvEntry, long ttl) {
  262 + long partition = toPartitionTs(tsKvEntry.getTs());
263 263 DataType type = tsKvEntry.getDataType();
264 264 BoundStatement stmt = (ttl == 0 ? getSaveStmt(type) : getSaveTtlStmt(type)).bind();
265 265 stmt.setString(0, entityId.getEntityType().name())
... ... @@ -275,7 +275,8 @@ public class CassandraBaseTimeseriesDao extends CassandraAbstractAsyncDao implem
275 275 }
276 276
277 277 @Override
278   - public ListenableFuture<Void> savePartition(EntityId entityId, long partition, String key, long ttl) {
  278 + public ListenableFuture<Void> savePartition(EntityId entityId, long tsKvEntryTs, String key, long ttl) {
  279 + long partition = toPartitionTs(tsKvEntryTs);
279 280 log.debug("Saving partition {} for the entity [{}-{}] and key {}", partition, entityId.getEntityType(), entityId.getId(), key);
280 281 BoundStatement stmt = (ttl == 0 ? getPartitionInsertStmt() : getPartitionInsertTtlStmt()).bind();
281 282 stmt = stmt.setString(0, entityId.getEntityType().name())
... ...
... ... @@ -27,17 +27,15 @@ import java.util.List;
27 27 */
28 28 public interface TimeseriesDao {
29 29
30   - long toPartitionTs(long ts);
31   -
32 30 ListenableFuture<List<TsKvEntry>> findAllAsync(EntityId entityId, List<TsKvQuery> queries);
33 31
34 32 ListenableFuture<TsKvEntry> findLatest(EntityId entityId, String key);
35 33
36 34 ListenableFuture<List<TsKvEntry>> findAllLatest(EntityId entityId);
37 35
38   - ListenableFuture<Void> save(EntityId entityId, long partition, TsKvEntry tsKvEntry, long ttl);
  36 + ListenableFuture<Void> save(EntityId entityId, TsKvEntry tsKvEntry, long ttl);
39 37
40   - ListenableFuture<Void> savePartition(EntityId entityId, long partition, String key, long ttl);
  38 + ListenableFuture<Void> savePartition(EntityId entityId, long tsKvEntryTs, String key, long ttl);
41 39
42 40 ListenableFuture<Void> saveLatest(EntityId entityId, TsKvEntry tsKvEntry);
43 41 }
... ...
... ... @@ -51,18 +51,6 @@ CREATE TABLE IF NOT EXISTS alarm (
51 51 );
52 52 ALTER TABLE alarm OWNER TO postgres;
53 53
54   -CREATE TABLE IF NOT EXISTS relation (
55   - from_id uuid,
56   - from_type character varying(255),
57   - to_id uuid,
58   - to_type character varying(255),
59   - relation_type_group character varying(255),
60   - relation_type character varying(255),
61   - additional_info jsonb,
62   - CONSTRAINT relation_unq_key UNIQUE (from_id, from_type, relation_type_group, relation_type, to_id, to_type)
63   -);
64   -ALTER TABLE relation OWNER TO postgres;
65   -
66 54 CREATE TABLE IF NOT EXISTS asset (
67 55 id uuid NOT NULL CONSTRAINT asset_pkey PRIMARY KEY,
68 56 additional_info jsonb,
... ... @@ -172,6 +160,18 @@ CREATE TABLE IF NOT EXISTS plugin (
172 160 );
173 161 ALTER TABLE plugin OWNER TO postgres;
174 162
  163 +CREATE TABLE IF NOT EXISTS relation (
  164 + from_id uuid,
  165 + from_type character varying(255),
  166 + to_id uuid,
  167 + to_type character varying(255),
  168 + relation_type_group character varying(255),
  169 + relation_type character varying(255),
  170 + additional_info jsonb,
  171 + CONSTRAINT relation_unq_key UNIQUE (from_id, from_type, relation_type_group, relation_type, to_id, to_type)
  172 +);
  173 +ALTER TABLE relation OWNER TO postgres;
  174 +
175 175 CREATE TABLE IF NOT EXISTS rule (
176 176 id uuid NOT NULL CONSTRAINT rule_pkey PRIMARY KEY,
177 177 action jsonb,
... ... @@ -217,6 +217,32 @@ CREATE TABLE IF NOT EXISTS tenant (
217 217 );
218 218 ALTER TABLE tenant OWNER TO postgres;
219 219
  220 +CREATE TABLE IF NOT EXISTS ts_kv (
  221 + entity_type character varying(255) NOT NULL,
  222 + entity_id uuid NOT NULL,
  223 + key character varying(255) NOT NULL,
  224 + ts bigint NOT NULL,
  225 + bool_v boolean,
  226 + str_v character varying(255),
  227 + long_v bigint,
  228 + dbl_v double precision,
  229 + CONSTRAINT ts_kv_unq_key UNIQUE (entity_type, entity_id, key, ts)
  230 +);
  231 +ALTER TABLE ts_kv OWNER TO postgres;
  232 +
  233 +CREATE TABLE IF NOT EXISTS ts_kv_latest (
  234 + entity_type character varying(255) NOT NULL,
  235 + entity_id uuid NOT NULL,
  236 + key character varying(255) NOT NULL,
  237 + ts bigint NOT NULL,
  238 + bool_v boolean,
  239 + str_v character varying(255),
  240 + long_v bigint,
  241 + dbl_v double precision,
  242 + CONSTRAINT ts_kv_latest_unq_key UNIQUE (entity_type, entity_id, key)
  243 +);
  244 +ALTER TABLE ts_kv OWNER TO postgres;
  245 +
220 246 CREATE TABLE IF NOT EXISTS user_credentials (
221 247 id uuid NOT NULL CONSTRAINT user_credentials_pkey PRIMARY KEY,
222 248 activate_token character varying(255) UNIQUE,
... ...