Fix entity duplication on duplicated attributes

application/src/main/java/org/thingsboard/server/service/subscription/TbAbstractDataSubCtx.java

@@ -201,7 +201,7 @@ public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends
         return result;
     }
 
-    protected synchronized void update(){
+    protected synchronized void update() {
         long start = System.currentTimeMillis();
         PageData<EntityData> newData = findEntityData();
         long end = System.currentTimeMillis();
@@ -209,11 +209,11 @@ public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends
         stats.getRegularQueryTimeSpent().addAndGet(end - start);
         Map<EntityId, EntityData> oldDataMap;
         if (data != null && !data.getData().isEmpty()) {
-            oldDataMap = data.getData().stream().collect(Collectors.toMap(EntityData::getEntityId, Function.identity()));
+            oldDataMap = data.getData().stream().collect(Collectors.toMap(EntityData::getEntityId, Function.identity(), (a, b) -> a));
         } else {
             oldDataMap = Collections.emptyMap();
         }
-        Map<EntityId, EntityData> newDataMap = newData.getData().stream().collect(Collectors.toMap(EntityData::getEntityId, Function.identity()));
+        Map<EntityId, EntityData> newDataMap = newData.getData().stream().collect(Collectors.toMap(EntityData::getEntityId, Function.identity(), (a,b)-> a));
         if (oldDataMap.size() == newDataMap.size() && oldDataMap.keySet().equals(newDataMap.keySet())) {
             log.trace("[{}][{}] No updates to entity data found", sessionRef.getSessionId(), cmdId);
         } else {
@@ -337,7 +337,7 @@ public abstract class TbAbstractDataSubCtx<T extends AbstractDataQuery<? extends
         }
     }
 
-    public void clearDynamicValueSubscriptions(){
+    public void clearDynamicValueSubscriptions() {
         if (subToDynamicValueKeySet != null) {
             for (Integer subId : subToDynamicValueKeySet) {
                 localSubscriptionService.cancelSubscription(sessionRef.getSessionId(), subId);

dao/src/main/java/org/thingsboard/server/dao/sql/query/DefaultEntityQueryRepository.java

@@ -49,6 +49,7 @@ import org.thingsboard.server.common.data.relation.EntitySearchDirection;
 import org.thingsboard.server.common.data.relation.EntityTypeFilter;
 import org.thingsboard.server.dao.util.SqlDao;
 
+import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
@@ -230,6 +231,8 @@ public class DefaultEntityQueryRepository implements EntityQueryRepository {
         ctx.append(addEntityTableQuery(ctx, query.getEntityFilter(), entityType));
         ctx.append(" e where ");
         ctx.append(buildEntityWhere(ctx, tenantId, customerId, query.getEntityFilter(), Collections.emptyList(), entityType));
+//        log.error("QUERY: {}", ctx.getQuery());
+//        Arrays.asList(ctx.getParameterNames()).forEach(param -> log.error("QUERY PARAM: {}->{}", param, ctx.getValue(param)));
         return transactionTemplate.execute(status -> jdbcTemplate.queryForObject(ctx.getQuery(), ctx, Long.class));
     }
 
@@ -312,6 +315,8 @@ public class DefaultEntityQueryRepository implements EntityQueryRepository {
             if (pageLink.getPageSize() > 0) {
                 dataQuery = String.format("%s limit %s offset %s", dataQuery, pageLink.getPageSize(), startIndex);
             }
+//            log.error("QUERY: {}", dataQuery);
+//            Arrays.asList(ctx.getParameterNames()).forEach(param -> log.error("QUERY PARAM: {}->{}", param, ctx.getValue(param)));
             List<Map<String, Object>> rows = jdbcTemplate.queryForList(dataQuery, ctx);
             return EntityDataAdapter.createEntityData(pageLink, selectionMapping, rows, totalElements);
         });

dao/src/main/java/org/thingsboard/server/dao/sql/query/EntityKeyMapping.java

@@ -231,15 +231,17 @@ public class EntityKeyMapping {
         } else {
             entityTypeStr = "'" + entityType.name() + "'";
         }
-        String join = hasFilter() ? "left join" : "left outer join";
         ctx.addStringParameter(alias + "_key_id", entityKey.getKey());
         if (entityKey.getType().equals(EntityKeyType.TIME_SERIES)) {
+            String join = hasFilter() ? "left join" : "left outer join";
             return String.format("%s ts_kv_latest %s ON %s.entity_id=entities.id AND %s.key = (select key_id from ts_kv_dictionary where key = :%s_key_id)",
                     join, alias, alias, alias, alias);
         } else {
-            String query = String.format("%s attribute_kv %s ON %s.entity_id=entities.id AND %s.entity_type=%s AND %s.attribute_key=:%s_key_id",
-                    join, alias, alias, alias, entityTypeStr, alias, alias);
+            String query;
             if (!entityKey.getType().equals(EntityKeyType.ATTRIBUTE)) {
+                String join = hasFilter() ? "left join" : "left outer join";
+                query = String.format("%s attribute_kv %s ON %s.entity_id=entities.id AND %s.entity_type=%s AND %s.attribute_key=:%s_key_id",
+                        join, alias, alias, alias, entityTypeStr, alias, alias);
                 String scope;
                 if (entityKey.getType().equals(EntityKeyType.CLIENT_ATTRIBUTE)) {
                     scope = DataConstants.CLIENT_SCOPE;
@@ -249,6 +251,11 @@ public class EntityKeyMapping {
                     scope = DataConstants.SERVER_SCOPE;
                 }
                 query = String.format("%s AND %s.attribute_type='%s'", query, alias, scope);
+            } else {
+                String join = hasFilter() ? "join LATERAL" : "left join LATERAL";
+                query = String.format("%s (select * from attribute_kv %s WHERE %s.entity_id=entities.id AND %s.entity_type=%s AND %s.attribute_key=:%s_key_id " +
+                                "ORDER BY %s.last_update_ts DESC limit 1) as %s ON true",
+                        join, alias, alias, alias, entityTypeStr, alias, alias, alias, alias);
             }
             return query;
         }