From f3eb19228b1c78845852daf5e1de59022a1270b4 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 18 Aug 2026 10:26:02 +0200 Subject: [PATCH 1/7] Add CityHash v1.0.2, the checksum the compression frame needs The frame's 16-byte checksum is CityHash128 from the historical 1.0.2 release, not the modern Google one: release 1.1 changed the 128-bit functions, so a 1.1+ implementation disagrees on every input and the server rejects every frame it writes. No NuGet package ships 1.0.2, so this is a port of the reference city.cc that ClickHouse vendors as contrib/cityhash102. Only the 128-bit entry point is ported. CityHash64 and HashLen17to32 are unreachable from it, so porting them would add untested surface. Two details a port gets wrong silently. CityMurmur compares a signed ssize_t, so a length below 16 must compare as negative. The tail of a long input walks backwards from the current position and reads bytes before it, which is in range only because the main loop already consumed 128 of them. The expected values in the tests come from compiling that reference and hashing the same inputs, so they pin the port against the implementation the server uses rather than against our own output. The lengths cover every branch, including the 8-to-15 case that seeds from an empty span and the lengths past 128 whose tail walks backwards. Co-Authored-By: Claude --- .../Compression/CityHash102.cs | 233 ++++++++++++++++++ ClickHouse.Driver.Tests/CityHash102Tests.cs | 134 ++++++++++ 2 files changed, 367 insertions(+) create mode 100644 ClickHouse.Driver.Common/Compression/CityHash102.cs create mode 100644 ClickHouse.Driver.Tests/CityHash102Tests.cs 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.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; + } +} From dd83eceff3215fa6f7621e4dcb522cec4b664ee7 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 18 Aug 2026 10:36:26 +0200 Subject: [PATCH 2/7] Add the compression frame codec, shared by both transports A frame is a 16-byte CityHash128 checksum, a 9-byte header of method and two sizes, then the body. The two spans it declares are not the same and are easy to confuse: the checksum covers the header plus the body, while compressed_size counts the header plus the body and excludes the checksum itself. Both are asserted separately. Pure span work with the caller owning the I/O, because the same frame format appears on the native protocol's packet bodies and on HTTP's compress/decompress stream, so the HTTP path can reuse this later. Decoding follows the frame's own method byte, not what the caller configured to write: the server picks its codec per query from network_compression_method, so a client that asked for LZ4 can still be sent ZSTD. Both declared sizes come from the peer, so both are range-checked before they size a rent, and an unknown method byte names the ones we do support rather than failing opaquely. Co-Authored-By: Claude --- .../Compression/CompressionFrame.cs | 199 ++++++++++++++ .../CompressionFrameTests.cs | 257 ++++++++++++++++++ 2 files changed, 456 insertions(+) create mode 100644 ClickHouse.Driver.Common/Compression/CompressionFrame.cs create mode 100644 ClickHouse.Driver.Tests/CompressionFrameTests.cs diff --git a/ClickHouse.Driver.Common/Compression/CompressionFrame.cs b/ClickHouse.Driver.Common/Compression/CompressionFrame.cs new file mode 100644 index 000000000..47d78537b --- /dev/null +++ b/ClickHouse.Driver.Common/Compression/CompressionFrame.cs @@ -0,0 +1,199 @@ +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. Cap them well + // above any legitimate frame (the server flushes at ~1 MiB) so a bad length cannot drive a huge rent. + private const int MaxBodySize = 1 << 30; + private const int MaxPlaintextSize = 1 << 30; + + /// + /// 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.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)); + } +} From 2451e8d787228f028241d90f5f421cd68702d29f Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 18 Aug 2026 10:36:40 +0200 Subject: [PATCH 3/7] Add the frame reader and writer that carry a block body A body arrives and leaves as a stream of frames rather than one frame, so these stream them, and the block and column codecs above read and write plaintext without learning that compression is on. Only the body is framed: the packet type code and the table name stay on the raw stream, which is why the switch happens between them rather than around the whole packet. The reader pulls frames through the connection's raw reader instead of the socket, so there stays one owner of buffered socket bytes. Its read-ahead stops at the current frame and it serves a short read at the boundary, pulling the next frame only when a caller asks for more: reading ahead past a block's last frame would consume the next packet's uncompressed envelope as frame bytes. EndBlock asserts a block consumed its frames exactly. The sender flushes at each block end, so a block's last frame ends where the block does, and anything left over means the decoders and the peer disagree about the body's length. The writer cuts plaintext into frames of at most the server's own ~1 MiB and flushes each one, so peak buffering is a single frame however large the block. A frame boundary inside a body is legal, since a reader finds a block's end from its own dimensions rather than from the framing. Co-Authored-By: Claude --- ClickHouse.Driver.Common/AssemblyInfo.cs | 4 + .../ClickHouse.Driver.Tcp.csproj | 6 + .../Compression/CompressedFrameReader.cs | 196 ++++++++++++++++++ .../Compression/CompressedFrameWriter.cs | 154 ++++++++++++++ 4 files changed, 360 insertions(+) create mode 100644 ClickHouse.Driver.Tcp/Compression/CompressedFrameReader.cs create mode 100644 ClickHouse.Driver.Tcp/Compression/CompressedFrameWriter.cs 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.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/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(); + } +} From 283d2e8a9d41db61f8e25a35d67d6472b4d77999 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 18 Aug 2026 11:54:07 +0200 Subject: [PATCH 4/7] Wire compression through the connection, off by default MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The Query packet's compression flag stops being a hardcoded false, and a Compressor option chooses the codec (Compression=lz4|zstd|none on a connection string). The default stays off in this commit so nothing about existing behaviour moves; the next one flips it. Only a body is framed, so the name a packet carries is split from it on both sides: BlockReader and BlockWriter grow body-only entry points, and the connection reads the name from the raw reader before choosing which reader decodes the rest. That leaves the block reader and all forty column codecs untouched — they are handed a reader and cannot tell which one it is. Not every block-bearing packet is framed. Log and ProfileEvents carry blocks and go through the same block reader, yet the server sends them uncompressed at our protocol target, so framing them would read a block name as a frame checksum. FramedPackets states the list once — Data, Totals, Extremes — and a test pins it, including that every other packet stays out. The Go client's ServerCode.Compressible admits the same three. The codec is fixed for a connection's life rather than per query, because the option is client-level; the wire flag is still written per query. A per-query override needs an API that can say "off for this query" when the client default is on, which a nullable codec cannot express, so it is left for its own change. An HTTP-only codec is refused when the client is built: GZip and Brotli throw from MethodByte, and finding that mid-query would leave the Query packet already promising the server frames. IsReusable now also refuses a connection holding undelivered plaintext, the same fault as leftover raw bytes one layer up. An abandoned response needed no new handling: the query path already terminates a connection that did not complete, so frame state cannot outlive it. Verified against a real server: 3015 pass, 0 fail. The new integration cases cover both codecs over bodies spanning several frames in both directions, Log packets arriving mid-stream while compressed, framed Totals, sequential compressed queries on one pooled connection, and the server answering ZSTD to a client that asked for LZ4. Co-Authored-By: Claude --- .../ClickHouse.Driver.Tcp.Tests.csproj | 3 + .../Client/ClickHouseTcpClientOptionsTests.cs | 4 + .../Client/CompressionOptionTests.cs | 84 ++++++ .../CompressedFrameStreamingTests.cs | 257 ++++++++++++++++++ .../Compression/FramedPacketsTests.cs | 62 +++++ .../CompressionIntegrationTests.cs | 252 +++++++++++++++++ .../Client/ClickHouseTcpClientOptions.cs | 60 ++++ .../ClickHouseTcpConnectionStringBuilder.cs | 11 + .../Client/IConnectionFactory.cs | 2 +- .../Compression/FramedPackets.cs | 26 ++ ClickHouse.Driver.Tcp/Format/BlockReader.cs | 27 ++ ClickHouse.Driver.Tcp/Format/BlockWriter.cs | 83 +++++- .../Protocol/ClickHouseTcpConnection.cs | 169 ++++++++++-- ClickHouse.Driver.Tcp/Protocol/Query.cs | 9 +- 14 files changed, 1012 insertions(+), 37 deletions(-) create mode 100644 ClickHouse.Driver.Tcp.Tests/Client/CompressionOptionTests.cs create mode 100644 ClickHouse.Driver.Tcp.Tests/Compression/CompressedFrameStreamingTests.cs create mode 100644 ClickHouse.Driver.Tcp.Tests/Compression/FramedPacketsTests.cs create mode 100644 ClickHouse.Driver.Tcp.Tests/Integration/CompressionIntegrationTests.cs create mode 100644 ClickHouse.Driver.Tcp/Compression/FramedPackets.cs 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..6f898316f --- /dev/null +++ b/ClickHouse.Driver.Tcp.Tests/Client/CompressionOptionTests.cs @@ -0,0 +1,84 @@ +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 +{ + [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..81355f8da --- /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 InsertAsync_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.InsertAsync($"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 InsertAsync_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.InsertAsync($"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/Client/ClickHouseTcpClientOptions.cs b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs index 307270254..4dc2dfa86 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,13 @@ 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. + internal const string DefaultCompression = CompressionNone; + + 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 +263,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 @@ -283,8 +309,42 @@ public static ClickHouseTcpClientOptions FromConnectionString(string connectionS /// 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. + /// + /// 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)), + }; + 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..d198960fe 100644 --- a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs +++ b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs @@ -254,6 +254,16 @@ public ClickHouseTcpPoolReusePolicy PoolReusePolicy set => this["PoolReusePolicy"] = value.ToString(); } + /// + /// Frame codec for the native protocol: lz4, zstd, or none. Defaults to + /// none. 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 +307,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/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..ebc64b231 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; } @@ -345,9 +368,8 @@ internal async IAsyncEnumerable QueryAsync( // 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); + await WriteEndOfInputBlockAsync(cancellationToken).ConfigureAwait(false); } catch { @@ -386,7 +408,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 +573,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 +752,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 +912,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 +1055,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 +1196,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 +1227,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); From 27de5c2f1cbbff3b8332f5067b20250f3ca5f6f3 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 18 Aug 2026 11:59:28 +0200 Subject: [PATCH 5/7] Turn compression on by default, with LZ4 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit One line, so that if this breaks something the cause is not in doubt. LZ4 is what clickhouse-client uses on this protocol: cheapest in CPU and lightest on the server. Compression=none, or a null Compressor, turns it off. The whole TCP suite now runs compressed: 3015 pass, 0 fail, so every existing integration test round-trips through the frame layer rather than only the cases written for it. A test pins the default in all three places it is observable — the options record, a connection string with no Compression key, and the builder. Co-Authored-By: Claude --- .../Client/CompressionOptionTests.cs | 13 +++++++++++++ .../Client/ClickHouseTcpClientOptions.cs | 8 ++++++-- 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/ClickHouse.Driver.Tcp.Tests/Client/CompressionOptionTests.cs b/ClickHouse.Driver.Tcp.Tests/Client/CompressionOptionTests.cs index 6f898316f..da8d0a9ce 100644 --- a/ClickHouse.Driver.Tcp.Tests/Client/CompressionOptionTests.cs +++ b/ClickHouse.Driver.Tcp.Tests/Client/CompressionOptionTests.cs @@ -12,6 +12,19 @@ namespace ClickHouse.Driver.Tcp.Tests.Client; [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)] diff --git a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs index 4dc2dfa86..20ee1a358 100644 --- a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs +++ b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs @@ -30,8 +30,12 @@ public sealed record ClickHouseTcpClientOptions internal const int DefaultMaxPoolSize = 20; internal const ClickHouseTcpPoolReusePolicy DefaultPoolReusePolicy = ClickHouseTcpPoolReusePolicy.Lifo; - /// The Compression connection-string value used when the key is absent. - internal const string DefaultCompression = CompressionNone; + /// + /// 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"; From da67f76c08457fa9fc4713bfb3d5b92cf71da19b Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Tue, 18 Aug 2026 12:45:42 +0200 Subject: [PATCH 6/7] Address the review: a send failure must not look reusable MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four findings from the Copilot review, one of them a real bug. **A failed framed end-of-input marker left the connection reusable.** The query path encodes the Query packet into the buffer inside a try whose catch resets the buffer and returns the connection to Ready, on the premise that nothing can have been sent yet. Framing the marker broke that premise: each frame flushes as it is emitted, so a socket failure or a cancellation there left the Query packet on the wire while the connection went back to the pool looking healthy, and the next caller to lease it would read the previous query's response. Only Query.Write stays in that block now; the marker moved into the outer try, whose failure terminates. The insert path already terminated on any failure, so it needed no change. A test pins it, and the old structure was restored to confirm the test fails on it — Ready instead of Terminated, IsReusable true instead of false. The insert case passes either way, so it is a guard rather than a regression test. **The frame size caps allowed two ~1 GiB rents.** Both sizes come from the peer and each one sizes a rent before the checksum can be checked, so 1 GiB apiece contradicted the comment claiming a bad length could not drive a huge rent. Now 128 MiB, ample over the server's ~1 MiB flush, and the same figure the Go client uses. **Two documentation defects.** The connection-string builder still documented the default as none after the flip to lz4, and Validate's summary and exception tags had been left stranded above ResolveCompressor, so Validate was undocumented and the tags described the wrong member. Declined: a changelog.d fragment. Every PR in this stack deliberately adds none (R3, confirmed 2026-08-17), and the client is experimental with no packaging yet, so there is no released behaviour to describe. R3 now names this default explicitly so the sweep cannot miss it. Also filed T1a: the default is on with LZ4 but no benchmark backs it. Wire size is a measured 1.85-2.00x win, while a loopback probe was inconclusive and possibly negative — it pays the CPU with no bandwidth to save. T1/T2 must measure over a realistic path and revisit the default if LZ4 costs there. Co-Authored-By: Claude --- .../Compression/CompressionFrame.cs | 10 +- .../Protocol/CompressedSendFailureTests.cs | 126 ++++++++++++++++++ .../Client/ClickHouseTcpClientOptions.cs | 6 +- .../ClickHouseTcpConnectionStringBuilder.cs | 3 +- .../Protocol/ClickHouseTcpConnection.cs | 8 +- 5 files changed, 143 insertions(+), 10 deletions(-) create mode 100644 ClickHouse.Driver.Tcp.Tests/Protocol/CompressedSendFailureTests.cs diff --git a/ClickHouse.Driver.Common/Compression/CompressionFrame.cs b/ClickHouse.Driver.Common/Compression/CompressionFrame.cs index 47d78537b..de8f6cafd 100644 --- a/ClickHouse.Driver.Common/Compression/CompressionFrame.cs +++ b/ClickHouse.Driver.Common/Compression/CompressionFrame.cs @@ -43,10 +43,12 @@ internal static class CompressionFrame /// 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. Cap them well - // above any legitimate frame (the server flushes at ~1 MiB) so a bad length cannot drive a huge rent. - private const int MaxBodySize = 1 << 30; - private const int MaxPlaintextSize = 1 << 30; + // 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 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/Client/ClickHouseTcpClientOptions.cs b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs index 20ee1a358..7a73361d6 100644 --- a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs +++ b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpClientOptions.cs @@ -310,9 +310,6 @@ public static ClickHouseTcpClientOptions FromConnectionString(string connectionS return new ClickHouseTcpConnectionStringBuilder(connectionString).ToOptions(); } - /// 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. /// /// Maps a Compression connection-string value to its codec. none yields /// , meaning the query is not compressed at all. @@ -330,6 +327,9 @@ public static ClickHouseTcpClientOptions FromConnectionString(string connectionS 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 diff --git a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs index d198960fe..a6208d295 100644 --- a/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs +++ b/ClickHouse.Driver.Tcp/Client/ClickHouseTcpConnectionStringBuilder.cs @@ -256,7 +256,8 @@ public ClickHouseTcpPoolReusePolicy PoolReusePolicy /// /// Frame codec for the native protocol: lz4, zstd, or none. Defaults to - /// none. Maps to . + /// lz4, so compression is on unless this key turns it off. Maps to + /// . /// public string Compression { diff --git a/ClickHouse.Driver.Tcp/Protocol/ClickHouseTcpConnection.cs b/ClickHouse.Driver.Tcp/Protocol/ClickHouseTcpConnection.cs index ebc64b231..d24bcb5f4 100644 --- a/ClickHouse.Driver.Tcp/Protocol/ClickHouseTcpConnection.cs +++ b/ClickHouse.Driver.Tcp/Protocol/ClickHouseTcpConnection.cs @@ -363,13 +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, compressor is not null); - await WriteEndOfInputBlockAsync(cancellationToken).ConfigureAwait(false); } catch { @@ -380,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) From bf3a3a7d954498cad3df0cef28f0bc1a70291770 Mon Sep 17 00:00:00 2001 From: Alex Soffronow-Pagonidis Date: Fri, 28 Aug 2026 13:04:05 +0200 Subject: [PATCH 7/7] Update compressed row inserts for InsertRowsAsync --- .../Integration/CompressionIntegrationTests.cs | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/ClickHouse.Driver.Tcp.Tests/Integration/CompressionIntegrationTests.cs b/ClickHouse.Driver.Tcp.Tests/Integration/CompressionIntegrationTests.cs index 81355f8da..78f6ca2ef 100644 --- a/ClickHouse.Driver.Tcp.Tests/Integration/CompressionIntegrationTests.cs +++ b/ClickHouse.Driver.Tcp.Tests/Integration/CompressionIntegrationTests.cs @@ -46,7 +46,7 @@ public async Task QueryAsync_Compressed_ReturnsEveryRow(IClickHouseCompressor co } [TestCaseSource(nameof(Codecs))] - public async Task InsertAsync_Compressed_RoundTripsThroughSelect(IClickHouseCompressor codec) + public async Task InsertRowsAsync_Compressed_RoundTripsThroughSelect(IClickHouseCompressor codec) { await using var client = new ClickHouseTcpClient(TcpServerFixture.Options() with { Compressor = codec }); string table = UniqueTableName(); @@ -57,7 +57,7 @@ public async Task InsertAsync_Compressed_RoundTripsThroughSelect(IClickHouseComp object[][] rows = Enumerable.Range(0, 2000) .Select(i => new object[] { (ulong)i, $"name-{i}" }) .ToArray(); - await client.InsertAsync($"INSERT INTO {table} (id, name) VALUES", rows, null, None); + 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)) @@ -101,7 +101,7 @@ public async Task QueryAsync_ABodyLargerThanOneFrame_ReassemblesAcrossFrameBound } [TestCaseSource(nameof(Codecs))] - public async Task InsertAsync_ABlockLargerThanOneFrame_RoundTripsThroughSelect(IClickHouseCompressor codec) + 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. @@ -112,7 +112,7 @@ public async Task InsertAsync_ABlockLargerThanOneFrame_RoundTripsThroughSelect(I 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.InsertAsync($"INSERT INTO {table} (id) VALUES", rows, null, None); + 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))