diff --git a/ClickHouse.Driver.Common/AssemblyInfo.cs b/ClickHouse.Driver.Common/AssemblyInfo.cs index 06960c16f..3218f1158 100644 --- a/ClickHouse.Driver.Common/AssemblyInfo.cs +++ b/ClickHouse.Driver.Common/AssemblyInfo.cs @@ -4,3 +4,7 @@ [assembly:AssemblyKeyFile("../sgKey.snk")] [assembly: InternalsVisibleTo("ClickHouse.Driver, PublicKey=00240000048000009400000006020000002400005253413100040000010001000968a6468f9d0397a051f167a25dcee773c674cf7a67629f78e884d232df23ff773fbfaba602e03eede6056b39bd6a4cddcd7e5b3ca9484bd83401d14a5e9ac5c98cbe676a1e89149816f5304f617b658440b2bd775e5ece71b5a38ceeb88e844869a376ceea71cbb6393b2ac14e506b92267a3cbcd6e7dc93ff6c750d53a5c7")] [assembly: InternalsVisibleTo("ClickHouse.Driver.Tests, PublicKey=00240000048000009400000006020000002400005253413100040000010001000968a6468f9d0397a051f167a25dcee773c674cf7a67629f78e884d232df23ff773fbfaba602e03eede6056b39bd6a4cddcd7e5b3ca9484bd83401d14a5e9ac5c98cbe676a1e89149816f5304f617b658440b2bd775e5ece71b5a38ceeb88e844869a376ceea71cbb6393b2ac14e506b92267a3cbcd6e7dc93ff6c750d53a5c7")] + +// The native-TCP client frames its packet bodies with the same codec and checksum. +[assembly: InternalsVisibleTo("ClickHouse.Driver.Tcp, PublicKey=00240000048000009400000006020000002400005253413100040000010001000968a6468f9d0397a051f167a25dcee773c674cf7a67629f78e884d232df23ff773fbfaba602e03eede6056b39bd6a4cddcd7e5b3ca9484bd83401d14a5e9ac5c98cbe676a1e89149816f5304f617b658440b2bd775e5ece71b5a38ceeb88e844869a376ceea71cbb6393b2ac14e506b92267a3cbcd6e7dc93ff6c750d53a5c7")] +[assembly: InternalsVisibleTo("ClickHouse.Driver.Tcp.Tests, PublicKey=00240000048000009400000006020000002400005253413100040000010001000968a6468f9d0397a051f167a25dcee773c674cf7a67629f78e884d232df23ff773fbfaba602e03eede6056b39bd6a4cddcd7e5b3ca9484bd83401d14a5e9ac5c98cbe676a1e89149816f5304f617b658440b2bd775e5ece71b5a38ceeb88e844869a376ceea71cbb6393b2ac14e506b92267a3cbcd6e7dc93ff6c750d53a5c7")] diff --git a/ClickHouse.Driver.Common/Compression/CityHash102.cs b/ClickHouse.Driver.Common/Compression/CityHash102.cs new file mode 100644 index 000000000..b59db4610 --- /dev/null +++ b/ClickHouse.Driver.Common/Compression/CityHash102.cs @@ -0,0 +1,233 @@ +using System; +using System.Buffers.Binary; +using System.Numerics; + +namespace ClickHouse.Driver.Compression; + +/// +/// CityHash v1.0.2, the historical variant ClickHouse uses for the compression frame's 128-bit checksum. +/// +/// This is not modern Google CityHash. Release 1.1 changed the 128-bit functions, so a 1.1+ +/// implementation disagrees with ClickHouse on every input and every frame it writes is rejected as +/// corrupt. A faithful port of the reference city.cc that ClickHouse vendors as +/// contrib/cityhash102. +/// +/// +/// Only the 128-bit entry point is ported, because the frame checksum is all the drivers need. +/// +/// +internal static class CityHash102 +{ + // Some primes between 2^63 and 2^64, named as in the reference. + private const ulong K0 = 0xc3a5c85c97cb3127UL; + private const ulong K1 = 0xb492b66fbe98f273UL; + private const ulong K2 = 0x9ae16a3b2f90404fUL; + private const ulong K3 = 0xc949d7c7509e6557UL; + + /// + /// Computes the CityHash v1.0.2 128-bit hash of . The frame writes + /// Low then High, each little-endian. + /// + /// The bytes to hash. May be empty. + /// The low and high halves of the 128-bit result. + public static (ulong Low, ulong High) Hash128(ReadOnlySpan data) + { + int len = data.Length; + if (len >= 16) + { + return Hash128WithSeed(data.Slice(16), Fetch64(data, 0) ^ K3, Fetch64(data, 8)); + } + + if (len >= 8) + { + // The reference passes a null pointer and length 0 on this branch; an empty span is the same. + return Hash128WithSeed(default, Fetch64(data, 0) ^ ((ulong)len * K0), Fetch64(data, len - 8) ^ K1); + } + + return Hash128WithSeed(data, K0, K1); + } + + /// Hashes with a 128-bit seed, per the reference CityHash128WithSeed. + private static (ulong Low, ulong High) Hash128WithSeed(ReadOnlySpan s, ulong seedLow, ulong seedHigh) + { + if (s.Length < 128) + { + return CityMurmur(s, seedLow, seedHigh); + } + + // 56 bytes of state: v, w, x, y and z. + ulong x = seedLow; + ulong y = seedHigh; + ulong z = (ulong)s.Length * K1; + (ulong First, ulong Second) v; + (ulong First, ulong Second) w; + v.First = Rotate(y ^ K1, 49) * K1 + Fetch64(s, 0); + v.Second = Rotate(v.First, 42) * K1 + Fetch64(s, 8); + w.First = Rotate(y + z, 35) * K1 + x; + w.Second = Rotate(x + Fetch64(s, 88), 53) * K1; + + // The same inner loop as CityHash64, manually unrolled to two 64-byte halves per pass. + int pos = 0; + int remaining = s.Length; + do + { + x = Rotate(x + y + v.First + Fetch64(s, pos + 16), 37) * K1; + y = Rotate(y + v.Second + Fetch64(s, pos + 48), 42) * K1; + x ^= w.Second; + y ^= v.First; + z = Rotate(z ^ w.First, 33); + v = WeakHashLen32WithSeeds(s, pos, v.Second * K1, x + w.First); + w = WeakHashLen32WithSeeds(s, pos + 32, z + w.Second, y); + (z, x) = (x, z); + pos += 64; + + x = Rotate(x + y + v.First + Fetch64(s, pos + 16), 37) * K1; + y = Rotate(y + v.Second + Fetch64(s, pos + 48), 42) * K1; + x ^= w.Second; + y ^= v.First; + z = Rotate(z ^ w.First, 33); + v = WeakHashLen32WithSeeds(s, pos, v.Second * K1, x + w.First); + w = WeakHashLen32WithSeeds(s, pos + 32, z + w.Second, y); + (z, x) = (x, z); + pos += 64; + + remaining -= 128; + } + while (remaining >= 128); + + y += Rotate(w.First, 37) * K0 + z; + x += Rotate(v.First + z, 49) * K0; + + // Hash up to four 32-byte chunks from the end. These indices walk backwards and can land before + // `pos`, which is in range only because the loop above ran at least once, so pos >= 128. + for (int tailDone = 0; tailDone < remaining;) + { + tailDone += 32; + y = Rotate(y - x, 42) * K0 + v.Second; + w.First += Fetch64(s, pos + remaining - tailDone + 16); + x = Rotate(x, 49) * K0 + w.First; + w.First += v.First; + v = WeakHashLen32WithSeeds(s, pos + remaining - tailDone, v.First, v.Second); + } + + x = HashLen16(x, v.First); + y = HashLen16(y, w.First); + return (HashLen16(x + v.Second, w.Second) + y, HashLen16(x + w.Second, y + v.Second)); + } + + /// The reference CityMurmur: a 128-bit hash for any length, used below 128 bytes. + private static (ulong Low, ulong High) CityMurmur(ReadOnlySpan s, ulong seedLow, ulong seedHigh) + { + ulong a = seedLow; + ulong b = seedHigh; + ulong c = 0; + ulong d = 0; + + // Signed on purpose: the reference uses ssize_t, so a length below 16 must compare as negative. + long l = (long)s.Length - 16; + if (l <= 0) + { + a = ShiftMix(a * K1) * K1; + c = (b * K1) + HashLen0to16(s); + d = ShiftMix(a + (s.Length >= 8 ? Fetch64(s, 0) : c)); + } + else + { + c = HashLen16(Fetch64(s, s.Length - 8) + K1, a); + d = HashLen16(b + (ulong)s.Length, c + Fetch64(s, s.Length - 16)); + a += d; + + int pos = 0; + do + { + a ^= ShiftMix(Fetch64(s, pos) * K1) * K1; + a *= K1; + b ^= a; + c ^= ShiftMix(Fetch64(s, pos + 8) * K1) * K1; + c *= K1; + d ^= c; + pos += 16; + l -= 16; + } + while (l > 0); + } + + a = HashLen16(a, c); + b = HashLen16(d, b); + return (a ^ b, HashLen16(b, a)); + } + + /// Hashes 0 to 16 bytes, per the reference HashLen0to16. + private static ulong HashLen0to16(ReadOnlySpan s) + { + int len = s.Length; + if (len > 8) + { + ulong a = Fetch64(s, 0); + ulong b = Fetch64(s, len - 8); + return HashLen16(a, Rotate(b + (ulong)len, len)) ^ b; + } + + if (len >= 4) + { + ulong a = Fetch32(s, 0); + return HashLen16((ulong)len + (a << 3), Fetch32(s, len - 4)); + } + + if (len > 0) + { + uint y = s[0] + ((uint)s[len >> 1] << 8); + uint z = (uint)len + ((uint)s[len - 1] << 2); + return ShiftMix((y * K2) ^ (z * K3)) * K2; + } + + return K2; + } + + /// A 16-byte hash of 32 bytes at plus two seeds. Quick and dirty, per the reference. + private static (ulong First, ulong Second) WeakHashLen32WithSeeds(ReadOnlySpan s, int pos, ulong a, ulong b) + => WeakHashLen32WithSeeds(Fetch64(s, pos), Fetch64(s, pos + 8), Fetch64(s, pos + 16), Fetch64(s, pos + 24), a, b); + + /// The six-word core of WeakHashLen32WithSeeds. + private static (ulong First, ulong Second) WeakHashLen32WithSeeds(ulong w, ulong x, ulong y, ulong z, ulong a, ulong b) + { + a += w; + b = Rotate(b + a + z, 21); + ulong c = a; + a += x; + a += y; + b += Rotate(a, 44); + return (a + z, b + c); + } + + /// Collapses a 128-bit value to 64 bits (Murmur-inspired), per the reference Hash128to64. + private static ulong Hash128to64(ulong low, ulong high) + { + const ulong Mul = 0x9ddfea08eb382d69UL; + ulong a = (low ^ high) * Mul; + a ^= a >> 47; + ulong b = (high ^ a) * Mul; + b ^= b >> 47; + b *= Mul; + return b; + } + + /// Hashes two words, treating them as the low and high halves of a 128-bit value. + private static ulong HashLen16(ulong u, ulong v) => Hash128to64(u, v); + + /// + /// Bitwise right rotate. Covers both the reference's Rotate and its RotateByAtLeast1: + /// the former guards against a shift of 0 only because shifting a 64-bit value by 64 is undefined in + /// C++, which is not. + /// + private static ulong Rotate(ulong value, int shift) => BitOperations.RotateRight(value, shift); + + /// The reference ShiftMix. + private static ulong ShiftMix(ulong value) => value ^ (value >> 47); + + /// Reads a little-endian 64-bit word, as the reference's Fetch64 does on a little-endian host. + private static ulong Fetch64(ReadOnlySpan s, int index) => BinaryPrimitives.ReadUInt64LittleEndian(s.Slice(index)); + + /// Reads a little-endian 32-bit word, as the reference's Fetch32 does on a little-endian host. + private static uint Fetch32(ReadOnlySpan s, int index) => BinaryPrimitives.ReadUInt32LittleEndian(s.Slice(index)); +} diff --git a/ClickHouse.Driver.Common/Compression/CompressionFrame.cs b/ClickHouse.Driver.Common/Compression/CompressionFrame.cs new file mode 100644 index 000000000..de8f6cafd --- /dev/null +++ b/ClickHouse.Driver.Common/Compression/CompressionFrame.cs @@ -0,0 +1,201 @@ +using System; +using System.Buffers.Binary; +using System.IO; + +namespace ClickHouse.Driver.Compression; + +/// +/// Encodes and decodes a single ClickHouse compression frame. Pure span work: callers own the I/O and the +/// buffers, so the same codec serves the native protocol's per-packet framing and the HTTP +/// compress/decompress stream. +/// +/// The layout is a 16-byte checksum, a 9-byte header, then the body: +/// +/// +/// [16 bytes: CityHash128 over the 9-byte header + body] +/// [1 byte : method] 0x02 NONE, 0x82 LZ4, 0x90 ZSTD +/// [4 bytes: compressed_size] little-endian; counts the 9-byte header, not the checksum +/// [4 bytes: uncompressed_size] little-endian +/// [N bytes: body] N = compressed_size - 9 +/// +/// +/// The two spans differ and are easy to confuse: the checksum covers the header plus the body, while +/// compressed_size counts the header plus the body but excludes the checksum itself. +/// +/// +internal static class CompressionFrame +{ + /// Size of the leading CityHash128 checksum. + public const int ChecksumSize = 16; + + /// Size of the header the checksum covers: method plus the two sizes. + public const int HeaderSize = 9; + + /// Bytes before the body: the checksum plus the header. + public const int PrefixSize = ChecksumSize + HeaderSize; + + /// Method byte for an uncompressed body. The frame and its checksum are still written. + public const byte MethodNone = 0x02; + + /// Method byte for an LZ4 body, in the LZ4 block format (no frame, no magic number). + public const byte MethodLz4 = 0x82; + + /// Method byte for a ZSTD body, a raw single zstd frame including its magic number. + public const byte MethodZstd = 0x90; + + // Both sizes arrive from the peer, so a corrupt or hostile stream can declare anything, and a decoder rents + // a buffer for each before it can check the checksum. The server flushes at ~1 MiB, so 128 MiB is ample + // headroom while keeping the worst case a peer can force to a size a process can absorb. The Go client caps + // both at the same figure. + private const int MaxBodySize = 128 * 1024 * 1024; + private const int MaxPlaintextSize = 128 * 1024 * 1024; + + /// + /// The largest frame can produce for bytes, so a + /// caller can size its destination buffer. + /// + /// The codec that will encode the body. + /// The number of plaintext bytes to be framed. + /// The maximum total frame size in bytes. + public static int MaxFrameSize(IClickHouseCompressor codec, int plaintextLength) + { + ArgumentNullException.ThrowIfNull(codec); + return PrefixSize + codec.MaxEncodedLength(plaintextLength); + } + + /// + /// Frames into , which must hold at least + /// bytes. + /// + /// The bytes to compress and frame. + /// The codec supplying the method byte and the body encoding. + /// The buffer to write the whole frame into. + /// The number of bytes written: plus the encoded body. + /// is too small. + public static int Write(ReadOnlySpan plaintext, IClickHouseCompressor codec, Span destination) + { + ArgumentNullException.ThrowIfNull(codec); + + int required = MaxFrameSize(codec, plaintext.Length); + if (destination.Length < required) + { + throw new ArgumentException($"Destination holds {destination.Length} bytes; framing {plaintext.Length} plaintext bytes needs up to {required}.", nameof(destination)); + } + + // Encode straight into place after the prefix, so the body is never copied. + int bodyLength = codec.Encode(plaintext, destination.Slice(PrefixSize)); + int compressedSize = HeaderSize + bodyLength; + + Span header = destination.Slice(ChecksumSize, HeaderSize); + header[0] = codec.MethodByte; + BinaryPrimitives.WriteInt32LittleEndian(header.Slice(1, 4), compressedSize); + BinaryPrimitives.WriteInt32LittleEndian(header.Slice(5, 4), plaintext.Length); + + // The checksum covers the header and the body, but not itself. + var (low, high) = CityHash102.Hash128(destination.Slice(ChecksumSize, compressedSize)); + BinaryPrimitives.WriteUInt64LittleEndian(destination.Slice(0, 8), low); + BinaryPrimitives.WriteUInt64LittleEndian(destination.Slice(8, 8), high); + + return ChecksumSize + compressedSize; + } + + /// Reads the 9-byte header, validating both declared sizes. + /// Exactly bytes, the ones following the checksum. + /// The declared method byte. + /// The body length, that is compressed_size minus the header. + /// The declared uncompressed size. + /// A declared size is negative, too small to include the header, or implausibly large. + public static void ReadHeader(ReadOnlySpan header, out byte method, out int bodySize, out int plaintextSize) + { + if (header.Length != HeaderSize) + { + throw new ArgumentException($"A frame header is exactly {HeaderSize} bytes.", nameof(header)); + } + + method = header[0]; + int compressedSize = BinaryPrimitives.ReadInt32LittleEndian(header.Slice(1, 4)); + plaintextSize = BinaryPrimitives.ReadInt32LittleEndian(header.Slice(5, 4)); + + // compressed_size counts the header, so anything below it is corrupt rather than merely empty. + if (compressedSize < HeaderSize) + { + throw new InvalidDataException($"Compression frame declares compressed_size {compressedSize}, which does not cover its own {HeaderSize}-byte header (corrupt stream)."); + } + + bodySize = compressedSize - HeaderSize; + if (bodySize > MaxBodySize) + { + throw new InvalidDataException($"Compression frame declares a {bodySize}-byte body, above the {MaxBodySize}-byte limit (corrupt stream)."); + } + + if (plaintextSize < 0 || plaintextSize > MaxPlaintextSize) + { + throw new InvalidDataException($"Compression frame declares uncompressed_size {plaintextSize}, outside the supported range (corrupt stream)."); + } + } + + /// + /// Recomputes the checksum over and compares it with the frame's. + /// + /// The frame's leading bytes. + /// The 9 header bytes followed by the body, contiguous. + /// The recomputed checksum differs, so the frame is corrupt. + public static void VerifyChecksum(ReadOnlySpan checksum, ReadOnlySpan headerAndBody) + { + var (low, high) = CityHash102.Hash128(headerAndBody); + ulong actualLow = BinaryPrimitives.ReadUInt64LittleEndian(checksum.Slice(0, 8)); + ulong actualHigh = BinaryPrimitives.ReadUInt64LittleEndian(checksum.Slice(8, 8)); + + if (low != actualLow || high != actualHigh) + { + throw new InvalidDataException( + $"Compression frame checksum mismatch: the frame carries {actualHigh:x16}{actualLow:x16} but its bytes hash to {high:x16}{low:x16}. " + + "The stream is corrupt, or the peer used a CityHash other than v1.0.2."); + } + } + + /// + /// Decodes a frame body into , which must be exactly the declared + /// uncompressed size. + /// + /// The frame's method byte. + /// The frame body. + /// The destination, sized to the declared uncompressed size. + /// The method byte is unknown, or the body decoded to the wrong length. + public static void Decode(byte method, ReadOnlySpan body, Span plaintext) + { + if (method == MethodNone) + { + if (body.Length != plaintext.Length) + { + throw new InvalidDataException($"An uncompressed frame declares {plaintext.Length} plaintext bytes but carries a {body.Length}-byte body."); + } + + body.CopyTo(plaintext); + return; + } + + IClickHouseCompressor codec = ResolveCodec(method); + int written = codec.Decode(body, plaintext); + if (written != plaintext.Length) + { + throw new InvalidDataException($"Compression frame declares {plaintext.Length} plaintext bytes but its body decoded to {written} (corrupt stream)."); + } + } + + /// + /// The codec for a frame's method byte. The peer chooses the codec per frame, so this must follow the + /// frame rather than whatever the caller configured for its own writes. + /// + /// The frame's method byte. + /// The matching codec. + /// The method byte is not one this client can decode. + public static IClickHouseCompressor ResolveCodec(byte method) => method switch + { + MethodLz4 => Lz4Compressor.Default, + MethodZstd => ZstdCompressor.Default, + _ => throw new InvalidDataException( + $"Compression frame declares method byte 0x{method:X2}, which this client cannot decode " + + $"(supported: 0x{MethodNone:X2} NONE, 0x{MethodLz4:X2} LZ4, 0x{MethodZstd:X2} ZSTD)."), + }; +} diff --git a/ClickHouse.Driver.Tcp.Tests/ClickHouse.Driver.Tcp.Tests.csproj b/ClickHouse.Driver.Tcp.Tests/ClickHouse.Driver.Tcp.Tests.csproj index 761edcad5..73757ec44 100644 --- a/ClickHouse.Driver.Tcp.Tests/ClickHouse.Driver.Tcp.Tests.csproj +++ b/ClickHouse.Driver.Tcp.Tests/ClickHouse.Driver.Tcp.Tests.csproj @@ -39,6 +39,9 @@ + + diff --git a/ClickHouse.Driver.Tcp.Tests/Client/ClickHouseTcpClientOptionsTests.cs b/ClickHouse.Driver.Tcp.Tests/Client/ClickHouseTcpClientOptionsTests.cs index 48b0e2ba4..35bde1e6c 100644 --- a/ClickHouse.Driver.Tcp.Tests/Client/ClickHouseTcpClientOptionsTests.cs +++ b/ClickHouse.Driver.Tcp.Tests/Client/ClickHouseTcpClientOptionsTests.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Reflection; +using ClickHouse.Driver.Compression; namespace ClickHouse.Driver.Tcp.Tests.Client; @@ -469,6 +470,9 @@ public void WithOwnedCustomSettings_CopiesEveryPropertyAndSnapshotsTheSettings() IdleTimeout = TimeSpan.FromSeconds(7), SweepInterval = TimeSpan.FromSeconds(8), PoolReusePolicy = ClickHouseTcpPoolReusePolicy.Fifo, + + // Zstd rather than Lz4 so this stays non-default whichever codec the default becomes. + Compressor = ZstdCompressor.Default, }; var defaults = new ClickHouseTcpClientOptions(); diff --git a/ClickHouse.Driver.Tcp.Tests/Client/CompressionOptionTests.cs b/ClickHouse.Driver.Tcp.Tests/Client/CompressionOptionTests.cs new file mode 100644 index 000000000..da8d0a9ce --- /dev/null +++ b/ClickHouse.Driver.Tcp.Tests/Client/CompressionOptionTests.cs @@ -0,0 +1,97 @@ +using System; +using System.Threading.Tasks; +using ClickHouse.Driver.Compression; + +namespace ClickHouse.Driver.Tcp.Tests.Client; + +/// +/// The compression option's surface: how a connection-string value maps to a codec, and the codecs the client +/// refuses. none means the query carries no compression at all, which is not the same as a frame whose +/// method byte is NONE — that byte only ever appears on the read side, chosen by the server. +/// +[TestFixture] +public class CompressionOptionTests +{ + [Test] + public void Compressor_WhenNotOverridden_DefaultsToLz4() + { + // The default is on. LZ4 is what clickhouse-client uses on this protocol: cheapest in CPU and lightest + // on the server. A change here alters what every query puts on the wire, so it is pinned. + Assert.Multiple(() => + { + Assert.That(new ClickHouseTcpClientOptions().Compressor, Is.SameAs(Lz4Compressor.Default)); + Assert.That(ClickHouseTcpClientOptions.FromConnectionString("Host=localhost").Compressor, Is.SameAs(Lz4Compressor.Default)); + Assert.That(new ClickHouseTcpConnectionStringBuilder().Compression, Is.EqualTo("lz4")); + }); + } + + [TestCase("lz4", CompressionFrame.MethodLz4)] + [TestCase("LZ4", CompressionFrame.MethodLz4)] + [TestCase(" lz4 ", CompressionFrame.MethodLz4)] + [TestCase("zstd", CompressionFrame.MethodZstd)] + [TestCase("ZSTD", CompressionFrame.MethodZstd)] + public void FromConnectionString_ACodecName_ResolvesToThatCodec(string value, byte expectedMethod) + { + var options = ClickHouseTcpClientOptions.FromConnectionString($"Host=localhost;Compression={value}"); + + Assert.That(options.Compressor, Is.Not.Null); + Assert.That(options.Compressor.MethodByte, Is.EqualTo(expectedMethod)); + } + + [TestCase("none")] + [TestCase("NONE")] + public void FromConnectionString_None_LeavesNoCompressor(string value) + { + var options = ClickHouseTcpClientOptions.FromConnectionString($"Host=localhost;Compression={value}"); + + Assert.That(options.Compressor, Is.Null); + } + + [Test] + public void FromConnectionString_AnUnknownCodecName_ThrowsNamingTheSupportedOnes() + { + var failure = Assert.Throws( + () => ClickHouseTcpClientOptions.FromConnectionString("Host=localhost;Compression=snappy")); + + Assert.That(failure.Message, Does.Contain("snappy").And.Contains("lz4").And.Contains("zstd")); + } + + [Test] + public void Compression_RoundTripsThroughTheBuilder() + { + var builder = new ClickHouseTcpConnectionStringBuilder { Compression = "zstd" }; + + Assert.Multiple(() => + { + Assert.That(builder.Compression, Is.EqualTo("zstd")); + Assert.That(builder.ToOptions().Compressor, Is.SameAs(ZstdCompressor.Default)); + }); + } + + [Test] + public void Validate_ACodecWithoutTheNativeBlockPath_IsRefusedAtConstruction() + { + // GZip and Brotli implement only the HTTP body path, so their MethodByte throws. Catching that when the + // client is built beats discovering it mid-query, after the Query packet has promised the server frames. + var options = new ClickHouseTcpClientOptions { Host = "localhost", Compressor = new GZipCompressor() }; + + var failure = Assert.Throws(() => new ClickHouseTcpClient(options)); + + Assert.That(failure.Message, Does.Contain("native block path").And.Contains(nameof(Lz4Compressor))); + } + + [Test] + public async Task Validate_TheBuiltInBlockCodecsAndNone_AreAccepted() + { + // Construction validates without connecting, so this needs no server. + IClickHouseCompressor[] accepted = [Lz4Compressor.Default, ZstdCompressor.Default, null]; + + foreach (IClickHouseCompressor codec in accepted) + { + var client = new ClickHouseTcpClient(new ClickHouseTcpClientOptions { Host = "localhost", Compressor = codec }); + await client.DisposeAsync(); + } + + Assert.Pass(); + } +} diff --git a/ClickHouse.Driver.Tcp.Tests/Compression/CompressedFrameStreamingTests.cs b/ClickHouse.Driver.Tcp.Tests/Compression/CompressedFrameStreamingTests.cs new file mode 100644 index 000000000..8672ff522 --- /dev/null +++ b/ClickHouse.Driver.Tcp.Tests/Compression/CompressedFrameStreamingTests.cs @@ -0,0 +1,257 @@ +using System; +using System.Collections.Generic; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using ClickHouse.Driver.Compression; +using ClickHouse.Driver.Tcp.Compression; +using ClickHouse.Driver.Tcp.Protocol; + +namespace ClickHouse.Driver.Tcp.Tests.Compression; + +/// +/// Unit tests for the streaming frame reader and writer over a memory stream. These cover what a server round +/// trip cannot pin down: how many frames a body is cut into, that a value split across a boundary reassembles, +/// that the reader stops at a boundary instead of reading into whatever follows, and the leftover assertion. +/// +[TestFixture] +public class CompressedFrameStreamingTests +{ + private static readonly CancellationToken None = CancellationToken.None; + + private static IEnumerable Codecs() + { + yield return new TestCaseData(Lz4Compressor.Default).SetName("{m}(LZ4)"); + yield return new TestCaseData(ZstdCompressor.Default).SetName("{m}(ZSTD)"); + } + + [TestCaseSource(nameof(Codecs))] + public async Task WriteThenRead_ABodySpanningManyFrames_RecoversEveryByte(IClickHouseCompressor codec) + { + // A frame target far below the payload forces many frames, so values land across the boundaries. + byte[] payload = Pattern(50_000); + var stream = new MemoryStream(); + + using (var rawWriter = new ClickHouseBinaryWriter(stream)) + using (var frames = new CompressedFrameWriter(rawWriter, codec, frameTarget: 700)) + { + frames.Writer.WriteBytes(payload); + await frames.EndBlockAsync(None); + } + + stream.Position = 0; + var readBack = new byte[payload.Length]; + using (var rawReader = new ClickHouseBinaryReader(stream)) + using (var frames = new CompressedFrameReader(rawReader)) + { + // Both read paths: a few single bytes through the buffer, then the bulk path for the remainder. + for (int i = 0; i < 5; i++) + { + readBack[i] = await frames.Reader.ReadByteAsync(None); + } + + await frames.Reader.ReadBytesAsync(readBack.AsMemory(5), None); + frames.EndBlock(); + } + + Assert.That(readBack, Is.EqualTo(payload)); + } + + [TestCaseSource(nameof(Codecs))] + public async Task WriteThenRead_AFixedWidthValueStraddlingAFrameBoundary_DecodesWhole(IClickHouseCompressor codec) + { + // The frame target is deliberately not a multiple of 8, so UInt64s straddle boundaries. + var stream = new MemoryStream(); + const int Count = 5000; + + using (var rawWriter = new ClickHouseBinaryWriter(stream)) + using (var frames = new CompressedFrameWriter(rawWriter, codec, frameTarget: 101)) + { + for (ulong i = 0; i < Count; i++) + { + frames.Writer.WriteUInt64(i); + } + + await frames.EndBlockAsync(None); + } + + stream.Position = 0; + using (var rawReader = new ClickHouseBinaryReader(stream)) + using (var frames = new CompressedFrameReader(rawReader)) + { + for (ulong i = 0; i < Count; i++) + { + ulong value = await frames.Reader.ReadUInt64Async(None); + if (value != i) + { + Assert.Fail($"value {i} decoded as {value}, so a frame boundary split it wrongly"); + } + } + + frames.EndBlock(); + } + + Assert.Pass(); + } + + [Test] + public async Task Read_AtTheEndOfABody_LeavesWhateverFollowsOnTheRawStream() + { + // The hazard the short reads exist for: after a body, the next bytes are an uncompressed envelope. The + // reader must not have consumed them while satisfying the last read. + byte[] payload = Pattern(64); + var stream = new MemoryStream(); + + using (var rawWriter = new ClickHouseBinaryWriter(stream)) + { + using (var frames = new CompressedFrameWriter(rawWriter, codec: Lz4Compressor.Default)) + { + frames.Writer.WriteBytes(payload); + await frames.EndBlockAsync(None); + } + + // An uncompressed "next packet" straight after the frames, exactly as the server would write it. + rawWriter.WriteVarUInt(42); + rawWriter.WriteString("next-packet"); + await rawWriter.FlushAsync(None); + } + + stream.Position = 0; + using var rawReader = new ClickHouseBinaryReader(stream); + using (var frames = new CompressedFrameReader(rawReader)) + { + var readBack = new byte[payload.Length]; + await frames.Reader.ReadBytesAsync(readBack, None); + frames.EndBlock(); + Assert.That(readBack, Is.EqualTo(payload)); + } + + // The envelope survived: it was never pulled in as frame bytes. + Assert.Multiple(async () => + { + Assert.That(await rawReader.ReadVarUIntAsync(None), Is.EqualTo(42UL)); + Assert.That(await rawReader.ReadStringAsync(None), Is.EqualTo("next-packet")); + }); + } + + [Test] + public async Task EndBlock_WithDecodedPlaintextLeftUnread_Throws() + { + // Stands in for a column decoder that consumed too few bytes: the frames carried more than the block's + // dimensions accounted for, which must fail loudly rather than desync the next block. + var stream = new MemoryStream(); + using (var rawWriter = new ClickHouseBinaryWriter(stream)) + using (var frames = new CompressedFrameWriter(rawWriter, Lz4Compressor.Default)) + { + frames.Writer.WriteBytes(Pattern(500)); + await frames.EndBlockAsync(None); + } + + stream.Position = 0; + using var rawReader = new ClickHouseBinaryReader(stream); + using var reader = new CompressedFrameReader(rawReader); + + var partial = new byte[100]; + await reader.Reader.ReadBytesAsync(partial, None); + + var failure = Assert.Throws(() => reader.EndBlock()); + Assert.That(failure.Message, Does.Contain("unread")); + } + + [Test] + public async Task EndBlock_AfterTheWholeBody_DoesNotThrow() + { + var stream = new MemoryStream(); + using (var rawWriter = new ClickHouseBinaryWriter(stream)) + using (var frames = new CompressedFrameWriter(rawWriter, Lz4Compressor.Default)) + { + frames.Writer.WriteBytes(Pattern(500)); + await frames.EndBlockAsync(None); + } + + stream.Position = 0; + using var rawReader = new ClickHouseBinaryReader(stream); + using var reader = new CompressedFrameReader(rawReader); + + var all = new byte[500]; + await reader.Reader.ReadBytesAsync(all, None); + + Assert.Multiple(() => + { + Assert.That(() => reader.EndBlock(), Throws.Nothing); + Assert.That(reader.PendingPlaintext, Is.Zero); + }); + } + + [Test] + public async Task Read_AFrameWhoseBodyWasCorruptedInTransit_Throws() + { + var stream = new MemoryStream(); + using (var rawWriter = new ClickHouseBinaryWriter(stream)) + using (var frames = new CompressedFrameWriter(rawWriter, Lz4Compressor.Default)) + { + frames.Writer.WriteBytes(Pattern(500)); + await frames.EndBlockAsync(None); + } + + byte[] wire = stream.ToArray(); + wire[CompressionFrame.PrefixSize + 2] ^= 0x01; + + using var rawReader = new ClickHouseBinaryReader(new MemoryStream(wire)); + using var reader = new CompressedFrameReader(rawReader); + + var destination = new byte[500]; + Assert.That( + async () => await reader.Reader.ReadBytesAsync(destination, None), + Throws.TypeOf().With.Message.Contains("checksum mismatch")); + } + + [Test] + public void Constructor_ANullRawReader_Throws() + => Assert.That(() => new CompressedFrameReader(null), Throws.ArgumentNullException); + + [Test] + public void Constructor_ANullCodecOrWriter_Throws() + { + using var stream = new MemoryStream(); + using var rawWriter = new ClickHouseBinaryWriter(stream); + + Assert.Multiple(() => + { + Assert.That(() => new CompressedFrameWriter(null, Lz4Compressor.Default), Throws.ArgumentNullException); + Assert.That(() => new CompressedFrameWriter(rawWriter, null), Throws.ArgumentNullException); + Assert.That(() => new CompressedFrameWriter(rawWriter, Lz4Compressor.Default, frameTarget: 0), Throws.InstanceOf()); + }); + } + + [Test] + public void Dispose_CalledTwice_IsANoOp() + { + var stream = new MemoryStream(); + using var rawWriter = new ClickHouseBinaryWriter(stream); + using var rawReader = new ClickHouseBinaryReader(stream); + + var reader = new CompressedFrameReader(rawReader); + var writer = new CompressedFrameWriter(rawWriter, Lz4Compressor.Default); + + reader.Dispose(); + writer.Dispose(); + + Assert.Multiple(() => + { + Assert.That(() => reader.Dispose(), Throws.Nothing); + Assert.That(() => writer.Dispose(), Throws.Nothing); + }); + } + + private static byte[] Pattern(int length) + { + var data = new byte[length]; + for (int i = 0; i < length; i++) + { + data[i] = unchecked((byte)((i * 31) + 7)); + } + + return data; + } +} diff --git a/ClickHouse.Driver.Tcp.Tests/Compression/FramedPacketsTests.cs b/ClickHouse.Driver.Tcp.Tests/Compression/FramedPacketsTests.cs new file mode 100644 index 000000000..0eec4f741 --- /dev/null +++ b/ClickHouse.Driver.Tcp.Tests/Compression/FramedPacketsTests.cs @@ -0,0 +1,62 @@ +using System; +using System.Linq; +using ClickHouse.Driver.Tcp.Compression; +using ClickHouse.Driver.Tcp.Protocol; + +namespace ClickHouse.Driver.Tcp.Tests.Compression; + +/// +/// Pins which packet bodies are framed. This list is not "every packet that carries a block": Log and +/// ProfileEvents carry blocks and go through the same block reader, yet arrive uncompressed at our +/// protocol target. Adding either here desynchronizes any query against a server with +/// send_logs_level set, so the list is asserted whole rather than case by case. +/// +/// The cases are not a TestCaseSource because ServerPacketType is internal and so cannot appear +/// in a public test method's signature. +/// +/// +[TestFixture] +public class FramedPacketsTests +{ + [Test] + public void CarriesFramedBody_TheCompressiblePackets_AreFramed() + { + Assert.Multiple(() => + { + Assert.That(FramedPackets.CarriesFramedBody(ServerPacketType.Data), Is.True, "Data"); + Assert.That(FramedPackets.CarriesFramedBody(ServerPacketType.Totals), Is.True, "Totals"); + Assert.That(FramedPackets.CarriesFramedBody(ServerPacketType.Extremes), Is.True, "Extremes"); + }); + } + + [Test] + public void CarriesFramedBody_TheBlockBearingPacketsTheServerDoesNotCompress_AreNotFramed() + { + Assert.Multiple(() => + { + Assert.That(FramedPackets.CarriesFramedBody(ServerPacketType.Log), Is.False, "Log"); + Assert.That(FramedPackets.CarriesFramedBody(ServerPacketType.ProfileEvents), Is.False, "ProfileEvents"); + }); + } + + [Test] + public void CarriesFramedBody_EveryOtherPacket_IsNotFramed() + { + // A packet with no block body cannot be framed, so a new enum member must default to false rather than + // silently join the framed set. + ServerPacketType[] framed = + [ + ServerPacketType.Data, + ServerPacketType.Totals, + ServerPacketType.Extremes, + ]; + + Assert.Multiple(() => + { + foreach (ServerPacketType packet in Enum.GetValues().Except(framed)) + { + Assert.That(FramedPackets.CarriesFramedBody(packet), Is.False, $"{packet} must not be framed"); + } + }); + } +} diff --git a/ClickHouse.Driver.Tcp.Tests/Integration/CompressionIntegrationTests.cs b/ClickHouse.Driver.Tcp.Tests/Integration/CompressionIntegrationTests.cs new file mode 100644 index 000000000..78f6ca2ef --- /dev/null +++ b/ClickHouse.Driver.Tcp.Tests/Integration/CompressionIntegrationTests.cs @@ -0,0 +1,252 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using ClickHouse.Driver.Compression; + +namespace ClickHouse.Driver.Tcp.Tests.Integration; + +/// +/// Compression against a real server, in both directions. The frame codec and its checksum are unit-tested +/// against the reference implementation, so these cover what only a server can settle: that the server accepts +/// the frames we write, that we decode the frames it writes, and that the framing survives the shapes the +/// codec tests cannot produce — bodies that span several frames, packets whose bodies are not framed +/// arriving mid-stream, and a pooled connection carrying one compressed query after another. +/// +[TestFixture] +[Category("Integration")] +public class CompressionIntegrationTests +{ + private static readonly CancellationToken None = CancellationToken.None; + + private static IEnumerable Codecs() + { + yield return new TestCaseData(Lz4Compressor.Default).SetName("{m}(LZ4)"); + yield return new TestCaseData(ZstdCompressor.Default).SetName("{m}(ZSTD)"); + } + + [TestCaseSource(nameof(Codecs))] + public async Task QueryAsync_Compressed_ReturnsEveryRow(IClickHouseCompressor codec) + { + await using var client = new ClickHouseTcpClient(TcpServerFixture.Options() with { Compressor = codec }); + + var values = new List(); + await foreach (object[] row in client.QueryAsync("SELECT number FROM numbers(5000)", null, None)) + { + values.Add((ulong)row[0]); + } + + Assert.Multiple(() => + { + Assert.That(values, Has.Count.EqualTo(5000)); + Assert.That(values[0], Is.Zero); + Assert.That(values[^1], Is.EqualTo(4999UL)); + }); + } + + [TestCaseSource(nameof(Codecs))] + public async Task InsertRowsAsync_Compressed_RoundTripsThroughSelect(IClickHouseCompressor codec) + { + await using var client = new ClickHouseTcpClient(TcpServerFixture.Options() with { Compressor = codec }); + string table = UniqueTableName(); + try + { + await client.ExecuteAsync($"CREATE TABLE {table} (id UInt64, name String) ENGINE = Memory", null, None); + + object[][] rows = Enumerable.Range(0, 2000) + .Select(i => new object[] { (ulong)i, $"name-{i}" }) + .ToArray(); + await client.InsertRowsAsync($"INSERT INTO {table} (id, name) VALUES", rows, null, None); + + var readBack = new List<(ulong Id, string Name)>(); + await foreach (object[] row in client.QueryAsync($"SELECT id, name FROM {table} ORDER BY id", null, None)) + { + readBack.Add(((ulong)row[0], (string)row[1])); + } + + Assert.Multiple(() => + { + Assert.That(readBack, Has.Count.EqualTo(2000)); + Assert.That(readBack[0], Is.EqualTo((0UL, "name-0"))); + Assert.That(readBack[^1], Is.EqualTo((1999UL, "name-1999"))); + }); + } + finally + { + await DropAsync(client, table); + } + } + + [TestCaseSource(nameof(Codecs))] + public async Task QueryAsync_ABodyLargerThanOneFrame_ReassemblesAcrossFrameBoundaries(IClickHouseCompressor codec) + { + // 400k UInt64 is ~3.2 MB of plaintext, so the server's ~1 MiB buffer emits several frames for one block + // and values land across the boundaries. Summing every row fails if a single byte is dropped or repeated. + await using var client = new ClickHouseTcpClient(TcpServerFixture.Options() with { Compressor = codec }); + + ulong count = 0; + ulong sum = 0; + await foreach (object[] row in client.QueryAsync("SELECT number FROM numbers(400000)", null, None)) + { + count++; + sum += (ulong)row[0]; + } + + Assert.Multiple(() => + { + Assert.That(count, Is.EqualTo(400000UL)); + Assert.That(sum, Is.EqualTo(400000UL * 399999UL / 2UL)); + }); + } + + [TestCaseSource(nameof(Codecs))] + public async Task InsertRowsAsync_ABlockLargerThanOneFrame_RoundTripsThroughSelect(IClickHouseCompressor codec) + { + // The write side's mirror: one block whose plaintext exceeds the frame target, so the client emits + // several frames for it and the server must accept every one. + await using var client = new ClickHouseTcpClient(TcpServerFixture.Options() with { Compressor = codec }); + string table = UniqueTableName(); + try + { + await client.ExecuteAsync($"CREATE TABLE {table} (id UInt64) ENGINE = Memory", null, None); + + object[][] rows = Enumerable.Range(0, 200000).Select(i => new object[] { (ulong)i }).ToArray(); + await client.InsertRowsAsync($"INSERT INTO {table} (id) VALUES", rows, null, None); + + var stored = new List(); + await foreach (object[] row in client.QueryAsync($"SELECT count(), sum(id) FROM {table}", null, None)) + { + stored.Add((ulong)row[0]); + stored.Add((ulong)row[1]); + } + + Assert.Multiple(() => + { + Assert.That(stored[0], Is.EqualTo(200000UL), "row count"); + Assert.That(stored[1], Is.EqualTo(200000UL * 199999UL / 2UL), "sum of ids"); + }); + } + finally + { + await DropAsync(client, table); + } + } + + [TestCaseSource(nameof(Codecs))] + public async Task QueryAsync_CompressedWithServerLogs_ReadsTheUnframedLogPacketsAndTheFramedData(IClickHouseCompressor codec) + { + // Log packets carry a block and are decoded by the same block reader, but the server does not compress + // them at our protocol target. Framing them would read a block name as a frame checksum, so this is the + // case that fails loudly if the framed-packet list ever grows to include them. + await using var client = new ClickHouseTcpClient(TcpServerFixture.Options() with { Compressor = codec }); + var options = new ClickHouseTcpQueryOptions + { + Settings = new Dictionary(StringComparer.Ordinal) { ["send_logs_level"] = "trace" }, + }; + + var values = new List(); + await foreach (object[] row in client.QueryAsync("SELECT number FROM numbers(1000)", options, None)) + { + values.Add((ulong)row[0]); + } + + Assert.That(values, Has.Count.EqualTo(1000)); + } + + [TestCaseSource(nameof(Codecs))] + public async Task QueryAsync_CompressedWithTotals_ReadsTheFramedTotalsBlock(IClickHouseCompressor codec) + { + // Totals is framed, unlike Log, so this covers the other side of the same predicate. + await using var client = new ClickHouseTcpClient(TcpServerFixture.Options() with { Compressor = codec }); + + var groups = new List(); + await foreach (object[] row in client.QueryAsync( + "SELECT number % 3 AS bucket, count() FROM numbers(100) GROUP BY bucket WITH TOTALS ORDER BY bucket", null, None)) + { + groups.Add((ulong)row[1]); + } + + Assert.That(groups, Is.Not.Empty); + } + + [TestCaseSource(nameof(Codecs))] + public async Task QueryAsync_SequentialCompressedQueries_ReuseThePooledConnection(IClickHouseCompressor codec) + { + // The frame reader and writer live for the connection's life and are reused across queries, so a second + // compressed query on the same pooled connection proves no frame state leaked out of the first. + await using var client = new ClickHouseTcpClient( + TcpServerFixture.Options() with { Compressor = codec, MaxPoolSize = 1 }); + + for (int attempt = 0; attempt < 3; attempt++) + { + var values = new List(); + await foreach (object[] row in client.QueryAsync($"SELECT number FROM numbers({100 + attempt})", null, None)) + { + values.Add((ulong)row[0]); + } + + Assert.That(values, Has.Count.EqualTo(100 + attempt), $"attempt {attempt}"); + } + } + + [Test] + public async Task QueryAsync_CompressedThenUncompressedClients_BothReadTheSameRows() + { + // The flag is per query on the wire, so a compressed and an uncompressed client must both work against + // the same server without either leaving it in a state the other cannot use. + await using var compressed = new ClickHouseTcpClient(TcpServerFixture.Options() with { Compressor = Lz4Compressor.Default }); + await using var plain = new ClickHouseTcpClient(TcpServerFixture.Options() with { Compressor = null }); + + ulong compressedSum = await SumAsync(compressed); + ulong plainSum = await SumAsync(plain); + + Assert.That(compressedSum, Is.EqualTo(plainSum)); + + static async ValueTask SumAsync(ClickHouseTcpClient client) + { + ulong sum = 0; + await foreach (object[] row in client.QueryAsync("SELECT number FROM numbers(2000)", null, None)) + { + sum += (ulong)row[0]; + } + + return sum; + } + } + + [Test] + public async Task QueryAsync_CompressedWithZstdAgainstAnLz4Request_DecodesWhicheverCodecArrives() + { + // The client's codec chooses what it writes, never what it reads: the server picks its own from + // network_compression_method. Asking for ZSTD while writing LZ4 exercises the reader's per-frame + // dispatch rather than the configured codec. + await using var client = new ClickHouseTcpClient(TcpServerFixture.Options() with { Compressor = Lz4Compressor.Default }); + var options = new ClickHouseTcpQueryOptions + { + Settings = new Dictionary(StringComparer.Ordinal) { ["network_compression_method"] = "ZSTD" }, + }; + + var values = new List(); + await foreach (object[] row in client.QueryAsync("SELECT number FROM numbers(3000)", options, None)) + { + values.Add((ulong)row[0]); + } + + Assert.That(values, Has.Count.EqualTo(3000)); + } + + private static async ValueTask DropAsync(ClickHouseTcpClient client, string table) + { + try + { + await client.ExecuteAsync($"DROP TABLE IF EXISTS {table}", null, None); + } + catch (Exception) + { + // The test's own assertions own the verdict; a failed cleanup must not mask them. + } + } + + private static string UniqueTableName() => $"tcp_compression_test_{Guid.NewGuid():N}"; +} diff --git a/ClickHouse.Driver.Tcp.Tests/Protocol/CompressedSendFailureTests.cs b/ClickHouse.Driver.Tcp.Tests/Protocol/CompressedSendFailureTests.cs new file mode 100644 index 000000000..3a7b1e2bf --- /dev/null +++ b/ClickHouse.Driver.Tcp.Tests/Protocol/CompressedSendFailureTests.cs @@ -0,0 +1,126 @@ +using System; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using ClickHouse.Driver.Compression; +using ClickHouse.Driver.Tcp.Format; +using ClickHouse.Driver.Tcp.Protocol; +using ClickHouse.Driver.Tcp.Tests.Utilities; +using ClickHouse.Driver.Tcp.Types; + +namespace ClickHouse.Driver.Tcp.Tests.Protocol; + +/// +/// What happens when sending a compressed request fails part-way. Framing the end-of-input marker is not +/// buffer-only work — each frame is flushed as it is emitted — so a failure there can leave the Query packet on +/// the wire. Such a connection must terminate rather than return to the pool looking reusable, or the next +/// caller to lease it reads the previous query's response. +/// +[TestFixture] +public class CompressedSendFailureTests +{ + private static readonly CancellationToken None = CancellationToken.None; + + [Test] + public async Task QueryAsync_Compressed_WhenTheFramedEndOfInputMarkerFailsToSend_TerminatesTheConnection() + { + var transport = new FailOnDemandStream(await FakeConnectionFactory.ServerHelloBytesAsync(None)); + var connection = new ClickHouseTcpConnection(transport, socket: null, Lz4Compressor.Default); + await connection.HandshakeAsync(new ClientHandshakeParameters { Username = "default" }, None); + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Ready), "guard: the handshake must have completed"); + + transport.FailWrites = true; + + Assert.ThrowsAsync(async () => + { + await foreach (Block _ in connection.QueryAsync("SELECT 1", cancellationToken: None)) + { + } + }); + + Assert.Multiple(() => + { + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Terminated)); + Assert.That(connection.IsReusable, Is.False); + }); + } + + [Test] + public async Task InsertAsync_Compressed_WhenTheFramedEndOfInputMarkerFailsToSend_TerminatesTheConnection() + { + var transport = new FailOnDemandStream(await FakeConnectionFactory.ServerHelloBytesAsync(None)); + var connection = new ClickHouseTcpConnection(transport, socket: null, Lz4Compressor.Default); + await connection.HandshakeAsync(new ClientHandshakeParameters { Username = "default" }, None); + + transport.FailWrites = true; + + Assert.ThrowsAsync( + async () => await connection.InsertAsync("INSERT INTO t VALUES", Array.Empty(), cancellationToken: None)); + + Assert.That(connection.State, Is.EqualTo(TcpConnectionState.Terminated)); + } + + /// Serves a fixed read script, and fails writes once armed. Reads keep working, so only the send path breaks. + private sealed class FailOnDemandStream : Stream + { + private readonly MemoryStream script; + + public FailOnDemandStream(byte[] readScript) => script = new MemoryStream(readScript); + + public bool FailWrites { get; set; } + + public override bool CanRead => true; + + public override bool CanSeek => false; + + public override bool CanWrite => true; + + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public override ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + => script.ReadAsync(buffer, cancellationToken); + + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + => script.ReadAsync(buffer, offset, count, cancellationToken); + + public override int Read(byte[] buffer, int offset, int count) => script.Read(buffer, offset, count); + + public override ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) + => FailWrites + ? ValueTask.FromException(new IOException("simulated send failure")) + : ValueTask.CompletedTask; + + public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + => WriteAsync(new ReadOnlyMemory(buffer, offset, count), cancellationToken).AsTask(); + + public override void Write(byte[] buffer, int offset, int count) + { + if (FailWrites) + { + throw new IOException("simulated send failure"); + } + } + + public override void Flush() + { + } + + public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + + protected override void Dispose(bool disposing) + { + script.Dispose(); + base.Dispose(disposing); + } + } +} diff --git a/ClickHouse.Driver.Tcp/ClickHouse.Driver.Tcp.csproj b/ClickHouse.Driver.Tcp/ClickHouse.Driver.Tcp.csproj index aefdc7b93..97e1dff77 100644 --- a/ClickHouse.Driver.Tcp/ClickHouse.Driver.Tcp.csproj +++ b/ClickHouse.Driver.Tcp/ClickHouse.Driver.Tcp.csproj @@ -24,4 +24,10 @@ + + + + + diff --git a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs index 307270254..7a73361d6 100644 --- a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs +++ b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs @@ -1,6 +1,7 @@ using System; using System.Collections.Generic; using System.Net.Security; +using ClickHouse.Driver.Compression; using ClickHouse.Driver.Tcp.Protocol; namespace ClickHouse.Driver.Tcp; @@ -28,6 +29,17 @@ public sealed record ClickHouseTcpClientOptions internal const int DefaultMinPoolSize = 0; internal const int DefaultMaxPoolSize = 20; internal const ClickHouseTcpPoolReusePolicy DefaultPoolReusePolicy = ClickHouseTcpPoolReusePolicy.Lifo; + + /// + /// The Compression connection-string value used when the key is absent. LZ4 is what + /// clickhouse-client uses on this protocol: the cheapest codec in CPU and the lightest on the + /// server. Pass Compression=none, or a null , to turn it off. + /// + internal const string DefaultCompression = CompressionLz4; + + internal const string CompressionNone = "none"; + internal const string CompressionLz4 = "lz4"; + internal const string CompressionZstd = "zstd"; internal static readonly TimeSpan DefaultDialTimeout = TimeSpan.FromSeconds(30); internal static readonly TimeSpan DefaultReadTimeout = TimeSpan.FromSeconds(300); internal static readonly TimeSpan DefaultPoolTimeout = TimeSpan.FromSeconds(30); @@ -255,6 +267,24 @@ public sealed record ClickHouseTcpClientOptions /// Which idle connection the pool hands out next. Defaults to . public ClickHouseTcpPoolReusePolicy PoolReusePolicy { get; init; } = DefaultPoolReusePolicy; + /// + /// Codec for the native protocol's compression frames, or to exchange blocks + /// uncompressed. Use (cheapest, lowest server-side load) or + /// (smaller, more CPU); a custom works if + /// it implements the native block path. + /// + /// Compression is requested per query, so this is the default for every query the client runs. It governs + /// both directions: the server compresses the blocks it sends and expects the client's own blocks framed + /// the same way. Null means the request carries no compression at all, which is not the same as a frame + /// whose method byte is NONE. + /// + /// + /// A codec chooses the method byte and the body encoding, but never the decoding: the server picks its own + /// codec, so a client that asks for LZ4 can still be sent ZSTD and must decode whatever arrives. + /// + /// + public IClickHouseCompressor Compressor { get; init; } = ResolveCompressor(DefaultCompression); + /// /// These options with replaced by a private snapshot, or this instance when there /// are none to copy. A client holds its options for its lifetime and merges the settings on every operation, so @@ -280,11 +310,45 @@ public static ClickHouseTcpClientOptions FromConnectionString(string connectionS return new ClickHouseTcpConnectionStringBuilder(connectionString).ToOptions(); } + /// + /// Maps a Compression connection-string value to its codec. none yields + /// , meaning the query is not compressed at all. + /// + /// The codec name, case-insensitive. Null or empty means none. + /// The codec, or null for no compression. + /// names no known codec. + internal static IClickHouseCompressor ResolveCompressor(string name) => name?.Trim().ToLowerInvariant() switch + { + null or "" or CompressionNone => null, + CompressionLz4 => Lz4Compressor.Default, + CompressionZstd => ZstdCompressor.Default, + _ => throw new ArgumentException( + $"Compression '{name}' is not a known codec; expected '{CompressionLz4}', '{CompressionZstd}' or '{CompressionNone}'.", + nameof(name)), + }; + /// Validates the options, throwing if any value is unusable. Runs at client construction. /// , , or is null or empty. /// is out of range, or a timeout / buffer size is not positive. internal void Validate() { + // A codec that only implements the HTTP body path cannot frame a block. Refuse it here rather than + // mid-query, where the Query packet has already promised the server compressed blocks. + if (Compressor is not null) + { + try + { + _ = Compressor.MethodByte; + } + catch (NotSupportedException e) + { + throw new ArgumentException( + $"{Compressor.GetType().Name} does not support the native block path, so it cannot frame a block; use {nameof(Lz4Compressor)} or {nameof(ZstdCompressor)}.", + nameof(Compressor), + e); + } + } + if (string.IsNullOrEmpty(Host)) { throw new ArgumentException("Host must not be null or empty.", nameof(Host)); diff --git a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs index d680d1614..a6208d295 100644 --- a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs +++ b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs @@ -254,6 +254,17 @@ public ClickHouseTcpPoolReusePolicy PoolReusePolicy set => this["PoolReusePolicy"] = value.ToString(); } + /// + /// Frame codec for the native protocol: lz4, zstd, or none. Defaults to + /// lz4, so compression is on unless this key turns it off. Maps to + /// . + /// + public string Compression + { + get => GetStringOrDefault("Compression", ClickHouseTcpClientOptions.DefaultCompression); + set => this["Compression"] = value; + } + /// Materializes these keys into a , folding set_* keys into . /// The equivalent options. public ClickHouseTcpClientOptions ToOptions() @@ -297,6 +308,7 @@ public ClickHouseTcpClientOptions ToOptions() IdleTimeout = IdleTimeout, SweepInterval = SweepInterval, PoolReusePolicy = PoolReusePolicy, + Compressor = ClickHouseTcpClientOptions.ResolveCompressor(Compression), CustomSettings = customSettings, }; } diff --git a/ClickHouse.Driver.Tcp/Client/IConnectionFactory.cs b/ClickHouse.Driver.Tcp/Client/IConnectionFactory.cs index 57570d2a4..9c5e92e77 100644 --- a/ClickHouse.Driver.Tcp/Client/IConnectionFactory.cs +++ b/ClickHouse.Driver.Tcp/Client/IConnectionFactory.cs @@ -52,7 +52,7 @@ public async ValueTask CreateAsync(CancellationToken ca try { return await ClickHouseTcpConnection.ConnectAsync( - options.Host, options.ResolvedPort, options.ToHandshakeParameters(), tls, linked.Token).ConfigureAwait(false); + options.Host, options.ResolvedPort, options.ToHandshakeParameters(), tls, linked.Token, options.Compressor).ConfigureAwait(false); } catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested && linked.IsCancellationRequested) { diff --git a/ClickHouse.Driver.Tcp/Compression/CompressedFrameReader.cs b/ClickHouse.Driver.Tcp/Compression/CompressedFrameReader.cs new file mode 100644 index 000000000..5921a13ac --- /dev/null +++ b/ClickHouse.Driver.Tcp/Compression/CompressedFrameReader.cs @@ -0,0 +1,196 @@ +using System; +using System.Buffers; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using ClickHouse.Driver.Compression; +using ClickHouse.Driver.Tcp.Protocol; + +namespace ClickHouse.Driver.Tcp.Compression; + +/// +/// Serves a block body that arrives as a stream of compression frames, so the block and column decoders +/// above it read plaintext and never learn that compression is on. +/// +/// Frames are pulled through the connection's raw reader rather than from the socket directly, so +/// there stays exactly one owner of buffered socket bytes. is a second reader over the +/// decoded plaintext; the packet envelope (the type code and the table name) is read from the raw one, +/// because the server writes those outside the frames. +/// +/// +/// Read-ahead is deliberately limited to the current frame: a read is served short at a frame boundary and +/// the next frame is pulled only when a caller asks for more. Reading ahead past a block's last frame would +/// consume the next packet's uncompressed envelope as if it were frame bytes. +/// +/// One per connection, reused across the blocks of a query. Not thread-safe. +/// +internal sealed class CompressedFrameReader : IDisposable +{ + private readonly ClickHouseBinaryReader raw; + private byte[] prefix; // the 16-byte checksum plus the 9-byte header + private byte[] headerAndBody; // the header and body, contiguous, as the checksum spans them + private byte[] plaintext; // the current frame, decoded + private int position; // bytes of `plaintext` already served + private int length; // bytes of `plaintext` that are valid + private bool disposed; + + /// Initializes a frame reader that pulls its frames from . + /// The connection's reader, positioned at a frame boundary when a body starts. + /// Capacity of the buffer that serves decoded plaintext to . + /// is null. + public CompressedFrameReader(ClickHouseBinaryReader raw, int bufferSize = 16384) + { + this.raw = raw ?? throw new ArgumentNullException(nameof(raw)); + prefix = ArrayPool.Shared.Rent(CompressionFrame.PrefixSize); + headerAndBody = ArrayPool.Shared.Rent(CompressionFrame.HeaderSize); + plaintext = ArrayPool.Shared.Rent(1); + Reader = new ClickHouseBinaryReader(new ReadBuffer(new PlaintextStream(this), bufferSize), ownsBuffer: true); + } + + /// Reads the decoded block body. Everything after the packet's table name comes from here. + public ClickHouseBinaryReader Reader { get; } + + /// Decoded bytes of the current frame that no caller has taken yet. + public int PendingPlaintext => length - position; + + /// + /// Asserts that a finished block consumed its frames exactly. The sender flushes at every block end, so + /// a block's last frame ends where the block does; anything left means the decoders and the peer + /// disagree about the body's length, and the connection can no longer be trusted. + /// + /// Decoded plaintext was left unread. + public void EndBlock() + { + int leftover = PendingPlaintext + Reader.BufferedBytes; + if (leftover != 0) + { + throw new ClickHouseProtocolException( + $"A compressed block left {leftover} decoded byte(s) unread, so its frames carried more than the block declared. The connection is out of step."); + } + } + + /// + public void Dispose() + { + if (disposed) + { + return; + } + + disposed = true; + Reader.Dispose(); + ArrayPool.Shared.Return(prefix); + ArrayPool.Shared.Return(headerAndBody); + ArrayPool.Shared.Return(plaintext); + prefix = Array.Empty(); + headerAndBody = Array.Empty(); + plaintext = Array.Empty(); + } + + /// + /// Reads the next frame: its prefix, its body, then verifies the checksum and decodes the body. + /// + /// A token to observe for cancellation. + /// The frame is malformed, its checksum fails, or it declares no plaintext. + private async ValueTask PullFrameAsync(CancellationToken cancellationToken) + { + await raw.ReadBytesAsync(prefix.AsMemory(0, CompressionFrame.PrefixSize), cancellationToken).ConfigureAwait(false); + CompressionFrame.ReadHeader( + prefix.AsSpan(CompressionFrame.ChecksumSize, CompressionFrame.HeaderSize), + out byte method, + out int bodySize, + out int plaintextSize); + + if (plaintextSize == 0) + { + // A block body is never empty (it carries at least the block info and two counts), and serving + // zero bytes would spin the read loop rather than make progress. + throw new InvalidDataException("Compression frame declares no plaintext, which cannot occur inside a block body (corrupt stream)."); + } + + // The checksum covers the header and body as one run, so keep them contiguous. + int framed = CompressionFrame.HeaderSize + bodySize; + Grow(ref headerAndBody, framed); + prefix.AsSpan(CompressionFrame.ChecksumSize, CompressionFrame.HeaderSize).CopyTo(headerAndBody); + await raw.ReadBytesAsync(headerAndBody.AsMemory(CompressionFrame.HeaderSize, bodySize), cancellationToken).ConfigureAwait(false); + + CompressionFrame.VerifyChecksum(prefix.AsSpan(0, CompressionFrame.ChecksumSize), headerAndBody.AsSpan(0, framed)); + + Grow(ref plaintext, plaintextSize); + CompressionFrame.Decode(method, headerAndBody.AsSpan(CompressionFrame.HeaderSize, bodySize), plaintext.AsSpan(0, plaintextSize)); + position = 0; + length = plaintextSize; + } + + /// Replaces with a larger pooled one when it cannot hold bytes. + private static void Grow(ref byte[] buffer, int needed) + { + if (buffer.Length >= needed) + { + return; + } + + ArrayPool.Shared.Return(buffer); + buffer = ArrayPool.Shared.Rent(needed); + } + + /// + /// The stream face the plaintext fills from. Each read serves bytes from the + /// current frame only, pulling the next frame when the current one runs out. + /// + private sealed class PlaintextStream : Stream + { + private readonly CompressedFrameReader owner; + + public PlaintextStream(CompressedFrameReader owner) => this.owner = owner; + + public override bool CanRead => true; + + public override bool CanSeek => false; + + public override bool CanWrite => false; + + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public override async ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + { + if (buffer.IsEmpty) + { + return 0; + } + + if (owner.PendingPlaintext == 0) + { + await owner.PullFrameAsync(cancellationToken).ConfigureAwait(false); + } + + // Never crosses a frame boundary: a short read keeps the next frame unread until it is needed. + int taken = Math.Min(buffer.Length, owner.PendingPlaintext); + owner.plaintext.AsMemory(owner.position, taken).CopyTo(buffer); + owner.position += taken; + return taken; + } + + public override Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + => ReadAsync(new Memory(buffer, offset, count), cancellationToken).AsTask(); + + public override int Read(byte[] buffer, int offset, int count) + => throw new NotSupportedException("The frame reader is asynchronous; use ReadAsync."); + + public override void Flush() + { + } + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + + public override void Write(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + } +} diff --git a/ClickHouse.Driver.Tcp/Compression/CompressedFrameWriter.cs b/ClickHouse.Driver.Tcp/Compression/CompressedFrameWriter.cs new file mode 100644 index 000000000..5abf03de4 --- /dev/null +++ b/ClickHouse.Driver.Tcp/Compression/CompressedFrameWriter.cs @@ -0,0 +1,154 @@ +using System; +using System.Buffers; +using System.IO; +using System.Threading; +using System.Threading.Tasks; +using ClickHouse.Driver.Compression; +using ClickHouse.Driver.Tcp.Protocol; + +namespace ClickHouse.Driver.Tcp.Compression; + +/// +/// Frames an outgoing block body, so the block and column encoders above it write plaintext and never learn +/// that compression is on. +/// +/// With compression requested, the server expects the client's own Data blocks framed the same way it frames +/// its own — the end-of-input marker, external tables and INSERT rows alike. Only the body is framed: the +/// packet type code and the table name go to the raw writer, ahead of the frames. +/// +/// +/// Plaintext is cut into frames of at most , matching the server's own +/// buffer, and each finished frame is flushed. A frame boundary inside a body is legal — the reader finds a +/// block's end from its own dimensions, not from the framing — and flushing per frame keeps at most one +/// frame buffered however large the block is. +/// +/// One per connection, reused across the blocks of a query. Not thread-safe. +/// +internal sealed class CompressedFrameWriter : IDisposable +{ + /// Plaintext per frame, matching the server's DBMS_DEFAULT_BUFFER_SIZE. + public const int DefaultFrameTargetBytes = 1024 * 1024; + + private readonly ClickHouseBinaryWriter raw; + private readonly IClickHouseCompressor codec; + private readonly int frameTarget; + private byte[] frame; + private bool disposed; + + /// Initializes a frame writer that emits its frames into . + /// The connection's writer, positioned after the packet envelope when a body starts. + /// The codec supplying the method byte and body encoding. + /// Plaintext bytes per frame; defaults to . + /// Initial capacity of the plaintext writer's buffer. + /// or is null. + /// is not positive. + public CompressedFrameWriter(ClickHouseBinaryWriter raw, IClickHouseCompressor codec, int frameTarget = DefaultFrameTargetBytes, int bufferSize = 16384) + { + this.raw = raw ?? throw new ArgumentNullException(nameof(raw)); + this.codec = codec ?? throw new ArgumentNullException(nameof(codec)); + ArgumentOutOfRangeException.ThrowIfNegativeOrZero(frameTarget); + + this.frameTarget = frameTarget; + frame = ArrayPool.Shared.Rent(CompressionFrame.MaxFrameSize(codec, frameTarget)); + Writer = new ClickHouseBinaryWriter(new PlaintextSink(this), bufferSize); + } + + /// Writes the block body. Everything after the packet's table name goes here. + public ClickHouseBinaryWriter Writer { get; } + + /// + /// Ends a block: flushes the plaintext writer, so whatever remains becomes the body's final frame. The + /// reader relies on this — a block end must coincide with a frame boundary. + /// + /// A token to observe for cancellation. + public async ValueTask EndBlockAsync(CancellationToken cancellationToken) + { + await Writer.FlushAsync(cancellationToken).ConfigureAwait(false); + Writer.TrimBuffer(); + } + + /// + public void Dispose() + { + if (disposed) + { + return; + } + + disposed = true; + Writer.Dispose(); + ArrayPool.Shared.Return(frame); + frame = Array.Empty(); + } + + /// Compresses one frame's worth of plaintext and writes the frame to the raw writer. + /// At most bytes. + /// A token to observe for cancellation. + private async ValueTask EmitFrameAsync(ReadOnlyMemory plaintext, CancellationToken cancellationToken) + { + int required = CompressionFrame.MaxFrameSize(codec, plaintext.Length); + if (frame.Length < required) + { + ArrayPool.Shared.Return(frame); + frame = ArrayPool.Shared.Rent(required); + } + + int written = CompressionFrame.Write(plaintext.Span, codec, frame); + raw.WriteBytes(frame.AsSpan(0, written)); + await raw.FlushAsync(cancellationToken).ConfigureAwait(false); + } + + /// + /// The stream face the plaintext flushes into. Each flush is cut + /// into frames of at most the frame target. + /// + private sealed class PlaintextSink : Stream + { + private readonly CompressedFrameWriter owner; + + public PlaintextSink(CompressedFrameWriter owner) => this.owner = owner; + + public override bool CanRead => false; + + public override bool CanSeek => false; + + public override bool CanWrite => true; + + public override long Length => throw new NotSupportedException(); + + public override long Position + { + get => throw new NotSupportedException(); + set => throw new NotSupportedException(); + } + + public override async ValueTask WriteAsync(ReadOnlyMemory buffer, CancellationToken cancellationToken = default) + { + while (!buffer.IsEmpty) + { + int take = Math.Min(owner.frameTarget, buffer.Length); + await owner.EmitFrameAsync(buffer.Slice(0, take), cancellationToken).ConfigureAwait(false); + buffer = buffer.Slice(take); + } + } + + public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + => WriteAsync(new ReadOnlyMemory(buffer, offset, count), cancellationToken).AsTask(); + + public override void Write(byte[] buffer, int offset, int count) + => throw new NotSupportedException("The frame writer is asynchronous; use WriteAsync."); + + public override void Flush() + { + } + + // The raw writer is flushed as each frame is emitted, so there is nothing buffered here to push. + public override Task FlushAsync(CancellationToken cancellationToken) => Task.CompletedTask; + + public override int Read(byte[] buffer, int offset, int count) => throw new NotSupportedException(); + + public override long Seek(long offset, SeekOrigin origin) => throw new NotSupportedException(); + + public override void SetLength(long value) => throw new NotSupportedException(); + } +} diff --git a/ClickHouse.Driver.Tcp/Compression/FramedPackets.cs b/ClickHouse.Driver.Tcp/Compression/FramedPackets.cs new file mode 100644 index 000000000..87df5a6c8 --- /dev/null +++ b/ClickHouse.Driver.Tcp/Compression/FramedPackets.cs @@ -0,0 +1,26 @@ +using ClickHouse.Driver.Tcp.Protocol; + +namespace ClickHouse.Driver.Tcp.Compression; + +/// +/// Which packet bodies travel in compression frames when a query requests compression. +/// +/// Not every block-bearing packet is framed, which is the trap here: Log and ProfileEvents +/// carry blocks and are decoded by the same block reader, yet the server sends them uncompressed at our +/// protocol target (they become compressible only at revision 54481). Framing them would read a block name +/// as a frame checksum and desynchronize the connection the first time a server has send_logs_level +/// set. +/// +/// +/// The list matches the server's own notion of a compressible message, and the Go client's +/// ServerCode.Compressible, which admits the same three. +/// +/// +internal static class FramedPackets +{ + /// Whether this server packet's block body arrives in frames while compression is active. + /// The packet type just read from the envelope. + /// True when the body is framed; false when it is written straight to the raw stream. + public static bool CarriesFramedBody(ServerPacketType packet) + => packet is ServerPacketType.Data or ServerPacketType.Totals or ServerPacketType.Extremes; +} diff --git a/ClickHouse.Driver.Tcp/Format/BlockReader.cs b/ClickHouse.Driver.Tcp/Format/BlockReader.cs index 77a930ff8..f3cafd4a5 100644 --- a/ClickHouse.Driver.Tcp/Format/BlockReader.cs +++ b/ClickHouse.Driver.Tcp/Format/BlockReader.cs @@ -28,6 +28,33 @@ public static async ValueTask ReadBlockAsync( CancellationToken cancellationToken) { string name = await reader.ReadStringAsync(cancellationToken).ConfigureAwait(false); + return await ReadBodyAsync(reader, name, negotiated, registry, context, cancellationToken).ConfigureAwait(false); + } + + /// + /// Reads a block whose name has already been consumed, starting at the block info. + /// + /// The split exists because the name belongs to the packet envelope rather than to the block: with + /// compression on, the name arrives uncompressed and everything from here on arrives in frames, so the + /// caller reads the name from the raw reader and passes a plaintext reader here. + /// + /// + /// The reader positioned at the block info, plaintext if the body is framed. + /// The block name the caller already read. + /// The negotiated protocol, for version-gated header fields. + /// The registry that resolves each column's type string to a codec. + /// The resolution context (e.g. the server timezone) passed to each column's codec factory. + /// A token to observe for cancellation. + /// The decoded block. + /// A column uses unsupported custom serialization, or a count is implausible. + public static async ValueTask ReadBodyAsync( + ClickHouseBinaryReader reader, + string name, + NegotiatedProtocol negotiated, + ColumnCodecRegistry registry, + ResolveContext context, + CancellationToken cancellationToken) + { BlockInfo info = await ReadBlockInfoAsync(reader, cancellationToken).ConfigureAwait(false); int columnCount = ToColumnCount(await reader.ReadVarUIntAsync(cancellationToken).ConfigureAwait(false)); diff --git a/ClickHouse.Driver.Tcp/Format/BlockWriter.cs b/ClickHouse.Driver.Tcp/Format/BlockWriter.cs index 7df08a9cc..efc293548 100644 --- a/ClickHouse.Driver.Tcp/Format/BlockWriter.cs +++ b/ClickHouse.Driver.Tcp/Format/BlockWriter.cs @@ -27,6 +27,20 @@ internal static class BlockWriter public static void WriteEmptyBlock(ClickHouseBinaryWriter writer) { writer.WriteString(string.Empty); + WriteEmptyBlockBody(writer); + } + + /// + /// Writes the empty end-of-input block without its name, which the caller has already written. + /// + /// The split exists because the name belongs to the packet envelope rather than to the block: with + /// compression on the name is written raw and the body is framed, so the caller writes the name to the + /// raw writer and passes a plaintext writer here. + /// + /// + /// The writer to encode the body into, plaintext if the body is framed. + public static void WriteEmptyBlockBody(ClickHouseBinaryWriter writer) + { WriteBlockInfo(writer, BlockInfo.Default); writer.WriteVarUInt(0); // num_columns writer.WriteVarUInt(0); // num_rows @@ -97,18 +111,46 @@ public static async ValueTask WriteDataBlockAsync( int flushThresholdBytes, CancellationToken cancellationToken) { - // The requested range must lie within every column, or the body would run past the values. Catch it first. - foreach (InsertColumn column in columns) - { - if (start < 0 || rowCount < 0 || start + (long)rowCount > column.Values.RowCount) - { - throw new ArgumentException( - $"Column '{column.Name}' cannot supply rows [{start}, {start + (long)rowCount}) of its {column.Values.RowCount} row(s).", - nameof(columns)); - } - } + EnsureRangeWithinColumns(columns, start, rowCount); writer.WriteString(string.Empty); // table_name: empty for the INSERT row stream + await WriteDataBlockBodyCoreAsync(writer, negotiated, columns, start, rowCount, flushThresholdBytes, cancellationToken).ConfigureAwait(false); + } + + /// + /// Writes a populated block's body without its name, which the caller has already written. See + /// for why the name is written separately. + /// + /// The writer to encode the body into, plaintext if the body is framed. + /// The negotiated protocol, gating the has_custom_serialization byte. + /// The columns to write, in header order. + /// The zero-based first row of the range each column contributes. + /// The number of rows the block holds. + /// The buffered-byte cap that triggers a between-column flush. + /// A token to observe for cancellation. + public static async ValueTask WriteDataBlockBodyAsync( + ClickHouseBinaryWriter writer, + NegotiatedProtocol negotiated, + IReadOnlyList columns, + int start, + int rowCount, + int flushThresholdBytes, + CancellationToken cancellationToken) + { + EnsureRangeWithinColumns(columns, start, rowCount); + await WriteDataBlockBodyCoreAsync(writer, negotiated, columns, start, rowCount, flushThresholdBytes, cancellationToken).ConfigureAwait(false); + } + + /// Writes the block info, the counts and every column. The row range is already validated. + private static async ValueTask WriteDataBlockBodyCoreAsync( + ClickHouseBinaryWriter writer, + NegotiatedProtocol negotiated, + IReadOnlyList columns, + int start, + int rowCount, + int flushThresholdBytes, + CancellationToken cancellationToken) + { WriteBlockInfo(writer, BlockInfo.Default); writer.WriteVarUInt((ulong)columns.Count); // num_columns writer.WriteVarUInt((ulong)rowCount); // num_rows @@ -149,6 +191,27 @@ public static async ValueTask WriteDataBlockAsync( } } + /// + /// Checks that the requested row range lies within every column, before anything is written: a range past + /// the values would run the body off the end of a column. + /// + /// The columns the block will draw from. + /// The zero-based first row of the range. + /// The number of rows requested. + /// A column cannot supply the requested range. + private static void EnsureRangeWithinColumns(IReadOnlyList columns, int start, int rowCount) + { + foreach (InsertColumn column in columns) + { + if (start < 0 || rowCount < 0 || start + (long)rowCount > column.Values.RowCount) + { + throw new ArgumentException( + $"Column '{column.Name}' cannot supply rows [{start}, {start + (long)rowCount}) of its {column.Values.RowCount} row(s).", + nameof(columns)); + } + } + } + /// Writes the field-id-tagged block info: is_overflows, bucket_number, then the terminator. /// The writer to encode into. /// The block info to write. diff --git a/ClickHouse.Driver.Tcp/Protocol/ClickHouseTcpConnection.cs b/ClickHouse.Driver.Tcp/Protocol/ClickHouseTcpConnection.cs index 38a8126ae..d24bcb5f4 100644 --- a/ClickHouse.Driver.Tcp/Protocol/ClickHouseTcpConnection.cs +++ b/ClickHouse.Driver.Tcp/Protocol/ClickHouseTcpConnection.cs @@ -6,6 +6,8 @@ using System.Runtime.ExceptionServices; using System.Threading; using System.Threading.Tasks; +using ClickHouse.Driver.Compression; +using ClickHouse.Driver.Tcp.Compression; using ClickHouse.Driver.Tcp.Format; using ClickHouse.Driver.Tcp.Types; @@ -42,6 +44,17 @@ internal sealed class ClickHouseTcpConnection : IDisposable, IAsyncDisposable private readonly Stream stream; private readonly ClickHouseBinaryReader reader; private readonly ClickHouseBinaryWriter writer; + + // Null means every query on this connection is uncompressed. Compression is per-query on the wire, but the + // codec is a client-level option today, so it is fixed for a connection's life; a per-query override would + // move this to the operation entry points. + private readonly IClickHouseCompressor compressor; + + // Created on the first compressed block and kept for the connection's life, so their pooled buffers are + // reused across blocks and queries rather than rented per block. + private CompressedFrameReader frameReader; + private CompressedFrameWriter frameWriter; + private ServerHandshake server; private ClientMetadata clientMetadata; private TcpConnectionState state; @@ -53,10 +66,12 @@ internal sealed class ClickHouseTcpConnection : IDisposable, IAsyncDisposable /// /// The duplex transport stream (a network stream in production). /// The underlying socket, closed on termination; null when the stream owns teardown. - internal ClickHouseTcpConnection(Stream stream, Socket socket) + /// Frame codec for this connection's queries, or null to run them uncompressed. + internal ClickHouseTcpConnection(Stream stream, Socket socket, IClickHouseCompressor compressor = null) { this.stream = stream; this.socket = socket; + this.compressor = compressor; reader = new ClickHouseBinaryReader(stream); writer = new ClickHouseBinaryWriter(stream); state = TcpConnectionState.Handshaking; @@ -109,6 +124,13 @@ internal bool IsReusable return false; } + // Decoded plaintext nobody read is the same fault one layer up: the last response's frames carried + // more than its blocks declared, so this side's idea of the stream position is wrong. + if (frameReader is { PendingPlaintext: not 0 }) + { + return false; + } + // The scripted-stream seam has no socket; there is nothing to poll, so trust the state. if (socket is null) { @@ -177,7 +199,8 @@ public static async ValueTask ConnectAsync( int port, ClientHandshakeParameters handshake, TlsParameters tls, - CancellationToken cancellationToken) + CancellationToken cancellationToken, + IClickHouseCompressor compressor = null) { ArgumentNullException.ThrowIfNull(host); ArgumentNullException.ThrowIfNull(handshake); @@ -213,7 +236,7 @@ public static async ValueTask ConnectAsync( // HandshakeAsync terminates the connection (closing this socket) on any failure, so a throw here needs // no extra cleanup. - var connection = new ClickHouseTcpConnection(transport, socket); + var connection = new ClickHouseTcpConnection(transport, socket, compressor); await connection.HandshakeAsync(handshake, cancellationToken).ConfigureAwait(false); return connection; } @@ -340,14 +363,12 @@ internal async IAsyncEnumerable QueryAsync( Block current = null; bool completed = false; - // Encode the request into the write buffer before any of it reaches the socket. A failure here is a + // Encode the Query packet into the write buffer before any of it reaches the socket. A failure here is a // client-side error (e.g. parameters on a protocol revision that predates them): nothing has been sent, // so discard the partial packet and leave the connection Ready and reusable rather than terminating it. try { - Query.Write(writer, negotiated, clientMetadata, queryId, sql, settings, parameters); - writer.WriteClientPacketType(ClientPacketType.Data); - BlockWriter.WriteEmptyBlock(writer); + Query.Write(writer, negotiated, clientMetadata, queryId, sql, settings, parameters, compressor is not null); } catch { @@ -358,6 +379,11 @@ internal async IAsyncEnumerable QueryAsync( try { + // The end-of-input marker is written here rather than above, because framing it is not buffer-only + // work: each frame is flushed as it is emitted. A failure part-way through would leave the Query + // packet on the wire, so it must terminate the connection instead of returning it to the pool + // looking reusable — the reusable path above holds only work that cannot have sent anything. + await WriteEndOfInputBlockAsync(cancellationToken).ConfigureAwait(false); await writer.FlushAsync(cancellationToken).ConfigureAwait(false); while (true) @@ -386,7 +412,7 @@ internal async IAsyncEnumerable QueryAsync( if (packet == ServerPacketType.Data) { - Block block = await BlockReader.ReadBlockAsync(reader, negotiated, ColumnCodecRegistry.Default, readContext, cancellationToken).ConfigureAwait(false); + Block block = await ReadBlockAsync(ServerPacketType.Data, negotiated, readContext, cancellationToken).ConfigureAwait(false); if (block.RowCount != 0) { // Held as the current block so it is released when the consumer advances or stops. @@ -551,9 +577,8 @@ private async ValueTask InsertCoreAsync( { // The empty end-of-input block must follow the Query: the server waits for it before sending the // schema block, so omitting it deadlocks. - Query.Write(writer, negotiated, clientMetadata, queryId, sql, settings, parameters); - writer.WriteClientPacketType(ClientPacketType.Data); - BlockWriter.WriteEmptyBlock(writer); + Query.Write(writer, negotiated, clientMetadata, queryId, sql, settings, parameters, compressor is not null); + await WriteEndOfInputBlockAsync(cancellationToken).ConfigureAwait(false); await writer.FlushAsync(cancellationToken).ConfigureAwait(false); // Drain metadata until the schema block (the first Data packet) or a terminal packet. @@ -731,15 +756,13 @@ private async ValueTask StreamInsertRowsAsync( // Row count controls block splitting; the flush threshold bounds buffered output within each block. foreach ((int start, int length) in PlanInsertBlocks(rowCount, maxRowsPerBlock)) { - writer.WriteClientPacketType(ClientPacketType.Data); - await BlockWriter.WriteDataBlockAsync( - writer, negotiated, plan, start, length, flushThresholdBytes, cancellationToken).ConfigureAwait(false); + await WriteDataBlockPacketAsync( + negotiated, plan, start, length, flushThresholdBytes, cancellationToken).ConfigureAwait(false); await writer.FlushAsync(cancellationToken).ConfigureAwait(false); } } - writer.WriteClientPacketType(ClientPacketType.Data); - BlockWriter.WriteEmptyBlock(writer); + await WriteEndOfInputBlockAsync(cancellationToken).ConfigureAwait(false); await writer.FlushAsync(cancellationToken).ConfigureAwait(false); // Return the pooled buffer to baseline so an idle connection doesn't retain a large insert's peak size. @@ -893,18 +916,104 @@ private static string DescribeSchemaMismatch(IReadOnlyList columns, Blo return $"The insert columns do not match the target schema — {string.Join("; ", parts)}. Columns are matched to the target by name."; } + /// + /// Reads a block whose name is next on the raw stream. The name belongs to the packet envelope and is never + /// compressed, so it is read from the raw reader; the body comes from the frame reader when compression is + /// active and this packet is one whose body the server frames. + /// + /// This is the only place in the driver that knows two readers exist. Everything below it — the block + /// reader, every column codec — is handed one reader and cannot tell which. + /// + /// + /// The packet type just read from the envelope, which decides whether the body is framed. + /// The negotiated protocol, for version-gated header fields. + /// The resolution context passed to each column's codec factory. + /// A token to observe for cancellation. + /// The decoded block. + private async ValueTask ReadBlockAsync( + ServerPacketType packet, + NegotiatedProtocol negotiated, + ResolveContext context, + CancellationToken cancellationToken) + { + string name = await reader.ReadStringAsync(cancellationToken).ConfigureAwait(false); + + if (compressor is null || !FramedPackets.CarriesFramedBody(packet)) + { + return await BlockReader.ReadBodyAsync(reader, name, negotiated, ColumnCodecRegistry.Default, context, cancellationToken).ConfigureAwait(false); + } + + frameReader ??= new CompressedFrameReader(reader); + Block block = await BlockReader.ReadBodyAsync(frameReader.Reader, name, negotiated, ColumnCodecRegistry.Default, context, cancellationToken).ConfigureAwait(false); + + // A block end coincides with a frame boundary, so anything left decoded means the peer and the column + // decoders disagree about the body's length. + frameReader.EndBlock(); + return block; + } + + /// + /// Writes a Data packet carrying the empty end-of-input block. With compression on the server expects the + /// client's own blocks framed too, this marker included, so it is framed like any other body. + /// + /// A token to observe for cancellation. + private async ValueTask WriteEndOfInputBlockAsync(CancellationToken cancellationToken) + { + writer.WriteClientPacketType(ClientPacketType.Data); + writer.WriteString(string.Empty); // table_name: envelope, never framed + + if (compressor is null) + { + BlockWriter.WriteEmptyBlockBody(writer); + return; + } + + frameWriter ??= new CompressedFrameWriter(writer, compressor); + BlockWriter.WriteEmptyBlockBody(frameWriter.Writer); + await frameWriter.EndBlockAsync(cancellationToken).ConfigureAwait(false); + } + + /// Writes a Data packet carrying rows [start, start + rowCount), framed when compression is on. + /// The negotiated protocol, gating the has_custom_serialization byte. + /// The columns to write, in header order. + /// The zero-based first row of the range each column contributes. + /// The number of rows the block holds. + /// The buffered-byte cap that triggers a between-column flush. + /// A token to observe for cancellation. + private async ValueTask WriteDataBlockPacketAsync( + NegotiatedProtocol negotiated, + IReadOnlyList columns, + int start, + int rowCount, + int flushThresholdBytes, + CancellationToken cancellationToken) + { + writer.WriteClientPacketType(ClientPacketType.Data); + writer.WriteString(string.Empty); // table_name: empty for the INSERT row stream, and never framed + + if (compressor is null) + { + await BlockWriter.WriteDataBlockBodyAsync(writer, negotiated, columns, start, rowCount, flushThresholdBytes, cancellationToken).ConfigureAwait(false); + return; + } + + frameWriter ??= new CompressedFrameWriter(writer, compressor); + await BlockWriter.WriteDataBlockBodyAsync(frameWriter.Writer, negotiated, columns, start, rowCount, flushThresholdBytes, cancellationToken).ConfigureAwait(false); + await frameWriter.EndBlockAsync(cancellationToken).ConfigureAwait(false); + } + /// /// Reads a metadata block, lends it to the handler if one is set (borrowed only for the duration of the /// call), then releases its storage. A throwing handler propagates after the block has been released. /// - private static async ValueTask ReadMetadataBlockAsync( - ClickHouseBinaryReader reader, + private async ValueTask ReadMetadataBlockAsync( + ServerPacketType packet, NegotiatedProtocol negotiated, ResolveContext context, Action handler, CancellationToken cancellationToken) { - Block block = await BlockReader.ReadBlockAsync(reader, negotiated, ColumnCodecRegistry.Default, context, cancellationToken).ConfigureAwait(false); + Block block = await ReadBlockAsync(packet, negotiated, context, cancellationToken).ConfigureAwait(false); try { handler?.Invoke(block); @@ -950,7 +1059,23 @@ public void Terminate() } finally { - writer.Dispose(); + try + { + writer.Dispose(); + } + finally + { + // The frame buffers are pooled like the reader's and writer's, and this runs under the + // same guarantee: Terminate happens once the I/O that pointed at them has unwound. + try + { + frameReader?.Dispose(); + } + finally + { + frameWriter?.Dispose(); + } + } } } } @@ -1075,7 +1200,7 @@ internal async ValueTask HandshakeAsync(ClientHandshakeParameters handshake, Can return (null, await ClickHouseServerException.ReadAsync(reader, cancellationToken).ConfigureAwait(false)); case ServerPacketType.Data: - return (await BlockReader.ReadBlockAsync(reader, negotiated, ColumnCodecRegistry.Default, context, cancellationToken).ConfigureAwait(false), null); + return (await ReadBlockAsync(ServerPacketType.Data, negotiated, context, cancellationToken).ConfigureAwait(false), null); default: await ConsumeMetadataAsync(packet, negotiated, context, handlers, cancellationToken).ConfigureAwait(false); @@ -1106,19 +1231,19 @@ private async ValueTask ConsumeMetadataAsync( { // Block-bearing packets lend the borrowed block to the handler for the call, then release it. case ServerPacketType.Totals: - await ReadMetadataBlockAsync(reader, negotiated, context, handlers?.OnTotals, cancellationToken).ConfigureAwait(false); + await ReadMetadataBlockAsync(ServerPacketType.Totals, negotiated, context, handlers?.OnTotals, cancellationToken).ConfigureAwait(false); break; case ServerPacketType.Extremes: - await ReadMetadataBlockAsync(reader, negotiated, context, handlers?.OnExtremes, cancellationToken).ConfigureAwait(false); + await ReadMetadataBlockAsync(ServerPacketType.Extremes, negotiated, context, handlers?.OnExtremes, cancellationToken).ConfigureAwait(false); break; case ServerPacketType.ProfileEvents: - await ReadMetadataBlockAsync(reader, negotiated, context, handlers?.OnProfileEvents, cancellationToken).ConfigureAwait(false); + await ReadMetadataBlockAsync(ServerPacketType.ProfileEvents, negotiated, context, handlers?.OnProfileEvents, cancellationToken).ConfigureAwait(false); break; case ServerPacketType.Log: - await ReadMetadataBlockAsync(reader, negotiated, context, handlers?.OnLog, cancellationToken).ConfigureAwait(false); + await ReadMetadataBlockAsync(ServerPacketType.Log, negotiated, context, handlers?.OnLog, cancellationToken).ConfigureAwait(false); break; case ServerPacketType.Progress: diff --git a/ClickHouse.Driver.Tcp/Protocol/Query.cs b/ClickHouse.Driver.Tcp/Protocol/Query.cs index 476521dc3..14bb73748 100644 --- a/ClickHouse.Driver.Tcp/Protocol/Query.cs +++ b/ClickHouse.Driver.Tcp/Protocol/Query.cs @@ -30,6 +30,10 @@ internal static class Query /// The SQL text. /// Per-query settings as textual values, or null for none. /// Query parameter values (already in SQL representation), or null for none. + /// + /// Whether this query's block bodies are compressed, in both directions. Setting it commits the client to + /// framing its own Data blocks — including the empty end-of-input marker — for this query. + /// public static void Write( ClickHouseBinaryWriter writer, NegotiatedProtocol negotiated, @@ -37,7 +41,8 @@ public static void Write( string queryId, string sql, IReadOnlyDictionary settings, - IReadOnlyDictionary queryParameters) + IReadOnlyDictionary queryParameters, + bool compressed = false) { writer.WriteClientPacketType(ClientPacketType.Query); writer.WriteString(queryId ?? string.Empty); @@ -52,7 +57,7 @@ public static void Write( } writer.WriteByte(StageComplete); - writer.WriteBool(false); // compression disabled + writer.WriteBool(compressed); // compression, for this query only, both directions writer.WriteString(sql); diff --git a/ClickHouse.Driver.Tests/CityHash102Tests.cs b/ClickHouse.Driver.Tests/CityHash102Tests.cs new file mode 100644 index 000000000..f4fb73b94 --- /dev/null +++ b/ClickHouse.Driver.Tests/CityHash102Tests.cs @@ -0,0 +1,134 @@ +using System; +using System.Collections.Generic; +using ClickHouse.Driver.Compression; +using NUnit.Framework; + +namespace ClickHouse.Driver.Tests; + +/// +/// Unit tests for , the CityHash v1.0.2 port that computes the compression +/// frame's checksum. +/// +/// Every expected value below was generated by compiling the reference city.cc that ClickHouse +/// vendors as contrib/cityhash102 and hashing the same input, so these pin the port against the +/// implementation the server itself uses rather than against our own output. A modern (1.1+) CityHash +/// fails all of them, which is the confusion this fixture exists to catch. +/// +/// +/// The lengths cover every branch: 0, below 4, 4 to 7, the 8 to 15 case that seeds from an empty span, +/// 16, 17 to 32, the CityMurmur range up to 127, 128 where the unrolled loop starts, and lengths +/// past it whose tail walks backwards over already-consumed bytes. +/// +/// +[TestFixture] +public class CityHash102Tests +{ + /// Reference vectors over inputs: length, then the expected low and high halves. + private static IEnumerable PatternVectors() + { + yield return new TestCaseData(0, 0x3df09dfc64c09a2bUL, 0x3cb540c392e51e29UL); + yield return new TestCaseData(1, 0x8be55b379cce0e40UL, 0x05ab4744478b1cbaUL); + yield return new TestCaseData(2, 0x701c733aee3382d4UL, 0xba4eae41e0e6dc7aUL); + yield return new TestCaseData(3, 0xe407deda7bb0f294UL, 0x0a3a0d146e27f1bdUL); + yield return new TestCaseData(4, 0x905dcb2063ae124fUL, 0x57fd5b5954a593c5UL); + yield return new TestCaseData(5, 0x69a56895d9ac3ddcUL, 0x955b14ec8ac620d5UL); + yield return new TestCaseData(7, 0xd7ba2459ab7c2913UL, 0x53dd430b513ca460UL); + yield return new TestCaseData(8, 0xf5a4ca47208136a0UL, 0x3dd4575b3d46e5abUL); + yield return new TestCaseData(9, 0x338434c54a2565e1UL, 0x1001bbb07bd33771UL); + yield return new TestCaseData(12, 0x8450aab120e0ccdcUL, 0x8a990d2011469548UL); + yield return new TestCaseData(15, 0xd956065c63d9ab15UL, 0x9623caadd0037b73UL); + yield return new TestCaseData(16, 0x3f3a3275564b7f48UL, 0xb48a2a7a16bac60bUL); + yield return new TestCaseData(17, 0xaee34bf68bb2e0a2UL, 0xe0d6910555f00b87UL); + yield return new TestCaseData(24, 0x5adfe1a6eab14c05UL, 0xf76c0804fe96ed75UL); + yield return new TestCaseData(31, 0x9b7f93e5ecfaed63UL, 0xd646e40154fb0d6bUL); + yield return new TestCaseData(32, 0x250e8c8007bfad61UL, 0xe42486b6776ae8dbUL); + yield return new TestCaseData(33, 0x4e495b057af3b520UL, 0xabae1d24f18410b5UL); + yield return new TestCaseData(63, 0x361db35b3402a110UL, 0x51bf6fff2d8235c5UL); + yield return new TestCaseData(64, 0xe225cf33b373dc02UL, 0x8d252ce5152d6d96UL); + yield return new TestCaseData(100, 0x49c189a397bdadeeUL, 0x0e8155bdfd58d3d3UL); + yield return new TestCaseData(127, 0xc244dc8a06ce8c41UL, 0xf30e2cf6b33629a6UL); + yield return new TestCaseData(128, 0xba863c0c70f32346UL, 0x24b156fc6f3c170aUL); + yield return new TestCaseData(129, 0x238598064c8ee2f1UL, 0x6c89b9263ef7410bUL); + yield return new TestCaseData(160, 0x5cb38c23f218820bUL, 0x28eec84c96793879UL); + yield return new TestCaseData(191, 0x2f4f8ba535aa5f81UL, 0xb5694954efdb1784UL); + yield return new TestCaseData(192, 0x45c173eb249e540eUL, 0x64bcb83693c2d72bUL); + yield return new TestCaseData(255, 0xc87d0f456f6a514bUL, 0xf35b13ec94eb8a52UL); + yield return new TestCaseData(256, 0xb1cfa883f958dcfaUL, 0x39f0dfb3be182292UL); + yield return new TestCaseData(1000, 0x597ff4957972ad7eUL, 0x7f39a4abd7c010e1UL); + yield return new TestCaseData(1048576, 0xd7a7dbcf65ef0f08UL, 0x7b45ba5f2ffeff9aUL); + } + + /// Reference vectors over uniform inputs, which catch sign and carry mistakes a varied pattern can hide. + private static IEnumerable UniformVectors() + { + yield return new TestCaseData(25, (byte)0x00, 0xc8596b8135f22993UL, 0xdb2985aef08357b8UL); + yield return new TestCaseData(25, (byte)0xFF, 0xc5d3cd20b17776e0UL, 0x1bd608e76fe125afUL); + yield return new TestCaseData(137, (byte)0x00, 0xbfaa109cc5cc6f50UL, 0xb57476dec812081cUL); + yield return new TestCaseData(137, (byte)0xFF, 0x3381b9cf974f40bcUL, 0x045d59f3d0fa0d6dUL); + } + + [TestCaseSource(nameof(PatternVectors))] + public void Hash128_PatternOfLength_MatchesTheReferenceImplementation(int length, ulong expectedLow, ulong expectedHigh) + { + var (low, high) = CityHash102.Hash128(Pattern(length)); + + Assert.Multiple(() => + { + Assert.That(low, Is.EqualTo(expectedLow), $"low half for length {length}"); + Assert.That(high, Is.EqualTo(expectedHigh), $"high half for length {length}"); + }); + } + + [TestCaseSource(nameof(UniformVectors))] + public void Hash128_UniformBytes_MatchesTheReferenceImplementation(int length, byte fill, ulong expectedLow, ulong expectedHigh) + { + var data = new byte[length]; + data.AsSpan().Fill(fill); + + var (low, high) = CityHash102.Hash128(data); + + Assert.Multiple(() => + { + Assert.That(low, Is.EqualTo(expectedLow), $"low half for {length} x 0x{fill:X2}"); + Assert.That(high, Is.EqualTo(expectedHigh), $"high half for {length} x 0x{fill:X2}"); + }); + } + + [Test] + public void Hash128_EmptyInput_DoesNotReadTheSpanAndReturnsTheSeededConstant() + { + var (low, high) = CityHash102.Hash128(ReadOnlySpan.Empty); + + Assert.Multiple(() => + { + Assert.That(low, Is.EqualTo(0x3df09dfc64c09a2bUL)); + Assert.That(high, Is.EqualTo(0x3cb540c392e51e29UL)); + }); + } + + [Test] + public void Hash128_OffsetSliceOfALargerBuffer_HashesOnlyTheSlice() + { + // Guards the port's absolute indexing: the tail of a long input walks backwards from the current + // position, so hashing a slice must not reach bytes outside it. + var padded = new byte[64 + 200 + 64]; + Pattern(200).CopyTo(padded.AsSpan(64)); + + var fromSlice = CityHash102.Hash128(padded.AsSpan(64, 200)); + var standalone = CityHash102.Hash128(Pattern(200)); + + Assert.That(fromSlice, Is.EqualTo(standalone)); + } + + /// The same deterministic input the reference harness hashed: byte i is (i * 31 + 7) & 0xFF. + private static byte[] Pattern(int length) + { + var data = new byte[length]; + for (int i = 0; i < length; i++) + { + data[i] = unchecked((byte)((i * 31) + 7)); + } + + return data; + } +} diff --git a/ClickHouse.Driver.Tests/CompressionFrameTests.cs b/ClickHouse.Driver.Tests/CompressionFrameTests.cs new file mode 100644 index 000000000..74b1beb6c --- /dev/null +++ b/ClickHouse.Driver.Tests/CompressionFrameTests.cs @@ -0,0 +1,257 @@ +using System; +using System.Buffers.Binary; +using System.Collections.Generic; +using System.IO; +using System.Text; +using ClickHouse.Driver.Compression; +using NUnit.Framework; + +namespace ClickHouse.Driver.Tests; + +/// +/// Unit tests for : the exact wire bytes of a frame, and the error paths a +/// round trip cannot reach. The two spans a frame declares are easy to confuse, so they are asserted +/// separately — the checksum covers the header plus the body, while compressed_size counts the header +/// plus the body and excludes the checksum. +/// +[TestFixture] +public class CompressionFrameTests +{ + private static IEnumerable BlockCodecs() + { + yield return new TestCaseData(Lz4Compressor.Default, CompressionFrame.MethodLz4).SetName("{m}(LZ4)"); + yield return new TestCaseData(ZstdCompressor.Default, CompressionFrame.MethodZstd).SetName("{m}(ZSTD)"); + } + + [TestCaseSource(nameof(BlockCodecs))] + public void Write_ABlockCodec_EmitsTheDocumentedHeader(IClickHouseCompressor codec, byte expectedMethod) + { + byte[] plaintext = Payload(500); + var destination = new byte[CompressionFrame.MaxFrameSize(codec, plaintext.Length)]; + + int total = CompressionFrame.Write(plaintext, codec, destination); + + int compressedSize = BinaryPrimitives.ReadInt32LittleEndian(destination.AsSpan(17, 4)); + Assert.Multiple(() => + { + Assert.That(destination[16], Is.EqualTo(expectedMethod), "method byte"); + Assert.That(compressedSize, Is.EqualTo(total - CompressionFrame.ChecksumSize), "compressed_size counts the header but not the checksum"); + Assert.That(BinaryPrimitives.ReadInt32LittleEndian(destination.AsSpan(21, 4)), Is.EqualTo(plaintext.Length), "uncompressed_size"); + Assert.That(compressedSize, Is.GreaterThanOrEqualTo(CompressionFrame.HeaderSize), "compressed_size covers its own header"); + }); + } + + [TestCaseSource(nameof(BlockCodecs))] + public void Write_ABlockCodec_ChecksumsTheHeaderAndBodyButNotItself(IClickHouseCompressor codec, byte expectedMethod) + { + _ = expectedMethod; + byte[] plaintext = Payload(500); + var destination = new byte[CompressionFrame.MaxFrameSize(codec, plaintext.Length)]; + + int total = CompressionFrame.Write(plaintext, codec, destination); + + // Recompute over exactly the span the format specifies: from the method byte to the end of the body. + var (low, high) = CityHash102.Hash128(destination.AsSpan(CompressionFrame.ChecksumSize, total - CompressionFrame.ChecksumSize)); + Assert.Multiple(() => + { + Assert.That(BinaryPrimitives.ReadUInt64LittleEndian(destination.AsSpan(0, 8)), Is.EqualTo(low), "low half, little-endian first"); + Assert.That(BinaryPrimitives.ReadUInt64LittleEndian(destination.AsSpan(8, 8)), Is.EqualTo(high), "high half"); + }); + } + + [TestCaseSource(nameof(BlockCodecs))] + public void WriteThenDecode_ABlockCodec_RecoversThePlaintext(IClickHouseCompressor codec, byte expectedMethod) + { + byte[] plaintext = Payload(4096); + var destination = new byte[CompressionFrame.MaxFrameSize(codec, plaintext.Length)]; + int total = CompressionFrame.Write(plaintext, codec, destination); + + CompressionFrame.ReadHeader( + destination.AsSpan(CompressionFrame.ChecksumSize, CompressionFrame.HeaderSize), + out byte method, + out int bodySize, + out int plaintextSize); + CompressionFrame.VerifyChecksum(destination.AsSpan(0, CompressionFrame.ChecksumSize), destination.AsSpan(CompressionFrame.ChecksumSize, total - CompressionFrame.ChecksumSize)); + var decoded = new byte[plaintextSize]; + CompressionFrame.Decode(method, destination.AsSpan(CompressionFrame.PrefixSize, bodySize), decoded); + + Assert.Multiple(() => + { + Assert.That(method, Is.EqualTo(expectedMethod)); + Assert.That(plaintextSize, Is.EqualTo(plaintext.Length)); + Assert.That(decoded, Is.EqualTo(plaintext)); + }); + } + + [Test] + public void Write_EmptyPlaintext_StillEmitsAFullPrefix() + { + var destination = new byte[CompressionFrame.MaxFrameSize(Lz4Compressor.Default, 0)]; + + int total = CompressionFrame.Write(ReadOnlySpan.Empty, Lz4Compressor.Default, destination); + + Assert.Multiple(() => + { + Assert.That(total, Is.GreaterThanOrEqualTo(CompressionFrame.PrefixSize)); + Assert.That(BinaryPrimitives.ReadInt32LittleEndian(destination.AsSpan(21, 4)), Is.Zero, "uncompressed_size"); + }); + } + + [Test] + public void Write_DestinationBelowTheCodecBound_Throws() + { + byte[] plaintext = Payload(100); + var tooSmall = new byte[CompressionFrame.PrefixSize]; + + Assert.That( + () => CompressionFrame.Write(plaintext, Lz4Compressor.Default, tooSmall), + Throws.ArgumentException.With.Message.Contains("Destination holds")); + } + + [Test] + public void MaxFrameSize_IsThePrefixPlusTheCodecsBound() + { + int bound = CompressionFrame.MaxFrameSize(Lz4Compressor.Default, 1000); + + Assert.That(bound, Is.EqualTo(CompressionFrame.PrefixSize + Lz4Compressor.Default.MaxEncodedLength(1000))); + } + + [Test] + public void ReadHeader_CompressedSizeBelowTheHeaderItCounts_Throws() + { + var header = new byte[CompressionFrame.HeaderSize]; + header[0] = CompressionFrame.MethodLz4; + BinaryPrimitives.WriteInt32LittleEndian(header.AsSpan(1, 4), CompressionFrame.HeaderSize - 1); + BinaryPrimitives.WriteInt32LittleEndian(header.AsSpan(5, 4), 10); + + Assert.That( + () => CompressionFrame.ReadHeader(header, out _, out _, out _), + Throws.TypeOf().With.Message.Contains("does not cover its own")); + } + + [Test] + public void ReadHeader_NegativeUncompressedSize_Throws() + { + var header = new byte[CompressionFrame.HeaderSize]; + header[0] = CompressionFrame.MethodLz4; + BinaryPrimitives.WriteInt32LittleEndian(header.AsSpan(1, 4), CompressionFrame.HeaderSize + 4); + BinaryPrimitives.WriteInt32LittleEndian(header.AsSpan(5, 4), -1); + + Assert.That( + () => CompressionFrame.ReadHeader(header, out _, out _, out _), + Throws.TypeOf().With.Message.Contains("outside the supported range")); + } + + [Test] + public void ReadHeader_ImplausibleBodySize_Throws() + { + var header = new byte[CompressionFrame.HeaderSize]; + header[0] = CompressionFrame.MethodLz4; + BinaryPrimitives.WriteInt32LittleEndian(header.AsSpan(1, 4), int.MaxValue); + BinaryPrimitives.WriteInt32LittleEndian(header.AsSpan(5, 4), 10); + + Assert.That( + () => CompressionFrame.ReadHeader(header, out _, out _, out _), + Throws.TypeOf().With.Message.Contains("above the")); + } + + [Test] + public void ReadHeader_WrongLength_Throws() + { + Assert.That( + () => CompressionFrame.ReadHeader(new byte[CompressionFrame.HeaderSize - 1], out _, out _, out _), + Throws.ArgumentException); + } + + [Test] + public void VerifyChecksum_ABodyBitFlippedInTransit_Throws() + { + byte[] plaintext = Payload(500); + var frame = new byte[CompressionFrame.MaxFrameSize(Lz4Compressor.Default, plaintext.Length)]; + int total = CompressionFrame.Write(plaintext, Lz4Compressor.Default, frame); + frame[CompressionFrame.PrefixSize + 3] ^= 0x01; + + Assert.That( + () => CompressionFrame.VerifyChecksum(frame.AsSpan(0, CompressionFrame.ChecksumSize), frame.AsSpan(CompressionFrame.ChecksumSize, total - CompressionFrame.ChecksumSize)), + Throws.TypeOf().With.Message.Contains("checksum mismatch")); + } + + [Test] + public void VerifyChecksum_AHeaderBitFlippedInTransit_Throws() + { + byte[] plaintext = Payload(500); + var frame = new byte[CompressionFrame.MaxFrameSize(Lz4Compressor.Default, plaintext.Length)]; + int total = CompressionFrame.Write(plaintext, Lz4Compressor.Default, frame); + + // The uncompressed_size field, which the checksum covers even though it is not body content. + frame[21] ^= 0x02; + + Assert.That( + () => CompressionFrame.VerifyChecksum(frame.AsSpan(0, CompressionFrame.ChecksumSize), frame.AsSpan(CompressionFrame.ChecksumSize, total - CompressionFrame.ChecksumSize)), + Throws.TypeOf().With.Message.Contains("checksum mismatch")); + } + + [Test] + public void Decode_TheNoneMethod_CopiesTheBodyVerbatim() + { + byte[] body = Payload(64); + var plaintext = new byte[body.Length]; + + CompressionFrame.Decode(CompressionFrame.MethodNone, body, plaintext); + + Assert.That(plaintext, Is.EqualTo(body)); + } + + [Test] + public void Decode_TheNoneMethodWithABodyThatDisagreesWithTheDeclaredSize_Throws() + { + Assert.That( + () => CompressionFrame.Decode(CompressionFrame.MethodNone, Payload(64), new byte[63]), + Throws.TypeOf().With.Message.Contains("carries a 64-byte body")); + } + + [Test] + public void Decode_AMethodByteThisClientCannotDecode_NamesTheSupportedOnes() + { + Assert.That( + () => CompressionFrame.Decode(0x99, Payload(8), new byte[8]), + Throws.TypeOf().With.Message.Contains("0x99").And.Message.Contains("0x82 LZ4")); + } + + [Test] + public void ResolveCodec_TheKnownMethodBytes_ReturnTheMatchingCodec() + { + Assert.Multiple(() => + { + Assert.That(CompressionFrame.ResolveCodec(CompressionFrame.MethodLz4).MethodByte, Is.EqualTo(CompressionFrame.MethodLz4)); + Assert.That(CompressionFrame.ResolveCodec(CompressionFrame.MethodZstd).MethodByte, Is.EqualTo(CompressionFrame.MethodZstd)); + }); + } + + [Test] + public void Decode_AFrameWhoseBodyDecodesShort_Throws() + { + // A body encoded from 100 bytes, then decoded as if it held 200: the codec stops early and the + // declared size is the only thing that catches it. + byte[] plaintext = Payload(100); + var frame = new byte[CompressionFrame.MaxFrameSize(Lz4Compressor.Default, plaintext.Length)]; + int total = CompressionFrame.Write(plaintext, Lz4Compressor.Default, frame); + int bodySize = total - CompressionFrame.PrefixSize; + + Assert.That( + () => CompressionFrame.Decode(CompressionFrame.MethodLz4, frame.AsSpan(CompressionFrame.PrefixSize, bodySize), new byte[200]), + Throws.TypeOf()); + } + + /// Compressible but not trivially so, which keeps the encoded body a realistic size. + private static byte[] Payload(int length) + { + var text = new StringBuilder(length + 16); + while (text.Length < length) + { + text.Append("the quick brown fox jumps over the lazy dog 0123456789 "); + } + + return Encoding.UTF8.GetBytes(text.ToString(0, length)); + } +}