如何构建SQL Server使用扫描器以获取历史查询使用信息？

针对你需要构建类似Snowflake QUERY_HISTORY的查询扫描器需求，以下是几种可行的持久化收集方案：

方案一：自定义捕获+持久化表 #

先创建存储历史数据的专用表，再通过定时任务抓取DMV数据存入表中：

1. 创建存储表

CREATE TABLE dbo.QueryHistory (
    QueryHistoryID BIGINT IDENTITY(1,1) PRIMARY KEY,
    SessionID INT,
    LoginName NVARCHAR(128),
    QueryText NVARCHAR(MAX),
    StartTime DATETIME,
    EndTime DATETIME,
    DurationMs BIGINT,
    CPUUsageMs BIGINT,
    Reads BIGINT,
    Writes BIGINT,
    Status NVARCHAR(30)
);

2. 编写捕获数据的存储过程

CREATE PROCEDURE dbo.CaptureQueryHistory
AS
BEGIN
    SET NOCOUNT ON;

    INSERT INTO dbo.QueryHistory (SessionID, LoginName, QueryText, StartTime, EndTime, DurationMs, CPUUsageMs, Reads, Writes, Status)
    SELECT
        s.session_id,
        s.login_name,
        t.text,
        r.start_time,
        COALESCE(r.end_time, GETDATE()),
        DATEDIFF(MILLISECOND, r.start_time, COALESCE(r.end_time, GETDATE())),
        r.cpu_time,
        r.reads,
        r.writes,
        r.status
    FROM sys.dm_exec_requests r
    JOIN sys.dm_exec_sessions s ON r.session_id = s.session_id
    CROSS APPLY sys.dm_exec_sql_text(r.sql_handle) t
    WHERE s.is_user_process = 1 -- 仅捕获用户进程，排除系统进程
    AND NOT EXISTS (
        SELECT 1 FROM dbo.QueryHistory qh WHERE qh.SessionID = r.session_id AND qh.StartTime = r.start_time
    ); -- 避免重复插入
END;

3. 配置SQL Server代理作业，设置每分钟执行一次该存储过程，实现持续数据捕获。

方案二：使用扩展事件（Extended Events） #

扩展事件是轻量级监控方案，性能开销远低于传统SQL Trace，可精准捕获查询细节：

1. 创建扩展事件会话

CREATE EVENT SESSION QueryCapture ON SERVER 
ADD EVENT sqlserver.sql_statement_completed(
    ACTION(sqlserver.session_id, sqlserver.login_name, sqlserver.sql_text)
    WHERE sqlserver.is_system = 0) -- 排除系统语句
ADD TARGET package0.event_file(SET filename=N'QueryCapture.xel', max_file_size=(100), max_rollover_files=(5))
WITH (STARTUP_STATE=ON); -- 服务重启后自动启动

2. 启动会话

ALTER EVENT SESSION QueryCapture ON SERVER STATE = START;

3. 创建视图解析事件文件数据

CREATE VIEW dbo.QueryHistoryExtended
AS
SELECT
    event_data.value('(event/@timestamp)[1]', 'DATETIME') AS EventTime,
    event_data.value('(event/action[@name="session_id"]/value)[1]', 'INT') AS SessionID,
    event_data.value('(event/action[@name="login_name"]/value)[1]', 'NVARCHAR(128)') AS LoginName,
    event_data.value('(event/action[@name="sql_text"]/value)[1]', 'NVARCHAR(MAX)') AS QueryText,
    event_data.value('(event/data[@name="duration"]/value)[1]', 'BIGINT')/1000 AS DurationMs,
    event_data.value('(event/data[@name="cpu_time"]/value)[1]', 'BIGINT')/1000 AS CPUUsageMs,
    event_data.value('(event/data[@name="logical_reads"]/value)[1]', 'BIGINT') AS Reads,
    event_data.value('(event/data[@name="writes"]/value)[1]', 'BIGINT') AS Writes
FROM (
    SELECT CONVERT(XML, event_data) AS event_data
    FROM sys.fn_xe_file_target_read_file('QueryCapture*.xel', NULL, NULL, NULL)
) AS x;

该方案可长期保存数据，且能灵活调整捕获的事件与字段。

方案三：启用SQL Server审计（SQL Server Audit） #

适合需要合规性审计记录的场景：

1. 创建服务器审计对象

CREATE SERVER AUDIT ServerQueryAudit
TO FILE (FILEPATH = N'C:\AuditLogs\', MAXSIZE = 100 MB, MAX_ROLLOVER_FILES = 5)
WITH (QUEUE_DELAY = 1000, ON_FAILURE = CONTINUE);

2. 创建数据库审计规范

CREATE DATABASE AUDIT SPECIFICATION QueryAuditSpec
FOR SERVER AUDIT ServerQueryAudit
ADD (EXECUTE ON DATABASE::[YourDatabase] BY [public])
WITH (STATE = ON);

3. 启动审计

ALTER SERVER AUDIT ServerQueryAudit WITH (STATE = ON);

4. 读取审计日志

SELECT
    event_time,
    session_id,
    server_principal_name AS LoginName,
    statement AS QueryText
FROM sys.fn_get_audit_file('C:\AuditLogs\ServerQueryAudit_*.sqlaudit', DEFAULT, DEFAULT);

此方案合规性强，但捕获的细节不如扩展事件丰富。

注意事项 #

• 性能优先级：扩展事件 < 自定义捕获 < SQL Trace（不推荐使用）
• 需定期清理历史数据，避免存储资源耗尽
• 针对长查询，可结合sys.dm_exec_input_buffer获取完整语句文本

内容的提问来源于stack exchange，提问作者Waqar Arshad