DBA.usp_Pull_Metadata

CREATE   PROCEDURE DBA.usp_Pull_Metadata
    @ServerID int, @DBLinkName sysname, @DatabaseName sysname
AS
BEGIN
    SET NOCOUNT ON;
    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));
    -- #1191: tarih metne gömülmez, @Arg1 (yyyymmdd) parametresiyle gider; gömülü sabit her gün yeni QDS query_id doğuruyordu
    DECLARE @snapDay bigint = CAST(CONVERT(char(8), CAST(DBA.fn_Now() AS date), 112) AS bigint);
    -- R05-3: kaynak saat dilimi (@Arg3) biliniyorsa tarihler satır bazında AT TIME ZONE ile çevrilir (DST doğru);
    -- bilinmiyorsa anlık offset (DST geçişinden önceki değerlerde 1 saat sapabilir).
    DECLARE @hdr nvarchar(400) = N'SET NOCOUNT ON; DECLARE @Arg1 bigint = ?, @Arg2 bigint = ?, @Arg3 nvarchar(max) = ?; '
                               + N'DECLARE @off int = DATEDIFF(minute, SYSDATETIME(), SYSUTCDATETIME()), @tz nvarchar(128) = NULLIF(@Arg3, N''''), '
                               + N'@snap datetime = CONVERT(datetime, CONVERT(char(8), @Arg1), 112); ';
    DECLARE @tz nvarchar(128) = NULLIF(DBA.fn_GetSettingNvarchar(N'Collect.SourceTimeZone', N''), N'');
    IF @tz IS NULL
    BEGIN
        CREATE TABLE #tz (tz nvarchar(128) NULL);
        EXEC DBA.usp_RemoteInsert @DBLinkName, N'#tz', N'tz',
            N'DECLARE @Arg1 bigint = ?, @Arg2 bigint = ?, @Arg3 nvarchar(max) = ?;
              IF CAST(SERVERPROPERTY(''ProductMajorVersion'') AS int) >= 16 EXEC (N''SELECT CURRENT_TIMEZONE_ID()'');
              ELSE SELECT CAST(NULL AS nvarchar(128));';
        SET @tz = (SELECT TOP (1) tz FROM #tz);
    END

    DECLARE @q TABLE (seq int, Target sysname, Cols nvarchar(max), Body nvarchar(max));
    INSERT @q VALUES
    (1, N'perf.Stg_MetaObject', N'object_id, SchemaName, ObjectName, ObjectType, CreateDateUtc, ModifyDateUtc',
     N'o.object_id, SCHEMA_NAME(o.schema_id), o.name, o.type, CASE WHEN @tz IS NULL THEN DATEADD(minute, @off, o.create_date) ELSE CONVERT(datetime, o.create_date AT TIME ZONE @tz AT TIME ZONE N''UTC'') END, CASE WHEN @tz IS NULL THEN DATEADD(minute, @off, o.modify_date) ELSE CONVERT(datetime, o.modify_date AT TIME ZONE @tz AT TIME ZONE N''UTC'') END
       FROM sys.objects AS o WHERE o.is_ms_shipped = 0 AND o.type IN (''U'',''V'',''P'',''FN'',''IF'',''TF'',''TR'')'),
    (2, N'perf.Stg_MetaColumn', N'object_id, column_id, ColumnName, TypeName, MaxLength, [Precision], Scale, IsNullable, IsIdentity, IsComputed, ComputedDefinition, CollationName, IsSparse',
     N'c.object_id, c.column_id, c.name, TYPE_NAME(c.user_type_id), c.max_length, c.precision, c.scale, c.is_nullable, c.is_identity, c.is_computed,
       cc.definition, c.collation_name, c.is_sparse
       FROM sys.columns AS c JOIN sys.objects AS o ON o.object_id = c.object_id
       LEFT JOIN sys.computed_columns AS cc ON cc.object_id = c.object_id AND cc.column_id = c.column_id
       WHERE o.is_ms_shipped = 0 AND o.type IN (''U'',''V'')'),
    (3, N'perf.Stg_MetaIndex', N'object_id, index_id, IndexName, IndexType, TypeDesc, IsUnique, IsPrimaryKey, IsUniqueConstraint, HasFilter, FilterDefinition, [FillFactor], IsDisabled, DataCompression, PartitionCount',
     N'i.object_id, i.index_id, i.name, i.type, i.type_desc, i.is_unique, i.is_primary_key, i.is_unique_constraint, i.has_filter, i.filter_definition,
       i.fill_factor, i.is_disabled,
       (SELECT TOP (1) p.data_compression_desc FROM sys.partitions AS p WHERE p.object_id = i.object_id AND p.index_id = i.index_id ORDER BY p.partition_number),
       (SELECT COUNT(*) FROM sys.partitions AS p WHERE p.object_id = i.object_id AND p.index_id = i.index_id)
       FROM sys.indexes AS i JOIN sys.objects AS o ON o.object_id = i.object_id
       WHERE o.is_ms_shipped = 0 AND o.type IN (''U'',''V'') AND i.is_hypothetical = 0'),
    (4, N'perf.Stg_MetaIndexColumn', N'object_id, index_id, index_column_id, column_id, key_ordinal, is_descending_key, is_included_column, partition_ordinal',
     N'ic.object_id, ic.index_id, ic.index_column_id, ic.column_id, ic.key_ordinal, ic.is_descending_key, ic.is_included_column, ic.partition_ordinal
       FROM sys.index_columns AS ic JOIN sys.objects AS o ON o.object_id = ic.object_id
       JOIN sys.indexes AS i ON i.object_id = ic.object_id AND i.index_id = ic.index_id
       WHERE o.is_ms_shipped = 0 AND o.type IN (''U'',''V'') AND i.is_hypothetical = 0'),
    (5, N'perf.Stg_MetaStatistics', N'object_id, stats_id, StatsName, AutoCreated, UserCreated, NoRecompute, HasFilter, FilterDefinition, LastUpdatedUtc, [Rows], RowsSampled, ModificationCounter',
     N's.object_id, s.stats_id, s.name, s.auto_created, s.user_created, s.no_recompute, s.has_filter, s.filter_definition,
       CASE WHEN @tz IS NULL THEN DATEADD(minute, @off, sp.last_updated) ELSE CONVERT(datetime, sp.last_updated AT TIME ZONE @tz AT TIME ZONE N''UTC'') END, sp.rows, sp.rows_sampled, sp.modification_counter
       FROM sys.stats AS s JOIN sys.objects AS o ON o.object_id = s.object_id
       OUTER APPLY sys.dm_db_stats_properties(s.object_id, s.stats_id) AS sp
       WHERE o.is_ms_shipped = 0 AND o.type = ''U'''),
    (6, N'perf.Stg_MetaStatisticsColumn', N'object_id, stats_id, stats_column_id, column_id',
     N'sc.object_id, sc.stats_id, sc.stats_column_id, sc.column_id
       FROM sys.stats_columns AS sc JOIN sys.objects AS o ON o.object_id = sc.object_id
       WHERE o.is_ms_shipped = 0 AND o.type = ''U'''),
    (7, N'perf.Stg_MetaModuleDefinition', N'object_id, Definition',
     N'm.object_id, m.definition FROM sys.sql_modules AS m JOIN sys.objects AS o ON o.object_id = m.object_id WHERE o.is_ms_shipped = 0'),
    (8, N'perf.Stg_MetaForeignKey', N'fk_object_id, FkName, ParentObjectId, ReferencedObjectId, IsDisabled, IsNotTrusted, DeleteAction, UpdateAction',
     N'fk.object_id, fk.name, fk.parent_object_id, fk.referenced_object_id, fk.is_disabled, fk.is_not_trusted,
       fk.delete_referential_action_desc, fk.update_referential_action_desc
       FROM sys.foreign_keys AS fk WHERE fk.is_ms_shipped = 0'),
    (9, N'perf.Stg_MetaForeignKeyColumn', N'fk_object_id, constraint_column_id, parent_column_id, referenced_column_id',
     N'fkc.constraint_object_id, fkc.constraint_column_id, fkc.parent_column_id, fkc.referenced_column_id FROM sys.foreign_key_columns AS fkc'),
    (10, N'perf.Stg_MetaTableSize', N'object_id, SnapshotDateUtc, [RowCount], ReservedMB, DataMB, IndexMB, UnusedMB, PartitionCount',
     N'ps.object_id, @snap,
       SUM(CASE WHEN ps.index_id IN (0,1) THEN ps.row_count ELSE 0 END),
       SUM(ps.reserved_page_count) * 8 / 1024.0,
       SUM(CASE WHEN ps.index_id IN (0,1) THEN ps.in_row_data_page_count + ps.lob_used_page_count + ps.row_overflow_used_page_count ELSE 0 END) * 8 / 1024.0,
       (SUM(ps.used_page_count) - SUM(CASE WHEN ps.index_id IN (0,1) THEN ps.in_row_data_page_count + ps.lob_used_page_count + ps.row_overflow_used_page_count ELSE 0 END)) * 8 / 1024.0,
       (SUM(ps.reserved_page_count) - SUM(ps.used_page_count)) * 8 / 1024.0,
       COUNT(DISTINCT ps.partition_number)
       FROM sys.dm_db_partition_stats AS ps JOIN sys.objects AS o ON o.object_id = ps.object_id
       WHERE o.is_ms_shipped = 0 AND o.type = ''U'' GROUP BY ps.object_id'),
    -- #1191: index başına iki ilişkili OUTER APPLY sys.sysrowsets üzerinde eager index spool kuruyordu (DB-10 2,6 sn);
    -- önceden gruplanmış LEFT JOIN aynı sonucu 0,5 sn'de verir
    (11, N'perf.Stg_MetaIndexUsage', N'object_id, index_id, SnapshotDateUtc, user_seeks, user_scans, user_lookups, user_updates, LastUserSeekUtc, LastUserScanUtc, LastUserLookupUtc, LastUserUpdateUtc, SqlServerStartUtc,
          IndexRowCount, IndexSizeMB, LeafInserts, LeafUpdates, LeafDeletes, LeafAllocations, RowLockWaitMs, PageLatchWaitMs, PageIoLatchWaitMs',
     N'i.object_id, i.index_id, @snap, ISNULL(u.user_seeks, 0), ISNULL(u.user_scans, 0), ISNULL(u.user_lookups, 0), ISNULL(u.user_updates, 0),
       CASE WHEN @tz IS NULL THEN DATEADD(minute, @off, u.last_user_seek) ELSE CONVERT(datetime, u.last_user_seek AT TIME ZONE @tz AT TIME ZONE N''UTC'') END, CASE WHEN @tz IS NULL THEN DATEADD(minute, @off, u.last_user_scan) ELSE CONVERT(datetime, u.last_user_scan AT TIME ZONE @tz AT TIME ZONE N''UTC'') END, CASE WHEN @tz IS NULL THEN DATEADD(minute, @off, u.last_user_lookup) ELSE CONVERT(datetime, u.last_user_lookup AT TIME ZONE @tz AT TIME ZONE N''UTC'') END, CASE WHEN @tz IS NULL THEN DATEADD(minute, @off, u.last_user_update) ELSE CONVERT(datetime, u.last_user_update AT TIME ZONE @tz AT TIME ZONE N''UTC'') END,
       (SELECT CASE WHEN @tz IS NULL THEN DATEADD(minute, @off, sqlserver_start_time) ELSE CONVERT(datetime, sqlserver_start_time AT TIME ZONE @tz AT TIME ZONE N''UTC'') END FROM sys.dm_os_sys_info),
       sz.RowCnt, sz.SizeMB, op.Ins, op.Upd, op.Del, op.Alloc, op.LockMs, op.LatchMs, op.IoMs
       FROM sys.indexes AS i JOIN sys.objects AS o ON o.object_id = i.object_id
       LEFT JOIN sys.dm_db_index_usage_stats AS u ON u.database_id = DB_ID() AND u.object_id = i.object_id AND u.index_id = i.index_id
       LEFT JOIN (SELECT ps.object_id, ps.index_id, SUM(ps.row_count) AS RowCnt, CAST(SUM(ps.used_page_count) * 8 / 1024.0 AS decimal(18,2)) AS SizeMB
                  FROM sys.dm_db_partition_stats AS ps GROUP BY ps.object_id, ps.index_id) AS sz ON sz.object_id = i.object_id AND sz.index_id = i.index_id
       LEFT JOIN (SELECT x.object_id, x.index_id, SUM(x.leaf_insert_count) AS Ins, SUM(x.leaf_update_count) AS Upd, SUM(x.leaf_delete_count) AS Del,
                         SUM(x.leaf_allocation_count) AS Alloc, SUM(x.row_lock_wait_in_ms) AS LockMs, SUM(x.page_latch_wait_in_ms) AS LatchMs,
                         SUM(x.page_io_latch_wait_in_ms) AS IoMs
                  FROM sys.dm_db_index_operational_stats(DB_ID(), NULL, NULL, NULL) AS x GROUP BY x.object_id, x.index_id) AS op ON op.object_id = i.object_id AND op.index_id = i.index_id
       WHERE o.is_ms_shipped = 0 AND o.type IN (''U'',''V'') AND i.is_hypothetical = 0'),
    (12, N'perf.Stg_MetaMissingIndex', N'index_handle, SnapshotDateUtc, object_id, EqualityColumns, InequalityColumns, IncludedColumns, UserSeeks, UserScans, AvgTotalUserCost, AvgUserImpact, LastUserSeekUtc',
     N'd.index_handle, @snap, d.object_id, d.equality_columns, d.inequality_columns, d.included_columns,
       s.user_seeks, s.user_scans, s.avg_total_user_cost, s.avg_user_impact, CASE WHEN @tz IS NULL THEN DATEADD(minute, @off, s.last_user_seek) ELSE CONVERT(datetime, s.last_user_seek AT TIME ZONE @tz AT TIME ZONE N''UTC'') END
       FROM sys.dm_db_missing_index_details AS d
       JOIN sys.dm_db_missing_index_groups AS g ON g.index_handle = d.index_handle
       JOIN sys.dm_db_missing_index_group_stats AS s ON s.group_handle = g.index_group_handle
       WHERE d.database_id = DB_ID()'),
    -- Partition: RANGE RIGHT/LEFT fark etmeksizin p. partition'ın alt sınırı (p-1). boundary'dir
    (13, N'perf.Stg_MetaPartition', N'object_id, index_id, SchemeName, FunctionName, PartitionColumn, BoundaryType, BoundaryOnRight, PartitionCount, NonEmptyPartitions, FirstBoundary, LastBoundary, FirstDataPartition, LastDataPartition, DataLowerBoundary, LastPartitionRows, MaxPartitionRows',
     N'i.object_id, i.index_id, ds.name, pf.name,
       (SELECT c.name FROM sys.index_columns AS ic JOIN sys.columns AS c ON c.object_id = ic.object_id AND c.column_id = ic.column_id
         WHERE ic.object_id = i.object_id AND ic.index_id = i.index_id AND ic.partition_ordinal = 1),
       (SELECT TYPE_NAME(pp.user_type_id) FROM sys.partition_parameters AS pp WHERE pp.function_id = pf.function_id AND pp.parameter_id = 1),
       pf.boundary_value_on_right, pf.fanout, st.NonEmpty,
       (SELECT CASE WHEN CAST(SQL_VARIANT_PROPERTY(v.value, ''BaseType'') AS sysname) IN (''date'', ''datetime'', ''datetime2'', ''smalldatetime'', ''datetimeoffset'') THEN CONVERT(nvarchar(100), CAST(v.value AS datetime2), 121) ELSE CAST(v.value AS nvarchar(100)) END FROM sys.partition_range_values AS v WHERE v.function_id = pf.function_id AND v.boundary_id = 1),
       (SELECT CASE WHEN CAST(SQL_VARIANT_PROPERTY(v.value, ''BaseType'') AS sysname) IN (''date'', ''datetime'', ''datetime2'', ''smalldatetime'', ''datetimeoffset'') THEN CONVERT(nvarchar(100), CAST(v.value AS datetime2), 121) ELSE CAST(v.value AS nvarchar(100)) END FROM sys.partition_range_values AS v WHERE v.function_id = pf.function_id AND v.boundary_id = pf.fanout - 1),
       st.FirstData, st.LastData,
       (SELECT CASE WHEN CAST(SQL_VARIANT_PROPERTY(v.value, ''BaseType'') AS sysname) IN (''date'', ''datetime'', ''datetime2'', ''smalldatetime'', ''datetimeoffset'') THEN CONVERT(nvarchar(100), CAST(v.value AS datetime2), 121) ELSE CAST(v.value AS nvarchar(100)) END FROM sys.partition_range_values AS v WHERE v.function_id = pf.function_id AND v.boundary_id = st.FirstData - 1),
       (SELECT ps.row_count FROM sys.dm_db_partition_stats AS ps WHERE ps.object_id = i.object_id AND ps.index_id = i.index_id AND ps.partition_number = pf.fanout),
       st.MaxRows
       FROM sys.indexes AS i JOIN sys.objects AS o ON o.object_id = i.object_id
       JOIN sys.partition_schemes AS ds ON ds.data_space_id = i.data_space_id
       JOIN sys.partition_functions AS pf ON pf.function_id = ds.function_id
       OUTER APPLY (SELECT SUM(CASE WHEN ps.row_count > 0 THEN 1 ELSE 0 END) AS NonEmpty,
                           MIN(CASE WHEN ps.row_count > 0 THEN ps.partition_number END) AS FirstData,
                           MAX(CASE WHEN ps.row_count > 0 THEN ps.partition_number END) AS LastData,
                           MAX(ps.row_count) AS MaxRows
                    FROM sys.dm_db_partition_stats AS ps WHERE ps.object_id = i.object_id AND ps.index_id = i.index_id) AS st
       WHERE o.is_ms_shipped = 0 AND o.type IN (''U'',''V'') AND i.is_hypothetical = 0'),
    -- Nesne düzeyi bağımlılık (kolon düzeyi referencing_minor_id gruplanır); DB'ler arası/linked server referansları adla gelir
    (14, N'perf.Stg_MetaDependency', N'referencing_id, ReferencedServer, ReferencedDatabase, ReferencedSchema, ReferencedEntity, referenced_id, ReferencedClass, IsAmbiguous, IsCallerDependent',
     N'd.referencing_id, d.referenced_server_name, d.referenced_database_name, d.referenced_schema_name, d.referenced_entity_name,
       d.referenced_id, d.referenced_class, MAX(CAST(d.is_ambiguous AS tinyint)), MAX(CAST(d.is_caller_dependent AS tinyint))
       FROM sys.sql_expression_dependencies AS d JOIN sys.objects AS o ON o.object_id = d.referencing_id
       WHERE d.referencing_class = 1 AND o.is_ms_shipped = 0 AND d.referenced_entity_name IS NOT NULL
       GROUP BY d.referencing_id, d.referenced_server_name, d.referenced_database_name, d.referenced_schema_name, d.referenced_entity_name,
                d.referenced_id, d.referenced_class');

    -- K1-3: histogram (sys.dm_db_stats_histogram) SQL 2017+; eski sürümde metadata toplamasının geri kalanı düşmesin
    DECLARE @ver int;
    EXEC DBA.usp_GetRemoteVersion @DBLinkName, @ver OUTPUT;
    IF @ver >= 14
        INSERT @q VALUES
        (15, N'perf.Stg_MetaStatisticsHistogram', N'object_id, stats_id, step_number, RangeHighKey, RangeRows, EqualRows, DistinctRangeRows, AverageRangeRows',
         N's.object_id, s.stats_id, h.step_number,
           CASE WHEN CAST(SQL_VARIANT_PROPERTY(h.range_high_key, ''BaseType'') AS sysname) IN (''date'', ''datetime'', ''datetime2'', ''smalldatetime'', ''datetimeoffset'')
                THEN CONVERT(nvarchar(400), CAST(h.range_high_key AS datetime2), 121)
                WHEN CAST(SQL_VARIANT_PROPERTY(h.range_high_key, ''BaseType'') AS sysname) IN (''binary'', ''varbinary'')
                THEN CONVERT(nvarchar(400), CAST(h.range_high_key AS varbinary(200)), 1)
                ELSE LEFT(CONVERT(nvarchar(4000), h.range_high_key), 400) END,
           h.range_rows, h.equal_rows, h.distinct_range_rows, h.average_range_rows
           FROM sys.stats AS s JOIN sys.objects AS o ON o.object_id = s.object_id
           CROSS APPLY sys.dm_db_stats_histogram(s.object_id, s.stats_id) AS h
           WHERE o.is_ms_shipped = 0 AND o.type = ''U''');

    DECLARE @target sysname, @cols nvarchar(max), @body nvarchar(max), @remote nvarchar(max), @del nvarchar(max), @allCols nvarchar(max);
    DECLARE c CURSOR LOCAL FAST_FORWARD FOR SELECT Target, Cols, Body FROM @q ORDER BY seq;
    OPEN c; FETCH NEXT FROM c INTO @target, @cols, @body;
    WHILE @@FETCH_STATUS = 0
    BEGIN
        SET @del = N'DELETE ' + @target + N' WHERE ServerID = @s AND DatabaseName = @d;';
        EXEC sys.sp_executesql @del, N'@s int, @d sysname', @s = @ServerID, @d = @DatabaseName;
        SET @remote = @hdr + @use + N'SELECT ' + @sid + N', ' + @lit + N', ' + @body + N';';
        SET @allCols = N'ServerID, DatabaseName, ' + @cols;
        EXEC DBA.usp_RemoteInsert @DBLinkName, @target, @allCols, @remote, @snapDay, NULL, @tz;
        FETCH NEXT FROM c INTO @target, @cols, @body;
    END
    CLOSE c; DEALLOCATE c;
END;