DBA.usp_Pull_Qds

CREATE   PROCEDURE DBA.usp_Pull_Qds
    @ServerID int, @DBLinkName sysname, @DatabaseName sysname,
    @LastIntervalId bigint, @MajorVersion int, @State nvarchar(60) OUTPUT, @Reset bit = 0 OUTPUT
AS
BEGIN
    SET NOCOUNT ON;
    DELETE perf.Stg_QdsRuntimeStats WHERE ServerID = @ServerID AND DatabaseName = @DatabaseName;
    DELETE perf.Stg_QdsWaitStats    WHERE ServerID = @ServerID AND DatabaseName = @DatabaseName;
    DELETE perf.Stg_QdsPlan         WHERE ServerID = @ServerID AND DatabaseName = @DatabaseName;
    DELETE perf.Stg_QdsPlanXml      WHERE ServerID = @ServerID AND DatabaseName = @DatabaseName;
    DELETE perf.Stg_QdsQuery        WHERE ServerID = @ServerID AND DatabaseName = @DatabaseName;
    DELETE perf.Stg_QdsQueryText    WHERE ServerID = @ServerID AND DatabaseName = @DatabaseName;

    -- DB adı: USE için QUOTENAME, çıktı kolonu için kaçışlı N'' literal. Asla ? parametresi değil.
    DECLARE @use nvarchar(400) = N'USE ' + QUOTENAME(@DatabaseName) + N'; ';
    DECLARE @lit nvarchar(400) = N'N''' + REPLACE(@DatabaseName, N'''', N'''''') + N'''';
    DECLARE @sid nvarchar(12) = CAST(@ServerID AS nvarchar(12));
    DECLARE @hdr nvarchar(200) = N'SET NOCOUNT ON; DECLARE @Arg1 bigint = ?, @Arg2 bigint = ?, @Arg3 nvarchar(max) = ?; ';
    DECLARE @ids nvarchar(400) = N'DECLARE @ids TABLE (id bigint PRIMARY KEY); INSERT @ids SELECT DISTINCT x.i.value(''.'', ''bigint'') FROM (SELECT CAST(N''<i>'' + REPLACE(@Arg3, N'','', N''</i><i>'') + N''</i>'' AS xml) AS d) AS s CROSS APPLY s.d.nodes(''/i'') AS x(i); ';
    DECLARE @remote nvarchar(max), @max bigint, @list nvarchar(max);

    -- 0) Durum ve ayarlar (K1-7): tamamı perf.Stg_QdsOptions'a; usp_Collect_Qds QDS kapalı olsa da saklar
    DELETE perf.Stg_QdsOptions WHERE ServerID = @ServerID AND DatabaseName = @DatabaseName;
    SET @remote = @hdr + @use + N'
        SELECT ' + @sid + N', ' + @lit + N', actual_state_desc, desired_state_desc, readonly_reason, query_capture_mode_desc,
               interval_length_minutes, max_storage_size_mb, current_storage_size_mb, stale_query_threshold_days, size_based_cleanup_mode_desc, '
        + CASE WHEN @MajorVersion >= 14 THEN N'wait_stats_capture_mode_desc' ELSE N'CAST(NULL AS nvarchar(60))' END + N'
        FROM sys.database_query_store_options;';
    EXEC DBA.usp_RemoteInsert @DBLinkName, N'perf.Stg_QdsOptions',
        N'ServerID, DatabaseName, ActualState, DesiredState, ReadonlyReason, CaptureMode, IntervalMinutes, MaxStorageMb, CurrentStorageMb,
          StaleQueryThresholdDays, SizeCleanupMode, WaitStatsCaptureMode', @remote;
    SET @State = (SELECT TOP (1) ActualState FROM perf.Stg_QdsOptions WHERE ServerID = @ServerID AND DatabaseName = @DatabaseName);
    SET @Reset = 0;
    IF @State IS NULL OR @State IN (N'OFF', N'ERROR') RETURN;

    -- R04-1: QDS CLEAR sonrası id'ler 1'den başlar; uzak max id watermark'tan küçükse sıfırlanmıştır.
    CREATE TABLE #mx (m bigint);
    SET @remote = @hdr + @use + N'SELECT ISNULL(MAX(runtime_stats_interval_id), 0) FROM sys.query_store_runtime_stats_interval;';
    EXEC DBA.usp_RemoteInsert @DBLinkName, N'#mx', N'm', @remote;
    IF @LastIntervalId > 0 AND (SELECT TOP (1) m FROM #mx) < @LastIntervalId
    BEGIN
        SET @Reset = 1;
        SET @LastIntervalId = 0;
    END

    -- 1) Kapanmış interval'lerin runtime istatistikleri
    SET @remote = @hdr + @use + N'
        SELECT ' + @sid + N', ' + @lit + N', rs.plan_id, rs.runtime_stats_interval_id, rs.execution_type,
               CONVERT(datetime, SWITCHOFFSET(i.start_time, 0)), CONVERT(datetime, SWITCHOFFSET(i.end_time, 0)),
               rs.count_executions, rs.avg_duration, rs.max_duration, rs.stdev_duration, rs.avg_cpu_time, rs.max_cpu_time,
               rs.avg_logical_io_reads, rs.max_logical_io_reads, rs.avg_physical_io_reads, rs.avg_logical_io_writes,
               rs.avg_query_max_used_memory, rs.avg_rowcount, rs.avg_dop, '
        + CASE WHEN @MajorVersion >= 14 THEN N'rs.avg_tempdb_space_used' ELSE N'CAST(NULL AS float)' END + N', rs.max_rowcount
        -- Kapanmis ama diske yazilmamis intervalde gorunum ayni satiri (ayni runtime_stats_id) bellekten ve diskten iki kez
        -- dondurebilir (2026-09-28 Nereus: 11 kopya, PK ihlali); anahtar basina en cok calismayi iceren satir alinir.
        FROM (SELECT x.*, ROW_NUMBER() OVER (PARTITION BY x.plan_id, x.runtime_stats_interval_id, x.execution_type
                                             ORDER BY x.count_executions DESC, x.last_execution_time DESC) AS rn
              FROM sys.query_store_runtime_stats AS x WHERE x.runtime_stats_interval_id > @Arg1) AS rs
        JOIN sys.query_store_runtime_stats_interval AS i ON i.runtime_stats_interval_id = rs.runtime_stats_interval_id
        WHERE rs.rn = 1 AND i.end_time < SYSDATETIMEOFFSET();';
    EXEC DBA.usp_RemoteInsert @DBLinkName, N'perf.Stg_QdsRuntimeStats',
        N'ServerID, DatabaseName, plan_id, runtime_stats_interval_id, execution_type, IntervalStartUtc, IntervalEndUtc,
          count_executions, avg_duration, max_duration, stdev_duration, avg_cpu_time, max_cpu_time,
          avg_logical_io_reads, max_logical_io_reads, avg_physical_io_reads, avg_logical_io_writes,
          avg_query_max_used_memory, avg_rowcount, avg_dop, avg_tempdb_space_used, max_rowcount',
        @remote, @LastIntervalId;

    SET @max = (SELECT MAX(runtime_stats_interval_id) FROM perf.Stg_QdsRuntimeStats WHERE ServerID = @ServerID AND DatabaseName = @DatabaseName);
    IF @max IS NULL RETURN;

    -- 2) Wait istatistikleri (2017+)
    IF @MajorVersion >= 14
    BEGIN
        SET @remote = @hdr + @use + N'
            SELECT ' + @sid + N', ' + @lit + N', w.plan_id, w.runtime_stats_interval_id, w.execution_type, w.wait_category,
                   CONVERT(datetime, SWITCHOFFSET(i.start_time, 0)), w.wait_category_desc, w.total_query_wait_time_ms, w.avg_query_wait_time_ms
            FROM (SELECT x.*, ROW_NUMBER() OVER (PARTITION BY x.plan_id, x.runtime_stats_interval_id, x.execution_type, x.wait_category
                                                 ORDER BY x.total_query_wait_time_ms DESC) AS rn
                  FROM sys.query_store_wait_stats AS x WHERE x.runtime_stats_interval_id > @Arg1 AND x.runtime_stats_interval_id <= @Arg2) AS w
            JOIN sys.query_store_runtime_stats_interval AS i ON i.runtime_stats_interval_id = w.runtime_stats_interval_id
            WHERE w.rn = 1;';
        EXEC DBA.usp_RemoteInsert @DBLinkName, N'perf.Stg_QdsWaitStats',
            N'ServerID, DatabaseName, plan_id, runtime_stats_interval_id, execution_type, wait_category, IntervalStartUtc, wait_category_desc, total_query_wait_time_ms, avg_query_wait_time_ms',
            @remote, @LastIntervalId, @max;
    END

    -- 3) Penceredeki planların metadata'sı (XML'siz)
    -- Zorlama durumu çalışmadan değişir: zorlanan plan NO_PLAN ile düşüyorsa bir daha hiç çalışmaz ve yalnız penceredeki
    -- planlar okunsaydı merkezde "zorlanmamış" kalırdı (ölçüldü 2026-09-29: DB-9 plan 521, sahada doğrulama
    -- yanlış NotFound verdi). Bu yüzden zorlanmış planlar ve pencerede çalışan sorguların TÜM planları da okunur;
    -- ikincisi zorlamanın kaldırılmasını (eski plan artık çalışmıyor) yakalar.
    SET @remote = @hdr + @use + N'
        SELECT ' + @sid + N', ' + @lit + N', p.plan_id, p.query_id, p.query_plan_hash, p.is_forced_plan, p.compatibility_level,
               CONVERT(datetime, SWITCHOFFSET(p.last_compile_start_time, 0)), CONVERT(datetime, SWITCHOFFSET(p.last_execution_time, 0)),
               p.force_failure_count, p.last_force_failure_reason_desc
        FROM sys.query_store_plan AS p
        WHERE p.is_forced_plan = 1
           OR p.query_id IN (SELECT pp.query_id FROM sys.query_store_plan AS pp
                             JOIN sys.query_store_runtime_stats AS rs ON rs.plan_id = pp.plan_id
                             WHERE rs.runtime_stats_interval_id > @Arg1 AND rs.runtime_stats_interval_id <= @Arg2);';
    EXEC DBA.usp_RemoteInsert @DBLinkName, N'perf.Stg_QdsPlan',
        N'ServerID, DatabaseName, plan_id, query_id, query_plan_hash, is_forced_plan, compatibility_level, LastCompileUtc, LastExecutionUtc, force_failure_count, last_force_failure_reason_desc',
        @remote, @LastIntervalId, @max;

    -- 4) Merkezde XML'i olmayan planların XML'i
    -- R04-3: çalıştırma başına en fazla 2000 id; kalanlar sonraki çalıştırmalarda gelir. K1: CPU'su yüksek plan önce (fn_QdsPlanXmlCandidates)
    SELECT @list = STRING_AGG(CAST(s.plan_id AS nvarchar(max)), N',')
    FROM (SELECT TOP (2000) c.plan_id FROM DBA.fn_QdsPlanXmlCandidates(@ServerID, @DatabaseName, @Reset) AS c
          ORDER BY c.CpuUs DESC, c.plan_id DESC) AS s;
    IF @list IS NOT NULL
    BEGIN
        SET @remote = @hdr + @ids + @use + N'
            SELECT ' + @sid + N', ' + @lit + N', p.plan_id, p.query_plan FROM sys.query_store_plan AS p JOIN @ids AS x ON x.id = p.plan_id;';
        EXEC DBA.usp_RemoteInsert @DBLinkName, N'perf.Stg_QdsPlanXml', N'ServerID, DatabaseName, plan_id, PlanXml', @remote, NULL, NULL, @list;
    END

    -- 5) Bu planların sorguları
    SET @remote = @hdr + @use + N'
        SELECT ' + @sid + N', ' + @lit + N', q.query_id, q.query_hash, q.query_text_id, q.object_id,
               CASE WHEN q.object_id > 0 THEN OBJECT_SCHEMA_NAME(q.object_id) + N''.'' + OBJECT_NAME(q.object_id) END,
               q.context_settings_id, CONVERT(datetime, SWITCHOFFSET(q.last_execution_time, 0))
        FROM sys.query_store_query AS q
        WHERE q.query_id IN (SELECT p.query_id FROM sys.query_store_plan AS p
                             WHERE p.plan_id IN (SELECT rs.plan_id FROM sys.query_store_runtime_stats AS rs
                                                 WHERE rs.runtime_stats_interval_id > @Arg1 AND rs.runtime_stats_interval_id <= @Arg2));';
    EXEC DBA.usp_RemoteInsert @DBLinkName, N'perf.Stg_QdsQuery',
        N'ServerID, DatabaseName, query_id, query_hash, query_text_id, object_id, ObjectName, context_settings_id, LastExecutionUtc',
        @remote, @LastIntervalId, @max;

    -- 6) Merkezde metni olmayan sorguların metni
    SET @list = NULL;
    SELECT @list = STRING_AGG(CAST(s.query_text_id AS nvarchar(max)), N',')
    FROM (SELECT DISTINCT TOP (2000) query_text_id FROM perf.Stg_QdsQuery AS g   -- R04-3: üst sınır
          WHERE g.ServerID = @ServerID AND g.DatabaseName = @DatabaseName
            AND (@Reset = 1 OR NOT EXISTS (SELECT 1 FROM perf.QdsQuery AS q WHERE q.ServerID = g.ServerID AND q.DatabaseName = g.DatabaseName
                            AND q.query_text_id = g.query_text_id AND q.TextHash IS NOT NULL))
          ORDER BY query_text_id DESC) AS s;
    IF @list IS NOT NULL
    BEGIN
        SET @remote = @hdr + @ids + @use + N'
            SELECT ' + @sid + N', ' + @lit + N', t.query_text_id, t.query_sql_text FROM sys.query_store_query_text AS t JOIN @ids AS x ON x.id = t.query_text_id;';
        EXEC DBA.usp_RemoteInsert @DBLinkName, N'perf.Stg_QdsQueryText', N'ServerID, DatabaseName, query_text_id, query_sql_text', @remote, NULL, NULL, @list;
    END
END;