rag.usp_PlanChange_Detect

-- Tespit (spec §5). @AsOfUtc test içindir; NULL: şimdi.
CREATE   PROCEDURE rag.usp_PlanChange_Detect
    @AsOfUtc datetime = NULL, @ServerID int = NULL, @DatabaseName sysname = NULL
AS
BEGIN
    SET NOCOUNT ON;
    SET XACT_ABORT ON;
    IF DBA.fn_GetSettingInt(N'PlanChange.Enabled', 1) = 0 RETURN;

    DECLARE @start datetime = DBA.fn_Now();
    DECLARE @asOf datetime = COALESCE(@AsOfUtc, @start);
    DECLARE @w int = DBA.fn_GetSettingInt(N'PlanChange.WindowDays', 14),
            @ret int = DBA.fn_GetSettingInt(N'Retention.QdsDays', 14),
            @minExec int = DBA.fn_GetSettingInt(N'PlanChange.MinExecutionsPerPlan', 20),
            @pct int = DBA.fn_GetSettingInt(N'PlanChange.DurationChangePercent', 30),
            @minDiffMs int = DBA.fn_GetSettingInt(N'PlanChange.MinDurationDiffMs', 5),
            @hours int = DBA.fn_GetSettingInt(N'PlanChange.EventWindowHours', 48),
            @altMin int = DBA.fn_GetSettingInt(N'PlanChange.AlternatingSwitches', 3),
            @maxNew int = DBA.fn_GetSettingInt(N'PlanChange.MaxNewCasesPerRun', 50);
    IF @w > @ret SET @w = @ret;
    DECLARE @from datetime = DATEADD(day, -@w, @asOf), @rows int = 0, @err nvarchar(4000);

    BEGIN TRY
        -- 1. Plan istatistikleri: yalnız normal tamamlanan çalışmaların ağırlıklı ortalaması; iptal/hata ayrı
        SELECT rs.ServerID, rs.DatabaseName, p.query_id, rs.plan_id,
               SUM(CASE WHEN rs.execution_type = 0 THEN rs.count_executions ELSE 0 END) AS Execs,
               SUM(CASE WHEN rs.execution_type = 3 THEN rs.count_executions ELSE 0 END) AS Aborted,
               SUM(CASE WHEN rs.execution_type = 4 THEN rs.count_executions ELSE 0 END) AS Failed,
               MIN(CASE WHEN rs.execution_type = 0 THEN rs.IntervalStartUtc END) AS FirstUtc,
               MAX(CASE WHEN rs.execution_type = 0 THEN rs.IntervalEndUtc END) AS LastUtc,
               SUM(CASE WHEN rs.execution_type = 0 THEN rs.avg_duration * rs.count_executions END) AS SumDur,
               MAX(CASE WHEN rs.execution_type = 0 THEN rs.max_duration END) AS MaxDur,
               SUM(CASE WHEN rs.execution_type = 0 THEN rs.count_executions * (SQUARE(ISNULL(rs.stdev_duration, 0)) + SQUARE(rs.avg_duration)) END) AS SumSq,
               SUM(CASE WHEN rs.execution_type = 0 THEN rs.avg_cpu_time * rs.count_executions END) AS SumCpu,
               MAX(CASE WHEN rs.execution_type = 0 THEN rs.max_cpu_time END) AS MaxCpu,
               SUM(CASE WHEN rs.execution_type = 0 THEN rs.avg_logical_io_reads * rs.count_executions END) AS SumReads,
               MAX(CASE WHEN rs.execution_type = 0 THEN rs.max_logical_io_reads END) AS MaxReads,
               SUM(CASE WHEN rs.execution_type = 0 THEN rs.avg_physical_io_reads * rs.count_executions END) AS SumPhys,
               SUM(CASE WHEN rs.execution_type = 0 THEN rs.avg_logical_io_writes * rs.count_executions END) AS SumWrites,
               SUM(CASE WHEN rs.execution_type = 0 THEN rs.avg_query_max_used_memory * rs.count_executions END) AS SumMem,
               SUM(CASE WHEN rs.execution_type = 0 THEN rs.avg_tempdb_space_used * rs.count_executions END) AS SumTempdb,
               SUM(CASE WHEN rs.execution_type = 0 THEN rs.avg_dop * rs.count_executions END) AS SumDop,
               SUM(CASE WHEN rs.execution_type = 0 THEN rs.avg_rowcount * rs.count_executions END) AS SumRows,
               MAX(CASE WHEN rs.execution_type = 0 THEN rs.max_rowcount END) AS MaxRows
        INTO #ps
        FROM perf.QdsRuntimeStats AS rs
        JOIN perf.QdsPlan AS p ON p.ServerID = rs.ServerID AND p.DatabaseName = rs.DatabaseName AND p.plan_id = rs.plan_id
        JOIN DBA.MonitoredDatabase AS md ON md.ServerID = rs.ServerID AND md.DatabaseName = rs.DatabaseName
        WHERE md.IsActive = 1 AND md.CollectQds = 1
          AND rs.IntervalStartUtc >= @from AND rs.IntervalStartUtc < @asOf
          AND (@ServerID IS NULL OR rs.ServerID = @ServerID)
          AND (@DatabaseName IS NULL OR rs.DatabaseName = @DatabaseName)
        GROUP BY rs.ServerID, rs.DatabaseName, p.query_id, rs.plan_id
        HAVING SUM(CASE WHEN rs.execution_type = 0 THEN rs.count_executions ELSE 0 END) >= @minExec;

        -- 2. Plan çiftleri: ilk görülme sırasıyla ardışık planlar; 3. eşik (oran + mutlak fark)
        ;WITH o AS (
            SELECT *, SumDur / Execs AS AvgDur,
                   ROW_NUMBER() OVER (PARTITION BY ServerID, DatabaseName, query_id ORDER BY FirstUtc, plan_id) AS rn
            FROM #ps)
        SELECT a.ServerID, a.DatabaseName, a.query_id, a.plan_id AS OldPlanId, b.plan_id AS NewPlanId,
               b.FirstUtc AS ChangeUtc, a.AvgDur AS OldAvgDur, b.AvgDur AS NewAvgDur, b.Execs AS NewExecs,
               CASE WHEN b.AvgDur > a.AvgDur THEN 'Regressed' ELSE 'Improved' END AS Direction,
               CASE WHEN b.AvgDur > a.AvgDur THEN b.AvgDur / NULLIF(a.AvgDur, 0) ELSE a.AvgDur / NULLIF(b.AvgDur, 0) END AS Ratio,
               b.Execs * ABS(b.AvgDur - a.AvgDur) / 1000.0 AS LoadMs,
               CAST(NULL AS datetime) AS ChangeEndUtc, 0 AS Switches, CAST(NULL AS bigint) AS CaseID, CAST(0 AS bit) AS IsNew
        INTO #pair
        FROM o AS a
        JOIN o AS b ON b.ServerID = a.ServerID AND b.DatabaseName = a.DatabaseName AND b.query_id = a.query_id AND b.rn = a.rn + 1
        WHERE ABS(b.AvgDur - a.AvgDur) / 1000.0 >= @minDiffMs
          AND (CASE WHEN b.AvgDur > a.AvgDur THEN b.AvgDur / NULLIF(a.AvgDur, 0) ELSE a.AvgDur / NULLIF(b.AvgDur, 0) END) >= 1 + @pct / 100.0;

        -- Değişim interval'inin sonu
        UPDATE pr SET ChangeEndUtc = COALESCE((
            SELECT MIN(rs.IntervalEndUtc) FROM perf.QdsRuntimeStats AS rs
            WHERE rs.ServerID = pr.ServerID AND rs.DatabaseName = pr.DatabaseName AND rs.plan_id = pr.NewPlanId
              AND rs.IntervalStartUtc = pr.ChangeUtc), DATEADD(hour, 1, pr.ChangeUtc))
        FROM #pair AS pr;

        -- Dönüşümlü desen: interval başına baskın plan (çiftin iki planı arasında) kaç kez el değiştirdi
        ;WITH iv AS (
            SELECT pr.ServerID, pr.DatabaseName, pr.query_id, pr.OldPlanId, pr.NewPlanId, rs.IntervalStartUtc,
                   SUM(CASE WHEN rs.plan_id = pr.NewPlanId THEN rs.count_executions ELSE 0 END) AS NewCnt,
                   SUM(CASE WHEN rs.plan_id = pr.OldPlanId THEN rs.count_executions ELSE 0 END) AS OldCnt
            FROM #pair AS pr
            JOIN perf.QdsRuntimeStats AS rs ON rs.ServerID = pr.ServerID AND rs.DatabaseName = pr.DatabaseName
                 AND rs.plan_id IN (pr.OldPlanId, pr.NewPlanId) AND rs.execution_type = 0
                 AND rs.IntervalStartUtc >= @from AND rs.IntervalStartUtc < @asOf
            GROUP BY pr.ServerID, pr.DatabaseName, pr.query_id, pr.OldPlanId, pr.NewPlanId, rs.IntervalStartUtc),
        d AS (
            SELECT *, CASE WHEN NewCnt > OldCnt THEN 1 ELSE 0 END AS Dom FROM iv WHERE NewCnt <> OldCnt),
        s AS (
            SELECT ServerID, DatabaseName, query_id, OldPlanId, NewPlanId,
                   CASE WHEN Dom <> LAG(Dom) OVER (PARTITION BY ServerID, DatabaseName, query_id, OldPlanId, NewPlanId ORDER BY IntervalStartUtc)
                        THEN 1 ELSE 0 END AS Sw
            FROM d)
        UPDATE pr SET Switches = x.n
        FROM #pair AS pr
        JOIN (SELECT ServerID, DatabaseName, query_id, OldPlanId, NewPlanId, SUM(Sw) AS n FROM s
              GROUP BY ServerID, DatabaseName, query_id, OldPlanId, NewPlanId) AS x
          ON x.ServerID = pr.ServerID AND x.DatabaseName = pr.DatabaseName AND x.query_id = pr.query_id
         AND x.OldPlanId = pr.OldPlanId AND x.NewPlanId = pr.NewPlanId;

        -- 6-7. Hangi çiftler yazılır: mevcut açık vakalar (New/Pending/Unsure) + en fazla @maxNew yeni vaka (toplam yük farkına göre)
        UPDATE pr SET CaseID = c.CaseID
        FROM #pair AS pr
        JOIN rag.PlanChangeCase AS c ON c.ServerID = pr.ServerID AND c.DatabaseName = pr.DatabaseName AND c.query_id = pr.query_id
             AND c.OldPlanId = pr.OldPlanId AND c.NewPlanId = pr.NewPlanId;
        DELETE pr FROM #pair AS pr
        WHERE pr.CaseID IS NOT NULL
          AND NOT EXISTS (SELECT 1 FROM rag.PlanChangeCase AS c WHERE c.CaseID = pr.CaseID AND c.Status IN ('New','Pending','Unsure'));
        ;WITH n AS (SELECT *, ROW_NUMBER() OVER (ORDER BY LoadMs DESC, ServerID, DatabaseName, query_id, NewPlanId) AS r FROM #pair WHERE CaseID IS NULL)
        DELETE FROM n WHERE r > @maxNew;
        UPDATE #pair SET IsNew = 1 WHERE CaseID IS NULL;

        BEGIN TRAN;

        INSERT rag.PlanChangeCase (ServerID, DatabaseName, query_id, OldPlanId, NewPlanId, ChangeUtc, ChangeEndUtc, Direction,
                                   DurationRatio, TotalLoadDeltaMs, Pattern, AlternatingSwitches, Status, DetectedUtc, EvidenceUpdatedUtc)
        SELECT ServerID, DatabaseName, query_id, OldPlanId, NewPlanId, ChangeUtc, ChangeEndUtc, Direction, Ratio, LoadMs,
               CASE WHEN Switches >= @altMin THEN 'Alternating' ELSE 'Switch' END, Switches, 'New', @start, @start
        FROM #pair WHERE IsNew = 1;
        UPDATE pr SET CaseID = c.CaseID
        FROM #pair AS pr
        JOIN rag.PlanChangeCase AS c ON c.ServerID = pr.ServerID AND c.DatabaseName = pr.DatabaseName AND c.query_id = pr.query_id
             AND c.OldPlanId = pr.OldPlanId AND c.NewPlanId = pr.NewPlanId
        WHERE pr.IsNew = 1;

        -- Başlık ve dondurulacak plan/metin: açık vakalar güncellenir; Unsure durumunu korur (insan kararı), diğerleri yeniden puanlanır
        UPDATE c SET
            ChangeUtc = pr.ChangeUtc, ChangeEndUtc = pr.ChangeEndUtc, Direction = pr.Direction, DurationRatio = pr.Ratio,
            TotalLoadDeltaMs = pr.LoadMs, Pattern = CASE WHEN pr.Switches >= @altMin THEN 'Alternating' ELSE 'Switch' END,
            AlternatingSwitches = pr.Switches,
            query_hash = q.query_hash, ObjectName = q.ObjectName, SqlText = t.query_sql_text,
            OldQueryPlanHash = po.query_plan_hash, NewQueryPlanHash = pn.query_plan_hash,
            OldIsForced = po.is_forced_plan, NewIsForced = pn.is_forced_plan,
            OldCompat = po.compatibility_level, NewCompat = pn.compatibility_level,
            OldPlanXml = xo.PlanXmlCompressed, NewPlanXml = xn.PlanXmlCompressed,
            Status = CASE WHEN c.Status = 'Unsure' THEN 'Unsure' ELSE 'New' END,
            RulesVersion = NULL, EvidenceUpdatedUtc = @start
        FROM rag.PlanChangeCase AS c
        JOIN #pair AS pr ON pr.CaseID = c.CaseID
        LEFT JOIN perf.QdsQuery AS q ON q.ServerID = c.ServerID AND q.DatabaseName = c.DatabaseName AND q.query_id = c.query_id
        LEFT JOIN perf.QdsQueryText AS t ON t.TextHash = q.TextHash
        LEFT JOIN perf.QdsPlan AS po ON po.ServerID = c.ServerID AND po.DatabaseName = c.DatabaseName AND po.plan_id = c.OldPlanId
        LEFT JOIN perf.QdsPlan AS pn ON pn.ServerID = c.ServerID AND pn.DatabaseName = c.DatabaseName AND pn.plan_id = c.NewPlanId
        LEFT JOIN perf.QdsPlanXml AS xo ON xo.PlanHash = po.PlanHash
        LEFT JOIN perf.QdsPlanXml AS xn ON xn.PlanHash = pn.PlanHash;

        -- Kanıt: önce eskisi silinir
        DELETE m FROM rag.PlanChangeMetric AS m WHERE m.CaseID IN (SELECT CaseID FROM #pair);
        DELETE s FROM rag.PlanChangeSeries AS s WHERE s.CaseID IN (SELECT CaseID FROM #pair);
        DELETE w FROM rag.PlanChangeWait AS w WHERE w.CaseID IN (SELECT CaseID FROM #pair);
        DELETE e FROM rag.PlanChangeEvent AS e WHERE e.CaseID IN (SELECT CaseID FROM #pair);

        -- Rol → plan eşlemesi
        SELECT pr.CaseID, r.PlanRole, r.plan_id, pr.ServerID, pr.DatabaseName
        INTO #role
        FROM #pair AS pr
        CROSS APPLY (VALUES ('Old', pr.OldPlanId), ('New', pr.NewPlanId)) AS r (PlanRole, plan_id);

        -- Etki özeti (QDS: süre/CPU mikrosaniye, bellek/tempdb 8 KB sayfa)
        INSERT rag.PlanChangeMetric (CaseID, PlanRole, Executions, Aborted, Failed, AvgDurationMs, MaxDurationMs, StdevDurationMs,
               AvgCpuMs, MaxCpuMs, AvgLogicalReads, MaxLogicalReads, AvgPhysicalReads, AvgWrites, AvgMemoryKb, AvgTempdbKb, AvgDop,
               AvgRows, MaxRows, FirstUtc, LastUtc)
        SELECT r.CaseID, r.PlanRole, p.Execs, p.Aborted, p.Failed,
               p.SumDur / p.Execs / 1000.0, p.MaxDur / 1000.0,
               SQRT(CASE WHEN p.SumSq / p.Execs - SQUARE(p.SumDur / p.Execs) > 0 THEN p.SumSq / p.Execs - SQUARE(p.SumDur / p.Execs) ELSE 0 END) / 1000.0,
               p.SumCpu / p.Execs / 1000.0, p.MaxCpu / 1000.0, p.SumReads / p.Execs, p.MaxReads, p.SumPhys / p.Execs, p.SumWrites / p.Execs,
               p.SumMem / p.Execs * 8, p.SumTempdb / p.Execs * 8, p.SumDop / p.Execs, p.SumRows / p.Execs, p.MaxRows, p.FirstUtc, p.LastUtc
        FROM #role AS r
        JOIN #ps AS p ON p.ServerID = r.ServerID AND p.DatabaseName = r.DatabaseName AND p.plan_id = r.plan_id;

        -- Interval serisi (grafik)
        INSERT rag.PlanChangeSeries (CaseID, PlanRole, IntervalStartUtc, Executions, AvgDurationMs, AvgCpuMs, AvgLogicalReads, AvgRows, TotalWaitMs)
        SELECT r.CaseID, r.PlanRole, rs.IntervalStartUtc, SUM(rs.count_executions),
               SUM(rs.avg_duration * rs.count_executions) / NULLIF(SUM(rs.count_executions), 0) / 1000.0,
               SUM(rs.avg_cpu_time * rs.count_executions) / NULLIF(SUM(rs.count_executions), 0) / 1000.0,
               SUM(rs.avg_logical_io_reads * rs.count_executions) / NULLIF(SUM(rs.count_executions), 0),
               SUM(rs.avg_rowcount * rs.count_executions) / NULLIF(SUM(rs.count_executions), 0),
               (SELECT SUM(ws.total_query_wait_time_ms) FROM perf.QdsWaitStats AS ws
                WHERE ws.ServerID = r.ServerID AND ws.DatabaseName = r.DatabaseName AND ws.plan_id = r.plan_id
                  AND ws.IntervalStartUtc = rs.IntervalStartUtc)
        FROM #role AS r
        JOIN perf.QdsRuntimeStats AS rs ON rs.ServerID = r.ServerID AND rs.DatabaseName = r.DatabaseName AND rs.plan_id = r.plan_id
             AND rs.execution_type = 0 AND rs.IntervalStartUtc >= @from AND rs.IntervalStartUtc < @asOf
        GROUP BY r.CaseID, r.PlanRole, r.ServerID, r.DatabaseName, r.plan_id, rs.IntervalStartUtc;

        -- Bekleme dağılımı
        INSERT rag.PlanChangeWait (CaseID, PlanRole, WaitCategory, TotalWaitMs, AvgWaitMsPerExec)
        SELECT r.CaseID, r.PlanRole, ISNULL(ws.wait_category_desc, N'Unknown'), SUM(ws.total_query_wait_time_ms),
               SUM(ws.total_query_wait_time_ms) * 1.0 / NULLIF(MAX(p.Execs), 0)
        FROM #role AS r
        JOIN #ps AS p ON p.ServerID = r.ServerID AND p.DatabaseName = r.DatabaseName AND p.plan_id = r.plan_id
        JOIN perf.QdsWaitStats AS ws ON ws.ServerID = r.ServerID AND ws.DatabaseName = r.DatabaseName AND ws.plan_id = r.plan_id
             AND ws.IntervalStartUtc >= @from AND ws.IntervalStartUtc < @asOf
        GROUP BY r.CaseID, r.PlanRole, ISNULL(ws.wait_category_desc, N'Unknown')
        HAVING SUM(ws.total_query_wait_time_ms) > 0;

        -- 4. Planlardaki nesneler (XQuery): tablo/index (RelOp Object) ve kullanılan istatistikler (StatisticsInfo)
        SELECT r.CaseID, r.PlanRole, r.ServerID, r.DatabaseName, TRY_CAST(CAST(DECOMPRESS(x.PlanXmlCompressed) AS nvarchar(max)) AS xml) AS PlanXml
        INTO #px
        FROM #role AS r
        JOIN perf.QdsPlan AS p ON p.ServerID = r.ServerID AND p.DatabaseName = r.DatabaseName AND p.plan_id = r.plan_id
        JOIN perf.QdsPlanXml AS x ON x.PlanHash = p.PlanHash;

        ;WITH XMLNAMESPACES (DEFAULT 'http://schemas.microsoft.com/sqlserver/2004/07/showplan')
        SELECT DISTINCT px.CaseID, px.PlanRole, px.ServerID, px.DatabaseName,
               PARSENAME(n.value('@Schema', 'nvarchar(260)'), 1) AS SchemaName,
               PARSENAME(n.value('@Table', 'nvarchar(260)'), 1) AS TableName,
               PARSENAME(n.value('@Index', 'nvarchar(260)'), 1) AS IndexName
        INTO #pobj
        FROM #px AS px
        CROSS APPLY px.PlanXml.nodes('//RelOp//Object') AS o (n)
        WHERE n.value('@Table', 'nvarchar(260)') IS NOT NULL AND n.value('@Table', 'nvarchar(260)') NOT LIKE N'[[]#%';

        ;WITH XMLNAMESPACES (DEFAULT 'http://schemas.microsoft.com/sqlserver/2004/07/showplan')
        SELECT DISTINCT px.CaseID, px.PlanRole,
               PARSENAME(n.value('@Schema', 'nvarchar(260)'), 1) AS SchemaName,
               PARSENAME(n.value('@Table', 'nvarchar(260)'), 1) AS TableName,
               PARSENAME(n.value('@Statistics', 'nvarchar(260)'), 1) AS StatsName
        INTO #pstat
        FROM #px AS px
        CROSS APPLY px.PlanXml.nodes('//OptimizerStatsUsage/StatisticsInfo') AS o (n);

        -- Planların dokunduğu tablolar (iki plan birleşimi) → object_id
        SELECT DISTINCT po.CaseID, po.ServerID, po.DatabaseName, mo.object_id, po.SchemaName, po.TableName
        INTO #ptab
        FROM #pobj AS po
        JOIN perf.MetaObject AS mo ON mo.ServerID = po.ServerID AND mo.DatabaseName = po.DatabaseName
             AND mo.SchemaName = po.SchemaName AND mo.ObjectName = po.TableName;

        -- Olay penceresi: [değişim − H, değişim interval'inin sonu]; aralık değişim anını aşıyorsa sıra belirsiz
        SELECT pr.CaseID, pr.ServerID, pr.DatabaseName, DATEADD(hour, -@hours, pr.ChangeUtc) AS WinStart, pr.ChangeUtc, pr.ChangeEndUtc
        INTO #win FROM #pair AS pr;

        -- DB'nin ilk metadata toplaması: o anda görülen index'ler "yeni" değildir
        SELECT DISTINCT w.ServerID, w.DatabaseName,
               (SELECT MIN(mo.FirstSeenUtc) FROM perf.MetaObject AS mo WHERE mo.ServerID = w.ServerID AND mo.DatabaseName = w.DatabaseName) AS FirstCollectUtc
        INTO #firstc FROM #win AS w;
        UPDATE c SET MetadataFirstUtc = f.FirstCollectUtc
        FROM rag.PlanChangeCase AS c JOIN #pair AS pr ON pr.CaseID = c.CaseID
        JOIN #firstc AS f ON f.ServerID = c.ServerID AND f.DatabaseName = c.DatabaseName;

        -- 5. Olaylar ─ IndexCreated: zaman = index istatistiğinin ilk güncellemesi (index oluşurken tam taramayla üretilir),
        -- toplama aralığında değilse [önceki toplama, FirstSeenUtc]
        ;WITH ic AS (
            SELECT w.CaseID, t.SchemaName, t.TableName, i.IndexName, i.FirstSeenUtc,
                   COALESCE((SELECT MAX(l.StartUtc) FROM DBA.CollectionRunLog AS l
                             WHERE l.SourceType = 'METADATA' AND l.Status = 'OK' AND l.ServerID = w.ServerID AND l.DatabaseName = w.DatabaseName
                               AND l.EndUtc < i.FirstSeenUtc), DATEADD(day, -1, i.FirstSeenUtc)) AS PrevCollect,
                   COALESCE((SELECT MIN(h.LastUpdatedUtc) FROM perf.MetaStatisticsHistory AS h
                             WHERE h.ServerID = w.ServerID AND h.DatabaseName = w.DatabaseName AND h.object_id = i.object_id AND h.stats_id = i.index_id),
                            (SELECT st.LastUpdatedUtc FROM perf.MetaStatistics AS st
                             WHERE st.ServerID = w.ServerID AND st.DatabaseName = w.DatabaseName AND st.object_id = i.object_id AND st.stats_id = i.index_id)) AS StatsUtc,
                   w.WinStart, w.ChangeUtc, w.ChangeEndUtc
            FROM #win AS w
            JOIN #ptab AS t ON t.CaseID = w.CaseID
            JOIN perf.MetaIndex AS i ON i.ServerID = w.ServerID AND i.DatabaseName = w.DatabaseName AND i.object_id = t.object_id
            JOIN #firstc AS f ON f.ServerID = w.ServerID AND f.DatabaseName = w.DatabaseName
            WHERE i.IndexName IS NOT NULL AND i.FirstSeenUtc > f.FirstCollectUtc),
        ict AS (
            SELECT *, CASE WHEN StatsUtc > PrevCollect AND StatsUtc <= FirstSeenUtc THEN StatsUtc ELSE PrevCollect END AS E,
                      CASE WHEN StatsUtc > PrevCollect AND StatsUtc <= FirstSeenUtc THEN StatsUtc ELSE FirstSeenUtc END AS L
            FROM ic)
        INSERT rag.PlanChangeEvent (CaseID, EventType, EarliestUtc, LatestUtc, ObjectName, UsedByOldPlan, UsedByNewPlan, OrderUncertain)
        SELECT c.CaseID, 'IndexCreated', c.E, c.L, CONCAT(c.SchemaName, N'.', c.TableName, N'.', c.IndexName),
               CASE WHEN EXISTS (SELECT 1 FROM #pobj AS o WHERE o.CaseID = c.CaseID AND o.PlanRole = 'Old' AND o.TableName = c.TableName AND o.IndexName = c.IndexName) THEN 1 ELSE 0 END,
               CASE WHEN EXISTS (SELECT 1 FROM #pobj AS o WHERE o.CaseID = c.CaseID AND o.PlanRole = 'New' AND o.TableName = c.TableName AND o.IndexName = c.IndexName) THEN 1 ELSE 0 END,
               CASE WHEN c.L > c.ChangeUtc THEN 1 ELSE 0 END
        FROM ict AS c
        WHERE c.L >= c.WinStart AND c.E <= c.ChangeEndUtc;

        -- IndexDropped: [LastSeenUtc, silindiğini gören toplama]
        ;WITH idr AS (
            SELECT w.CaseID, t.SchemaName, t.TableName, i.IndexName, i.LastSeenUtc AS E,
                   COALESCE((SELECT MIN(l.EndUtc) FROM DBA.CollectionRunLog AS l
                             WHERE l.SourceType = 'METADATA' AND l.Status = 'OK' AND l.ServerID = w.ServerID AND l.DatabaseName = w.DatabaseName
                               AND l.StartUtc > i.LastSeenUtc), DATEADD(day, 1, i.LastSeenUtc)) AS L,
                   w.WinStart, w.ChangeUtc, w.ChangeEndUtc
            FROM #win AS w
            JOIN #ptab AS t ON t.CaseID = w.CaseID
            JOIN perf.MetaIndex AS i ON i.ServerID = w.ServerID AND i.DatabaseName = w.DatabaseName AND i.object_id = t.object_id
            WHERE i.IsDeleted = 1 AND i.IndexName IS NOT NULL)
        INSERT rag.PlanChangeEvent (CaseID, EventType, EarliestUtc, LatestUtc, ObjectName, UsedByOldPlan, UsedByNewPlan, OrderUncertain)
        SELECT d.CaseID, 'IndexDropped', d.E, d.L, CONCAT(d.SchemaName, N'.', d.TableName, N'.', d.IndexName),
               CASE WHEN EXISTS (SELECT 1 FROM #pobj AS o WHERE o.CaseID = d.CaseID AND o.PlanRole = 'Old' AND o.TableName = d.TableName AND o.IndexName = d.IndexName) THEN 1 ELSE 0 END,
               CASE WHEN EXISTS (SELECT 1 FROM #pobj AS o WHERE o.CaseID = d.CaseID AND o.PlanRole = 'New' AND o.TableName = d.TableName AND o.IndexName = d.IndexName) THEN 1 ELSE 0 END,
               CASE WHEN d.L > d.ChangeUtc THEN 1 ELSE 0 END
        FROM idr AS d
        WHERE d.L >= d.WinStart AND d.E <= d.ChangeEndUtc;

        -- StatsUpdated: geçmiş + güncel satır; yeni index'in kendi istatistiği hariç (o olay IndexCreated'a ait).
        -- Plan StatisticsInfo taşıyorsa kullanım bayrağı ondan, taşımıyorsa NULL (bilinmiyor)
        ;WITH su AS (
            SELECT w.CaseID, t.SchemaName, t.TableName, t.object_id, h.stats_id, h.LastUpdatedUtc, h.[Rows], h.RowsSampled, w.WinStart, w.ChangeUtc, w.ChangeEndUtc
            FROM #win AS w
            JOIN #ptab AS t ON t.CaseID = w.CaseID
            JOIN perf.MetaStatisticsHistory AS h ON h.ServerID = w.ServerID AND h.DatabaseName = w.DatabaseName AND h.object_id = t.object_id
            UNION
            SELECT w.CaseID, t.SchemaName, t.TableName, t.object_id, s.stats_id, s.LastUpdatedUtc, s.[Rows], s.RowsSampled, w.WinStart, w.ChangeUtc, w.ChangeEndUtc
            FROM #win AS w
            JOIN #ptab AS t ON t.CaseID = w.CaseID
            JOIN perf.MetaStatistics AS s ON s.ServerID = w.ServerID AND s.DatabaseName = w.DatabaseName AND s.object_id = t.object_id
            WHERE s.LastUpdatedUtc IS NOT NULL)
        INSERT rag.PlanChangeEvent (CaseID, EventType, EarliestUtc, LatestUtc, ObjectName, BeforeValue, AfterValue, UsedByOldPlan, UsedByNewPlan, OrderUncertain)
        SELECT su.CaseID, 'StatsUpdated', su.LastUpdatedUtc, su.LastUpdatedUtc, CONCAT(su.SchemaName, N'.', su.TableName, N'.', st.StatsName),
               NULL, CONCAT(su.RowsSampled, N'/', su.[Rows]),
               CASE WHEN NOT EXISTS (SELECT 1 FROM #pstat AS ps WHERE ps.CaseID = su.CaseID AND ps.PlanRole = 'Old') THEN NULL
                    WHEN EXISTS (SELECT 1 FROM #pstat AS ps WHERE ps.CaseID = su.CaseID AND ps.PlanRole = 'Old' AND ps.TableName = su.TableName AND ps.StatsName = st.StatsName) THEN 1 ELSE 0 END,
               CASE WHEN NOT EXISTS (SELECT 1 FROM #pstat AS ps WHERE ps.CaseID = su.CaseID AND ps.PlanRole = 'New') THEN NULL
                    WHEN EXISTS (SELECT 1 FROM #pstat AS ps WHERE ps.CaseID = su.CaseID AND ps.PlanRole = 'New' AND ps.TableName = su.TableName AND ps.StatsName = st.StatsName) THEN 1 ELSE 0 END,
               CASE WHEN su.LastUpdatedUtc > su.ChangeUtc THEN 1 ELSE 0 END
        FROM su
        CROSS APPLY (SELECT COALESCE(
                        (SELECT TOP (1) s2.StatsName FROM perf.MetaStatistics AS s2 JOIN #win AS w2 ON w2.CaseID = su.CaseID
                         WHERE s2.ServerID = w2.ServerID AND s2.DatabaseName = w2.DatabaseName AND s2.object_id = su.object_id AND s2.stats_id = su.stats_id),
                        CONCAT(N'stats_id ', su.stats_id)) AS StatsName) AS st
        WHERE su.LastUpdatedUtc >= su.WinStart AND su.LastUpdatedUtc <= su.ChangeEndUtc
          AND NOT EXISTS (SELECT 1 FROM rag.PlanChangeEvent AS e
                          WHERE e.CaseID = su.CaseID AND e.EventType = 'IndexCreated'
                            AND e.ObjectName = CONCAT(su.SchemaName, N'.', su.TableName, N'.', st.StatsName));

        -- CompatChanged, PlanForced / PlanUnforced
        INSERT rag.PlanChangeEvent (CaseID, EventType, EarliestUtc, LatestUtc, BeforeValue, AfterValue)
        SELECT c.CaseID, 'CompatChanged', c.ChangeUtc, c.ChangeUtc, CAST(c.OldCompat AS nvarchar(10)), CAST(c.NewCompat AS nvarchar(10))
        FROM rag.PlanChangeCase AS c JOIN #pair AS pr ON pr.CaseID = c.CaseID
        WHERE c.OldCompat <> c.NewCompat;
        INSERT rag.PlanChangeEvent (CaseID, EventType)
        SELECT c.CaseID, CASE WHEN c.NewIsForced = 1 THEN 'PlanForced' ELSE 'PlanUnforced' END
        FROM rag.PlanChangeCase AS c JOIN #pair AS pr ON pr.CaseID = c.CaseID
        WHERE c.NewIsForced = 1 OR (c.OldIsForced = 1 AND c.NewIsForced = 0);

        -- ModuleChanged: sorgunun modülü (ilk görülme değişim sayılmaz)
        ;WITH mc AS (
            SELECT w.CaseID, q.ObjectName, md.LastChangedUtc AS L,
                   COALESCE((SELECT MAX(l.StartUtc) FROM DBA.CollectionRunLog AS l
                             WHERE l.SourceType = 'METADATA' AND l.Status = 'OK' AND l.ServerID = w.ServerID AND l.DatabaseName = w.DatabaseName
                               AND l.EndUtc < md.LastChangedUtc), DATEADD(day, -1, md.LastChangedUtc)) AS E,
                   w.WinStart, w.ChangeUtc, w.ChangeEndUtc
            FROM #win AS w
            JOIN #pair AS pr ON pr.CaseID = w.CaseID
            JOIN perf.QdsQuery AS q ON q.ServerID = pr.ServerID AND q.DatabaseName = pr.DatabaseName AND q.query_id = pr.query_id
            JOIN perf.MetaModuleDefinition AS md ON md.ServerID = q.ServerID AND md.DatabaseName = q.DatabaseName AND md.object_id = q.object_id
            WHERE md.LastChangedUtc > md.FirstSeenUtc)
        INSERT rag.PlanChangeEvent (CaseID, EventType, EarliestUtc, LatestUtc, ObjectName, OrderUncertain)
        SELECT CaseID, 'ModuleChanged', E, L, ObjectName, CASE WHEN L > ChangeUtc THEN 1 ELSE 0 END
        FROM mc WHERE L >= WinStart AND E <= ChangeEndUtc;

        -- TableGrowth: eski planın başındaki (ya da ondan önceki en yakın) snapshot → değişim anından önceki en yakın snapshot.
        -- Değerler "satır|MB" (değişmez kültür)
        ;WITH tg AS (
            SELECT t.CaseID, t.SchemaName, t.TableName, b.SnapshotDateUtc AS BDate, b.[RowCount] AS BRows, b.ReservedMB AS BMb,
                   a.SnapshotDateUtc AS ADate, a.[RowCount] AS ARows, a.ReservedMB AS AMb
            FROM #ptab AS t
            JOIN #pair AS pr ON pr.CaseID = t.CaseID
            JOIN rag.PlanChangeMetric AS om ON om.CaseID = t.CaseID AND om.PlanRole = 'Old'
            CROSS APPLY (SELECT TOP (1) s.SnapshotDateUtc, s.[RowCount], s.ReservedMB FROM perf.MetaTableSize AS s
                         WHERE s.ServerID = t.ServerID AND s.DatabaseName = t.DatabaseName AND s.object_id = t.object_id
                         ORDER BY CASE WHEN s.SnapshotDateUtc <= om.FirstUtc THEN 0 ELSE 1 END,
                                  CASE WHEN s.SnapshotDateUtc <= om.FirstUtc THEN s.SnapshotDateUtc END DESC, s.SnapshotDateUtc) AS b
            CROSS APPLY (SELECT TOP (1) s.SnapshotDateUtc, s.[RowCount], s.ReservedMB FROM perf.MetaTableSize AS s
                         WHERE s.ServerID = t.ServerID AND s.DatabaseName = t.DatabaseName AND s.object_id = t.object_id
                           AND s.SnapshotDateUtc <= pr.ChangeUtc
                         ORDER BY s.SnapshotDateUtc DESC) AS a)
        INSERT rag.PlanChangeEvent (CaseID, EventType, EarliestUtc, LatestUtc, ObjectName, BeforeValue, AfterValue)
        SELECT CaseID, 'TableGrowth', BDate, ADate, CONCAT(SchemaName, N'.', TableName),
               CONCAT(BRows, N'|', CONVERT(nvarchar(30), BMb)), CONCAT(ARows, N'|', CONVERT(nvarchar(30), AMb))
        FROM tg WHERE ADate > BDate;

        COMMIT;
        SET @rows = (SELECT COUNT(*) FROM #pair);
        INSERT DBA.CollectionRunLog (StartUtc, EndUtc, SourceType, ServerID, DatabaseName, Status, RowsLoaded)
        VALUES (@start, DBA.fn_Now(), 'PLANCHANGE', ISNULL(@ServerID, 0), ISNULL(@DatabaseName, N'(tümü)'), 'OK', @rows);
    END TRY
    BEGIN CATCH
        IF @@TRANCOUNT > 0 ROLLBACK;
        SET @err = LEFT(CONCAT(N'[', ERROR_NUMBER(), N'] ', ERROR_MESSAGE()), 4000);
        INSERT DBA.CollectionRunLog (StartUtc, EndUtc, SourceType, ServerID, DatabaseName, Status, ErrorMessage)
        VALUES (@start, DBA.fn_Now(), 'PLANCHANGE', ISNULL(@ServerID, 0), ISNULL(@DatabaseName, N'(tümü)'), 'ERROR', @err);
        THROW;
    END CATCH
END;