SQL Server 存储过程 - 在执行插入选择时插入好的记录并捕获错误记录

这里根据您在示例中提供的一些列名称逐步解释我的意思。由于我没有你的表定义，我假设了一些东西，你必须根据你的需要调整它。

创建StagingTable. 我选择使用 GUID 以便能够对所有 3 个表 ( StagingTable_Id = FinalTable_Id = ErrorTable_Id)使用相同的 ID 。使用 GUID 时，请确保不要将其作为表中记录的物理排序顺序（聚集键）。我选择记录存储在StagingTable所有 3 个表中的时间点）。

CREATE TABLE dbo.StagingTable(
    StagingTable_Id UNIQUEIDENTIFIER ROWGUIDCOL NOT NULL,
    StagingTable_ImportDateUtc DATETIME2(7) NOT NULL,
    StagingTable_DateTime NVARCHAR(MAX) NULL,
    StagingTable_BillNo NVARCHAR(MAX) NULL,
    StagingTable_VendorId NVARCHAR(MAX) NULL,
    StagingTable_RepFirstName NVARCHAR(MAX) NULL,
    StagingTable_RepLastName NVARCHAR(MAX) NULL,
    StagingTable_LocationId NVARCHAR(MAX) NULL,
    CONSTRAINT PK_StagingTable PRIMARY KEY NONCLUSTERED (StagingTable_Id ASC) ON [PRIMARY]
) ON [PRIMARY];
ALTER TABLE dbo.StagingTable ADD CONSTRAINT DF_StagingTable_StagingTable_ID DEFAULT (NEWID()) FOR StagingTable_Id;
ALTER TABLE dbo.StagingTable ADD CONSTRAINT DF_StagingTable_StagingTable_ImportDateUtc DEFAULT (SYSUTCDATETIME()) FOR StagingTable_ImportDateUtc;
ALTER TABLE dbo.StagingTable SET (LOCK_ESCALATION = DISABLE);
CREATE CLUSTERED INDEX IX_StagingTable01 ON dbo.StagingTable (StagingTable_ImportDateUtc ASC) ON [PRIMARY];

  错误表与 StagingTable 具有相同的结构：

CREATE TABLE dbo.ErrorTable(
    ErrorTable_Id UNIQUEIDENTIFIER ROWGUIDCOL NOT NULL,
    ErrorTable_ImportDateUtc DATETIME2(7) NOT NULL,
    ErrorTable_DateTime NVARCHAR(MAX) NULL,
    ErrorTable_BillNo NVARCHAR(MAX) NULL,
    ErrorTable_VendorId NVARCHAR(MAX) NULL,
    ErrorTable_RepFirstName NVARCHAR(MAX) NULL,
    ErrorTable_RepLastName NVARCHAR(MAX) NULL,
    ErrorTable_LocationId NVARCHAR(MAX) NULL,
    CONSTRAINT PK_ErrorTable PRIMARY KEY NONCLUSTERED (ErrorTable_Id ASC) ON [PRIMARY]
) ON [PRIMARY];
ALTER TABLE dbo.ErrorTable ADD CONSTRAINT DF_ErrorTable_ErrorTable_ID DEFAULT (NEWID()) FOR ErrorTable_Id;
ALTER TABLE dbo.ErrorTable SET (LOCK_ESCALATION = DISABLE);
CREATE CLUSTERED INDEX IX_ErrorTable01 ON dbo.ErrorTable (ErrorTable_ImportDateUtc ASC) ON [PRIMARY];

但是最终表已经有了转换后的数据类型的结构：

CREATE TABLE dbo.FinalTable(
    FinalTable_Id UNIQUEIDENTIFIER ROWGUIDCOL NOT NULL,
    FinalTable_ImportDateUtc DATETIME2(7) NOT NULL,
    FinalTable_DateTime DATETIME2(7) NOT NULL,
    FinalTable_BillNo INT NOT NULL,
    FinalTable_VendorId INT NOT NULL,
    FinalTable_RepName NVARCHAR(60) NOT NULL,
    FinalTable_LocationId INT NOT NULL,
    CONSTRAINT PK_FinalTable PRIMARY KEY NONCLUSTERED (FinalTable_Id ASC) ON [PRIMARY]
) ON [PRIMARY];
ALTER TABLE dbo.FinalTable ADD CONSTRAINT DF_FinalTable_FinalTable_Id DEFAULT (NEWID()) FOR FinalTable_Id;
ALTER TABLE dbo.FinalTable SET (LOCK_ESCALATION = DISABLE);
CREATE CLUSTERED INDEX IX_FinalTable01 ON dbo.FinalTable (FinalTable_ImportDateUtc ASC) ON [PRIMARY];

接下来我们创建转换视图，它有两个目的：

• 它过滤视图以仅包含尚未处理的记录（其 id 既不在FinalTable也不在 中ErrorTable）。

• 它尝试将StagingTable-records 的每个字段转换为其目标格式。对于转换失败的字段，它分配NULL.

CREATE VIEW dbo.V_StagingConversion
AS
SELECT 
    st.StagingTable_Id, 
    st.StagingTable_ImportDateUtc, 
    st.StagingTable_DateTime, 
    TRY_CONVERT(datetime2, st.StagingTable_DateTime, 121) AS FinalTable_DateTime,
    st.StagingTable_BillNo,
    TRY_CAST(st.StagingTable_BillNo AS INT) AS FinalTable_BillNo,
    st.StagingTable_VendorId,
    TRY_CAST(st.StagingTable_VendorId AS INT) AS FinalTable_VendorId,
    st.StagingTable_RepFirstName, 
    st.StagingTable_RepLastName,
    RTRIM(LTRIM(ISNULL(st.StagingTable_RepFirstName, '') + N' ' + ISNULL(st.StagingTable_RepLastName, ''))) AS FinalTable_RepName, 
    st.StagingTable_LocationId,
    TRY_CAST(st.StagingTable_LocationId AS INT) AS FinalTable_LocationId
FROM StagingTable st
LEFT OUTER JOIN FinalTable ft ON st.StagingTable_Id = ft.FinalTable_Id
LEFT OUTER JOIN ErrorTable et ON st.StagingTable_Id = et.ErrorTable_Id
WHERE (ft.FinalTable_Id IS NULL) AND (et.ErrorTable_Id IS NULL);

然后我们创建一个基于前一个视图并决定记录是否有效的视图：

CREATE VIEW dbo.V_StagingValidation
AS
SELECT 
    *, 
    CAST(CASE WHEN FinalTable_DateTime IS NULL THEN 0 ELSE 
            CASE WHEN FinalTable_BillNo IS NULL THEN 0 ELSE 
                CASE WHEN FinalTable_VendorId IS NULL THEN 0 ELSE 
                    CASE WHEN (LEN(FinalTable_RepName) = 0) OR (LEN(FinalTable_RepName) > 60) THEN 0 ELSE 
                        CASE WHEN FinalTable_LocationId IS NULL THEN 0 ELSE 1 END 
                    END 
                END 
            END 
         END AS BIT) AS StagingTable_IsValid 
FROM dbo.V_StagingConversion;

接下来，我们创建表示未来状态并用于相应INSERT查询的视图：

CREATE VIEW dbo.V_StagingPublicationFinal
AS
SELECT StagingTable_Id AS FinalTable_Id, StagingTable_ImportDateUtc AS FinalTable_ImportDateUtc, FinalTable_DateTime, FinalTable_BillNo, FinalTable_VendorId, FinalTable_RepName, FinalTable_LocationId 
FROM dbo.V_StagingValidation WHERE StagingTable_IsValid = 1;

和

CREATE VIEW dbo.V_StagingPublicationError
AS
SELECT StagingTable_Id AS ErrorTable_Id, StagingTable_ImportDateUtc AS ErrorTable_ImportDateUtc, StagingTable_DateTime AS ErrorTable_DateTime, StagingTable_BillNo AS ErrorTable_BillNo, StagingTable_VendorId AS ErrorTable_VendorId, StagingTable_RepFirstName AS ErrorTable_RepFirstName, StagingTable_RepLastName AS ErrorTable_RepLastName, StagingTable_LocationId AS ErrorTable_LocationId 
FROM dbo.V_StagingValidation WHERE StagingTable_IsValid = 0;

以及显示DELETE查询所需的所有已处理记录的视图（如果您不想StagingTable在插入和处理后立即删除记录，请取消注释存储过程中的相应语句）。

CREATE VIEW dbo.V_StagingPublicationDone
AS
SELECT st.*, CAST(CASE WHEN ft.FinalTable_Id IS NULL THEN 0 ELSE 1 END AS BIT) AS StagingTable_IsValid FROM StagingTable st
LEFT OUTER JOIN FinalTable ft ON st.StagingTable_Id = ft.FinalTable_Id
LEFT OUTER JOIN ErrorTable et ON st.StagingTable_Id = et.ErrorTable_Id
WHERE (ft.FinalTable_Id IS NOT NULL) OR (et.ErrorTable_Id IS NOT NULL);

然后我决定将导入逻辑放入一个存储过程中，这样可以灵活地手动执行它，例如，如果触发器被禁用，因为必须尝试一些东西或调整一些东西。

CREATE PROCEDURE dbo.ImportStagingTable
AS
BEGIN
    SET NOCOUNT ON;
    INSERT INTO dbo.FinalTable 
      SELECT FinalTable_Id, FinalTable_ImportDateUtc, FinalTable_DateTime, FinalTable_BillNo, FinalTable_VendorId, FinalTable_RepName, FinalTable_LocationId FROM V_StagingPublicationFinal p ORDER BY p.FinalTable_ImportDateUtc;
    INSERT INTO dbo.ErrorTable 
      SELECT ErrorTable_Id, ErrorTable_ImportDateUtc, ErrorTable_DateTime, ErrorTable_BillNo, ErrorTable_VendorId, ErrorTable_RepFirstName, ErrorTable_RepLastName, ErrorTable_LocationId FROM V_StagingPublicationError p ORDER BY p.ErrorTable_ImportDateUtc;
    DELETE st FROM dbo.V_StagingPublicationDone spd INNER JOIN dbo.StagingTable st ON spd.StagingTable_Id = st.StagingTable_Id;
END

每个人都应该能够执行这个存储过程：

GRANT EXECUTE ON dbo.ImportStagingTable TO PUBLIC;

最后我们创建触发器。我添加了一个 COMMIT/BEGIN TRANSACTION 以确保在 after 触发器失败的情况下不会回滚已经写入的 StagingTable 记录：

CREATE TRIGGER dbo.StagingTableAfterInsert ON dbo.StagingTable AFTER INSERT
AS 
BEGIN
    SET NOCOUNT ON;
    COMMIT TRANSACTION;
    BEGIN TRANSACTION;
    EXECUTE dbo.ImportStagingTable;
END

并启用它（但它可能已经启用）：

ALTER TABLE dbo.StagingTable ENABLE TRIGGER StagingTableAfterInsert

现在您可以将记录插入到 中StagingTable，它们会自动分发到FinalTable或 中ErrorTable。