使用从并行线程运行的创建索引、表、过程的 SQL 查询

Question

我遇到了从并行线程执行的查询的问题。 几乎总是我得到关于该索引的错误 |表类型 |程序 |表已经存在。

我很确定我的 sql 查询受到保护，不会出现该错误，并且它永远不会出现。

此查询应该能够从并发线程运行而不会出现任何错误。

我在做什么：

1. 我有数百个文件，每个文件包含几百万行。

2. 每一行都包含我需要插入到特定表中的对象。

3. 如果表/索引/过程不存在，我需要创建它。

4. 目前它工作得很好，除了这个不应该出现的错误。

我的查询代码（创建索引）：

public void CreateIndex(string tableName)
{
  try
  {
    StringBuilder sb = new StringBuilder();

    var connectionString = ConnectionString;
    using (SqlConnection conn = new SqlConnection(connectionString))
    {
        conn.Open();

        // Creating INDEX
        sb.AppendFormat(@" IF NOT EXISTS(SELECT * FROM sys.indexes 
                                         WHERE name='TimeSeries_DateStamp_{0}' AND object_id = OBJECT_ID('{0}'))
                            exec(' CREATE UNIQUE CLUSTERED INDEX TimeSeries_DateStamp_{0} ON [dbo].[{0}]
                                              (
                                                [TimeSeriesID] ASC,
                                                [DateStamp]    ASC
                                              ) WITH (
                                              PAD_INDEX = OFF, 
                                              STATISTICS_NORECOMPUTE = OFF, 
                                              SORT_IN_TEMPDB = OFF, 
                                              IGNORE_DUP_KEY = OFF, 
                                              DROP_EXISTING = OFF, 
                                              ONLINE = OFF,
                                              ALLOW_ROW_LOCKS = ON, 
                                              ALLOW_PAGE_LOCKS = ON) 
                                              ON [PRIMARY] '); ", tableName);

        using (SqlCommand cmd = new SqlCommand(sb.ToString().Trim(), conn))
        {
            cmd.ExecuteNonQuery();
        }

        conn.Close();
    }

}
catch (Exception ex)
{
    Log.Error(string.Format("Error occured while adding index: {0}", tableName), ex);
}

}

我已经标记了 (...) 代码，其中实际上没有什么重要的事情发生 - 只是一些数据处理。

DB 是实体框架 6 上下文，也包含此 CreateIndex 方法。

Parallel.ForEach(timeSeriesFiles, file =>
{
    (...)

    // FOREACH LINE 
    foreach (string line in File.ReadLines(file))
    {
        (...)
        using (var db = Context.DB)
        {
            db.CreateIndex(tableName(line));
        }
        (...)
    }
}

我将不胜感激任何建议或线索为什么会发生这种情况。

Answer 1

您收到此错误是因为至少有 2 个线程同时尝试创建索引/存储过程/表。我在您的 SQL 中看不到任何可以防止此错误的内容。我建议实施某种锁定机制来处理此错误。您可以在 SQL 或代码中执行此操作 - 无论哪种方式，逻辑基本相同：

1. 检查索引是否存在。如果索引不存在，则获取锁。在 SQL 中类似于：SELECT 1 FROM tableName WITH (TABLOCKX)

2. 一旦锁定被授予，检查索引是否不存在（是的，您需要检查两次。也许另一个线程在您等待被授予锁定时创建了索引）。

3. 解除锁定

Answer 2

使用事务。 例如修改你的脚本：

sb.AppendFormat(@" 
begin tran
IF NOT EXISTS(SELECT * FROM sys.indexes 
    WHERE name='TimeSeries_DateStamp_{0}' AND object_id = OBJECT_ID('{0}'))
    exec(' CREATE UNIQUE CLUSTERED INDEX TimeSeries_DateStamp_{0} ON [dbo].[{0}]
        (
        [TimeSeriesID] ASC,
        [DateStamp]    ASC
        ) WITH (
        PAD_INDEX = OFF, 
        STATISTICS_NORECOMPUTE = OFF, 
        SORT_IN_TEMPDB = OFF, 
        IGNORE_DUP_KEY = OFF, 
        DROP_EXISTING = OFF, 
        ONLINE = OFF,
        ALLOW_ROW_LOCKS = ON, 
        ALLOW_PAGE_LOCKS = ON) 
        ON [PRIMARY] '); 
commit", tableName);

或者你可以使用Entity Framework Transactions

Answer 3

感谢 SteveR 和 Maksim Eroshkin，我终于取得了成就 是：

创建新存储过程的代码（我遇到了最大的问题）

sb.AppendFormat(@"BEGIN TRAN
IF NOT EXISTS ( SELECT * 
                            FROM sys.procedures AS T 
                            WITH (TABLOCKX)
                            INNER JOIN sys.schemas AS S  ON T.schema_id = S.schema_id
                            WHERE S.Name = 'dbo' AND T.Name = 'BulkMerge{0}' ) 
BEGIN
SELECT 1 FROM sys.procedures WITH (TABLOCKX)
IF NOT EXISTS ( SELECT * 
                            FROM sys.procedures AS T 
                            WITH (TABLOCKX)
                            INNER JOIN sys.schemas AS S  ON T.schema_id = S.schema_id
                            WHERE S.Name = 'dbo' AND T.Name = 'BulkMerge{0}' ) 
BEGIN TRY
          EXEC(' 
            CREATE PROCEDURE [dbo].[BulkMerge{0}]
                @table [Bulk{0}] READONLY
            AS
            BEGIN
                SET NOCOUNT ON;

                MERGE INTO {0} t1
                USING @table t2
                ON t1.[TimeSeriesID] = t2.[TimeSeriesID] AND t1.[DateStamp] = t2.[DateStamp]
                WHEN MATCHED 
                THEN UPDATE SET {1} 
                WHEN NOT MATCHED 
                THEN INSERT VALUES (t2.TimeSeriesID, t2.DateStamp {2} ); 
            END
            ')
END TRY BEGIN CATCH END CATCH
END 
COMMIT ", tableName, fieldsSB, insertSB);

重要的是添加锁以及存储过程 BEGIN TRY 和 BEGIN CATCH。 这是我找到的最好方法。