Merge pull request #14631 from AndriiLandiak/fix-drop-partitions-event

Fixed partition cleanup for non-public PostgreSQL schemas

application/src/main/java/org/thingsboard/server/service/housekeeper/processor/EventsDeletionTaskProcessor.java

@@ -17,10 +17,13 @@ package org.thingsboard.server.service.housekeeper.processor;
 
 import lombok.RequiredArgsConstructor;
 import org.springframework.stereotype.Component;
+import org.thingsboard.server.common.data.event.EventType;
 import org.thingsboard.server.common.data.housekeeper.HousekeeperTask;
 import org.thingsboard.server.common.data.housekeeper.HousekeeperTaskType;
 import org.thingsboard.server.dao.event.EventService;
 
+import java.util.Arrays;
+
 @Component
 @RequiredArgsConstructor
 public class EventsDeletionTaskProcessor extends HousekeeperTaskProcessor<HousekeeperTask> {
@@ -29,7 +32,9 @@ public class EventsDeletionTaskProcessor extends HousekeeperTaskProcessor<Housek
 
     @Override
     public void process(HousekeeperTask task) throws Exception {
-        eventService.removeEvents(task.getTenantId(), task.getEntityId(), null, 0L, System.currentTimeMillis());
+        // Only delete non-debug events for deleted entities.
+        EventType[] nonDebugEventTypes = Arrays.stream(EventType.values()).filter(eventType -> !eventType.isDebug()).toArray(EventType[]::new);
+        eventService.removeEvents(task.getTenantId(), task.getEntityId(), 0L, System.currentTimeMillis(), nonDebugEventTypes);
     }
 
     @Override

application/src/main/java/org/thingsboard/server/service/ttl/EdgeEventsCleanUpService.java

@@ -20,7 +20,6 @@ import org.springframework.beans.factory.annotation.Value;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
-import org.thingsboard.server.dao.edge.EdgeEventDao;
 import org.thingsboard.server.dao.edge.EdgeEventService;
 import org.thingsboard.server.dao.sqlts.insert.sql.SqlPartitioningRepository;
 import org.thingsboard.server.queue.discovery.PartitionService;

common/dao-api/src/main/java/org/thingsboard/server/dao/event/EventService.java

@@ -41,6 +41,8 @@ public interface EventService {
 
     void removeEvents(TenantId tenantId, EntityId entityId);
 
+    void removeEvents(TenantId tenantId, EntityId entityId, Long startTime, Long endTime, EventType... types);
+
     void removeEvents(TenantId tenantId, EntityId entityId, EventFilter eventFilter, Long startTime, Long endTime);
 
     void cleanupEvents(long regularEventExpTs, long debugEventExpTs, boolean cleanupDb);

dao/src/main/java/org/thingsboard/server/dao/event/BaseEventService.java

@@ -47,11 +47,6 @@ import java.util.stream.Collectors;
 @Slf4j
 public class BaseEventService implements EventService {
 
-    @Value("${sql.ttl.events.events_ttl:0}")
-    private long ttlInSec;
-    @Value("${sql.ttl.events.debug_events_ttl:604800}")
-    private long debugTtlInSec;
-
     @Value("${event.debug.max-symbols:4096}")
     private int maxDebugEventSymbols;
 
@@ -125,7 +120,12 @@ public class BaseEventService implements EventService {
 
     @Override
     public void removeEvents(TenantId tenantId, EntityId entityId) {
-        removeEvents(tenantId, entityId, null, null, null);
+        removeEvents(tenantId, entityId, null, null);
+    }
+
+    @Override
+    public void removeEvents(TenantId tenantId, EntityId entityId, Long startTime, Long endTime, EventType... types) {
+        eventDao.removeEvents(tenantId.getId(), entityId.getId(), startTime, endTime, types);
     }
 
     @Override

dao/src/main/java/org/thingsboard/server/dao/event/EventDao.java

@@ -25,80 +25,24 @@ import org.thingsboard.server.common.data.page.TimePageLink;
 import java.util.List;
 import java.util.UUID;
 
-/**
- * The Interface EventDao.
- */
 public interface EventDao {
 
-    /**
-     * Save or update event object async
-     *
-     * @param event the event object
-     * @return saved event object future
-     */
     ListenableFuture<Void> saveAsync(Event event);
 
-    /**
-     * Find events by tenantId, entityId, eventType and pageLink.
-     *
-     * @param tenantId the tenantId
-     * @param entityId the entityId
-     * @param eventType the eventType
-     * @param pageLink the pageLink
-     * @return the event list
-     */
     PageData<? extends Event> findEvents(UUID tenantId, UUID entityId, EventType eventType, TimePageLink pageLink);
 
     PageData<? extends Event> findEventByFilter(UUID tenantId, UUID entityId, EventFilter eventFilter, TimePageLink pageLink);
 
-    /**
-     * Find latest events by tenantId, entityId and eventType.
-     *
-     * @param tenantId the tenantId
-     * @param entityId the entityId
-     * @param eventType the eventType
-     * @param limit the limit
-     * @return the event list
-     */
     List<? extends Event> findLatestEvents(UUID tenantId, UUID entityId, EventType eventType, int limit);
 
-    /**
-     * Find latest debug IN event by tenantId, entityId.
-     *
-     * @param tenantId the tenantId
-     * @param entityId the entityId
-     * @return the latest debug IN event
-     */
     Event findLatestDebugRuleNodeInEvent(UUID tenantId, UUID entityId);
 
-    /**
-     * Executes stored procedure to cleanup old events. Uses separate ttl for debug and other events.
-     * @param regularEventExpTs the expiration time of the regular events
-     * @param debugEventExpTs the expiration time of the debug events
-     * @param cleanupDb
-     */
     void cleanupEvents(long regularEventExpTs, long debugEventExpTs, boolean cleanupDb);
 
-    /**
-     * Removes all events for the specified entity and time interval
-     *
-     * @param tenantId
-     * @param entityId
-     * @param startTime
-     * @param endTime
-     */
     void removeEvents(UUID tenantId, UUID entityId, Long startTime, Long endTime);
 
-    /**
-     *
-     * Removes all events for the specified entity, event filter and time interval
-     *
-     * @param tenantId
-     * @param entityId
-     * @param eventFilter
-     * @param startTime
-     * @param endTime
-     */
+    void removeEvents(UUID tenantId, UUID entityId, Long startTime, Long endTime, EventType... types);
+
     void removeEvents(UUID tenantId, UUID entityId, EventFilter eventFilter, Long startTime, Long endTime);
 
 }

dao/src/main/java/org/thingsboard/server/dao/sql/asset/AssetRepository.java

@@ -30,9 +30,6 @@ import org.thingsboard.server.dao.model.sql.AssetInfoEntity;
 import java.util.List;
 import java.util.UUID;
 
-/**
- * Created by Valerii Sosliuk on 5/21/2017.
- */
 public interface AssetRepository extends JpaRepository<AssetEntity, UUID>, ExportableEntityRepository<AssetEntity> {
 
     @Query("SELECT new org.thingsboard.server.dao.model.sql.AssetInfoEntity(a, c.title, c.additionalInfo, p.name) " +

dao/src/main/java/org/thingsboard/server/dao/sql/edge/JpaBaseEdgeEventDao.java

@@ -86,9 +86,6 @@ public class JpaBaseEdgeEventDao extends JpaPartitionedAbstractDao<EdgeEventEnti
     @Value("${sql.edge_events.partition_size:168}")
     private int partitionSizeInHours;
 
-    @Value("${sql.ttl.edge_events.edge_events_ttl:2628000}")
-    private long edgeEventsTtl;
-
     private static final String TABLE_NAME = ModelConstants.EDGE_EVENT_TABLE_NAME;
 
     private TbSqlBlockingQueueWrapper<EdgeEventEntity, Void> queue;

dao/src/main/java/org/thingsboard/server/dao/sql/entityview/EntityViewRepository.java

@@ -29,9 +29,6 @@ import org.thingsboard.server.dao.model.sql.EntityViewInfoEntity;
 import java.util.List;
 import java.util.UUID;
 
-/**
- * Created by Victor Basanets on 8/31/2017.
- */
 public interface EntityViewRepository extends JpaRepository<EntityViewEntity, UUID>, ExportableEntityRepository<EntityViewEntity> {
 
     @Query("SELECT new org.thingsboard.server.dao.model.sql.EntityViewInfoEntity(e, c.title, c.additionalInfo) " +

dao/src/main/java/org/thingsboard/server/dao/sql/entityview/JpaEntityViewDao.java

@@ -44,9 +44,6 @@ import java.util.UUID;
 
 import static org.thingsboard.server.dao.DaoUtil.convertTenantEntityTypesToDto;
 
-/**
- * Created by Victor Basanets on 8/31/2017.
- */
 @Component
 @Slf4j
 @SqlDao

dao/src/main/java/org/thingsboard/server/dao/sql/event/JpaBaseEventDao.java

@@ -173,8 +173,14 @@ public class JpaBaseEventDao implements EventDao {
 
     @Override
     public void removeEvents(UUID tenantId, UUID entityId, Long startTime, Long endTime) {
+        removeEvents(tenantId, entityId, startTime, endTime, EventType.values());
+    }
+
+    @Override
+    public void removeEvents(UUID tenantId, UUID entityId, Long startTime, Long endTime, EventType... types) {
         log.debug("[{}][{}] Remove events [{}-{}] ", tenantId, entityId, startTime, endTime);
-        for (EventType eventType : EventType.values()) {
+        EventType[] eventTypes = (types == null || types.length == 0) ? EventType.values() : types;
+        for (EventType eventType : eventTypes) {
             getEventRepository(eventType).removeEvents(tenantId, entityId, startTime, endTime);
         }
     }

dao/src/main/java/org/thingsboard/server/dao/sql/relation/JpaRelationDao.java

@@ -49,9 +49,6 @@ import static org.thingsboard.server.dao.model.ModelConstants.RELATION_TYPE_GROU
 import static org.thingsboard.server.dao.model.ModelConstants.RELATION_TYPE_PROPERTY;
 import static org.thingsboard.server.dao.model.ModelConstants.VERSION_COLUMN;
 
-/**
- * Created by Valerii Sosliuk on 5/29/2017.
- */
 @Slf4j
 @Component
 @SqlDao
@@ -256,9 +253,7 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple
 
         if (!CollectionUtils.isEmpty(relationTypeGroups)) {
             sqlBuilder.append("AND relation_type_group IN (?");
-            for (int i = 1; i < relationTypeGroups.size(); i++) {
-                sqlBuilder.append(", ?");
-            }
+            sqlBuilder.append(", ?".repeat(Math.max(0, relationTypeGroups.size() - 1)));
             sqlBuilder.append(")");
             params.addAll(relationTypeGroups);
         }
@@ -285,12 +280,12 @@ public class JpaRelationDao extends JpaAbstractDaoListeningExecutorService imple
                     relation.setTypeGroup(RelationTypeGroup.valueOf((String) relationTypeGroup));
                     relation.setVersion((Long) version);
                     return relation;
-                })
-                .collect(Collectors.toList());
+                }).toList();
     }
 
     @Override
     public List<EntityRelation> findRuleNodeToRuleChainRelations(RuleChainType ruleChainType, int limit) {
         return DaoUtil.convertDataList(relationRepository.findRuleNodeToRuleChainRelations(ruleChainType, PageRequest.of(0, limit)));
     }
+
 }

dao/src/main/java/org/thingsboard/server/dao/sql/tenant/TenantRepository.java

@@ -28,9 +28,6 @@ import org.thingsboard.server.dao.model.sql.TenantInfoEntity;
 import java.util.List;
 import java.util.UUID;
 
-/**
- * Created by Valerii Sosliuk on 4/30/2017.
- */
 public interface TenantRepository extends JpaRepository<TenantEntity, UUID> {
 
     @Query("SELECT new org.thingsboard.server.dao.model.sql.TenantInfoEntity(t, p.name) " +

dao/src/main/java/org/thingsboard/server/dao/sql/usagerecord/ApiUsageStateRepository.java

@@ -29,9 +29,6 @@ import org.thingsboard.server.dao.model.sql.ApiUsageStateEntity;
 import java.util.List;
 import java.util.UUID;
 
-/**
- * @author Valerii Sosliuk
- */
 public interface ApiUsageStateRepository extends JpaRepository<ApiUsageStateEntity, UUID> {
 
     @Query("SELECT ur FROM ApiUsageStateEntity ur WHERE ur.tenantId = :tenantId " +

dao/src/main/java/org/thingsboard/server/dao/sql/usagerecord/JpaApiUsageStateDao.java

@@ -34,9 +34,6 @@ import org.thingsboard.server.dao.util.SqlDao;
 import java.util.List;
 import java.util.UUID;
 
-/**
- * @author Andrii Shvaika
- */
 @Component
 @SqlDao
 public class JpaApiUsageStateDao extends JpaAbstractDao<ApiUsageStateEntity, ApiUsageState> implements ApiUsageStateDao {

dao/src/main/java/org/thingsboard/server/dao/sql/widget/JpaWidgetTypeDao.java

@@ -53,9 +53,6 @@ import java.util.stream.Collectors;
 
 import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID;
 
-/**
- * Created by Valerii Sosliuk on 4/29/2017.
- */
 @Component
 @SqlDao
 public class JpaWidgetTypeDao extends JpaAbstractDao<WidgetTypeDetailsEntity, WidgetTypeDetails> implements WidgetTypeDao, TenantEntityDao<WidgetTypeDetails> {

dao/src/main/java/org/thingsboard/server/dao/sql/widget/JpaWidgetsBundleDao.java

@@ -42,9 +42,6 @@ import java.util.UUID;
 
 import static org.thingsboard.server.dao.model.ModelConstants.NULL_UUID;
 
-/**
- * Created by Valerii Sosliuk on 4/23/2017.
- */
 @Component
 @SqlDao
 public class JpaWidgetsBundleDao extends JpaAbstractDao<WidgetsBundleEntity, WidgetsBundle> implements WidgetsBundleDao, TenantEntityDao<WidgetsBundle> {

dao/src/main/java/org/thingsboard/server/dao/sql/widget/WidgetsBundleRepository.java

@@ -21,7 +21,6 @@ import org.springframework.data.domain.Pageable;
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.data.repository.query.Param;
-import org.thingsboard.server.common.data.edqs.fields.WidgetTypeFields;
 import org.thingsboard.server.common.data.edqs.fields.WidgetsBundleFields;
 import org.thingsboard.server.dao.ExportableEntityRepository;
 import org.thingsboard.server.dao.model.sql.WidgetsBundleEntity;
@@ -29,9 +28,6 @@ import org.thingsboard.server.dao.model.sql.WidgetsBundleEntity;
 import java.util.List;
 import java.util.UUID;
 
-/**
- * Created by Valerii Sosliuk on 4/23/2017.
- */
 public interface WidgetsBundleRepository extends JpaRepository<WidgetsBundleEntity, UUID>, ExportableEntityRepository<WidgetsBundleEntity> {
 
     WidgetsBundleEntity findWidgetsBundleByTenantIdAndAlias(UUID tenantId, String alias);

dao/src/main/java/org/thingsboard/server/dao/sqlts/insert/sql/SqlPartitioningRepository.java

@@ -41,7 +41,7 @@ public class SqlPartitioningRepository {
     @Autowired
     private JdbcTemplate jdbcTemplate;
 
-    private static final String SELECT_PARTITIONS_STMT = "SELECT tablename from pg_tables WHERE schemaname = 'public' and tablename like concat(?, '_%')";
+    private static final String SELECT_PARTITIONS_STMT = "SELECT tablename from pg_tables WHERE schemaname = current_schema() and tablename like concat(?, '_%')";
 
     private static final int PSQL_VERSION_14 = 140000;
     private volatile Integer currentServerVersion;

dao/src/main/resources/sql/schema-ts-psql.sql

@@ -72,7 +72,7 @@ BEGIN
                 IF partition_type = 'YEARS' THEN
                     FOR partition IN SELECT tablename
                                      FROM pg_tables
-                                     WHERE schemaname = 'public'
+                                     WHERE schemaname = current_schema()
                                        AND tablename like 'ts_kv_' || '%'
                                        AND tablename != 'ts_kv_latest'
                                        AND tablename != 'key_dictionary'
@@ -93,7 +93,7 @@ BEGIN
                         ELSE
                             FOR partition IN SELECT tablename
                                              FROM pg_tables
-                                             WHERE schemaname = 'public'
+                                             WHERE schemaname = current_schema()
                                                AND tablename like 'ts_kv_' || '%'
                                                AND tablename != 'ts_kv_latest'
                                                AND tablename != 'key_dictionary'
@@ -135,7 +135,7 @@ BEGIN
                                 ELSE
                                     FOR partition IN SELECT tablename
                                                      FROM pg_tables
-                                                     WHERE schemaname = 'public'
+                                                     WHERE schemaname = current_schema()
                                                        AND tablename like 'ts_kv_' || '%'
                                                        AND tablename != 'ts_kv_latest'
                                                        AND tablename != 'key_dictionary'
@@ -205,7 +205,7 @@ BEGIN
     IF partition IS NOT NULL THEN
         IF NOT EXISTS(SELECT
                       FROM pg_tables
-                      WHERE schemaname = 'public'
+                      WHERE schemaname = current_schema()
                         AND tablename = partition) THEN
             partition := NULL;
             RAISE NOTICE 'Failed to found partition by ttl';