如何实现接口 IDataReader 以便在插入之前处理数据？

Question

我有一个存储过程，它给我一个结果集，该结果集由一个单个列组成，其中包含数百万个未处理的行。我需要使用 SqlBulkCopy 将这些数据传输到另一台服务器，但问题是我不能简单地执行以下操作：

using (var con = new SqlConnection(sqlConnectionStringSource))
{
    using (var cmd = new SqlCommand("usp_GetUnprocessedData", con))
    {
        cmd.CommandType = CommandType.StoredProcedure;
        con.Open();
        using (var reader = cmd.ExecuteReader())
        {
            using (var sqlBulk = new SqlBulkCopy(sqlConnectionStringDestination))
            {
                sqlBulk.DestinationTableName = "BulkCopy";
                sqlBulk.BulkCopyTimeout = 0;
                sqlBulk.BatchSize = 200000;
                sqlBulk.WriteToServer(reader);
            }
        }
    }
}

因为根本不会处理数据。

在我的例子中，结果集的第 n 行如下所示：

value1_n,value2_n,value3_n

其中 n 只是我引入的下标，用于区分各个行。

在我命名为 BulkCopy 的目的地 table 中，我想要：

╔══════════╦══════════╦══════════╗
║  Field1  ║  Field2  ║  Field3  ║
╠══════════╬══════════╬══════════╣
║ Value1_1 ║ Value2_1 ║ Value3_1 ║
║ Value1_2 ║ Value2_2 ║ Value3_2 ║
║ ...      ║ ...      ║ ...      ║
║ Value1_n ║ Value2_n ║ Value3_n ║
╚══════════╩══════════╩══════════╝

有人告诉我通过 IDataReader 接口的实现使用自定义 DataReader，以便在 SqlBulkCopy 从中复制数据之前逐行处理数据，使用 EnableStreamingProperty = true 确保内存中只有少量数据，但我不知道从哪里开始。你能帮帮我吗？

Answer 1

我找到了以下代码项目：https://www.codeproject.com/script/Articles/ViewDownloads.aspx?aid=1095790。看起来您必须获取 csv 数据并拆分为对象。我用下面的代码修改了代码项目。有很多类型没有实现，您可能需要实现一些额外的方法。也不确定结果值应该是什么类型。

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Data;
using System.Data.SqlClient;



namespace ConsoleApplication108
{
    class Program
    {
        static void Main(string[] args)
        {

        }
    }
    public class MyDataReader : IDataReader 
    {
        private SqlConnection conn { get; set; }
        private SqlCommand cmd { get; set; }
        private SqlDataReader reader { get; set; }
        private DataTable schemaTable { get; set; }

        private string data { get; set; }
        private object[] arrayData { get; set; }
        private IEnumerator<object> m_dataEnumerator { get; set; }


        public MyDataReader(string commandText, string connectionString, List<KeyValuePair<string, Type>> columns)
        {
            conn = new SqlConnection(connectionString);
            conn.Open();
            cmd = new SqlCommand(commandText, conn);
            reader = cmd.ExecuteReader();

            schemaTable = new DataTable();
            foreach(KeyValuePair<string,Type> col in columns)
            {
                schemaTable.Columns.Add(col.Key, col.Value);
            }
        }
        public Boolean NextResult()
        {
            return reader.Read();
        }
        public int RecordsAffected
        {
            get { return -1; }
        }
        public int Depth
        {
            get { return -1; }
        }
        public void Dispose()
        {
            Dispose(true);
            GC.SuppressFinalize(this);
        }
        private void Dispose(bool disposing)
        {
            if (disposing)
            {
                if (m_dataEnumerator != null)
                {
                    m_dataEnumerator.Dispose();
                    m_dataEnumerator = null;
                }
            }
        }

        public Boolean IsClosed {
            get { return reader.IsClosed; }
        }
        public Boolean Read()
        {

            if (IsClosed)
            {
                throw new ObjectDisposedException(GetType().Name);
            }
            else
            {
                arrayData = reader.GetString(0).Split(new char[] { ',' }).ToArray();
            }
            return m_dataEnumerator.MoveNext();

        }
        public DataTable GetSchemaTable()
        {
            return schemaTable;
        }
        public void Close()
        {
            Dispose();
        }


        public object this[string name]
        {
            get { throw new NotImplementedException(); }

        }

        public object this[int i]
        {
            get { return arrayData[i]; }
        }
        public int FieldCount
        {
            get { return arrayData.Length; }
        }
        public bool IsDBNull(int i)
        {
              throw new NotImplementedException();
        }
        public bool GetBoolean(int i)
        {
            throw new NotImplementedException();
        }

        public byte GetByte(int i)
        {
            throw new NotImplementedException();
        }

        public long GetBytes(int i, long fieldOffset, byte[] buffer, int bufferoffset, int length)
        {
            throw new NotImplementedException();
        }

        public char GetChar(int i)
        {
            throw new NotImplementedException();
        }

        public long GetChars(int i, long fieldoffset, char[] buffer, int bufferoffset, int length)
        {
            throw new NotImplementedException();
        }

        public IDataReader GetData(int i)
        {
            throw new NotImplementedException();
        }

        public string GetDataTypeName(int i)
        {
            throw new NotImplementedException();
        }

        public DateTime GetDateTime(int i)
        {
            throw new NotImplementedException();
        }

        public decimal GetDecimal(int i)
        {
            throw new NotImplementedException();
        }

        public double GetDouble(int i)
        {
            throw new NotImplementedException();
        }

        public Type GetFieldType(int i)
        {
            throw new NotImplementedException();
        }

        public float GetFloat(int i)
        {
            throw new NotImplementedException();
        }

        public Guid GetGuid(int i)
        {
            throw new NotImplementedException();
        }

        public short GetInt16(int i)
        {
            throw new NotImplementedException();
        }

        public int GetInt32(int i)
        {
            throw new NotImplementedException();
        }

        public long GetInt64(int i)
        {
            throw new NotImplementedException();
        }

        public string GetName(int i)
        {
            throw new NotImplementedException();
        }

        public string GetString(int i)
        {
            throw new NotImplementedException();
        }

        public int GetValues(object[] values)
        {
            values = arrayData;

            return arrayData.Length;
        }
        public int GetOrdinal(string name)
        {
            throw new NotImplementedException();
        }

        public object GetValue(int i)
        {
            return arrayData[i];
        }



    }
}

Answer 2

让我们把问题反过来。不要寻找通用的解决方案，而是针对这个问题创建一个特定的解决方案。花了几天时间创建 IDataReader 包装器后，我知道这不是一件小事。

我们知道有多少字段，我们不关心结果中的任何其他字段。我们可以创建一个迭代器方法来以流式方式将数据和 return 记录一个接一个地拆分，而不是尝试正确地实现 IDataReader 包装器。 FastMember's ObjectReader 可以在任何 IEnumerable 上包装一个 IDataReader 接口：

class MyDTO
{
    public string Field1{get;set;}
    public string Field2{get;set;}
    public string Field3{get;set;}
}

public IEnumerable<MyDTO> ReaderToStream(IDataReader reader)
{
    while(reader.Read())
    {
        var line=reader.GetString(0);
        var fields=String.Split(",",line);
        yield return new MyDTO{Field1=fields[0];Field2=fields[1];Field3=fields[2]};
    }
}

导入方式可以改为：

using (var con = new SqlConnection(sqlConnectionStringSource))
{
    ...
    using (var reader = cmd.ExecuteReader())
    {
        var recordStream=ReaderToStream(reader);
        using(var rd=ObjectReader(recordStream))
        using (var sqlBulk = new SqlBulkCopy(sqlConnectionStringDestination))
        {
            ...
            sqlBulk.WriteToServer(rd);
        }
    }
}

迭代器仅在 SqlBulkCopy 请求新记录时调用 Read()，因此我们不会最终将所有内容加载到内存中。

和 IDataReader 包装器

Resharper 和 Visual Studio 2019 提供通过将调用委托给包装的 class 来实现接口。在 Visual Studio 2019 年，这被称为 Implement interface through 'field_name'。

从这段代码开始：

class ReaderWrapper:IDataReader
{
    private readonly IDataReader _inner ;
    public ReaderWrapper(IDataReader inner)
    {
        _inner = inner;
    }
}

应用重构得到：

class ReaderWrapper:IDataReader
{
    private readonly IDataReader _inner ;
    public ReaderWrapper(IDataReader inner)
    {
        _inner = inner;
    }

    public object this[int i] => _inner[i];

    public object this[string name] => _inner[name];

    public int Depth => _inner.Depth;

    public bool IsClosed => _inner.IsClosed;

    public int RecordsAffected => _inner.RecordsAffected;

    public int FieldCount => _inner.FieldCount;

    public void Close() => _inner.Close();
    public void Dispose() => _inner.Dispose();
    public bool GetBoolean(int i) => _inner.GetBoolean(i);
    public byte GetByte(int i) => _inner.GetByte(i);
    public long GetBytes(int i, long fieldOffset, byte[] buffer, int bufferoffset, int length) => _inner.GetBytes(i, fieldOffset, buffer, bufferoffset, length);
    public char GetChar(int i) => _inner.GetChar(i);
    public long GetChars(int i, long fieldoffset, char[] buffer, int bufferoffset, int length) => _inner.GetChars(i, fieldoffset, buffer, bufferoffset, length);
    public IDataReader GetData(int i) => _inner.GetData(i);
    public string GetDataTypeName(int i) => _inner.GetDataTypeName(i);
    public DateTime GetDateTime(int i) => _inner.GetDateTime(i);
    public decimal GetDecimal(int i) => _inner.GetDecimal(i);
    public double GetDouble(int i) => _inner.GetDouble(i);
    public Type GetFieldType(int i) => _inner.GetFieldType(i);
    public float GetFloat(int i) => _inner.GetFloat(i);
    public Guid GetGuid(int i) => _inner.GetGuid(i);
    public short GetInt16(int i) => _inner.GetInt16(i);
    public int GetInt32(int i) => _inner.GetInt32(i);
    public long GetInt64(int i) => _inner.GetInt64(i);
    public string GetName(int i) => _inner.GetName(i);
    public int GetOrdinal(string name) => _inner.GetOrdinal(name);
    public DataTable GetSchemaTable() => _inner.GetSchemaTable();
    public string GetString(int i) => _inner.GetString(i);
    public object GetValue(int i) => _inner.GetValue(i);
    public int GetValues(object[] values) => _inner.GetValues(values);
    public bool IsDBNull(int i) => _inner.IsDBNull(i);
    public bool NextResult() => _inner.NextResult();
    public bool Read() => _inner.Read();
}

要创建拆分包装器，我们需要将 Read() 替换为我们自己的版本：

    private string[] _values;

    public bool Read()
    {
        var ok = _inner.Read();
        if (ok)
        {
            //It *could be null*
            if (_inner.IsDBNull(0))
            {
                //What to do? Store an empty array for now
                _values = new string[0];
            }
            var fieldValue = _inner.GetString(0);                
            _values= fieldValue.Split(',');
        }
        return ok;
    }

这会拆分 CSV 值并将它们存储在一个字符串中。这说明了为什么实现包装器有点麻烦 - 我们需要处理很多事情并决定在意外情况（如 null、空字符串等）下做什么。

之后，我们需要为SqlBulkCopy调用的方法添加自己的实现。 GetValue() 肯定被调用，FieldCount 也是如此。其他成员根据列映射类型按名称或序号调用。

public int FieldCount => _values.Length;

public string GetString(int ordinal) => _values[ordinal];

public object GetValue(int ordinal)=> _values[ordinal];

//What if we have more values than expected?
public int GetValues(object[] values)
{
    if (_values.Length > 0)
    {
        Array.Copy(_values, values,_values.Length);
        return _values.Length;
    }
    return 0;
}

现在是 "funny" 部分。 GetName() 呢？大概：

public string GetName(int ordinal) => $"Field{ordinal}";

GetOrdinal ？它可以在名称映射中调用。变得棘手:

public int GetOrdinal(string name) => int.Parse(name.Substring(5));

希望这能奏效。

我们还需要覆盖索引：

    public object this[string name] => _values[GetOrdinal(name)];

    public object this[int i] => _values[i];

我忘记了什么？ ...仍然需要处理任意值的数字。需要处理空值。没有 GetSchemaTable 这可能意味着必须明确指定列映射，可能按序数指定。

快速且简单的 IsDbNull 实现可以是：

public bool IsDBNull(int i)
{  
    //Covers the "null" case too, when `Length` is 0
    if (i>_values.Length-1)
    {
        return true;
    }
    return _inner.IsDBNull(i);
}

GetSchemaTable 更难，因为我们真的不知道每个记录中有多少个值。 table 有 20 多列，所以我宁愿 而不是 编写该代码，直到我发现需要它为止。

public DataTable GetSchemaTable() => throw new NotImplementedException();

Leave it as an excercise to the reader 正如他们所说

PPS：默认接口实现，因为为什么不

所有这一切都可能是一个不错的情况，如果复杂的话，可以使用 C# 8 的默认接口方法来创建包装的 reader 特征。默认情况下，遵循包装内部 reader。这将消除实现中的所有延迟调用。

interface IReaderWrapper:IDataReader
{
    //Gives access to the wrapped reader in the concrete classes
    abstract IDataReader Inner();

    override object this[int i] => Inner()[i];

    override object this[string name] => Inner()[name];

    override int Depth => Inner().Depth;

    override bool IsClosed => Inner().IsClosed;
    ...
}

class SplitterWrapper:IReaderWrapper
{

    private readonly IDataReader _inner ;
    public SplitterWrapper(IDataReader inner)
    {
        _inner = inner;
    }

    IDataReader Inner()=> _inner;

    string[] _values;
    public object this[int i] => _values[i];
    ...
}

此功能在 VS 2019 附带的 C#8 编译器中不起作用，并且会以某种方式崩溃 Sharplab.io。不知道它是否会编译或者是否真的需要覆盖。

如何实现接口 IDataReader 以便在插入之前处理数据？

How to implement the interface IDataReader in order to process the data before being insert?

c#

sqldatareader

sqlbulkcopy

idatareader