c# - 使用 SqlBulkCopy 时提供流作为二进制列的数据源

Question

如果需要以流方式从 SqlServer 中 读取数据，可以使用一些功能。例如将 SqlDataReader 与 CommandBehavior.SequentialAccess 结合使用，特别是当需要访问二进制列数据时，可以使用 GetStream(int) 方法:

var cmd = new SqlCommand();
cmd.Connection = connection;
cmd.CommandText = @"select 0x0123456789 as Data";

using (var dr = cmd.ExecuteReader(CommandBehavior.SequentialAccess))
{
    dr.Read();

    var stream = dr.GetStream(0);
    // access stream
}

但是当需要使用 SqlBulkCopy 将数据 提供给 SqlServer 时，特别是如果需要提供流作为源二进制列的数据？

我试着跟随

var cmd2 = new SqlCommand();
cmd2.Connection = connection;
cmd2.CommandText = @"create table #Test (ID int, Data varbinary(max))";
cmd2.ExecuteNonQuery();

using (SqlBulkCopy sbc = new SqlBulkCopy(connection, SqlBulkCopyOptions.TableLock, null))
{
    sbc.DestinationTableName = "#Test";
    sbc.EnableStreaming = true;

    sbc.ColumnMappings.Add(0, "ID");
    sbc.ColumnMappings.Add(1, "Data");

    sbc.WriteToServer(new TestDataReader());
}

其中TestDataReader实现IDataReader如下:

class TestDataReader : IDataReader
{
    public int FieldCount { get { return 2; } }
    int rowCount = 1;
    public bool Read() { return (rowCount++) < 3; }
    public bool IsDBNull(int i) { return false; }

    public object GetValue(int i)
    {
        switch (i)
        {
            case 0: return rowCount;
            case 1: return new byte[] { 0x01, 0x23, 0x45, 0x67, 0x89 };
            default: throw new Exception();
        }
    }

    //the rest members of IDataReader
}

它按预期工作。

然而改变

case 1: return new byte[] { 0x01, 0x23, 0x45, 0x67, 0x89 };

到

case 1: return new MemoryStream(new byte[] { 0x01, 0x23, 0x45, 0x67, 0x89 });

引发异常 System.InvalidOperationException 消息

The given value of type MemoryStream from the data source cannot be converted to type varbinary of the specified target column.

有没有办法将IDataReader(或者可能是DbDataReader)的Stream作为源提供给SqlBulkCopy二进制列的数据，而不先将其所有数据复制到内存(字节数组)中？

Answer 1

不确定这是否记录在任何地方，但如果对 SqlBulkCopy 进行简短检查查看源代码，您可能会发现它以不同的方式处理不同的数据读取器。一、SqlBulkCopy确实支持流媒体和GetStream ，但您可能会注意到 IDataReader接口(interface)不包含GetStream方法。因此，当您提供自定义 IDataReader 时实现到 SqlBulkCopy - 它不会将二进制列视为流式处理，也不会接受 Stream 的值类型。

另一方面 - DbDataReader 确实有这个方法。如果你喂 SqlBulkCopy DbDataReader 的实例- 继承类 - 它将以流方式处理所有二进制列，并将调用 DbDataReader.GetStream .

所以要解决您的问题 - 继承自 DbDataReader像这样:

class TestDataReader : DbDataReader
{
    public override bool IsDBNull(int ordinal) {
        return false;
    }

    public override int FieldCount { get; } = 2;
    int rowCount = 1;

    public override bool HasRows { get; } = true;
    public override bool IsClosed { get; } = false;

    public override bool Read()
    {
        return (rowCount++) < 3;
    }

    public override object GetValue(int ordinal) {
        switch (ordinal) {
            // do not return anything for binary column here - it will not be called
            case 0:
                return rowCount;
            default:
                throw new Exception();
        }
    }

    public override Stream GetStream(int ordinal) {
        // instead - return your stream here
        if (ordinal == 1)
            return new MemoryStream(new byte[] {0x01, 0x23, 0x45, 0x67, 0x89});
        throw new Exception();
    }
    // bunch of irrelevant stuff

}