如何在 C# 中比较 SQL Server CDC LSN 值？

Question

在 SQL 中，它很容易，因为它支持二进制 (10) LSN 值进行比较：

SELECT *, __$start_lsn, __$seqval
FROM cdc.fn_cdc_get_all_changes_dbo_sometable(@startLsn, @endLsn, 'all update old') 
WHERE __$seqval > @seqval 
ORDER BY __$start_lsn, __$seqval

在 C# 中，它更困难：

byte[] mySeqval = ...
foreach(var row in cdcData)
{
    if(row.seqval > mySeqval) // Cannot perform this
        ...
}

LSN / SeqVal 值可以转换为可以轻松比较的数字吗？ 它们的大小为 10 字节（80 位）。

我的项目是.Net 3.5

Answer 1

当我过去几周查看时，我很惊讶没有在任何地方找到这个问题的半体面的答案。

LSN 的主要问题是它们是 10 个字节，因此不能简单地将它们转换为 Int64 并进行比较（旁白：你真的会生成那么多 LSN 吗？！ Int64 真的很大）。正如OP发现的那样，逐一比较字节有点痛苦/容易出错（比较相等性很好 - 比较大于/小于则更好）。然而，从 .Net Framework 4 开始，我们有了 BigInteger 类，它可以用来轻松比较超过 8 字节的整数。

所以问题只是如何将 varbinary(10) 从 LSN 转换为 BigInteger。从检查[1]来看，SQL似乎以big-endian格式存储LSN，因此您必须：

• 将varbinary(10)放入内存中。 LinqToSql 将为您提供 Binary，其他提供程序将直接映射到 byte[]。
• 如果您使用的是小端架构（提示：您是），则翻转字节。如果您不想自己进行反向循环，则 IEnumerable.Reverse().ToArray() 会执行此操作，
• 请调用 new BigInteger(bytes)
• 比较您的值休闲

这可能看起来像这样：

// https://gist.github.com/piers7/91141f39715a2ec133e5
// Example of how to interpret SQL server CDC LSNs in C# / .Net
// This is required when polling a server for updates in order to determine
// if a previously stored LSN is still valid (ie > min LSN available)

// Requires .Net 4 (or equivilent BigInteger implementation)
// Sample is a Linqpad script, but you get the idea

// NB: That SQL uses big-endian representation for it's LSNs is not
// (as best I know) something they guarantee not to change

Connection.Open();
var command = Connection.CreateCommand();
command.CommandText = @"select sys.fn_cdc_get_max_lsn() as maxLsn";
var bytes = (byte[])command.ExecuteScalar();

// dump bytes as hex
var hexString = string.Join(" ", bytes.Select(b => b.ToString("X2")))
    .Dump("Hex String");

if(BitConverter.IsLittleEndian)
    bytes = bytes.Reverse().ToArray();

var bigInt = new BigInteger(bytes)
    // dump Integer representation
    .Dump("Big Integer")
;

[1] 我进行了连续的更改，并查看了 LSN。最后一个字节明显递增，因此是大端字节序。

Answer 2

目前正在调查 http://intx.codeplex.com/ 作为其 .Net 2.0

Answer 3

最后写了我自己的LSN比较器：

public class CdcLsnValue : IEquatable<CdcLsnValue>
{
    public byte[] Bytes;
    private const int Size = 10;

    public CdcLsnValue()
    {
        Bytes = null;
    }

    public CdcLsnValue(byte[] lsn)
    {
        if (lsn == null)
        {
            Bytes = null;
            return;
        }
        if(lsn.Length != Size)
            throw new ArgumentOutOfRangeException("lsn");
        Bytes = (byte[]) lsn.Clone();
    }

    public static bool operator ==(CdcLsnValue left, CdcLsnValue right)
    {
        if (ReferenceEquals(left, right)) return true;
        if (ReferenceEquals(null, left)) return false;
        if (ReferenceEquals(null, right)) return false;

        for (int i = 0; i < Size; i++)
        {
            if (left.Bytes[i] == right.Bytes[i])
                continue;
            return false;
        }
        return true;

    }

    public static bool operator !=(CdcLsnValue left, CdcLsnValue right)
    {
        return !(left == right);
    }

    public static bool operator <=(CdcLsnValue left, CdcLsnValue right)
    {
        if (ReferenceEquals(null, left)) return false;
        if (ReferenceEquals(null, right)) return false;

        for (int i = 0; i < Size; i++)
        {
            if (left.Bytes[i] <= right.Bytes[i])
                continue;
            return false;
        }
        return true;
    }

    public static bool operator >=(CdcLsnValue left, CdcLsnValue right)
    {
        if (ReferenceEquals(null, left)) return false;
        if (ReferenceEquals(null, right)) return false;

        for (int i = 0; i < Size; i++)
        {
            if (left.Bytes[i] >= right.Bytes[i])
                continue;
            return false;
        }
        return true;
    }

    public static bool operator <(CdcLsnValue left, CdcLsnValue right)
    {
        if (ReferenceEquals(null, left)) return false;
        if (ReferenceEquals(null, right)) return false;

        if (left == right)
            return false;

        return left <= right;
    }

    public static bool operator >(CdcLsnValue left, CdcLsnValue right)
    {
        return !(left < right);
    }

    public bool Equals(CdcLsnValue other)
    {
        if (ReferenceEquals(null, other)) return false;
        if (ReferenceEquals(this, other)) return true;
        return Equals(other.Bytes, Bytes);
    }

    public override bool Equals(object obj)
    {
        if (ReferenceEquals(null, obj)) return false;
        if (ReferenceEquals(this, obj)) return true;
        if (obj.GetType() != typeof(CdcLsnValue)) return false;
        return Equals((CdcLsnValue)obj);
    }

    public override int GetHashCode()
    {
        return (Bytes != null ? Bytes.GetHashCode() : 0);
    }
}

Answer 4

最终不需要使用以上任何一个。只有我的同事最终解决了这个问题（感谢托尼·布罗迪）。这样做的方法是与 seqval 进行比较，然后取+1。简单。

SqlExecutor.ExecuteReader(cnn,
string.Format("SELECT {0} , __$start_lsn, __$seqval , __$update_mask " +
    "FROM cdc.fn_cdc_get_all_changes_{1}(@startLsn,@endLsn,'all update old') cdc {2} " +
    "where __$operation = {3} ORDER BY __$start_lsn, __$seqval", columns,
    captureInstance, joins, (int)operation), 
    reader =>
    {
        if (reader != null)
            items.Add(createEntity(reader));
    }, 5, 60, new SqlParameter("@startLsn", lsn), 
              new SqlParameter("@endLsn", endLsn));
});
startLsn = lsn;
seqVal = sequence;
var startIndex = sequence == null ? 0 : 
  items.FindIndex(0, item => item.Lsn.SequenceEqual(lsn)
    && item.Seqval.SequenceEqual(sequence)) + 1; // <---- Look here. See the +1
return items.Skip(startIndex).ToList();

Answer 5

受到 @casperOne 的固定长度解决方案的启发，我创建了一个扩展方法允许您像这样按字节数组排序：

allchanges.OrderByDescendingFixedLength(sortLength: 10, x => x.___seqval)

这是扩展方法：

public static IEnumerable<T> OrderByFixedLength<T>(this IEnumerable<T> items, int sortLength, Func<T, byte[]> fieldValue)
    {
        //this routine came from:
        //      https://stackoverflow.com/questions/10658709/linq-orderbybyte-values
        //  it was modified to be generic <T> instead of specific type

        // Validate parameters.
        if (items == null) throw new ArgumentNullException("items");
        if (sortLength < 0) throw
            new ArgumentOutOfRangeException("sortLength", sortLength,
                "The sortLength parameter must be a non-negative value.");

        // Shortcut, if sortLength is zero, return the sequence, as-is.
        if (sortLength == 0) return items;

        // The ordered enumerable.
        IOrderedEnumerable<T> ordered = items.OrderBy(x => fieldValue(x)[0]);

        // Cycle from the second index on.
        for (int index = 1; index < sortLength; index++)
        {
            // Copy the index.
            int indexCopy = index;

            // Sort by the next item in the array.
            ordered = ordered.ThenBy(x => fieldValue(x)[indexCopy]);
        }

        // Return the ordered enumerable.
        return ordered;
    }

  public static IEnumerable<T> OrderByDescendingFixedLength<T>(this IEnumerable<T> items, int sortLength, Func<T, byte[]> fieldValue)
    {
        //we could probably optimize this, but honestly it's used so little and already quite quick... so we'll just go with it
        return items.OrderByFixedLength(sortLength, fieldValue).Reverse();
    }