Руководство администратора

create or alter procedure dbo.GetBrowserMemoryStats

(

   @Days int = 3,

   @TopUsersCount int = 10,

   @DebugF bit = 0,

   @JsonResult nvarchar(max) = null output

)

as

begin

   set nocount on;

   set transaction isolation level read uncommitted;

   -- Находим минимальный ID для диапазона дат

   declare @MinId int;

   select

           @MinId = min(ID)

   from

           dbo.MobileClientStats

   where

           EventDate >= cast(dateadd(day, -@Days, getutcdate()) as date);

   -- Собираем данные с парсингом JSON один раз

   declare @MemData table

   (

       Version varchar(50),

       EventDay date,

       UserId int,

       HeapUsedMb decimal(10, 2)

   );

   insert into @MemData

   (

       Version,

       EventDay,

       UserId,

       HeapUsedMb

   )

   select

           Version,

           cast(EventDate as date),

           UserId,

           cast(json_value(Action, '$.mem.heapUsedMb') as decimal(10, 2))

   from

           dbo.MobileClientStats

   where

           ID >= @MinId and

           Type = 'browser_telemetry' and

           json_value(Action, '$.mem.heapUsedMb') is not null;

   if @DebugF = cast(1 as bit)

       select '@MemData rows' as Debug, count(*) as [Count] from @MemData;

   -- Статистика по версиям

   declare @VersionStats nvarchar(max);

   with percentiles as

   (

       select distinct

               Version,

               percentile_cont(0.5) within group (order by HeapUsedMb) over (partition by Version) as P50Mb,

               percentile_cont(0.95) within group (order by HeapUsedMb) over (partition by Version) as P95Mb,

               percentile_cont(0.99) within group (order by HeapUsedMb) over (partition by Version) as P99Mb

       from

               @MemData

   )

   select

           @VersionStats =

           (

               select

                       a.Version as [version],

                       a.EventCount as eventCount,

                       a.MinMb as minMb,

                       a.AvgMb as avgMb,

                       p.P50Mb as p50Mb,

                       p.P95Mb as p95Mb,

                       p.P99Mb as p99Mb,

                       a.MaxMb as maxMb

               from

                       percentiles p

                       inner join

                       (

                           select

                                   Version,

                                   count(*) as EventCount,

                                   min(HeapUsedMb) as MinMb,

                                   cast(avg(HeapUsedMb) as decimal(10, 2)) as AvgMb,

                                   max(HeapUsedMb) as MaxMb

                           from

                                   @MemData

                           group by

                                   Version

                       ) a

                           on a.Version = p.Version

               order by

                       a.EventCount desc

               for json path

           );

   -- Статистика по дням

   declare @DailyStats nvarchar(max);

   with percentiles as

   (

       select distinct

               EventDay,

               percentile_cont(0.5) within group (order by HeapUsedMb) over (partition by EventDay) as P50Mb,

               percentile_cont(0.95) within group (order by HeapUsedMb) over (partition by EventDay) as P95Mb,

               percentile_cont(0.99) within group (order by HeapUsedMb) over (partition by EventDay) as P99Mb

       from

               @MemData

   )

   select

           @DailyStats =

           (

               select

                       a.EventDay as eventDay,

                       a.EventCount as eventCount,

                       a.UniqueUsers as uniqueUsers,

                       a.MinMb as minMb,

                       a.AvgMb as avgMb,

                       p.P50Mb as p50Mb,

                       p.P95Mb as p95Mb,

                       p.P99Mb as p99Mb,

                       a.MaxMb as maxMb

               from

                       percentiles p

                       inner join

                       (

                           select

                                   EventDay,

                                   count(*) as EventCount,

                                   count(distinct UserId) as UniqueUsers,

                                   min(HeapUsedMb) as MinMb,

                                   cast(avg(HeapUsedMb) as decimal(10, 2)) as AvgMb,

                                   max(HeapUsedMb) as MaxMb

                           from

                                   @MemData

                           group by

                                   EventDay

                       ) a

                           on a.EventDay = p.EventDay

               order by

                       a.EventDay

               for json path

           );

   -- Топ проблемных пользователей

   declare @TopUsers nvarchar(max);

   with percentiles as

   (

       select distinct

               UserId,

               percentile_cont(0.5) within group (order by HeapUsedMb) over (partition by UserId) as P50Mb,

               percentile_cont(0.95) within group (order by HeapUsedMb) over (partition by UserId) as P95Mb,

               percentile_cont(0.99) within group (order by HeapUsedMb) over (partition by UserId) as P99Mb

       from

               @MemData

   )

   select

           @TopUsers =

           (

               select top (@TopUsersCount)

                       a.UserId as userId,

                       u.DisplayName as displayName,

                       a.EventCount as eventCount,

                       a.MinMb as minMb,

                       a.AvgMb as avgMb,

                       p.P50Mb as p50Mb,

                       p.P95Mb as p95Mb,

                       p.P99Mb as p99Mb,

                       a.MaxMb as maxMb

               from

                       percentiles p

                       inner join

                       (

                           select

                                   UserId,

                                   count(*) as EventCount,

                                   min(HeapUsedMb) as MinMb,

                                   cast(avg(HeapUsedMb) as decimal(10, 2)) as AvgMb,

                                   max(HeapUsedMb) as MaxMb

                           from

                                   @MemData

                           group by

                                   UserId

                       ) a

                           on a.UserId = p.UserId

                       left join

                       dbo.Users u

                           on u.UserID = a.UserId

               order by

                       p.P95Mb desc

               for json path

           );

   -- Собираем итоговый JSON

   set @JsonResult = '{

       "generatedAt": "' + convert(varchar(30), getutcdate(), 127) + 'Z",

       "periodDays": ' + cast(@Days as varchar(10)) + ',

       "minId": ' + cast(@MinId as varchar(20)) + ',

       "byVersion": ' + isnull(@VersionStats, '[]') + ',

       "byDay": ' + isnull(@DailyStats, '[]') + ',

       "topProblematicUsers": ' + isnull(@TopUsers, '[]') + '

}';

   -- Вывод как XML чтобы SSMS не обрезал

   select

           cast('<json><![CDATA[' + @JsonResult + ']]></json>' as xml) as JsonResult;

   /*

end;

GO

Примеры использования:

-- Базовый вызов (3 дня, топ 10 юзеров)

   exec dbo.GetBrowserMemoryStats;

-- За неделю с топ 20 юзеров

   exec dbo.GetBrowserMemoryStats @Days = 7, @TopUsersCount = 20;

-- С отладкой

   exec dbo.GetBrowserMemoryStats @DebugF = 1;

-- Получить JSON в переменную

   declare @Result nvarchar(max);

   exec dbo.GetBrowserMemoryStats @JsonResult = @Result output;

   select @Result;

   */

call dbo.droproutineifexists('dbo', 'get_browser_memory_stats');

create or replace function dbo.get_browser_memory_stats

(

   p_days int = 3,

   p_top_users_count int = 10,

   p_debug_f bool = false

)

returns table

(

   json_result jsonb

)

as

$$

declare

   v_min_id int;

   v_mem_data jsonb;

   v_version_stats jsonb;

   v_daily_stats jsonb;

   v_top_users jsonb;

begin

   -- Находим минимальный ID для диапазона дат

   select

           min(id)

   into

           v_min_id

   from

           dbo.mobileclientstats

   where

           eventdate >= (current_date - p_days);

   -- Собираем данные с парсингом JSON один раз

   with mem_data as

   (

       select

               version,

               eventdate::date as event_day,

               userid,

               ((action::jsonb) -> 'mem' ->> 'heapUsedMb')::decimal(10, 2) as heap_used_mb

       from

               dbo.mobileclientstats

       where

               id >= v_min_id and

               type = 'browser_telemetry' and

               (action::jsonb) -> 'mem' ->> 'heapUsedMb' is not null

   )

   select

           jsonb_agg(row_to_json(t)::jsonb)

   into

           v_mem_data

   from

           mem_data t;

   if p_debug_f then

       raise notice '@mem_data rows: %', jsonb_array_length(v_mem_data);

   end if;

   -- Статистика по версиям

   with mem_data as

   (

       select

               (item ->> 'version')::varchar as version,

               (item ->> 'heap_used_mb')::decimal(10, 2) as heap_used_mb

       from

               jsonb_array_elements(v_mem_data) as item

   )

   select

           jsonb_agg(row_to_json(t)::jsonb order by t.event_count desc)

   into

           v_version_stats

   from

           (

               select

                       version,

                       count(*) as event_count,

                       min(heap_used_mb) as min_mb,

                       round(avg(heap_used_mb), 2) as avg_mb,

                       round(percentile_cont(0.5) within group (order by heap_used_mb)::numeric, 2) as p50_mb,

                       round(percentile_cont(0.95) within group (order by heap_used_mb)::numeric, 2) as p95_mb,

                       round(percentile_cont(0.99) within group (order by heap_used_mb)::numeric, 2) as p99_mb,

                       max(heap_used_mb) as max_mb

               from

                       mem_data

               group by

                       version

           ) t;

   -- Статистика по дням

   with mem_data as

   (

       select

               (item ->> 'event_day')::date as event_day,

               (item ->> 'userid')::int as userid,

               (item ->> 'heap_used_mb')::decimal(10, 2) as heap_used_mb

       from

               jsonb_array_elements(v_mem_data) as item

   )

   select

           jsonb_agg(row_to_json(t)::jsonb order by t.event_day)

   into

           v_daily_stats

   from

           (

               select

                       event_day,

                       count(*) as event_count,

                       count(distinct userid) as unique_users,

                       min(heap_used_mb) as min_mb,

                       round(avg(heap_used_mb), 2) as avg_mb,

                       round(percentile_cont(0.5) within group (order by heap_used_mb)::numeric, 2) as p50_mb,

                       round(percentile_cont(0.95) within group (order by heap_used_mb)::numeric, 2) as p95_mb,

                       round(percentile_cont(0.99) within group (order by heap_used_mb)::numeric, 2) as p99_mb,

                       max(heap_used_mb) as max_mb

               from

                       mem_data

               group by

                       event_day

           ) t;

   -- Топ проблемных пользователей

   with mem_data as

   (

       select

               (item ->> 'userid')::int as userid,

               (item ->> 'heap_used_mb')::decimal(10, 2) as heap_used_mb

       from

               jsonb_array_elements(v_mem_data) as item

   ),

   user_stats as

   (

       select

               m.userid,

               u.displayname,

               count(*) as event_count,

               min(heap_used_mb) as min_mb,

               round(avg(heap_used_mb), 2) as avg_mb,

               round(percentile_cont(0.5) within group (order by heap_used_mb)::numeric, 2) as p50_mb,

               round(percentile_cont(0.95) within group (order by heap_used_mb)::numeric, 2) as p95_mb,

               round(percentile_cont(0.99) within group (order by heap_used_mb)::numeric, 2) as p99_mb,

               max(heap_used_mb) as max_mb

       from

               mem_data m

               left join

               dbo.users u

                   on u.userid = m.userid

       group by

               m.userid,

               u.displayname

       order by

               p95_mb desc

       limit

               p_top_users_count

   )

   select

           jsonb_agg(row_to_json(t)::jsonb)

   into

           v_top_users

   from

           user_stats t;

   -- Собираем итоговый JSON

   return query

   select

           jsonb_build_object

           (

               'generatedAt', to_char(now() at time zone 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'),

               'periodDays', p_days,

               'minId', v_min_id,

               'byVersion', coalesce(v_version_stats, '[]'::jsonb),

               'byDay', coalesce(v_daily_stats, '[]'::jsonb),

               'topProblematicUsers', coalesce(v_top_users, '[]'::jsonb)

           );

   /*

end;

$$

language plpgsql;

Примеры использования:

   -- Базовый вызов (3 дня, топ 10 юзеров)

   select * from dbo.get_browser_memory_stats();

   -- За неделю с топ 20 юзеров

   select * from dbo.get_browser_memory_stats(p_days => 7, p_top_users_count => 20);

   -- С отладкой

   select * from dbo.get_browser_memory_stats(p_debug_f => true);

   -- Красивый вывод

   select jsonb_pretty(json_result) from dbo.get_browser_memory_stats();

   */

create or alter procedure dbo.tc_NightlyMaintance

(

   -- параметры для передачу в dbo.IndexRebuild

    @index_rebuild_object_name     nvarchar(max) = null, -- имя таблиц/индексированных представлений (можно несколько через запятую), для которого запускается переиндексация (или null для всех)

    @index_rebuild_index_name      sysname = null, -- имя индекса, для которого запускается переиндексация (или null для всех)

    @index_rebuild_frglvl_low      float = 10, -- минимальный процент фрагментации для реорганизации индекса

    @index_rebuild_frglvl_hi       float = 25, -- минимальный процент фрагментации для перестройки индекса

    @index_rebuild_max_duration    int = 120, -- количество минут, которое будет выполняться процедура (плюс время для завершения перестройки очередного индекса)

    @index_rebuild_min_page_count  int = 8, -- минимальное к-во страниц затрагиваемого индекса

    @index_rebuild_max_log_size    decimal(18,2) = NULL, -- максимальный размер лога в ГБ

    @index_rebuild_maxdop          int = null, -- опция maxdop() в alter index reorganize; null для исключения опции

    @index_rebuild_rebuild_online  bit = 1, -- по возможности перестраивать online

    @index_rebuild_qrtz_job_log_f  bit = 0, -- записывать в лог кварцевых джобов

    @index_rebuild_simple_mode_f   bit = 0, -- переводить ли на время выполнения в simple mode

    --

    @upd_stat_minute_duration      int = 120, -- количество минут, которое будет выполняться SP

    @upd_stat_log_size             int = 15, -- размер лога в ГБ

    @upd_stat_count_days           int = 7, -- количестов дней, за которые не обновляем статистику

    @upd_stat_main_tables_update   bit = 1, -- предварительное безусловное обновление главных таблиц

    @error_message                 nvarchar(max) = null output,

    @debug_f                       bit = 0 -- только вывод без реальных команд

)

as

begin

   /*

    *  0) удаление неиспользуемых индексов на денормализациях

  *        1) dbo.IndexRebuild

  *        2) dbo.tc_UpdateStatistics

  *        3) alter fulltext catalog TasksDescriptionFTCatalog rebuild - отменено

  *        4) dbcc freeproccache with no_infomsgs - отменено

    *  5) dbo.ShowTasksFeedExecMainCacheMaintance  

  *        6) dbcc checkdb with no_infomsgs

  */

   declare @ret int;

   set @error_message = null;

   -- 0) удаление неиспользуемых индексов на денормализациях

   exec dba_util.usp_get_unused_indexes_drop_script @DoActualDropF = 1;

   -- 1) dbo.IndexRebuild

   exec @ret = dbo.IndexRebuild

       @object_name    = @index_rebuild_object_name,

       @index_name     = @index_rebuild_index_name,

       @frglvl_low     = @index_rebuild_frglvl_low,

       @frglvl_hi      = @index_rebuild_frglvl_hi,

       @max_duration   = @index_rebuild_max_duration,

       @min_page_count = @index_rebuild_min_page_count,

       @max_log_size   = @index_rebuild_max_log_size,

       @maxdop         = @index_rebuild_maxdop,

       @rebuild_online = @index_rebuild_rebuild_online,

       @qrtz_job_log_f = @index_rebuild_qrtz_job_log_f,

       @debug_f        = @debug_f,

       @simple_mode_f  = @index_rebuild_simple_mode_f,

       @error_message  = @error_message output

   ;

   if @ret <> 0

       return @ret;

   -- 2) обновление статистик

   set @error_message = NULL;

   exec dbo.tc_UpdateStatistics

       @MinutDuration = @upd_stat_minute_duration,

       @LogSize = @upd_stat_log_size,

       @CountDays = @upd_stat_count_days,

       @MainTablesUpdate = @upd_stat_main_tables_update

   ;

   --alter fulltext catalog TasksDescriptionFTCatalog rebuild;

   -- очистка кешей

   exec dbo.ShowTasksFeedExecMainCacheMaintance;

   -- общая проверка

   dbcc checkdb with no_infomsgs;

end;

GO

create or alter procedure dba_util.usp_get_unused_indexes_drop_script

(

   @ExcludeIndexNameLike   nvarchar(128) = null,  -- доп. исключение по имени индекса (like)

   @MinIndexSizeKb         bigint        = 0,     -- минимальный размер индекса (кб)

   @ExcludeCanonicalF      bit           = 1,     -- исключать "канонические" индексы

   @DoActualDropF          bit           = 0,     -- не выводить резалтсет, а реально дропать

   @DebugF                 bit           = 0      -- выводить промежуточную диагностику

)

as

begin

   set nocount on;

   set transaction isolation level read uncommitted;

   declare @TableNameLike nvarchar(128) = N'TasksInSubcat%';

   -- целевые объекты

   declare @TargetObjects table

   (

         [object_id]   int             not null primary key clustered,

         [schema_name] sysname         not null,

         table_name    sysname         not null

   );

   insert into @TargetObjects ([object_id], [schema_name], table_name)

   select

           obj.[object_id],

           sch.name as [schema_name],

           obj.name as table_name

   from

           sys.objects as obj

           inner join

           sys.schemas as sch

               on sch.[schema_id] = obj.[schema_id]

   where

           obj.[type] = 'U' and

           sch.name = N'dbo' and

           obj.name like @TableNameLike;

   if @DebugF = cast(1 as bit)

   begin

       select '@TargetObjects' as debug_step, count(*) as row_count from @TargetObjects;

   end;

   -- индексы-кандидаты с базовой фильтрацией

   declare @CandidateIndexes table

   (

         [object_id]   int             not null,

         index_id      int             not null,

         index_name    sysname         not null,

         primary key clustered ([object_id], index_id)

   );

   insert into @CandidateIndexes ([object_id], index_id, index_name)

   select

           idx.[object_id],

           idx.index_id,

           idx.name

   from

           sys.indexes as idx

           inner join

           @TargetObjects as tgt

               on tgt.[object_id] = idx.[object_id]

   where

           idx.index_id > 0 and

           idx.name is not null and

           idx.is_disabled = cast(0 as bit) and

           idx.is_hypothetical = cast(0 as bit) and

           idx.is_primary_key = cast(0 as bit) and

           idx.is_unique_constraint = cast(0 as bit) and

           idx.is_unique = cast(0 as bit) and

           idx.has_filter = cast(0 as bit) and

           (@ExcludeIndexNameLike is null or idx.name not like @ExcludeIndexNameLike);

   if @DebugF = cast(1 as bit)

   begin

       select '@CandidateIndexes' as debug_step, count(*) as row_count from @CandidateIndexes;

   end;

   -- размеры индексов

   declare @IndexSizes table

   (

         [object_id]       int     not null,

         index_id          int     not null,

         index_size_kb     bigint  not null,

         primary key clustered ([object_id], index_id)

   );

   insert into @IndexSizes ([object_id], index_id, index_size_kb)

   select

           cand.[object_id],

           cand.index_id,

           isnull(sum(dps.used_page_count) * 8, 0)

   from

           @CandidateIndexes as cand

           inner join

           sys.dm_db_partition_stats as dps

               on dps.[object_id] = cand.[object_id] and

                  dps.index_id = cand.index_id

   group by

           cand.[object_id],

           cand.index_id;

   -- статистика использования

   declare @IndexUsage table

   (

         [object_id]       int     not null,

         index_id          int     not null,

         user_seeks        bigint  not null,

         user_scans        bigint  not null,

         user_lookups      bigint  not null,

         user_updates      bigint  not null,

         primary key clustered ([object_id], index_id)

   );

   insert into @IndexUsage ([object_id], index_id, user_seeks, user_scans, user_lookups, user_updates)

   select

           cand.[object_id],

           cand.index_id,

           isnull(ius.user_seeks, 0),

           isnull(ius.user_scans, 0),

           isnull(ius.user_lookups, 0),

           isnull(ius.user_updates, 0)

   from

           @CandidateIndexes as cand

           left join

           sys.dm_db_index_usage_stats as ius

               on ius.database_id = db_id() and

                  ius.[object_id] = cand.[object_id] and

                  ius.index_id = cand.index_id;

   -- канонические паттерны

   -- [_] экранирует подчёркивание, % в конце ловит trailing spaces

   declare @CanonicalPatterns table

   (

         pattern_id        int identity(1, 1) primary key,

         index_name_like   nvarchar(128) not null

   );

   if @ExcludeCanonicalF = cast(1 as bit)

   begin

       insert into @CanonicalPatterns (index_name_like)

       values

             (N'ix[_][_]dbo[_][_]%[_][_]TaskText%'),

             (N'ix[_][_]dbo[_][_]%[_][_]StateID%');

   end;

   -- результат

   select

           row_number()

           over

           (

               order by

               usg.user_updates desc,

               sz.index_size_kb desc,

               tgt.[schema_name],

               tgt.table_name,

               cand.index_name

           ) as rn,

           tgt.[schema_name],

           tgt.table_name,

           cand.index_name,

           usg.user_seeks,

           usg.user_scans,

           usg.user_lookups,

           usg.user_updates,

           sz.index_size_kb,

           N'drop index ' + quotename(cand.index_name) + N' on '

               + quotename(tgt.[schema_name]) + N'.' + quotename(tgt.table_name) + N';' as drop_script

   into

           #res

   from

           @CandidateIndexes as cand

           inner join

           @TargetObjects as tgt

               on tgt.[object_id] = cand.[object_id]

           inner join

           @IndexSizes as sz

               on sz.[object_id] = cand.[object_id] and

                  sz.index_id = cand.index_id

           inner join

           @IndexUsage as usg

               on usg.[object_id] = cand.[object_id] and

                  usg.index_id = cand.index_id

   where

           usg.user_seeks = 0 and

           usg.user_scans = 0 and

           usg.user_lookups = 0 and

           sz.index_size_kb >= @MinIndexSizeKb and

           not exists

               (

                   select 1

                   from @CanonicalPatterns as cp

                   where cand.index_name like cp.index_name_like

               )

   order by

           rn

   ;

   if @DoActualDropF = 0

       select  res.*

       from    #res res

       order by res.rn;

   else

   begin

       declare @stt nvarchar(max);

       select  @stt = string_agg(cast(res.drop_script as nvarchar(max)), nchar(13) + nchar(10)) within group(order by res.rn)

       from    #res res

       ;

       raiserror ('%s', 0, 0, @stt) with nowait;

       exec (@stt);

   end;

   /*

       примеры использования:

       exec dba_util.usp_get_unused_indexes_drop_script;

       exec dba_util.usp_get_unused_indexes_drop_script

             @MinIndexSizeKb = 10240,

             @DebugF = 1;

   */

end;

GO

Процедура обнаружения закольцовок в self-referencing таблицах

Требования: SQL Server 2019 (15.x) или выше

Особенности:

 - Рекурсивный CTE с дедупликацией результатов

 - Второй resultset: статистика по FK (самоссылки / закольцовки)

 - Лимит вывода: > 1000 записей не выводятся детально

-- Создаём схему если не существует

if schema_id('dba_util') is null

   exec('create schema dba_util');

go

-- Основная процедура

create or alter procedure dba_util.sp_DetectTableCycles

(

   @TableName      nvarchar(128),

   @SchemaName     nvarchar(128) = 'dbo',

   @HasCyclesF     bit = null output,

   @ShowDetailsF   bit = 0,

   @DebugF         bit = 0

)

as

begin

   set nocount on;

   set transaction isolation level read uncommitted;

   declare @SQL nvarchar(max);

   declare @FullTableName nvarchar(256) = concat(quotename(@SchemaName), '.', quotename(@TableName));

   declare @CycleCount int = 0;

   declare @ErrorMessage nvarchar(500);

   declare @StartTime datetime2 = sysdatetime();

   declare @Msg nvarchar(2048);

   set @HasCyclesF = 0;

   -- Проверка существования таблицы

   if object_id(@FullTableName) is null

   begin

       set @ErrorMessage = concat('Таблица ', @FullTableName, ' не существует.');

       throw 50001, @ErrorMessage, 1;

   end;

   if @DebugF = 1

   begin

       set @Msg = concat('Начало проверки: ', @FullTableName, ' в ', convert(nvarchar, @StartTime, 121));

       raiserror(@Msg, 0, 0) with nowait;

   end;

   -- Табличная переменная для self-referencing связей

   declare @SelfReferences table

   (

       PKColumn    nvarchar(128),

       FKColumn    nvarchar(128),

       FKName      nvarchar(128),

       index IX_Columns nonclustered (PKColumn, FKColumn)

   );

   -- Табличная переменная для результатов

   declare @CycleResults table

   (

       RecordId    int,

       ParentId    int,

       CycleDepth  int,

       FKName      nvarchar(128),

       index IX_RecordId nonclustered (RecordId)

   );

   -- Временная таблица для сырых результатов из динамического SQL

   declare @RawResults table

   (

       RecordId    int,

       ParentId    int,

       CycleDepth  int,

       FKName      nvarchar(128)

   );

   -- Находим все self-referencing foreign keys

   insert  into @SelfReferences (PKColumn, FKColumn, FKName)

   select  c_pk.name,

           c_fk.name,

           fk.name

   from    sys.foreign_keys fk

           join

           sys.foreign_key_columns fkc

               on fk.object_id = fkc.constraint_object_id

           join

           sys.columns c_fk

               on fkc.parent_object_id = c_fk.object_id and

                  fkc.parent_column_id = c_fk.column_id

           join

           sys.columns c_pk

               on fkc.referenced_object_id = c_pk.object_id and

                  fkc.referenced_column_id = c_pk.column_id

   where   fk.parent_object_id = fk.referenced_object_id and

           fk.parent_object_id = object_id(@FullTableName)

   option (recompile);

   if not exists (select * from @SelfReferences)

   begin

       set @Msg = concat('Таблица ', @FullTableName, ' не имеет self-referencing связей.');

       raiserror(@Msg, 0, 0) with nowait;

       return;

   end;

   -- Показываем найденные связи

   if @DebugF = 1 or @ShowDetailsF = 1

   begin

       set @Msg = concat('Найдены self-referencing связи в таблице ', @FullTableName, ':');

       raiserror(@Msg, 0, 0) with nowait;

       select  PKColumn as [PK колонка],

               FKColumn as [FK колонка],

               FKName as [Имя FK]

       from    @SelfReferences;

   end;

   -- Проверяем на циклы для КАЖДОЙ связи

   declare @CurrentPK nvarchar(128);

   declare @CurrentFK nvarchar(128);

   declare @CurrentFKName nvarchar(128);

   declare @RefCount int = 0;

   declare ref_cursor cursor local fast_forward for

       select  PKColumn, FKColumn, FKName

       from    @SelfReferences;

   open ref_cursor;

   fetch next from ref_cursor into @CurrentPK, @CurrentFK, @CurrentFKName;

   while @@fetch_status = 0

   begin

       set @RefCount = @RefCount + 1;

       if @DebugF = 1

       begin

           set @Msg = concat(char(13), char(10), 'Проверка связи #', @RefCount, ': ', @CurrentPK, ' <- ', @CurrentFK, ' (', @CurrentFKName, ')');

           raiserror(@Msg, 0, 0) with nowait;

       end;

      -- Динамический SQL для поиска циклов (рекурсивный CTE)

       -- Оптимизации: varchar вместо nvarchar, глубина 100

       set @SQL = concat(N'

       ;with RecursiveHierarchy as

       (

           select  ', quotename(@CurrentPK), N' as RecordId,

                   ', quotename(@CurrentFK), N' as ParentId,

                   cast(concat('','', cast(', quotename(@CurrentPK), N' as varchar(12)), '','') as varchar(max)) as VisitedIds,

                   0 as Level

           from    ', @FullTableName, N'

           where   ', quotename(@CurrentFK), N' is not null

           union all

           select  rh.RecordId,

                   t.', quotename(@CurrentFK), N',

                   cast(concat(rh.VisitedIds, cast(t.', quotename(@CurrentPK), N' as varchar(12)), '','') as varchar(max)),

                   rh.Level + 1

           from    ', @FullTableName, N' t

                   join

                   RecursiveHierarchy rh

                       on t.', quotename(@CurrentPK), N' = rh.ParentId

           where   t.', quotename(@CurrentFK), N' is not null and

                   rh.Level < 100 and

                   charindex(concat('','', cast(t.', quotename(@CurrentPK), N' as varchar(12)), '',''), rh.VisitedIds) = 0

       )

       select  RecordId,

               ParentId,

               Level as CycleDepth,

               N''', @CurrentFKName, N''' as FKName

       from    RecursiveHierarchy

       where   charindex(concat('','', cast(ParentId as varchar(12)), '',''), VisitedIds) > 0 or

               RecordId = ParentId

       option (maxrecursion 100);');

       if @DebugF = 1

       begin

           raiserror('--- Dynamic SQL ---', 0, 0) with nowait;

           print @SQL;

           raiserror('--- End SQL ---', 0, 0) with nowait;

       end;

       begin try

           insert  into @RawResults (RecordId, ParentId, CycleDepth, FKName)

           exec sp_executesql @SQL;

           -- Дедупликация: берём минимальную глубину для каждой записи

           insert  into @CycleResults (RecordId, ParentId, CycleDepth, FKName)

           select  RecordId, min(ParentId), min(CycleDepth), FKName

           from    @RawResults

           group by RecordId, FKName;

           delete from @RawResults;

       end try

       begin catch

           set @ErrorMessage = concat('Ошибка при проверке связи ', @CurrentFKName, ': ', error_message());

           if @DebugF = 1

               raiserror(@ErrorMessage, 0, 0) with nowait;

       end catch;

       fetch next from ref_cursor into @CurrentPK, @CurrentFK, @CurrentFKName;

   end;

   close ref_cursor;

   deallocate ref_cursor;

   -- Подсчёт результатов

   select @CycleCount = count(*) from @CycleResults;

   set @HasCyclesF = case when @CycleCount > 0 then 1 else 0 end;

   -- Вывод результатов

   if @CycleCount > 0

   begin

       set @Msg = concat(char(13), char(10), N'⚠ ОБНАРУЖЕНЫ ПРОБЛЕМЫ: ', @CycleCount);

       raiserror(@Msg, 0, 0) with nowait;

       if @ShowDetailsF = 1

       begin

           if @CycleCount <= 1000

           begin

               select  RecordId as [ID записи],

                       ParentId as [ID родителя],

                       CycleDepth as [Глубина цикла],

                       FKName as [FK связь]

               from    @CycleResults

               order by FKName, CycleDepth, RecordId;

           end

           else

           begin

               set @Msg = '  (детальный список не выводится - более 1000 записей)';

               raiserror(@Msg, 0, 0) with nowait;

           end;

       end;

   end

   else

   begin

       set @Msg = concat(char(13), char(10), N'✓ Закольцовок не обнаружено. Иерархия корректна.');

       raiserror(@Msg, 0, 0) with nowait;

   end;

   -- Resultset 1: Итоговый результат

   select  @FullTableName as [Таблица],

           @CycleCount as [Всего проблем],

           (select count(*) from @SelfReferences) as [Self-ref связей],

           datediff(millisecond, @StartTime, sysdatetime()) as [Время (мс)];

   -- Resultset 2: Статистика по FK

   select  FKName as [FK связь],

           sum(case when CycleDepth = 0 then 1 else 0 end) as [Самоссылок],

           sum(case when CycleDepth > 0 then 1 else 0 end) as [Закольцовок],

           count(*) as [Всего]

   from    @CycleResults

   group by FKName

   order by FKName;

   if @DebugF = 1

   begin

       set @Msg = concat('Завершено за ', datediff(millisecond, @StartTime, sysdatetime()), ' мс');

       raiserror(@Msg, 0, 0) with nowait;

   end;

end;

go

Примеры использования процедуры dba_util.sp_DetectTableCycles:

ПРИМЕР 1: Простая проверка БЕЗ OUTPUT параметра

-- Самый простой вариант - просто смотрим на результаты

exec dba_util.sp_DetectTableCycles

   @TableName = 'FileStorageFolders',

   @SchemaName = 'dbo';

go

ПРИМЕР 2: Проверка С OUTPUT параметром (для программного использования)

declare @HasCyclesF bit;

exec dba_util.sp_DetectTableCycles

   @TableName = 'FileStorageFolders',

   @SchemaName = 'dbo',

   @HasCyclesF = @HasCyclesF output;

-- Проверяем результат

if @HasCyclesF = 1

   raiserror(N'❌ В таблице FileStorageFolders обнаружены закольцовки!', 0, 0) with nowait;

else

   raiserror(N'✓ Таблица FileStorageFolders в порядке - закольцовок нет.', 0, 0) with nowait;

go

ПРИМЕР 3: Детальная проверка с выводом всех найденных циклов

declare @HasCyclesF bit;

exec dba_util.sp_DetectTableCycles

   @TableName = 'FileStorageFolders',

   @SchemaName = 'dbo',

   @HasCyclesF = @HasCyclesF output,

   @ShowDetailsF = 1; -- Включаем детальный вывод

select  case when @HasCyclesF = 1

           then 'Требуется исправление данных'

           else 'Данные корректны'

       end as [Рекомендация];

go

ПРИМЕР 4: Проверка другой таблицы (например, Categories)

-- Можно использовать без OUTPUT параметра

exec dba_util.sp_DetectTableCycles

   @TableName = 'Categories', -- Замените на название вашей таблицы

   @SchemaName = 'dbo';

go

ПРИМЕР 5: Массовая проверка всех таблиц с self-referencing связями

declare @TableName nvarchar(128);

declare @SchemaName nvarchar(128);

declare @HasCyclesF bit;

declare @Msg nvarchar(500);

-- Временная таблица для хранения результатов

create table #CheckResults

(

   SchemaName  nvarchar(128),

   TableName   nvarchar(128),

   HasCyclesF  bit,

   CheckTime   datetime

);

-- Курсор по всем таблицам с self-referencing FK

declare table_cursor cursor for

   select distinct

           object_schema_name(fk.parent_object_id) as SchemaName,

           object_name(fk.parent_object_id) as TableName

   from    sys.foreign_keys fk

   where   fk.parent_object_id = fk.referenced_object_id;

open table_cursor;

fetch next from table_cursor into @SchemaName, @TableName;

while @@fetch_status = 0

begin

   set @Msg = concat('=========================================', char(13), char(10),

                     'Проверка таблицы: ', @SchemaName, '.', @TableName, char(13), char(10),

                     '=========================================');

   raiserror(@Msg, 0, 0) with nowait;

   exec dba_util.sp_DetectTableCycles

       @TableName = @TableName,

       @SchemaName = @SchemaName,

       @HasCyclesF = @HasCyclesF output;

   -- Сохраняем результат

   insert  into #CheckResults (SchemaName, TableName, HasCyclesF, CheckTime)

   values  (@SchemaName, @TableName, @HasCyclesF, getdate());

   fetch next from table_cursor into @SchemaName, @TableName;

end;

close table_cursor;

deallocate table_cursor;

-- Итоговый отчет

raiserror('=========================================', 0, 0) with nowait;

raiserror('ИТОГОВЫЙ ОТЧЕТ', 0, 0) with nowait;

raiserror('=========================================', 0, 0) with nowait;

select  SchemaName + '.' + TableName as [Таблица],

       case when HasCyclesF = 1 then N'❌ ДА' else N'✓ НЕТ' end as [Есть закольцовки],

       CheckTime as [Время проверки]

from    #CheckResults

order by HasCyclesF desc, SchemaName, TableName;

-- Статистика

select  count(*) as [Всего проверено таблиц],

       sum(case when HasCyclesF = 1 then 1 else 0 end) as [Таблиц с проблемами],

       sum(case when HasCyclesF = 0 then 1 else 0 end) as [Таблиц без проблем]

from    #CheckResults;

drop table #CheckResults;

go

ПРИМЕР 6: Использование в T-SQL скриптах с обработкой ошибок

begin try

   declare @HasCyclesF bit;

   exec dba_util.sp_DetectTableCycles

       @TableName = 'FileStorageFolders',

       @HasCyclesF = @HasCyclesF output;

   if @HasCyclesF = 1

   begin

      -- Отправить уведомление администратору

       -- Записать в лог

       raiserror('Обнаружены закольцовки в таблице FileStorageFolders!', 16, 1);

   end

   else

   begin

       raiserror('Проверка целостности завершена успешно.', 0, 0) with nowait;

   end;

end try

begin catch

   declare @ErrMsg nvarchar(500) = concat('Ошибка при проверке: ', error_message());

   raiserror(@ErrMsg, 0, 0) with nowait;

end catch;

go

ПРИМЕР 7: Интеграция в SQL Server Agent Job (шаблон)

-- use msdb;

-- go

--

-- exec dbo.sp_add_job

--     @job_name = N'Daily Cycle Check',

--     @enabled = 1,

--     @description = N'Ежедневная проверка на закольцовки';

--

-- exec sp_add_jobstep

--     @job_name = N'Daily Cycle Check',

--     @step_name = N'Check FileStorageFolders',

--     @subsystem = N'TSQL',

--     @command = N'

--         declare @HasCyclesF bit;

--         exec dba_util.sp_DetectTableCycles

--             @TableName = ''FileStorageFolders'',

--             @HasCyclesF = @HasCyclesF output;

--        

--         if @HasCyclesF = 1

--             raiserror(''Закольцовки обнаружены!'', 16, 1);

--     ',

--     @on_fail_action = 3; -- Go to next step

--

-- exec dbo.sp_add_schedule

--     @schedule_name = N'Daily at 2 AM',

--     @freq_type = 4, -- Daily

--     @freq_interval = 1,

--     @active_start_time = 020000; -- 02:00:00

--

-- exec sp_attach_schedule

--     @job_name = N'Daily Cycle Check',

--     @schedule_name = N'Daily at 2 AM';

--

-- exec dbo.sp_add_jobserver

--     @job_name = N'Daily Cycle Check';

-- go

ПРИМЕР 8: Проверка перед выполнением критических операций

-- Например, перед реорганизацией структуры папок

declare @HasCyclesF bit;

declare @CanProceedF bit = 0;

-- Проверяем на циклы

exec dba_util.sp_DetectTableCycles

   @TableName = 'FileStorageFolders',

   @HasCyclesF = @HasCyclesF output;

if @HasCyclesF = 0

begin

   set @CanProceedF = 1;

   raiserror('Проверка пройдена. Можно продолжить операцию.', 0, 0) with nowait;

   -- Здесь ваш код для реорганизации

   -- begin transaction

   -- ...

   -- commit

end

else

begin

   raiserror('ВНИМАНИЕ! Обнаружены закольцовки. Операция отменена.', 0, 0) with nowait;

   raiserror('Сначала необходимо исправить данные.', 0, 0) with nowait;

end;

go

 Функция обнаружения закольцовок в self-referencing таблицах

 Требования: PostgreSQL 14+ (используется CYCLE clause)

 Особенности:

 - Использование sys.tables, sys.constraint_columns вместо pg_catalog напрямую

 - CYCLE clause вместо массивов (встроенная оптимизация PostgreSQL)

 - SQL-функции где возможно

 - Без временных таблиц (массивы типов)

Вспомогательная функция: поиск self-referencing FK

Возвращает список self-referencing FK для указанной таблицы

create or replace function dba_util.fn_get_self_ref_fks

(

   p_table_name    varchar,

   p_schema_name   varchar

)

returns table

(

   pk_column   varchar,

   fk_column   varchar,

   fk_name     varchar

)

stable

as

$$

   select  distinct

           cc.fk_parent_colname::varchar   as pk_column,

           cc.colname::varchar             as fk_column,

           cc.constraint_name::varchar     as fk_name

   from    sys.constraint_columns cc

   where   cc.contype = 'f'

     and   cc.parent_object_id = cc.referenced_object_id  -- self-referencing

     and   cc.schemaname = lower(p_schema_name)

     and   cc.tablename = lower(p_table_name)

$$

language sql;

comment on function dba_util.fn_get_self_ref_fks(varchar, varchar) is

'Возвращает список self-referencing FK для таблицы.';

Функция проверки наличия циклов (возвращает bool)

Аналог OUTPUT параметра @HasCyclesF из MS SQL версии.

create or replace function dba_util.fn_has_table_cycles

(

   p_table_name    varchar,

   p_schema_name   varchar

)

returns bool

as

$$

declare

   v_full_table_name   varchar;

   v_sql               varchar;

   v_has_cycles        bool := false;

   v_ref               record;

   v_table_oid         int;

begin

   v_full_table_name := quote_ident(p_schema_name) || '.' || quote_ident(p_table_name);

   -- Проверка существования таблицы через sys.tables

   select  t.object_id into v_table_oid

   from    sys.tables t

   where   t.schemaname = lower(p_schema_name)

     and   t.name = lower(p_table_name);

   if v_table_oid is null then

       raise exception 'Таблица % не существует.', v_full_table_name;

   end if;

   -- Для каждой self-referencing связи

   for v_ref in

       select  sr.pk_column, sr.fk_column, sr.fk_name

       from    dba_util.fn_get_self_ref_fks(p_table_name, p_schema_name) sr

   loop

       -- Быстрая проверка с LIMIT 1 и CYCLE clause

       v_sql :=

           'with recursive recursive_hierarchy as ' ||

           '( ' ||

               'select  ' || quote_ident(v_ref.pk_column) || '::bigint as record_id, ' ||

                       quote_ident(v_ref.fk_column) || '::bigint as parent_id, ' ||

                       '0 as level ' ||

               'from    ' || v_full_table_name || ' ' ||

               'where   ' || quote_ident(v_ref.fk_column) || ' is not null ' ||

               'union all ' ||

               'select  rh.record_id, ' ||

                       't.' || quote_ident(v_ref.fk_column) || '::bigint, ' ||

                       'rh.level + 1 ' ||

               'from    ' || v_full_table_name || ' t ' ||

                       'join recursive_hierarchy rh ' ||

                           'on t.' || quote_ident(v_ref.pk_column) || ' = rh.parent_id ' ||

               'where   t.' || quote_ident(v_ref.fk_column) || ' is not null ' ||

                 'and   rh.level < 100 ' ||

                 'and   not is_cycle ' ||

           ') ' ||

           'cycle parent_id set is_cycle using path ' ||

           'select  exists ' ||

           '( ' ||

               'select  1 ' ||

               'from    recursive_hierarchy ' ||

               'where   is_cycle ' ||

                  'or   record_id = parent_id ' ||

               'limit   1 ' ||

           ')';

       execute v_sql into v_has_cycles;

       if v_has_cycles then

           return true;

       end if;

   end loop;

   return false;

end;

$$

language plpgsql;

comment on function dba_util.fn_has_table_cycles(varchar, varchar) is

'Быстрая проверка наличия циклов. Возвращает true если циклы есть.';

Тип для хранения результатов (используется внутри fn_detect_table_cycles)

call dbo.drop_type('dba_util.t__fn_detect_table_cycles__cycle_row');

create type dba_util.t__fn_detect_table_cycles__cycle_row as

(

   record_id       bigint,

   parent_id       bigint,

   cycle_depth     int,

   fk_name         varchar

);

 Основная функция обнаружения циклов

 Параметры:

   p_table_name    - имя таблицы

   p_schema_name   - схема

   p_show_details  - выводить ли детальную информацию (по умолчанию false)

   p_debug         - режим отладки (по умолчанию false)

 Возвращает:

   Таблицу с записями, участвующими в циклах

create or replace function dba_util.fn_detect_table_cycles

(

   p_table_name    varchar,

   p_schema_name   varchar,

   p_show_details  bool = false,

   p_debug         bool = false

)

returns table

(

   record_id       bigint,

   parent_id       bigint,

   cycle_depth     int,

   fk_name         varchar

)

as

$$

declare

   v_full_table_name   varchar;

   v_sql               varchar;

   v_cycle_count       bigint := 0;

   v_start_time        pg_catalog.timestamptz := pg_catalog.clock_timestamp();

   v_ref               record;

   v_ref_count         int := 0;

   v_table_oid         int;

   v_results           dba_util.t__fn_detect_table_cycles__cycle_row[];

   v_raw_results       dba_util.t__fn_detect_table_cycles__cycle_row[];

   v_row               dba_util.t__fn_detect_table_cycles__cycle_row;

begin

   v_full_table_name := quote_ident(p_schema_name) || '.' || quote_ident(p_table_name);

   v_results := array[]::dba_util.t__fn_detect_table_cycles__cycle_row[];

   -- Проверка существования таблицы через sys.tables

   select  t.object_id into v_table_oid

   from    sys.tables t

   where   t.schemaname = lower(p_schema_name)

     and   t.name = lower(p_table_name);

   if v_table_oid is null then

       raise exception 'Таблица % не существует.', v_full_table_name;

   end if;

   if p_debug then

       raise notice 'Начало проверки: % в %', v_full_table_name, v_start_time;

   end if;

   -- Проверяем наличие self-referencing связей

   if not exists

   (

       select  1

       from    dba_util.fn_get_self_ref_fks(p_table_name, p_schema_name)

   ) then

       raise notice 'Таблица % не имеет self-referencing связей.', v_full_table_name;

       return;

   end if;

   -- Показываем найденные связи

   if p_debug or p_show_details then

       raise notice 'Найдены self-referencing связи в таблице %:', v_full_table_name;

       for v_ref in

           select  sr.pk_column, sr.fk_column, sr.fk_name

           from    dba_util.fn_get_self_ref_fks(p_table_name, p_schema_name) sr

       loop

           raise notice '  PK: %, FK: %, Constraint: %', v_ref.pk_column, v_ref.fk_column, v_ref.fk_name;

       end loop;

   end if;

   -- Проверяем на циклы для КАЖДОЙ связи

   for v_ref in

       select  sr.pk_column, sr.fk_column, sr.fk_name

       from    dba_util.fn_get_self_ref_fks(p_table_name, p_schema_name) sr

   loop

       v_ref_count := v_ref_count + 1;

       if p_debug then

           raise notice '';

           raise notice 'Проверка связи #%: % <- % (%)',

               v_ref_count, v_ref.pk_column, v_ref.fk_column, v_ref.fk_name;

       end if;

       -- Динамический SQL для поиска циклов с CYCLE clause

       v_sql :=

           'with recursive recursive_hierarchy as ' ||

           '( ' ||

               'select  ' || quote_ident(v_ref.pk_column) || '::bigint as record_id, ' ||

                       quote_ident(v_ref.fk_column) || '::bigint as parent_id, ' ||

                       '0 as level ' ||

               'from    ' || v_full_table_name || ' ' ||

               'where   ' || quote_ident(v_ref.fk_column) || ' is not null ' ||

               'union all ' ||

               'select  rh.record_id, ' ||

                       't.' || quote_ident(v_ref.fk_column) || '::bigint, ' ||

                       'rh.level + 1 ' ||

               'from    ' || v_full_table_name || ' t ' ||

                       'join recursive_hierarchy rh ' ||

                           'on t.' || quote_ident(v_ref.pk_column) || ' = rh.parent_id ' ||

               'where   t.' || quote_ident(v_ref.fk_column) || ' is not null ' ||

                 'and   rh.level < 100 ' ||

                 'and   not is_cycle ' ||

           ') ' ||

           'cycle parent_id set is_cycle using path ' ||

           'select  pg_catalog.array_agg ' ||

           '( ' ||

               'row(record_id, parent_id, level, ' || quote_literal(v_ref.fk_name) || ')::' ||

               'dba_util.t__fn_detect_table_cycles__cycle_row ' ||

           ') ' ||

           'from    recursive_hierarchy ' ||

           'where   is_cycle ' ||

              'or   record_id = parent_id';

       if p_debug then

           raise notice '--- Dynamic SQL ---';

           raise notice '%', v_sql;

           raise notice '--- End SQL ---';

       end if;

       begin

           execute v_sql into v_raw_results;

           if v_raw_results is not null then

               v_results := v_results || v_raw_results;

           end if;

       exception when others then

           if p_debug then

               raise notice 'Ошибка при проверке связи %: %', v_ref.fk_name, sqlerrm;

           end if;

       end;

   end loop;

   -- Дедупликация: оставляем записи с минимальной глубиной для каждой пары (record_id, fk_name)

   select  pg_catalog.array_agg

           (

               row(d.record_id, d.parent_id, d.cycle_depth, d.fk_name)::

               dba_util.t__fn_detect_table_cycles__cycle_row

           )

   into    v_results

   from  

           (

               select  distinct on (r.record_id, r.fk_name)

                       r.record_id,

                       r.parent_id,

                       r.cycle_depth,

                       r.fk_name

               from    pg_catalog.unnest(v_results) r

               order by r.record_id, r.fk_name, r.cycle_depth

           ) d;

   -- Подсчёт результатов

   v_cycle_count := coalesce(pg_catalog.array_length(v_results, 1), 0);

   -- Вывод результатов

   if v_cycle_count > 0 then

       raise notice '';

       raise notice '⚠ ОБНАРУЖЕНЫ ПРОБЛЕМЫ: %', v_cycle_count;

       if p_show_details then

           if v_cycle_count <= 1000 then

               return query

                   select  r.record_id, r.parent_id, r.cycle_depth, r.fk_name

                   from    pg_catalog.unnest(v_results) r

                   order by r.fk_name, r.cycle_depth, r.record_id;

           else

               raise notice '⚠ Найдено более 1000 записей (%). Выводим первые 100.', v_cycle_count;

               return query

                   select  r.record_id, r.parent_id, r.cycle_depth, r.fk_name

                   from    pg_catalog.unnest(v_results) r

                   order by r.fk_name, r.cycle_depth, r.record_id

                   limit   100;

           end if;

       end if;

   else

       raise notice '';

       raise notice '✓ Закольцовок не обнаружено в таблице %', v_full_table_name;

   end if;

   -- Статистика по FK

   if p_show_details or p_debug then

       raise notice '';

       raise notice '=== Статистика по FK ===';

       for v_ref in

           select  sr.fk_name,

                   sr.pk_column,

                   sr.fk_column,

                   coalesce

                   (

                       (

                           select  pg_catalog.count(*)

                           from    pg_catalog.unnest(v_results) u

                           where   u.fk_name = sr.fk_name

                       ),

                       0

                   ) as cycles_count

           from    dba_util.fn_get_self_ref_fks(p_table_name, p_schema_name) sr

       loop

           raise notice 'FK: % (% -> %): % циклов',

               v_ref.fk_name, v_ref.fk_column, v_ref.pk_column, v_ref.cycles_count;

       end loop;

   end if;

   if p_debug then

       raise notice '';

       raise notice 'Время выполнения: % мс',

           extract(milliseconds from pg_catalog.clock_timestamp() - v_start_time)::int;

   end if;

   return;

end;

$$

language plpgsql;

comment on function dba_util.fn_detect_table_cycles(varchar, varchar, bool, bool) is

'Обнаружение закольцовок в таблицах с self-referencing FK.

Параметры: p_table_name, p_schema_name, p_show_details, p_debug';

/*------------------------------------------------------------------------------

 Вспомогательная функция: статистика по FK (прямые самоссылки)

------------------------------------------------------------------------------*/

create or replace function dba_util.fn_get_fk_statistics

(

   p_table_name    varchar,

   p_schema_name   varchar

)

returns table

(

   fk_name             varchar,

   pk_column           varchar,

   fk_column           varchar,

   self_refs_count     bigint,

   cycles_count        bigint

)

as

$$

declare

   v_full_table_name   varchar;

   v_ref               record;

   v_self_refs         bigint;

   v_sql               varchar;

begin

   v_full_table_name := quote_ident(p_schema_name) || '.' || quote_ident(p_table_name);

   for v_ref in

       select  sr.pk_column, sr.fk_column, sr.fk_name

       from    dba_util.fn_get_self_ref_fks(p_table_name, p_schema_name) sr

   loop

       v_sql :=

           'select pg_catalog.count(*) ' ||

           'from   ' || v_full_table_name || ' ' ||

           'where  ' || quote_ident(v_ref.fk_column) || ' is not null ' ||

             'and  ' || quote_ident(v_ref.pk_column) || ' = ' || quote_ident(v_ref.fk_column);

       execute v_sql into v_self_refs;

       fk_name := v_ref.fk_name;

       pk_column := v_ref.pk_column;

       fk_column := v_ref.fk_column;

       self_refs_count := v_self_refs;

       cycles_count := v_self_refs;

       return next;

   end loop;

   return;

end;

$$

language plpgsql;

comment on function dba_util.fn_get_fk_statistics(varchar, varchar) is

'Возвращает статистику по self-referencing FK (прямые самоссылки).';

/*==============================================================================

 Примеры использования

==============================================================================

-- ============================================================================

-- ПРИМЕР 1: Простая проверка

-- ============================================================================

select * from dba_util.fn_detect_table_cycles('filestoragefolders', 'dbo');

-- ============================================================================

-- ПРИМЕР 2: Проверка с получением флага (аналог OUTPUT параметра)

-- ============================================================================

do

$$

declare

   v_has_cycles bool;

begin

   v_has_cycles := dba_util.fn_has_table_cycles('filestoragefolders', 'dbo');

   if v_has_cycles then

       raise notice '❌ В таблице filestoragefolders обнаружены закольцовки!';

   else

       raise notice '✓ Таблица filestoragefolders в порядке - закольцовок нет.';

   end if;

end;

$$

language plpgsql;

-- Или в одну строку:

select  case

           when dba_util.fn_has_table_cycles('filestoragefolders', 'dbo')

           then '❌ Есть циклы'

           else '✓ Циклов нет'

       end as result;

-- ============================================================================

-- ПРИМЕР 3: Детальная проверка с выводом всех найденных циклов

-- ============================================================================

select  *

from    dba_util.fn_detect_table_cycles

       (

           p_table_name   := 'filestoragefolders',

           p_schema_name  := 'dbo',

           p_show_details := true

       );

-- ============================================================================

-- ПРИМЕР 4: Проверка с режимом отладки

-- ============================================================================

select  *

from    dba_util.fn_detect_table_cycles

       (

           p_table_name   := 'categories',

           p_schema_name  := 'dbo',

           p_show_details := true,

           p_debug        := true

       );

-- ============================================================================

-- ПРИМЕР 5: Массовая проверка всех таблиц с self-referencing связями

-- ============================================================================

do

$$

declare

   v_table     record;

   v_has_cycles bool;

   v_results   jsonb := '[]'::jsonb;

begin

   -- Находим все таблицы с self-referencing FK через sys.constraint_columns

   for v_table in

       select  distinct

               cc.schemaname,

               cc.tablename

       from    sys.constraint_columns cc

       where   cc.contype = 'f'

         and   cc.parent_object_id = cc.referenced_object_id

   loop

       raise notice '=========================================';

       raise notice 'Проверка таблицы: %.%', v_table.schemaname, v_table.tablename;

       raise notice '=========================================';

       v_has_cycles := dba_util.fn_has_table_cycles(v_table.tablename, v_table.schemaname);

       v_results := v_results || pg_catalog.jsonb_build_object

       (

           'schema', v_table.schemaname,

           'table', v_table.tablename,

           'has_cycles', v_has_cycles,

           'check_time', pg_catalog.clock_timestamp()

       );

   end loop;

   raise notice '=========================================';

   raise notice 'ИТОГОВЫЙ ОТЧЕТ';

   raise notice '=========================================';

   -- Вывод итогов

   for v_table in

       select  r->>'schema' as schema_name,

               r->>'table' as table_name,

               (r->>'has_cycles')::bool as has_cycles,

               r->>'check_time' as check_time

       from    pg_catalog.jsonb_array_elements(v_results) r

       order by (r->>'has_cycles')::bool desc, r->>'schema', r->>'table'

   loop

       raise notice '%: %',

           v_table.schema_name || '.' || v_table.table_name,

           case when v_table.has_cycles then '❌ ДА' else '✓ НЕТ' end;

   end loop;

end;

$$

language plpgsql;

-- ============================================================================

-- ПРИМЕР 6: Использование в PL/pgSQL с обработкой ошибок

-- ============================================================================

do

$$

declare

   v_has_cycles bool;

begin

   begin

       v_has_cycles := dba_util.fn_has_table_cycles('filestoragefolders', 'dbo');

       if v_has_cycles then

           raise exception 'Обнаружены закольцовки в таблице filestoragefolders!';

       else

           raise notice 'Проверка целостности завершена успешно.';

       end if;

   exception when others then

       raise notice 'Ошибка при проверке: %', sqlerrm;

   end;

end;

$$

language plpgsql;

-- ============================================================================

-- ПРИМЕР 7: Проверка перед критическими операциями

-- ============================================================================

do

$$

declare

   v_has_cycles bool;

begin

   v_has_cycles := dba_util.fn_has_table_cycles('filestoragefolders', 'dbo');

   if not v_has_cycles then

       raise notice 'Проверка пройдена. Можно продолжить операцию.';

       -- Ваш код здесь

   else

       raise notice 'ВНИМАНИЕ! Обнаружены закольцовки. Операция отменена.';

   end if;

end;

$$

language plpgsql;

-- ============================================================================

-- ПРИМЕР 8: Статистика по FK

-- ============================================================================

select * from dba_util.fn_get_fk_statistics('filestoragefolders', 'dbo');

==============================================================================*/