CLR 表值函数如何“流式传输”？答案

【问题标题】：How is a CLR table valued function 'streaming''?CLR 表值函数如何“流式传输”？
【发布时间】：2011-08-04 23:13:39
【问题描述】：

MSDN Docs on table-valued Sql Clr functions 声明：

Transact-SQL 表值函数实现调用的结果函数到一个中间表。 ... 相比之下，CLR 表值函数代表一个流选择。没有要求整个结果集是在单个表中实现。这 IEnumerable 对象返回的托管函数直接调用查询的执行计划调用表值函数，并且结果被消耗在一个增量方式。 ...这也是一个如果你有更好的选择返回大量行，因为他们不必整体在记忆中具体化。

Then I find out that no data access is allowed in the 'Fill row' method。这意味着您仍然必须在 init 方法中完成所有数据访问并将其保存在内存中，等待调用“填充行”。我误解了什么吗？如果我不将结果强制放入数组或列表中，我会收到错误消息：'ExecuteReader 需要一个打开且可用的连接。连接的当前状态为关闭。'

代码示例：

[<SqlFunction(DataAccess = DataAccessKind.Read, FillRowMethodName = "Example8Row")>]
static member InitExample8() : System.Collections.IEnumerable = 
   let c = cn() // opens a context connection
   // I'd like to avoid forcing enumeration here:
   let data = getData c |> Array.ofSeq
   data :> System.Collections.IEnumerable

static member Example8Row ((obj : Object),(ssn: SqlChars byref)) = 
   do ssn <- new SqlChars(new SqlString(obj :?> string))
   ()

我在这里处理几百万行。有没有办法懒惰地做到这一点？

【问题讨论】：

我阅读文档的方式似乎暗示使用表值函数生成整个结果集并将其放在某个地方 - 如果它很小，可能是内存，否则是 tempdb - 在其结果返回到客户。使用 CLR，只要有一些记录可用，结果集就可以开始直接从内存缓冲区返回给客户端。我不知道这是否是您需要明确担心的事情。我认为 MSDN 只是在解释这两种表函数的内部工作原理。除非我误解了这篇文章。
我相信 yield return 在 C# 中工作。我希望seq { } 能够类似地工作。没有？
@Daniel - 这就是我正在尝试的。我想删除 '|> Array.ofSeq' 并改用 yield return ，但这会导致我出现这个错误。这就是这个问题的意义所在。让步时，在 Example8Row 函数中执行数据访问，这似乎是不允许的。
@Yuck and Robert：是的，“流式传输”部分是指在方法结束之前使用yield return 将结果发回的能力。这是您可以做的一个选项，或者您可以建立集合并在方法结束时将其全部发回。对于 T-SQL 多语句 TVF，别无选择。他们所能做的就是将结果建立在一个只能在函数结束时返回的表变量中。

标签： sql-server f# sqlclr

【解决方案1】：

我假设您使用的是 SQL Server 2008。正如一位 Microsoft 员工在 this page 上提到的，2008 年需要使用 DataAccessKind 标记的方法。比 2005 年更频繁地读取。其中一个是 TVF参与交易（当我测试时似乎总是如此）。解决方案是在连接字符串中指定enlist=false，可惜不能与context connection=true 组合。这意味着您的连接字符串需要采用典型的客户端格式：Data Source=.;Initial Catalog=MyDb;Integrated Security=sspi;Enlist=false，并且您的程序集必须至少使用permission_set=external_access 创建。以下作品：

using System;
using System.Collections;
using System.Data.SqlClient;
using System.Data.SqlTypes;
using Microsoft.SqlServer.Server;

namespace SqlClrTest {
    public static class Test {
        [SqlFunction(
            DataAccess = DataAccessKind.Read,
            SystemDataAccess = SystemDataAccessKind.Read,
            TableDefinition = "RowNumber int",
            FillRowMethodName = "FillRow"
            )]
        public static IEnumerable MyTest(SqlInt32 databaseID) {
            using (var con = new SqlConnection("data source=.;initial catalog=TEST;integrated security=sspi;enlist=false")) {
                con.Open();
                using (var cmd = new SqlCommand("select top (100) RowNumber from SSP1 where DatabaseID = @DatabaseID", con)) {
                    cmd.Parameters.AddWithValue("@DatabaseID", databaseID.IsNull ? (object)DBNull.Value : databaseID.Value);
                    using (var reader = cmd.ExecuteReader()) {
                        while (reader.Read())
                            yield return reader.GetInt32(0);
                    }
                }
            }
        }
        public static void FillRow(object obj, out SqlInt32 rowNumber) {
            rowNumber = (int)obj;
        }
    }
}

在 F# 中也是这样：

namespace SqlClrTest

module Test =

    open System
    open System.Data
    open System.Data.SqlClient
    open System.Data.SqlTypes
    open Microsoft.SqlServer.Server

    [<SqlFunction(
        DataAccess = DataAccessKind.Read,
        SystemDataAccess = SystemDataAccessKind.Read,
        TableDefinition = "RowNumber int",
        FillRowMethodName = "FillRow"
        )>]
    let MyTest (databaseID:SqlInt32) =
        seq {
            use con = new SqlConnection("data source=.;initial catalog=TEST;integrated security=sspi;enlist=false")
            con.Open()
            use cmd = new SqlCommand("select top (100) RowNumber from SSP1 where DatabaseID = @DatabaseID", con)
            cmd.Parameters.AddWithValue("@DatabaseID", if databaseID.IsNull then box DBNull.Value else box databaseID.Value) |> ignore
            use reader = cmd.ExecuteReader()
            while reader.Read() do
                yield reader.GetInt32(0)
        } :> System.Collections.IEnumerable

    let FillRow (obj:obj) (rowNumber:SqlInt32 byref) =
        rowNumber <- SqlInt32(unbox obj)

好消息是：Microsoft considers this a bug。

【讨论】：

感谢 Daniel，我在某处读到过关于能够使用不是上下文连接的连接来完成此操作，但认为它太混乱了。不幸的是，Clr 程序集需要知道如何连接到它所在的数据库。:)
顺便说一句，您不需要使用 --standalone 进行编译。事实上，如果您使用来自其他程序集的相同程序集，这可能会导致问题，FSharp.Core 中的类型本质上是复制的，并且不再与常规程序集使用的 FSharp.Core 版本兼容。我只是在 Sql Server 中注册 FSharp.Core + 我需要的任何其他东西。
@Robert：有道理。我没有想到这一点。我从我的答案中删除了它。
“Enlist=true”对我来说很新鲜。现在我可以使用收益回报了，太棒了！谢谢！
@BrankoDimitrijevic "Enlist=false;" 的效果是外部连接不是分布式事务的一部分。它是完全独立的，就像所有其他会话一样。因此，它可以看到尚未提交的更改的唯一方法是执行脏读（WITH (NOLOCK) 或 SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED）。

【解决方案2】：

是的，您需要将结果拉入内存，然后从那里返回。尽管这样做的目的是避免您进行此类操作。

您可以在您链接到的 MSDN 文档的某个部分中看到该方法的示例（“示例：返回 SQL 查询的结果”）

这些示例有点做作，尽管电子邮件验证的实际实现将使用标量而不是表函数 - 为每个输入电子邮件值返回一个布尔值，而不是无效的列表。

你能解释一下你想要达到的目标吗？可能有更好的方法来构造函数。

【讨论】：

-1 无需将所有内容都拉入内存。这个问题比这更微妙。

【解决方案3】：

您可以做的是用一个 IEnumerable 包装一个 SqlDataReader 类，该 IEnumerable 使用一个枚举器，当调用它的“Next”方法时，在 SqlDataReader 上执行 MoveNext 并返回 SqlDataReader。然后，您的 FillRow 方法需要 SqlDataReader 作为一个类。如果您的枚举器在无法再“下一步”时关闭数据库连接和 SqlDataReader，那么您已经有效地将输出流式传输到 FillRows 函数。您也可以使用 ContextConnection=true 来执行此操作...

...这里的问题是您必须能够返回实际查询的结果：如果您正在做更复杂的事情来创建结果集，那么您就不走运了。

【讨论】：