Merge pull request #4275 from ViacheslavKlimov/fix/timescale-ttl-procedure

Fix Timescale procedure

+CREATE OR REPLACE PROCEDURE cleanup_timeseries_by_ttl(IN null_uuid uuid,
+                                                      IN system_ttl bigint, INOUT deleted bigint)
+    LANGUAGE plpgsql AS
+$$
+DECLARE
+tenant_cursor CURSOR FOR select tenant.id as tenant_id
+                         from tenant;
+    tenant_id_record     uuid;
+    customer_id_record   uuid;
+    tenant_ttl           bigint;
+    customer_ttl         bigint;
+    deleted_for_entities bigint;
+    tenant_ttl_ts        bigint;
+    customer_ttl_ts      bigint;
+BEGIN
+OPEN tenant_cursor;
+FETCH tenant_cursor INTO tenant_id_record;
+WHILE FOUND
+        LOOP
+            EXECUTE format(
+                    'select attribute_kv.long_v from attribute_kv where attribute_kv.entity_id = %L and attribute_kv.attribute_key = %L',
+                    tenant_id_record, 'TTL') INTO tenant_ttl;
+            if tenant_ttl IS NULL THEN
+                tenant_ttl := system_ttl;
+END IF;
+            IF tenant_ttl > 0 THEN
+                tenant_ttl_ts := (EXTRACT(EPOCH FROM current_timestamp) * 1000 - tenant_ttl::bigint * 1000)::bigint;
+                deleted_for_entities := delete_device_records_from_ts_kv(tenant_id_record, null_uuid, tenant_ttl_ts);
+                deleted := deleted + deleted_for_entities;
+                RAISE NOTICE '% telemetry removed for devices where tenant_id = %', deleted_for_entities, tenant_id_record;
+                deleted_for_entities := delete_asset_records_from_ts_kv(tenant_id_record, null_uuid, tenant_ttl_ts);
+                deleted := deleted + deleted_for_entities;
+                RAISE NOTICE '% telemetry removed for assets where tenant_id = %', deleted_for_entities, tenant_id_record;
+END IF;
+FOR customer_id_record IN
+SELECT customer.id AS customer_id FROM customer WHERE customer.tenant_id = tenant_id_record
+    LOOP
+                    EXECUTE format(
+                            'select attribute_kv.long_v from attribute_kv where attribute_kv.entity_id = %L and attribute_kv.attribute_key = %L',
+                            customer_id_record, 'TTL') INTO customer_ttl;
+IF customer_ttl IS NULL THEN
+                        customer_ttl_ts := tenant_ttl_ts;
+ELSE
+                        IF customer_ttl > 0 THEN
+                            customer_ttl_ts :=
+                                    (EXTRACT(EPOCH FROM current_timestamp) * 1000 -
+                                     customer_ttl::bigint * 1000)::bigint;
+END IF;
+END IF;
+                    IF customer_ttl_ts IS NOT NULL AND customer_ttl_ts > 0 THEN
+                        deleted_for_entities :=
+                                delete_customer_records_from_ts_kv(tenant_id_record, customer_id_record,
+                                                                   customer_ttl_ts);
+                        deleted := deleted + deleted_for_entities;
+                        RAISE NOTICE '% telemetry removed for customer with id = % where tenant_id = %', deleted_for_entities, customer_id_record, tenant_id_record;
+                        deleted_for_entities :=
+                                delete_device_records_from_ts_kv(tenant_id_record, customer_id_record,
+                                                                 customer_ttl_ts);
+                        deleted := deleted + deleted_for_entities;
+                        RAISE NOTICE '% telemetry removed for devices where tenant_id = % and customer_id = %', deleted_for_entities, tenant_id_record, customer_id_record;
+                        deleted_for_entities := delete_asset_records_from_ts_kv(tenant_id_record,
+                                                                                customer_id_record,
+                                                                                customer_ttl_ts);
+                        deleted := deleted + deleted_for_entities;
+                        RAISE NOTICE '% telemetry removed for assets where tenant_id = % and customer_id = %', deleted_for_entities, tenant_id_record, customer_id_record;
+END IF;
+END LOOP;
+FETCH tenant_cursor INTO tenant_id_record;
+END LOOP;
+END
+$$;

@@ -178,7 +178,11 @@ public class TimescaleTsDatabaseUpgradeService extends AbstractSqlTsDatabaseUpgr
                 }
                 break;
             case "3.1.1":
+                break;
             case "3.2.1":
+                try (Connection conn = DriverManager.getConnection(dbUrl, dbUserName, dbPassword)) {
+                    loadSql(conn, LOAD_TTL_FUNCTIONS_SQL);
+                }
                 break;
             default:
                 throw new RuntimeException("Unable to upgrade SQL database, unsupported fromVersion: " + fromVersion);

@@ -91,8 +91,8 @@ $$
 DECLARE
     tenant_cursor CURSOR FOR select tenant.id as tenant_id
                              from tenant;
-    tenant_id_record     varchar;
-    customer_id_record   varchar;
+    tenant_id_record     uuid;
+    customer_id_record   uuid;
     tenant_ttl           bigint;
     customer_ttl         bigint;
     deleted_for_entities bigint;