求U-SQL中对应HASHBYTES的整行哈希函数及行变更追踪方案

Hey there! Let me walk you through how to handle row change tracking (including storing all change records) and get the latest row versions using U-SQL, with an equivalent to SQL Server's HASHBYTES for row-level hashing.

1. U-SQL中替代SQL Server HASHBYTES的哈希函数 #

U-SQL没有和HASHBYTES完全一致的函数，但有两个实用选项可以实现相同的行哈希效果：

选项1：通用哈希HASH函数 #

HASH函数可以快速对单个或多个值生成哈希，适合做行级对比。要计算整行哈希，直接把所有字段作为参数传入即可：

HASH(ROW(Col1, Col2, Col3, ...)) AS RowHash

注意：HASH输出是整数类型，适合快速比较；如果需要加密级别的哈希，建议用下面的算法特定函数。

选项2：加密哈希函数（MD5/SHA系列） #

如果要和SQL Server HASHBYTES的加密算法对齐（比如MD5、SHA256），U-SQL提供MD5、SHA1、SHA256等函数。不过这些函数需要先把行数据转为字节流，通常的做法是拼接所有字段为字符串（注意处理NULL值和字段歧义）：

// 拼接字段时用分隔符避免冲突，用COALESCE处理NULL
STRING rowString = String.Concat(
    COALESCE(Col1.ToString(), ""), 
    "|",
    COALESCE(Col2.ToString(), ""), 
    "|",
    COALESCE(Col3.ToString(), "")
);
// 等价于SQL Server的HASHBYTES('SHA256', rowString)
SHA256(Encoding.UTF8.GetBytes(rowString)) AS RowHash

这里用|做分隔符是为了避免不同字段内容拼接后产生歧义（比如字段A是"ab"、字段B是"c"，和字段A是"a"、字段B是"bc"会生成相同字符串，导致哈希冲突）。

2. 存储所有数据行变更记录 #

推荐采用主表+历史表的方案来留存所有变更：

• 先创建两张表：

  • 主表（MainTable）：存储最新版本数据，包含主键、业务字段、LastUpdatedTimestamp（最后更新时间）、RowHash（当前行哈希）。
  • 历史表（HistoryTable）：存储所有变更记录，结构和主表一致，额外增加ChangeType（比如"INSERT"/"UPDATE"/"DELETE"）和ChangeTimestamp（变更发生时间）。

• 同步逻辑简化步骤：

  1. 读取源数据，计算每行的RowHash和当前时间NewTimestamp。
  2. 对比源数据和主表：
     • 源有、主表无：插入主表，同时往历史表插一条INSERT记录。
     • 源有、主表有但哈希不同：把主表旧数据插入历史表（标记UPDATE），再更新主表的业务字段、时间戳和哈希。
     • 源无、主表有：把主表数据插入历史表（标记DELETE），再删除主表对应行。

示例U-SQL代码片段：

// 读取源数据并计算哈希和时间戳
@source =
    SELECT 
        Id, Col1, Col2,
        SHA256(Encoding.UTF8.GetBytes(String.Concat(COALESCE(Col1.ToString(), ""), "|", COALESCE(Col2.ToString(), "")))) AS RowHash,
        DateTime.UtcNow AS NewTimestamp
    FROM SourceData;

// 读取主表现有数据
@main =
    SELECT Id, Col1, Col2, RowHash, LastUpdatedTimestamp
    FROM MainTable;

// 处理新增记录
@inserts =
    SELECT s.Id, s.Col1, s.Col2, s.RowHash, s.NewTimestamp, "INSERT" AS ChangeType
    FROM @source s
    LEFT JOIN @main m ON s.Id = m.Id
    WHERE m.Id IS NULL;

// 处理更新记录：先捞旧数据存入历史表
@updates_old =
    SELECT m.Id, m.Col1, m.Col2, m.RowHash, m.LastUpdatedTimestamp, "UPDATE" AS ChangeType
    FROM @source s
    JOIN @main m ON s.Id = m.Id
    WHERE s.RowHash != m.RowHash;

// 准备要更新到主表的新数据
@updates_new =
    SELECT s.Id, s.Col1, s.Col2, s.RowHash, s.NewTimestamp
    FROM @source s
    JOIN @main m ON s.Id = m.Id
    WHERE s.RowHash != m.RowHash;

// 处理删除记录
@deletes =
    SELECT m.Id, m.Col1, m.Col2, m.RowHash, m.LastUpdatedTimestamp, "DELETE" AS ChangeType
    FROM @main m
    LEFT JOIN @source s ON m.Id = s.Id
    WHERE s.Id IS NULL;

// 把所有变更写入历史表
INSERT INTO HistoryTable
SELECT Id, Col1, Col2, RowHash, LastUpdatedTimestamp AS ChangeTimestamp, ChangeType
FROM (@inserts UNION ALL @updates_old UNION ALL @deletes);

// 更新主表：新增+修改
MERGE INTO MainTable AS target
USING (
    SELECT Id, Col1, Col2, RowHash, NewTimestamp FROM @inserts
    UNION ALL
    SELECT Id, Col1, Col2, RowHash, NewTimestamp FROM @updates_new
) AS source
ON target.Id = source.Id
WHEN MATCHED THEN
    UPDATE SET target.Col1 = source.Col1, target.Col2 = source.Col2, target.RowHash = source.RowHash, target.LastUpdatedTimestamp = source.NewTimestamp
WHEN NOT MATCHED THEN
    INSERT (Id, Col1, Col2, RowHash, LastUpdatedTimestamp)
    VALUES (source.Id, source.Col1, source.Col2, source.RowHash, source.NewTimestamp);

// 删除主表中已移除的记录
DELETE FROM MainTable
WHERE Id IN (SELECT Id FROM @deletes);

3. 结合时间戳获取最新版本数据 #

如果从主表取最新数据，直接查询即可（主表本身存的就是最新状态）。如果要从历史表中提取每个主键的最新变更记录，用窗口函数ROW_NUMBER()实现：

@latest_rows =
    SELECT 
        Id, Col1, Col2, ChangeType, ChangeTimestamp,
        ROW_NUMBER() OVER(PARTITION BY Id ORDER BY ChangeTimestamp DESC) AS RowNum
    FROM HistoryTable;

// 取每个主键的最新记录
SELECT Id, Col1, Col2, ChangeType, ChangeTimestamp
FROM @latest_rows
WHERE RowNum = 1;

这个查询会按主键分区，按变更时间降序排序，每个分区的第一行就是该行的最新状态。

内容的提问来源于stack exchange，提问作者Arnay