From 45ea0e41f13c0f9d7a494c546b921ad4a113da32 Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Thu, 30 Jul 2026 13:42:51 +0000 Subject: [PATCH 1/7] fix(read): detect mid-stream server exceptions on the streaming read path When a query fails after the HTTP response is committed (e.g. a throwIf partway through a large result), ClickHouse appends an in-band exception block framed as `__exception__\r\n\r\n...`. Two bugs kept the native reader from ever surfacing it: - ExceptionTagAwareStream searched for a contiguous `__exception__` marker, but the server writes a CRLF between `__exception__` and the tag, so TryExtractMidStreamException always returned null. - ClickHouseDataReader.Read() caught only EndOfStreamException, while a live truncated response surfaces as HttpIOException; its end-of-stream PeekChar probe also sat outside the try. Tolerate the server's CRLF marker framing and treat the truncation as an IOException so a mid-stream failure raises a ClickHouseServerException with the real server error instead of a bare HttpIOException/EndOfStreamException. Fixes: https://github.com/ClickHouse/clickhouse-cs/issues/476 Co-Authored-By: Claude Opus 4.7 --- CHANGELOG.md | 1 + .../ADO/MidStreamExceptionTests.cs | 37 +++++++++ .../Formats/ExceptionTagAwareStreamTests.cs | 47 +++++++++++ .../ADO/Readers/ClickHouseDataReader.cs | 18 ++-- .../Formats/ExceptionTagAwareStream.cs | 83 ++++++++++++++----- RELEASENOTES.md | 1 + 6 files changed, 158 insertions(+), 29 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6cdd61bc8..f7e4b65cd 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,7 @@ Unreleased * Fixed `InsertOptions.WithColumnTypes()` and `InsertOptions.WithQueryId()` silently dropping some caller-set options (such as `AcceptEncoding`) when copying. * Fixed `ClickHouseServerException` carrying a blank `Message` and an `ErrorCode` of `-1` when the server — or an upstream component such as a load balancer or the ClickHouse Cloud edge — returned a non-2xx HTTP response with an empty (or whitespace-only) body. The exception now reports the HTTP status code and reason phrase, and uses the `X-ClickHouse-Exception-Code` response header as the error code when the server sets it (issue #440). Non-empty error bodies are unaffected. * Fixed `ClickHouseDataReader.GetSchemaTable()` leaving `NumericScale` unset (`DBNull`) for `DateTime64(N)` and `Time64(N)` columns (including their `Nullable(...)` variants). The schema table now reports the fractional-seconds precision `N` in `NumericScale`, matching how `Decimal` columns are already reported (issue #438). +* Fixed mid-stream server exceptions never surfacing on the streaming read path (`ExecuteReader`/`ExecuteReaderAsync`). When a query fails after the HTTP response is committed (for example a `throwIf` partway through a large result), ClickHouse appends an in-band exception block that `ExceptionTagAwareStream` failed to detect: it searched for a contiguous `__exception__` marker, but the server writes a CRLF between `__exception__` and the tag, so `TryExtractMidStreamException()` always returned `null`. The reader also caught only `EndOfStreamException`, whereas a live truncated response surfaces as `HttpIOException`. The reader now tolerates the server's CRLF marker framing and treats the truncation as an `IOException`, so a mid-stream failure raises a `ClickHouseServerException` carrying the real server error instead of a bare `HttpIOException`/`EndOfStreamException` (issue #476). v1.3.0 --- diff --git a/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs b/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs index 79c23bbeb..dca01863a 100644 --- a/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs +++ b/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs @@ -37,6 +37,43 @@ FROM system.numbers Assert.That(ex.Message, Does.Contain("boom")); } + + [Test] + [FromVersion(25, 11)] + public void ShouldDetectMidStreamException_AfterResponseIsCommitted() + { + // ShouldDetectMidStreamException above throws so early that the server returns a plain + // HTTP 500 before committing a response, so it never exercises the in-band exception path. + // Here the server streams a committed 200 OK plus rows before the throwIf fires, so the + // failure is delivered in-band (X-ClickHouse-Exception-Tag) and reading past the truncated + // body raises an HttpIOException that ExceptionTagAwareStream must convert into the real + // server error. Compression is disabled so the server streams the response incrementally + // rather than buffering it (a buffered response instead fails pre-commit as a 500). + using var streamingClient = TestUtilities.GetTestClickHouseClient(compression: false); + using var streamingConnection = streamingClient.CreateConnection(); + using var command = streamingConnection.CreateCommand(); + command.CustomSettings["http_write_exception_in_output_format"] = 1; + command.CustomSettings["max_block_size"] = 1000; + command.CustomSettings["http_response_buffer_size"] = 0; + command.CustomSettings["wait_end_of_query"] = 0; + + command.CommandText = @" + SELECT toInt32(number) AS n, + throwIf(number = 200000, 'boom mid stream') AS e + FROM system.numbers + LIMIT 400000"; + + var ex = Assert.Throws(() => + { + using var reader = command.ExecuteReader(); + while (reader.Read()) + { + // Drain until the in-band mid-stream exception surfaces + } + }); + + Assert.That(ex.Message, Does.Contain("boom mid stream")); + } } /// diff --git a/ClickHouse.Driver.Tests/Formats/ExceptionTagAwareStreamTests.cs b/ClickHouse.Driver.Tests/Formats/ExceptionTagAwareStreamTests.cs index dee366fd3..d3f2ebbda 100644 --- a/ClickHouse.Driver.Tests/Formats/ExceptionTagAwareStreamTests.cs +++ b/ClickHouse.Driver.Tests/Formats/ExceptionTagAwareStreamTests.cs @@ -106,6 +106,31 @@ public void TryExtractMidStreamException_DetectsMarker_WithCompleteFormat() Assert.That(result.Message, Is.EqualTo("Test error message")); } + [Test] + public void TryExtractMidStreamException_DetectsMarker_WithServerCrlfFraming() + { + // Real ClickHouse (>= 25.11) frames the in-band block with a CRLF between the + // "__exception__" literal and the tag, which the contiguous fixtures do not cover: + // "\r\n__exception__\r\n\r\n\n \r\n__exception__\r\n". + var message = "Code: 395. DB::Exception: boom mid stream"; + var messageLength = Encoding.UTF8.GetByteCount(message); + var exceptionData = + $"\r\n__exception__\r\n{TestToken}\r\n{message}\n{messageLength} {TestToken}\r\n__exception__\r\n"; + var data = Encoding.UTF8.GetBytes("some rows before" + exceptionData); + + using var ms = new MemoryStream(data); + using var stream = new ExceptionTagAwareStream(ms, TestToken); + + var buffer = new byte[data.Length]; + _ = stream.Read(buffer, 0, buffer.Length); + + var result = stream.TryExtractMidStreamException(); + + Assert.That(result, Is.Not.Null); + Assert.That(result.Message, Is.EqualTo(message)); + Assert.That(result.ErrorCode, Is.EqualTo(395)); + } + [Test] public void TryExtractMidStreamException_DetectsMarker_WithMultilineMessage() { @@ -274,4 +299,26 @@ public void TryExtractMidStreamException_IgnoresWrongToken() Assert.That(result, Is.Null); // Should not match wrong token } + + [Test] + public void TryExtractMidStreamException_IgnoresWrongToken_WithServerCrlfFraming() + { + // The CRLF-tolerant matcher must not loosen tag matching: a real-framed block whose tag + // differs from the configured one must still be ignored. + var wrongToken = "WRONGTOKEN"; + var message = "Wrong token error"; + var exceptionData = + $"\r\n__exception__\r\n{wrongToken}\r\n{message}\n{message.Length} {wrongToken}\r\n__exception__\r\n"; + var data = Encoding.UTF8.GetBytes(exceptionData); + + using var ms = new MemoryStream(data); + using var stream = new ExceptionTagAwareStream(ms, TestToken); // Looking for TestToken + + var buffer = new byte[data.Length]; + _ = stream.Read(buffer, 0, buffer.Length); + + var result = stream.TryExtractMidStreamException(); + + Assert.That(result, Is.Null); // Should not match wrong token + } } diff --git a/ClickHouse.Driver/ADO/Readers/ClickHouseDataReader.cs b/ClickHouse.Driver/ADO/Readers/ClickHouseDataReader.cs index 7a4f15a00..59b0df88d 100644 --- a/ClickHouse.Driver/ADO/Readers/ClickHouseDataReader.cs +++ b/ClickHouse.Driver/ADO/Readers/ClickHouseDataReader.cs @@ -439,12 +439,6 @@ private static string BuildAssignmentErrorMessage( public override bool Read() { - if (reader.PeekChar() == -1) - { - hasCurrentRow = false; - return false; // End of stream reached - } - var count = RawTypes.Length; var data = CurrentRow; @@ -453,6 +447,11 @@ public override bool Read() hasCurrentRow = false; try { + // PeekChar is inside the try: a mid-stream failure truncates the body at a row + // boundary too, so the end-of-stream probe can itself hit the truncation. + if (reader.PeekChar() == -1) + return false; // End of stream reached + for (var i = 0; i < count; i++) { var rawType = RawTypes[i]; @@ -461,7 +460,12 @@ public override bool Read() hasCurrentRow = true; return true; } - catch (EndOfStreamException) when (exceptionTagStream != null) + // A mid-stream server failure truncates the HTTP body; reading past the truncation surfaces + // as an IOException — EndOfStreamException for a buffered body, but HttpIOException + // ("response ended prematurely") for a live streamed response. Both derive from IOException. + // When the server tagged the response (exceptionTagStream != null) the in-band exception block + // is captured in the ring buffer, so convert it to the real server error; otherwise re-throw. + catch (IOException) when (exceptionTagStream != null) { var serverEx = exceptionTagStream.TryExtractMidStreamException(); if (serverEx != null) diff --git a/ClickHouse.Driver/Formats/ExceptionTagAwareStream.cs b/ClickHouse.Driver/Formats/ExceptionTagAwareStream.cs index 395c4af71..3d65878c7 100644 --- a/ClickHouse.Driver/Formats/ExceptionTagAwareStream.cs +++ b/ClickHouse.Driver/Formats/ExceptionTagAwareStream.cs @@ -14,9 +14,8 @@ internal sealed class ExceptionTagAwareStream : Stream private const int BufferCapacity = 4096; // 4KB ring buffer private readonly Stream innerStream; - private readonly string exceptionToken; - private readonly byte[] exceptionMarker; // "__exception__" + token - private readonly byte[] closingMarker; // token + "__exception__" + private readonly byte[] exceptionPrefixBytes; // "__exception__" + private readonly byte[] tagBytes; // exception tag/token // Ring buffer for recent bytes private readonly byte[] recentBytes = new byte[BufferCapacity]; @@ -30,9 +29,8 @@ public ExceptionTagAwareStream(Stream innerStream, string exceptionTag) if (string.IsNullOrEmpty(exceptionTag)) throw new ArgumentException("Exception tag cannot be null or empty", nameof(exceptionTag)); - exceptionToken = exceptionTag; - exceptionMarker = Encoding.UTF8.GetBytes(ExceptionPrefix + exceptionTag); - closingMarker = Encoding.UTF8.GetBytes(exceptionTag + ExceptionPrefix); + exceptionPrefixBytes = Encoding.UTF8.GetBytes(ExceptionPrefix); + tagBytes = Encoding.UTF8.GetBytes(exceptionTag); } public override bool CanRead => innerStream.CanRead; @@ -101,16 +99,17 @@ private void RecordBytes(byte[] buffer, int offset, int count) /// ClickHouseServerException if marker found, null otherwise public ClickHouseServerException TryExtractMidStreamException() { - if (bytesRecorded < exceptionMarker.Length) + if (bytesRecorded < exceptionPrefixBytes.Length + tagBytes.Length) return null; byte[] buffer = GetLinearBuffer(); - int markerIndex = FindPattern(buffer, exceptionMarker); + // Opening marker: "__exception__" "". + int markerIndex = FindDelimitedMarker(buffer, exceptionPrefixBytes, tagBytes, 0, out int messageStart); if (markerIndex < 0) return null; - return ParseExceptionFormat(buffer, markerIndex); + return ParseExceptionFormat(buffer, messageStart); } private byte[] GetLinearBuffer() @@ -133,18 +132,16 @@ private byte[] GetLinearBuffer() return result; } - private ClickHouseServerException ParseExceptionFormat(byte[] buffer, int markerIndex) + private ClickHouseServerException ParseExceptionFormat(byte[] buffer, int messageStart) { - // Format: __exception__TOKEN\n\n TOKEN__exception__ - // We ignore the size - int messageStart = markerIndex + exceptionMarker.Length; - - // Skip newlines after opening marker + // Full block: __exception__\n __exception__ + // where is the CR/LF run the server writes. messageStart points just past the + // opening ""; skip the separator before the message text. We ignore . while (messageStart < buffer.Length && (buffer[messageStart] == '\n' || buffer[messageStart] == '\r')) messageStart++; - // Find closing marker: TOKEN__exception__ - int closingIndex = FindPattern(buffer, closingMarker, messageStart); + // Closing marker: "" "__exception__". + int closingIndex = FindDelimitedMarker(buffer, tagBytes, exceptionPrefixBytes, messageStart, out _); // Determine where message ends int messageEnd = closingIndex >= 0 ? closingIndex : buffer.Length; @@ -153,11 +150,16 @@ private ClickHouseServerException ParseExceptionFormat(byte[] buffer, int marker while (messageEnd > messageStart && char.IsWhiteSpace((char)buffer[messageEnd - 1])) messageEnd--; - // Also trim any trailing digits and space (the size number) - while (messageEnd > messageStart && char.IsDigit((char)buffer[messageEnd - 1])) - messageEnd--; - while (messageEnd > messageStart && char.IsWhiteSpace((char)buffer[messageEnd - 1])) - messageEnd--; + // The " " token sits between the message and the closing marker, so only strip a + // trailing number when a closing marker was actually found. Otherwise a message captured + // without its closing marker that legitimately ends in a digit would be mangled. + if (closingIndex >= 0) + { + while (messageEnd > messageStart && char.IsDigit((char)buffer[messageEnd - 1])) + messageEnd--; + while (messageEnd > messageStart && char.IsWhiteSpace((char)buffer[messageEnd - 1])) + messageEnd--; + } if (messageEnd <= messageStart) return ClickHouseServerException.FromMidStreamException("Unknown error (could not parse exception message)"); @@ -166,6 +168,43 @@ private ClickHouseServerException ParseExceptionFormat(byte[] buffer, int marker return ClickHouseServerException.FromMidStreamException(errorMessage); } + /// + /// Finds followed by , allowing an optional + /// run of CR/LF bytes between them. The ClickHouse server frames the in-band exception block as + /// "__exception__\r\n<tag>" (open) and "<tag>\r\n__exception__" (close); tolerating the + /// separator — and its absence — keeps detection robust across server framings. + /// + /// Index immediately past when found; -1 otherwise. + /// Index of when the delimited pair is found; -1 otherwise. + private static int FindDelimitedMarker(byte[] buffer, byte[] first, byte[] second, int startIndex, out int afterSecond) + { + afterSecond = -1; + int searchFrom = startIndex; + + while (true) + { + int firstIndex = FindPattern(buffer, first, searchFrom); + if (firstIndex < 0) + return -1; + + int pos = firstIndex + first.Length; + + // Skip an optional CR/LF separator run between the two parts. + while (pos < buffer.Length && (buffer[pos] == (byte)'\r' || buffer[pos] == (byte)'\n')) + pos++; + + if (pos + second.Length <= buffer.Length && + buffer.AsSpan(pos, second.Length).SequenceEqual(second)) + { + afterSecond = pos + second.Length; + return firstIndex; + } + + // This occurrence of `first` is not followed by `second`; keep searching. + searchFrom = firstIndex + 1; + } + } + private static int FindPattern(byte[] buffer, byte[] pattern, int startIndex = 0) { if (pattern.Length == 0 || buffer.Length < pattern.Length + startIndex) diff --git a/RELEASENOTES.md b/RELEASENOTES.md index 836ea3904..e4d1f1096 100644 --- a/RELEASENOTES.md +++ b/RELEASENOTES.md @@ -26,6 +26,7 @@ Unreleased * Fixed `InsertOptions.WithColumnTypes()` and `InsertOptions.WithQueryId()` silently dropping some caller-set options (such as `AcceptEncoding`) when copying. * Fixed `ClickHouseServerException` carrying a blank `Message` and an `ErrorCode` of `-1` when the server — or an upstream component such as a load balancer or the ClickHouse Cloud edge — returned a non-2xx HTTP response with an empty (or whitespace-only) body. The exception now reports the HTTP status code and reason phrase, and uses the `X-ClickHouse-Exception-Code` response header as the error code when the server sets it (issue #440). Non-empty error bodies are unaffected. * Fixed `ClickHouseDataReader.GetSchemaTable()` leaving `NumericScale` unset (`DBNull`) for `DateTime64(N)` and `Time64(N)` columns (including their `Nullable(...)` variants). The schema table now reports the fractional-seconds precision `N` in `NumericScale`, matching how `Decimal` columns are already reported (issue #438). +* Fixed mid-stream server exceptions never surfacing on the streaming read path (`ExecuteReader`/`ExecuteReaderAsync`). When a query fails after the HTTP response is committed (for example a `throwIf` partway through a large result), the driver now detects the server's in-band exception block and raises a `ClickHouseServerException` with the real error, instead of a bare `HttpIOException` ("The response ended prematurely") or `EndOfStreamException` (issue #476). v1.3.0 --- From f1d3498f055054de2edf6f0da45b24619f442394 Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Thu, 30 Jul 2026 14:19:12 +0000 Subject: [PATCH 2/7] Fix ClickHouseRawResult: surface in-band mid-stream server exceptions on the raw streaming path MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ExecuteRawResultAsync / ClickHouseRawResult handed the HTTP response body straight to the caller, so a query failing after the 200 OK was committed (server appends an in-band exception block, then drops the connection) surfaced as a System.Net.Http.HttpIOException ("The response ended prematurely") — or, with http_write_exception_in_output_format=1, the raw __exception__ block spliced into the caller's data — instead of a ClickHouseServerException. All four accessors (ReadAsStreamAsync, ReadAsByteArrayAsync, ReadAsStringAsync, CopyToAsync) now consult the X-ClickHouse-Exception-Tag header and wrap the body with a new opt-in throwAtEndOfStream mode on ExceptionTagAwareStream, reusing the in-band detector fixed in #476/#477. The default passive-observer mode (native reader path) and the untagged raw path are unchanged. Fixes: https://github.com/ClickHouse/clickhouse-cs/issues/475 --- CHANGELOG.md | 1 + .../ADO/MidStreamExceptionTests.cs | 143 +++++++++++++++++ .../Formats/ExceptionTagAwareStreamTests.cs | 145 ++++++++++++++++++ .../ADO/Readers/ClickHouseRawResult.cs | 72 ++++++++- .../Formats/ExceptionTagAwareStream.cs | 120 ++++++++++++++- RELEASENOTES.md | 1 + 6 files changed, 469 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f7e4b65cd..b2c89dfaa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -27,6 +27,7 @@ Unreleased * Fixed `ClickHouseServerException` carrying a blank `Message` and an `ErrorCode` of `-1` when the server — or an upstream component such as a load balancer or the ClickHouse Cloud edge — returned a non-2xx HTTP response with an empty (or whitespace-only) body. The exception now reports the HTTP status code and reason phrase, and uses the `X-ClickHouse-Exception-Code` response header as the error code when the server sets it (issue #440). Non-empty error bodies are unaffected. * Fixed `ClickHouseDataReader.GetSchemaTable()` leaving `NumericScale` unset (`DBNull`) for `DateTime64(N)` and `Time64(N)` columns (including their `Nullable(...)` variants). The schema table now reports the fractional-seconds precision `N` in `NumericScale`, matching how `Decimal` columns are already reported (issue #438). * Fixed mid-stream server exceptions never surfacing on the streaming read path (`ExecuteReader`/`ExecuteReaderAsync`). When a query fails after the HTTP response is committed (for example a `throwIf` partway through a large result), ClickHouse appends an in-band exception block that `ExceptionTagAwareStream` failed to detect: it searched for a contiguous `__exception__` marker, but the server writes a CRLF between `__exception__` and the tag, so `TryExtractMidStreamException()` always returned `null`. The reader also caught only `EndOfStreamException`, whereas a live truncated response surfaces as `HttpIOException`. The reader now tolerates the server's CRLF marker framing and treats the truncation as an `IOException`, so a mid-stream failure raises a `ClickHouseServerException` carrying the real server error instead of a bare `HttpIOException`/`EndOfStreamException` (issue #476). +* Fixed the raw / custom-FORMAT streaming surface (`ExecuteRawResultAsync` → `ClickHouseRawResult`) not surfacing in-band mid-stream server exceptions (issue #475). The native reader detects these via the `X-ClickHouse-Exception-Tag` header, but `ClickHouseRawResult` handed the response body straight to the caller, so a query that failed after the HTTP `200 OK` was committed (e.g. a `throwIf` partway through a large `FORMAT CSV`/`JSONEachRow`/`Arrow`/`Parquet` result) surfaced as a `System.Net.Http.HttpIOException` ("The response ended prematurely") — or, with `http_write_exception_in_output_format=1`, spliced the raw `__exception__` block into the caller's data — instead of a `ClickHouseServerException`. All four accessors (`ReadAsStreamAsync`, `ReadAsByteArrayAsync`, `ReadAsStringAsync`, `CopyToAsync`) now consult the exception-tag header and raise a `ClickHouseServerException` carrying the server's message; the untagged path is byte-for-byte unchanged. v1.3.0 --- diff --git a/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs b/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs index dca01863a..1dc932675 100644 --- a/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs +++ b/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs @@ -263,3 +263,146 @@ public async Task WithoutExceptionTagHeader_AndExceptionMarkerInStream_ShouldNot Assert.That(ex, Is.TypeOf()); } } + +/// +/// Tests that the raw / custom-FORMAT streaming surface () surfaces an +/// in-band mid-stream server exception as a across every accessor, +/// using mock HTTP responses (no ClickHouse server required). +/// +public class ClickHouseRawResultMidStreamMockTests +{ + private const string Token = "PU1FNUFH98"; + private const string ErrorMessage = "Code: 395. DB::Exception: boom"; + + public enum Accessor + { + Stream, + Bytes, + String, + CopyTo, + } + + // Real server framing: \r\n__exception__\r\n\r\n\n \r\n__exception__\r\n + private static byte[] MidStreamBody() => + Encoding.UTF8.GetBytes($"1,0\r\n2,0\r\n__exception__\r\n{Token}\r\n{ErrorMessage}\n{ErrorMessage.Length} {Token}\r\n__exception__\r\n"); + + private static HttpResponseMessage Response(byte[] content, string exceptionTag) + { + var response = new HttpResponseMessage(HttpStatusCode.OK) { Content = new ByteArrayContent(content) }; + if (exceptionTag != null) + response.Headers.Add("X-ClickHouse-Exception-Tag", exceptionTag); + return response; + } + + private static async Task DrainAsync(ClickHouseRawResult raw, Accessor accessor) + { + switch (accessor) + { + case Accessor.Bytes: + return await raw.ReadAsByteArrayAsync(); + case Accessor.String: + return Encoding.UTF8.GetBytes(await raw.ReadAsStringAsync()); + case Accessor.Stream: + { + using var stream = await raw.ReadAsStreamAsync(); + using var sink = new MemoryStream(); + await stream.CopyToAsync(sink); + return sink.ToArray(); + } + default: // CopyTo + { + using var sink = new MemoryStream(); + await raw.CopyToAsync(sink); + return sink.ToArray(); + } + } + } + + [TestCase(Accessor.Stream)] + [TestCase(Accessor.Bytes)] + [TestCase(Accessor.String)] + [TestCase(Accessor.CopyTo)] + public void Accessor_WithExceptionTag_AndInBandException_ThrowsServerException(Accessor accessor) + { + using var response = Response(MidStreamBody(), Token); + using var raw = new ClickHouseRawResult(response); + + var ex = Assert.ThrowsAsync(() => DrainAsync(raw, accessor)); + + Assert.That(ex.Message, Does.Contain("boom")); + Assert.That(ex.ErrorCode, Is.EqualTo(395)); + } + + [TestCase(Accessor.Stream)] + [TestCase(Accessor.Bytes)] + [TestCase(Accessor.String)] + [TestCase(Accessor.CopyTo)] + public async Task Accessor_WithExceptionTag_ButSuccessfulBody_ReturnsFullBody(Accessor accessor) + { + var body = Encoding.UTF8.GetBytes("1,0\r\n2,0\r\n3,0\r\n"); // tag present but query succeeded (no in-band block) + using var response = Response(body, Token); + using var raw = new ClickHouseRawResult(response); + + var read = await DrainAsync(raw, accessor); + + Assert.That(read, Is.EqualTo(body)); + } + + [Test] + public async Task ReadAsByteArrayAsync_WithoutExceptionTag_ReturnsBodyVerbatim() + { + // No tag header => the untagged path must be byte-for-byte unchanged (no wrapping, no detection), + // even if the body coincidentally contains "__exception__" bytes. + var body = MidStreamBody(); + using var response = Response(body, exceptionTag: null); + using var raw = new ClickHouseRawResult(response); + + var bytes = await raw.ReadAsByteArrayAsync(); + + Assert.That(bytes, Is.EqualTo(body)); + } +} + +/// +/// Integration test for mid-stream exception detection against a real server. +/// +public class ClickHouseRawResultMidStreamTests : AbstractConnectionTestFixture +{ + [Test] + [FromVersion(25, 11)] + public async Task ExecuteRawResultAsync_MidStreamException_SurfacesServerException() + { + // The query streams a committed 200 OK plus rows before throwIf fires, so the failure is delivered + // in-band (X-ClickHouse-Exception-Tag) and the truncated body surfaces as an HttpIOException that the + // raw path used to leak instead of the real server error. Compression is disabled and buffering + // minimized so the server streams incrementally (a buffered response instead fails pre-commit as 500). + using var streamingClient = TestUtilities.GetTestClickHouseClient(compression: false); + using var streamingConnection = streamingClient.CreateConnection(); + using var command = streamingConnection.CreateCommand(); + command.CustomSettings["http_write_exception_in_output_format"] = 1; + command.CustomSettings["max_block_size"] = 1000; + command.CustomSettings["http_response_buffer_size"] = 0; + command.CustomSettings["wait_end_of_query"] = 0; + + command.CommandText = @" + SELECT toInt32(number) AS n, + throwIf(number = 200000, 'boom mid stream') AS e + FROM system.numbers + LIMIT 400000 + FORMAT CSV"; + + using var result = await command.ExecuteRawResultAsync(default); + using var stream = await result.ReadAsStreamAsync(); + + var ex = Assert.ThrowsAsync(async () => + { + var buffer = new byte[64 * 1024]; + while (await stream.ReadAsync(buffer, 0, buffer.Length) > 0) + { + // Drain until the in-band mid-stream exception surfaces + } + }); + + Assert.That(ex.Message, Does.Contain("boom mid stream")); + } +} diff --git a/ClickHouse.Driver.Tests/Formats/ExceptionTagAwareStreamTests.cs b/ClickHouse.Driver.Tests/Formats/ExceptionTagAwareStreamTests.cs index d3f2ebbda..0d064974e 100644 --- a/ClickHouse.Driver.Tests/Formats/ExceptionTagAwareStreamTests.cs +++ b/ClickHouse.Driver.Tests/Formats/ExceptionTagAwareStreamTests.cs @@ -1,6 +1,7 @@ using System; using System.IO; using System.Text; +using System.Threading.Tasks; using ClickHouse.Driver.Formats; using NUnit.Framework; @@ -321,4 +322,148 @@ public void TryExtractMidStreamException_IgnoresWrongToken_WithServerCrlfFraming Assert.That(result, Is.Null); // Should not match wrong token } + + public enum ReadApi + { + SyncRead, + ReadByte, + AsyncReadArray, + AsyncReadMemory, + } + + private static async Task DrainAsync(ExceptionTagAwareStream stream, ReadApi api) + { + var buffer = new byte[64]; + switch (api) + { + case ReadApi.SyncRead: + while (stream.Read(buffer, 0, buffer.Length) > 0) { } + break; + case ReadApi.ReadByte: + while (stream.ReadByte() >= 0) { } + break; + case ReadApi.AsyncReadArray: + while (await stream.ReadAsync(buffer, 0, buffer.Length) > 0) { } + break; + default: // AsyncReadMemory + while (await stream.ReadAsync(buffer.AsMemory()) > 0) { } + break; + } + } + + // In throwAtEndOfStream mode the wrapper must surface the in-band exception no matter which read + // API drains it, and whether the body ends cleanly (a 0-byte read) or the connection drops + // mid-stream (an IOException — how a live truncated HTTP response actually terminates). + [TestCase(ReadApi.SyncRead, false)] + [TestCase(ReadApi.SyncRead, true)] + [TestCase(ReadApi.ReadByte, false)] + [TestCase(ReadApi.ReadByte, true)] + [TestCase(ReadApi.AsyncReadArray, false)] + [TestCase(ReadApi.AsyncReadArray, true)] + [TestCase(ReadApi.AsyncReadMemory, false)] + [TestCase(ReadApi.AsyncReadMemory, true)] + public void ThrowAtEndOfStream_SurfacesServerException_AcrossReadApisAndTerminations(ReadApi api, bool prematureClose) + { + var message = "Code: 395. DB::Exception: boom"; + var data = Encoding.UTF8.GetBytes( + $"data\r\n__exception__\r\n{TestToken}\r\n{message}\n{message.Length} {TestToken}\r\n__exception__\r\n"); + + Stream inner = prematureClose ? new ThrowAtEndStream(data) : new MemoryStream(data); + using var stream = new ExceptionTagAwareStream(inner, TestToken, throwAtEndOfStream: true); + + var ex = Assert.ThrowsAsync(() => DrainAsync(stream, api)); + + Assert.That(ex.Message, Does.Contain("boom")); + Assert.That(ex.ErrorCode, Is.EqualTo(395)); + } + + [Test] + public void ThrowAtEndOfStream_DetectsException_WhenBlockFollowsLargeData() + { + // The in-band block arrives after >4 KiB of row data (larger than the ring buffer) and the + // stream is drained in small chunks; the block at the tail must still survive and be detected. + var message = "Code: 395. DB::Exception: boom"; + var prefix = new string('x', 8192); + var data = Encoding.UTF8.GetBytes( + $"{prefix}\r\n__exception__\r\n{TestToken}\r\n{message}\n{message.Length} {TestToken}\r\n__exception__\r\n"); + + using var ms = new MemoryStream(data); + using var stream = new ExceptionTagAwareStream(ms, TestToken, throwAtEndOfStream: true); + + var ex = Assert.Throws(() => + { + var buffer = new byte[64]; + while (stream.Read(buffer, 0, buffer.Length) > 0) { } + }); + + Assert.That(ex.Message, Does.Contain("boom")); + } + + [Test] + public void Read_WithThrowAtEndOfStream_DoesNotThrow_WhenNoMarkerPresent() + { + // Tag present but the query succeeded: no in-band block, so the full body must pass through cleanly. + var data = Encoding.UTF8.GetBytes("clean,csv\r\ndata,rows\r\nno,error\r\n"); + + using var ms = new MemoryStream(data); + using var stream = new ExceptionTagAwareStream(ms, TestToken, throwAtEndOfStream: true); + using var sink = new MemoryStream(); + + Assert.DoesNotThrow(() => stream.CopyTo(sink)); + Assert.That(sink.ToArray(), Is.EqualTo(data)); + } + + [Test] + public void Read_WithoutThrowAtEndOfStream_DoesNotThrow_EvenWithMarker() + { + // Default (passive-observer) mode used by the native reader path must be unchanged: reading never + // throws by itself; the marker is only surfaced when the caller asks via TryExtractMidStreamException. + var message = "Code: 395. DB::Exception: boom"; + var data = Encoding.UTF8.GetBytes( + $"data\r\n__exception__\r\n{TestToken}\r\n{message}\n{message.Length} {TestToken}\r\n__exception__\r\n"); + + using var ms = new MemoryStream(data); + using var stream = new ExceptionTagAwareStream(ms, TestToken); // default: throwAtEndOfStream = false + + Assert.DoesNotThrow(() => + { + var buffer = new byte[64]; + while (stream.Read(buffer, 0, buffer.Length) > 0) { } + }); + + Assert.That(stream.TryExtractMidStreamException(), Is.Not.Null); + } + + /// Stream that yields its content, then throws an IOException at end-of-stream (like a dropped HTTP connection). + private sealed class ThrowAtEndStream : Stream + { + private readonly MemoryStream inner; + + public ThrowAtEndStream(byte[] data) => inner = new MemoryStream(data); + + public override int Read(byte[] buffer, int offset, int count) + { + int n = inner.Read(buffer, offset, count); + if (n == 0) + throw new IOException("The response ended prematurely."); + return n; + } + + 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 => inner.Position; set => throw new NotSupportedException(); } + 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(); + + protected override void Dispose(bool disposing) + { + if (disposing) + inner.Dispose(); + base.Dispose(disposing); + } + } } diff --git a/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs b/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs index f395050ce..19674f518 100644 --- a/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs +++ b/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs @@ -1,8 +1,10 @@ -using System; +using System; using System.IO; using System.Linq; using System.Net.Http; +using System.Net.Http.Headers; using System.Threading.Tasks; +using ClickHouse.Driver.Formats; namespace ClickHouse.Driver.ADO; @@ -16,10 +18,19 @@ namespace ClickHouse.Driver.ADO; public class ClickHouseRawResult : IDisposable { private readonly HttpResponseMessage response; + private readonly string exceptionTag; internal ClickHouseRawResult(HttpResponseMessage response) { this.response = response; + + // When the server-side setting http_write_exception_in_output_format is enabled it sends this + // leading header carrying a per-query token, and — if the query fails after the 200 OK is already + // committed and rows are streaming — appends an in-band exception block delimited by that token to + // the body before closing the connection. Capture the token so the accessors below can surface a + // ClickHouseServerException instead of leaking the raw block / a truncated body to the caller. + if (response.Headers.TryGetValues(ExceptionTagAwareStream.HeaderName, out var tagValues)) + exceptionTag = tagValues.FirstOrDefault(); } /// @@ -51,26 +62,77 @@ public string ContentEncoding /// Reads the response content as a stream. /// /// A task that resolves to the response content stream. - public Task ReadAsStreamAsync() => response.Content.ReadAsStreamAsync(); + /// + /// If the server reports an in-band mid-stream exception, reading the returned stream throws a + /// once the end of the body is reached. + /// + public Task ReadAsStreamAsync() => + string.IsNullOrEmpty(exceptionTag) ? response.Content.ReadAsStreamAsync() : WrapContentStreamAsync(); + + private async Task WrapContentStreamAsync() + { + var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false); + return new ExceptionTagAwareStream(stream, exceptionTag, throwAtEndOfStream: true); + } /// /// Reads the response content as a byte array. /// /// A task that resolves to the response content as bytes. - public Task ReadAsByteArrayAsync() => response.Content.ReadAsByteArrayAsync(); + /// + /// Throws a if the server reports an in-band mid-stream exception. + /// + public Task ReadAsByteArrayAsync() => + string.IsNullOrEmpty(exceptionTag) ? response.Content.ReadAsByteArrayAsync() : ReadAllBytesAsync(); + + private async Task ReadAllBytesAsync() + { + using var stream = await WrapContentStreamAsync().ConfigureAwait(false); + using var memory = new MemoryStream(); + await stream.CopyToAsync(memory).ConfigureAwait(false); + return memory.ToArray(); + } /// /// Reads the response content as a string. /// /// A task that resolves to the response content as a string. - public Task ReadAsStringAsync() => response.Content.ReadAsStringAsync(); + /// + /// Throws a if the server reports an in-band mid-stream exception. + /// + public Task ReadAsStringAsync() => + string.IsNullOrEmpty(exceptionTag) ? response.Content.ReadAsStringAsync() : ReadAllStringAsync(); + + private async Task ReadAllStringAsync() + { + var bytes = await ReadAllBytesAsync().ConfigureAwait(false); + + // Decode the buffered body exactly as HttpContent.ReadAsStringAsync would — honouring the + // response Content-Type charset and stripping a BOM — so a successful tagged response returns + // the same string the untagged path returns for identical bytes. + using var content = new ByteArrayContent(bytes); + var contentType = response.Content.Headers.ContentType; + if (contentType != null) + content.Headers.ContentType = new MediaTypeHeaderValue(contentType.MediaType) { CharSet = contentType.CharSet }; + return await content.ReadAsStringAsync().ConfigureAwait(false); + } /// /// Copies the response content to the specified stream. /// /// The destination stream to copy the content to. /// A task that completes when the copy operation is finished. - public Task CopyToAsync(Stream stream) => response.Content.CopyToAsync(stream); + /// + /// Throws a if the server reports an in-band mid-stream exception. + /// + public Task CopyToAsync(Stream stream) => + string.IsNullOrEmpty(exceptionTag) ? response.Content.CopyToAsync(stream) : CopyViaWrapperAsync(stream); + + private async Task CopyViaWrapperAsync(Stream destination) + { + using var source = await WrapContentStreamAsync().ConfigureAwait(false); + await source.CopyToAsync(destination).ConfigureAwait(false); + } public void Dispose() { diff --git a/ClickHouse.Driver/Formats/ExceptionTagAwareStream.cs b/ClickHouse.Driver/Formats/ExceptionTagAwareStream.cs index 3d65878c7..3ce36a4e8 100644 --- a/ClickHouse.Driver/Formats/ExceptionTagAwareStream.cs +++ b/ClickHouse.Driver/Formats/ExceptionTagAwareStream.cs @@ -1,6 +1,8 @@ using System; using System.IO; using System.Text; +using System.Threading; +using System.Threading.Tasks; namespace ClickHouse.Driver.Formats; @@ -10,19 +12,33 @@ namespace ClickHouse.Driver.Formats; /// internal sealed class ExceptionTagAwareStream : Stream { + /// Response header carrying the per-query exception token. + internal const string HeaderName = "X-ClickHouse-Exception-Tag"; + private const string ExceptionPrefix = "__exception__"; private const int BufferCapacity = 4096; // 4KB ring buffer private readonly Stream innerStream; private readonly byte[] exceptionPrefixBytes; // "__exception__" private readonly byte[] tagBytes; // exception tag/token + private readonly bool throwAtEndOfStream; // Ring buffer for recent bytes private readonly byte[] recentBytes = new byte[BufferCapacity]; private int writePosition; private int bytesRecorded; - public ExceptionTagAwareStream(Stream innerStream, string exceptionTag) + /// The stream to observe. + /// The per-query token from the response header. + /// + /// When , the wrapper proactively throws a + /// as soon as the inner stream ends (or fails with an ) and an in-band exception + /// block is present in the observed tail. This is required by consumers that read the stream directly + /// (e.g. the raw / custom-FORMAT streaming path), which have no decoder to translate the end-of-stream + /// into a server error. When (the default) the wrapper is a passive observer and + /// the caller surfaces the error itself by calling after a read failure. + /// + public ExceptionTagAwareStream(Stream innerStream, string exceptionTag, bool throwAtEndOfStream = false) { this.innerStream = innerStream ?? throw new ArgumentNullException(nameof(innerStream)); @@ -31,6 +47,7 @@ public ExceptionTagAwareStream(Stream innerStream, string exceptionTag) exceptionPrefixBytes = Encoding.UTF8.GetBytes(ExceptionPrefix); tagBytes = Encoding.UTF8.GetBytes(exceptionTag); + this.throwAtEndOfStream = throwAtEndOfStream; } public override bool CanRead => innerStream.CanRead; @@ -49,33 +66,120 @@ public override long Position public override int Read(byte[] buffer, int offset, int count) { - int bytesRead = innerStream.Read(buffer, offset, count); + int bytesRead; + try + { + bytesRead = innerStream.Read(buffer, offset, count); + } + catch (IOException) when (throwAtEndOfStream) + { + ThrowIfMidStreamException(); + throw; + } + + if (bytesRead > 0) + { + RecordBytes(buffer.AsSpan(offset, bytesRead)); + return bytesRead; + } + + if (throwAtEndOfStream) + ThrowIfMidStreamException(); + return bytesRead; + } + + public override async Task ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken) + { + int bytesRead; + try + { + bytesRead = await innerStream.ReadAsync(buffer, offset, count, cancellationToken).ConfigureAwait(false); + } + catch (IOException) when (throwAtEndOfStream) + { + ThrowIfMidStreamException(); + throw; + } + + if (bytesRead > 0) + { + RecordBytes(buffer.AsSpan(offset, bytesRead)); + return bytesRead; + } + + if (throwAtEndOfStream) + ThrowIfMidStreamException(); + return bytesRead; + } + + public override async ValueTask ReadAsync(Memory buffer, CancellationToken cancellationToken = default) + { + int bytesRead; + try + { + bytesRead = await innerStream.ReadAsync(buffer, cancellationToken).ConfigureAwait(false); + } + catch (IOException) when (throwAtEndOfStream) + { + ThrowIfMidStreamException(); + throw; + } if (bytesRead > 0) - RecordBytes(buffer, offset, bytesRead); + { + RecordBytes(buffer.Span.Slice(0, bytesRead)); + return bytesRead; + } + if (throwAtEndOfStream) + ThrowIfMidStreamException(); return bytesRead; } public override int ReadByte() { - int b = innerStream.ReadByte(); + int b; + try + { + b = innerStream.ReadByte(); + } + catch (IOException) when (throwAtEndOfStream) + { + ThrowIfMidStreamException(); + throw; + } + if (b >= 0) { recentBytes[writePosition] = (byte)b; writePosition = (writePosition + 1) % BufferCapacity; if (bytesRecorded < BufferCapacity) bytesRecorded++; + return b; } + + if (throwAtEndOfStream) + ThrowIfMidStreamException(); return b; } - private void RecordBytes(byte[] buffer, int offset, int count) + private void ThrowIfMidStreamException() { + var serverException = TryExtractMidStreamException(); + if (serverException != null) + throw serverException; + } + + private void RecordBytes(ReadOnlySpan data) + { + int count = data.Length; + if (count == 0) + return; + // If count >= buffer capacity, only keep last BufferCapacity bytes if (count >= BufferCapacity) { - Array.Copy(buffer, offset + count - BufferCapacity, recentBytes, 0, BufferCapacity); + data.Slice(count - BufferCapacity).CopyTo(recentBytes); writePosition = 0; bytesRecorded = BufferCapacity; return; @@ -83,10 +187,10 @@ private void RecordBytes(byte[] buffer, int offset, int count) // Copy into circular buffer, wrapping as needed int firstPart = Math.Min(count, BufferCapacity - writePosition); - Array.Copy(buffer, offset, recentBytes, writePosition, firstPart); + data.Slice(0, firstPart).CopyTo(recentBytes.AsSpan(writePosition)); if (firstPart < count) - Array.Copy(buffer, offset + firstPart, recentBytes, 0, count - firstPart); + data.Slice(firstPart).CopyTo(recentBytes.AsSpan(0)); writePosition = (writePosition + count) % BufferCapacity; bytesRecorded = Math.Min(bytesRecorded + count, BufferCapacity); diff --git a/RELEASENOTES.md b/RELEASENOTES.md index e4d1f1096..564af73db 100644 --- a/RELEASENOTES.md +++ b/RELEASENOTES.md @@ -27,6 +27,7 @@ Unreleased * Fixed `ClickHouseServerException` carrying a blank `Message` and an `ErrorCode` of `-1` when the server — or an upstream component such as a load balancer or the ClickHouse Cloud edge — returned a non-2xx HTTP response with an empty (or whitespace-only) body. The exception now reports the HTTP status code and reason phrase, and uses the `X-ClickHouse-Exception-Code` response header as the error code when the server sets it (issue #440). Non-empty error bodies are unaffected. * Fixed `ClickHouseDataReader.GetSchemaTable()` leaving `NumericScale` unset (`DBNull`) for `DateTime64(N)` and `Time64(N)` columns (including their `Nullable(...)` variants). The schema table now reports the fractional-seconds precision `N` in `NumericScale`, matching how `Decimal` columns are already reported (issue #438). * Fixed mid-stream server exceptions never surfacing on the streaming read path (`ExecuteReader`/`ExecuteReaderAsync`). When a query fails after the HTTP response is committed (for example a `throwIf` partway through a large result), the driver now detects the server's in-band exception block and raises a `ClickHouseServerException` with the real error, instead of a bare `HttpIOException` ("The response ended prematurely") or `EndOfStreamException` (issue #476). +* Fixed the raw / custom-FORMAT streaming surface (`ExecuteRawResultAsync` → `ClickHouseRawResult`) not surfacing in-band mid-stream server exceptions. A query that fails after the HTTP response is committed (e.g. a `throwIf` partway through a large `FORMAT CSV`/`Arrow`/`Parquet` result) now raises a `ClickHouseServerException` with the real error across all four accessors, instead of a bare `HttpIOException` ("The response ended prematurely") or a body containing the raw exception block (issue #475). v1.3.0 --- From 2384a767629a0a00da27b3dac6f8ccb70f8a7b06 Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Fri, 31 Jul 2026 10:24:17 +0000 Subject: [PATCH 3/7] Address review feedback: integration tests for the raw mid-stream surface @alex-clickhouse asked for integration tests over the mock-based unit tests and questioned the ReadAsStringAsync()/ReadAsByteArrayAsync() surface. AGENTS.md likewise says to strongly prefer tests that actually call the db. - Replace the mock-HTTP ClickHouseRawResultMidStreamMockTests with real-server coverage of all four accessors (stream, bytes, string, CopyTo), plus a successful-query contrast case. Verified these fail against the unpatched reader with the reported HttpIOException and pass with the fix. - Fix a repeat-read regression this PR had introduced: the buffered accessors disposed the response's cached content stream, so a second ReadAsByteArrayAsync()/ReadAsStringAsync() threw ObjectDisposedException where the untagged path returns the body again. Since the server sends X-ClickHouse-Exception-Tag on every response, that affected every raw result on 25.11+. The body is now buffered and the response-owned stream left open, restoring HttpContent's repeat-read behaviour, with a regression test. - Correct the changelog/release-note over-claim raised in review: the streaming accessors raise the exception at end-of-stream and do not retroactively filter bytes already handed to the caller. Same clarification added to the XML docs. --- CHANGELOG.md | 2 +- .../ADO/MidStreamExceptionTests.cs | 161 ++++++++---------- .../ADO/Readers/ClickHouseRawResult.cs | 31 +++- RELEASENOTES.md | 2 +- 4 files changed, 99 insertions(+), 97 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b1ee4f793..7408240b8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,7 +28,7 @@ Unreleased * Fixed `ClickHouseServerException` carrying a blank `Message` and an `ErrorCode` of `-1` when the server — or an upstream component such as a load balancer or the ClickHouse Cloud edge — returned a non-2xx HTTP response with an empty (or whitespace-only) body. The exception now reports the HTTP status code and reason phrase, and uses the `X-ClickHouse-Exception-Code` response header as the error code when the server sets it (issue #440). Non-empty error bodies are unaffected. * Fixed `ClickHouseDataReader.GetSchemaTable()` leaving `NumericScale` unset (`DBNull`) for `DateTime64(N)` and `Time64(N)` columns (including their `Nullable(...)` variants). The schema table now reports the fractional-seconds precision `N` in `NumericScale`, matching how `Decimal` columns are already reported (issue #438). * Fixed mid-stream server exceptions never surfacing on the streaming read path (`ExecuteReader`/`ExecuteReaderAsync`). A query that fails after the HTTP response is committed (for example a `throwIf` partway through a large result) now raises a `ClickHouseServerException` with the real server error, instead of a bare `HttpIOException` or `EndOfStreamException` (issue #476). -* Fixed the raw / custom-FORMAT streaming surface (`ExecuteRawResultAsync` → `ClickHouseRawResult`) not surfacing in-band mid-stream server exceptions (issue #475). The native reader detects these via the `X-ClickHouse-Exception-Tag` header, but `ClickHouseRawResult` handed the response body straight to the caller, so a query that failed after the HTTP `200 OK` was committed (e.g. a `throwIf` partway through a large `FORMAT CSV`/`JSONEachRow`/`Arrow`/`Parquet` result) surfaced as a `System.Net.Http.HttpIOException` ("The response ended prematurely") — or, with `http_write_exception_in_output_format=1`, spliced the raw `__exception__` block into the caller's data — instead of a `ClickHouseServerException`. All four accessors (`ReadAsStreamAsync`, `ReadAsByteArrayAsync`, `ReadAsStringAsync`, `CopyToAsync`) now consult the exception-tag header and raise a `ClickHouseServerException` carrying the server's message; the untagged path is byte-for-byte unchanged. +* Fixed the raw / custom-FORMAT streaming surface (`ExecuteRawResultAsync` → `ClickHouseRawResult`) not surfacing in-band mid-stream server exceptions (issue #475). The native reader detects these via the `X-ClickHouse-Exception-Tag` header, but `ClickHouseRawResult` handed the response body straight to the caller, so a query that failed after the HTTP `200 OK` was committed (e.g. a `throwIf` partway through a large `FORMAT CSV`/`JSONEachRow`/`Arrow`/`Parquet` result) surfaced as a `System.Net.Http.HttpIOException` ("The response ended prematurely") instead of a `ClickHouseServerException`. All four accessors (`ReadAsStreamAsync`, `ReadAsByteArrayAsync`, `ReadAsStringAsync`, `CopyToAsync`) now consult the exception-tag header and raise a `ClickHouseServerException` carrying the server's message: the buffered accessors (`ReadAsByteArrayAsync`/`ReadAsStringAsync`) throw instead of returning a truncated body, while the streaming accessors (`ReadAsStreamAsync`/`CopyToAsync`) raise it once the end of the response is reached — data already streamed to the caller before the failure is still delivered as-is and is not retroactively filtered. Successful responses are unaffected. v1.3.0 --- diff --git a/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs b/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs index 1dc932675..39fef92fc 100644 --- a/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs +++ b/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs @@ -265,15 +265,11 @@ public async Task WithoutExceptionTagHeader_AndExceptionMarkerInStream_ShouldNot } /// -/// Tests that the raw / custom-FORMAT streaming surface () surfaces an -/// in-band mid-stream server exception as a across every accessor, -/// using mock HTTP responses (no ClickHouse server required). +/// Integration tests for mid-stream exception detection against a real +/// server, covering every accessor of the raw / custom-FORMAT streaming surface. /// -public class ClickHouseRawResultMidStreamMockTests +public class ClickHouseRawResultMidStreamTests : AbstractConnectionTestFixture { - private const string Token = "PU1FNUFH98"; - private const string ErrorMessage = "Code: 395. DB::Exception: boom"; - public enum Accessor { Stream, @@ -282,127 +278,116 @@ public enum Accessor CopyTo, } - // Real server framing: \r\n__exception__\r\n\r\n\n \r\n__exception__\r\n - private static byte[] MidStreamBody() => - Encoding.UTF8.GetBytes($"1,0\r\n2,0\r\n__exception__\r\n{Token}\r\n{ErrorMessage}\n{ErrorMessage.Length} {Token}\r\n__exception__\r\n"); - - private static HttpResponseMessage Response(byte[] content, string exceptionTag) - { - var response = new HttpResponseMessage(HttpStatusCode.OK) { Content = new ByteArrayContent(content) }; - if (exceptionTag != null) - response.Headers.Add("X-ClickHouse-Exception-Tag", exceptionTag); - return response; - } - - private static async Task DrainAsync(ClickHouseRawResult raw, Accessor accessor) + /// + /// Drains a raw result through the given accessor and returns the bytes it produced. The buffered + /// accessors materialize the whole body, so a mid-stream failure surfaces before they hand anything + /// back; the streaming ones surface it while the caller is draining. + /// + private static async Task DrainAsync(ClickHouseRawResult result, Accessor accessor) { switch (accessor) { case Accessor.Bytes: - return await raw.ReadAsByteArrayAsync(); + return await result.ReadAsByteArrayAsync(); case Accessor.String: - return Encoding.UTF8.GetBytes(await raw.ReadAsStringAsync()); - case Accessor.Stream: + return Encoding.UTF8.GetBytes(await result.ReadAsStringAsync()); + case Accessor.CopyTo: { - using var stream = await raw.ReadAsStreamAsync(); using var sink = new MemoryStream(); - await stream.CopyToAsync(sink); + await result.CopyToAsync(sink); return sink.ToArray(); } - default: // CopyTo + default: // Stream, drained through the array ReadAsync overload { + using var stream = await result.ReadAsStreamAsync(); using var sink = new MemoryStream(); - await raw.CopyToAsync(sink); + var buffer = new byte[64 * 1024]; + int read; + while ((read = await stream.ReadAsync(buffer, 0, buffer.Length)) > 0) + sink.Write(buffer, 0, read); return sink.ToArray(); } } } + private static ClickHouseCommand CreateStreamingCommand(ClickHouseConnection streamingConnection) + { + // Compression is disabled by the caller and buffering minimized here so the server streams the + // response incrementally; a buffered response instead fails pre-commit as a plain HTTP 500, which + // never exercises the in-band path. + var command = streamingConnection.CreateCommand(); + command.CustomSettings["http_write_exception_in_output_format"] = 1; + command.CustomSettings["max_block_size"] = 1000; + command.CustomSettings["http_response_buffer_size"] = 0; + command.CustomSettings["wait_end_of_query"] = 0; + return command; + } + [TestCase(Accessor.Stream)] [TestCase(Accessor.Bytes)] [TestCase(Accessor.String)] [TestCase(Accessor.CopyTo)] - public void Accessor_WithExceptionTag_AndInBandException_ThrowsServerException(Accessor accessor) + [FromVersion(25, 11)] + public async Task ExecuteRawResultAsync_MidStreamException_SurfacesServerException(Accessor accessor) { - using var response = Response(MidStreamBody(), Token); - using var raw = new ClickHouseRawResult(response); + // The query streams a committed 200 OK plus rows before throwIf fires, so the failure is delivered + // in-band (X-ClickHouse-Exception-Tag) and the truncated body surfaced as an HttpIOException that + // the raw path used to leak instead of the real server error. + using var streamingClient = TestUtilities.GetTestClickHouseClient(compression: false); + using var streamingConnection = streamingClient.CreateConnection(); + using var command = CreateStreamingCommand(streamingConnection); + + command.CommandText = @" + SELECT toInt32(number) AS n, + throwIf(number = 200000, 'boom mid stream') AS e + FROM system.numbers + LIMIT 400000 + FORMAT CSV"; - var ex = Assert.ThrowsAsync(() => DrainAsync(raw, accessor)); + using var result = await command.ExecuteRawResultAsync(default); - Assert.That(ex.Message, Does.Contain("boom")); - Assert.That(ex.ErrorCode, Is.EqualTo(395)); + var ex = Assert.ThrowsAsync(() => DrainAsync(result, accessor)); + + Assert.That(ex.Message, Does.Contain("boom mid stream")); + Assert.That(ex.ErrorCode, Is.EqualTo(395)); // FUNCTION_THROW_IF_VALUE_IS_NON_ZERO } [TestCase(Accessor.Stream)] [TestCase(Accessor.Bytes)] [TestCase(Accessor.String)] [TestCase(Accessor.CopyTo)] - public async Task Accessor_WithExceptionTag_ButSuccessfulBody_ReturnsFullBody(Accessor accessor) + [FromVersion(25, 11)] + public async Task ExecuteRawResultAsync_SuccessfulQuery_ReturnsCompleteBody(Accessor accessor) { - var body = Encoding.UTF8.GetBytes("1,0\r\n2,0\r\n3,0\r\n"); // tag present but query succeeded (no in-band block) - using var response = Response(body, Token); - using var raw = new ClickHouseRawResult(response); - - var read = await DrainAsync(raw, accessor); - - Assert.That(read, Is.EqualTo(body)); - } + // Contrast case. The server sends X-ClickHouse-Exception-Tag on every response, so detection is + // engaged for successful queries too; the body must still come back complete and unmodified. + using var command = connection.CreateCommand(); + command.CustomSettings["http_write_exception_in_output_format"] = 1; + command.CommandText = "SELECT number, number * 2 FROM system.numbers LIMIT 3 FORMAT CSV"; - [Test] - public async Task ReadAsByteArrayAsync_WithoutExceptionTag_ReturnsBodyVerbatim() - { - // No tag header => the untagged path must be byte-for-byte unchanged (no wrapping, no detection), - // even if the body coincidentally contains "__exception__" bytes. - var body = MidStreamBody(); - using var response = Response(body, exceptionTag: null); - using var raw = new ClickHouseRawResult(response); + using var result = await command.ExecuteRawResultAsync(default); - var bytes = await raw.ReadAsByteArrayAsync(); + var body = await DrainAsync(result, accessor); - Assert.That(bytes, Is.EqualTo(body)); + Assert.That(Encoding.UTF8.GetString(body), Is.EqualTo("0,0\n1,2\n2,4\n")); } -} -/// -/// Integration test for mid-stream exception detection against a real server. -/// -public class ClickHouseRawResultMidStreamTests : AbstractConnectionTestFixture -{ - [Test] + [TestCase(Accessor.Bytes)] + [TestCase(Accessor.String)] [FromVersion(25, 11)] - public async Task ExecuteRawResultAsync_MidStreamException_SurfacesServerException() + public async Task ExecuteRawResultAsync_BufferedAccessorReadTwice_ReturnsSameBody(Accessor accessor) { - // The query streams a committed 200 OK plus rows before throwIf fires, so the failure is delivered - // in-band (X-ClickHouse-Exception-Tag) and the truncated body surfaces as an HttpIOException that the - // raw path used to leak instead of the real server error. Compression is disabled and buffering - // minimized so the server streams incrementally (a buffered response instead fails pre-commit as 500). - using var streamingClient = TestUtilities.GetTestClickHouseClient(compression: false); - using var streamingConnection = streamingClient.CreateConnection(); - using var command = streamingConnection.CreateCommand(); - command.CustomSettings["http_write_exception_in_output_format"] = 1; - command.CustomSettings["max_block_size"] = 1000; - command.CustomSettings["http_response_buffer_size"] = 0; - command.CustomSettings["wait_end_of_query"] = 0; - - command.CommandText = @" - SELECT toInt32(number) AS n, - throwIf(number = 200000, 'boom mid stream') AS e - FROM system.numbers - LIMIT 400000 - FORMAT CSV"; + // The buffering accessors materialize the whole body, so — as when reading straight off + // HttpContent — asking twice must hand back the same body rather than failing on a consumed stream. + using var command = connection.CreateCommand(); + command.CommandText = "SELECT number, number * 2 FROM system.numbers LIMIT 3 FORMAT CSV"; using var result = await command.ExecuteRawResultAsync(default); - using var stream = await result.ReadAsStreamAsync(); - var ex = Assert.ThrowsAsync(async () => - { - var buffer = new byte[64 * 1024]; - while (await stream.ReadAsync(buffer, 0, buffer.Length) > 0) - { - // Drain until the in-band mid-stream exception surfaces - } - }); + var first = await DrainAsync(result, accessor); + var second = await DrainAsync(result, accessor); - Assert.That(ex.Message, Does.Contain("boom mid stream")); + Assert.That(Encoding.UTF8.GetString(first), Is.EqualTo("0,0\n1,2\n2,4\n")); + Assert.That(second, Is.EqualTo(first)); } } diff --git a/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs b/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs index 19674f518..64593bc42 100644 --- a/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs +++ b/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs @@ -19,6 +19,7 @@ public class ClickHouseRawResult : IDisposable { private readonly HttpResponseMessage response; private readonly string exceptionTag; + private byte[] bufferedContent; internal ClickHouseRawResult(HttpResponseMessage response) { @@ -64,7 +65,9 @@ public string ContentEncoding /// A task that resolves to the response content stream. /// /// If the server reports an in-band mid-stream exception, reading the returned stream throws a - /// once the end of the body is reached. + /// once the end of the body is reached. Bytes read before + /// that point are returned as-is, so a caller that parses incrementally may already have consumed + /// a partial result — including the server's raw in-band exception block — before the throw. /// public Task ReadAsStreamAsync() => string.IsNullOrEmpty(exceptionTag) ? response.Content.ReadAsStreamAsync() : WrapContentStreamAsync(); @@ -80,17 +83,25 @@ private async Task WrapContentStreamAsync() /// /// A task that resolves to the response content as bytes. /// - /// Throws a if the server reports an in-band mid-stream exception. + /// Throws a if the server reports an in-band mid-stream + /// exception; no truncated body is returned. The body is buffered, so repeat calls return the same bytes. /// public Task ReadAsByteArrayAsync() => string.IsNullOrEmpty(exceptionTag) ? response.Content.ReadAsByteArrayAsync() : ReadAllBytesAsync(); private async Task ReadAllBytesAsync() { - using var stream = await WrapContentStreamAsync().ConfigureAwait(false); + if (bufferedContent != null) + return bufferedContent; + + // The wrapper is deliberately not disposed: doing so closes the response's content stream, which + // HttpContent caches and hands back on the next call. Buffering the body here instead preserves + // the repeat-read behaviour of HttpContent.ReadAsByteArrayAsync/ReadAsStringAsync that callers + // get on the untagged path. The stream is released when this instance disposes the response. + var stream = await WrapContentStreamAsync().ConfigureAwait(false); using var memory = new MemoryStream(); await stream.CopyToAsync(memory).ConfigureAwait(false); - return memory.ToArray(); + return bufferedContent = memory.ToArray(); } /// @@ -98,7 +109,8 @@ private async Task ReadAllBytesAsync() /// /// A task that resolves to the response content as a string. /// - /// Throws a if the server reports an in-band mid-stream exception. + /// Throws a if the server reports an in-band mid-stream + /// exception; no truncated body is returned. The body is buffered, so repeat calls return the same string. /// public Task ReadAsStringAsync() => string.IsNullOrEmpty(exceptionTag) ? response.Content.ReadAsStringAsync() : ReadAllStringAsync(); @@ -123,14 +135,19 @@ private async Task ReadAllStringAsync() /// The destination stream to copy the content to. /// A task that completes when the copy operation is finished. /// - /// Throws a if the server reports an in-band mid-stream exception. + /// Throws a if the server reports an in-band mid-stream + /// exception. The copy is streamed, so the destination may already hold the partial result — including + /// the server's raw in-band exception block — by the time the exception is raised; treat a destination + /// written by a failed copy as incomplete. /// public Task CopyToAsync(Stream stream) => string.IsNullOrEmpty(exceptionTag) ? response.Content.CopyToAsync(stream) : CopyViaWrapperAsync(stream); private async Task CopyViaWrapperAsync(Stream destination) { - using var source = await WrapContentStreamAsync().ConfigureAwait(false); + // Not disposed, for the same reason as ReadAllBytesAsync: HttpContent.CopyToAsync leaves the + // content stream open too, so closing it here would break any subsequent read of this result. + var source = await WrapContentStreamAsync().ConfigureAwait(false); await source.CopyToAsync(destination).ConfigureAwait(false); } diff --git a/RELEASENOTES.md b/RELEASENOTES.md index e5c18e6af..63b22140b 100644 --- a/RELEASENOTES.md +++ b/RELEASENOTES.md @@ -28,7 +28,7 @@ Unreleased * Fixed `ClickHouseServerException` carrying a blank `Message` and an `ErrorCode` of `-1` when the server — or an upstream component such as a load balancer or the ClickHouse Cloud edge — returned a non-2xx HTTP response with an empty (or whitespace-only) body. The exception now reports the HTTP status code and reason phrase, and uses the `X-ClickHouse-Exception-Code` response header as the error code when the server sets it (issue #440). Non-empty error bodies are unaffected. * Fixed `ClickHouseDataReader.GetSchemaTable()` leaving `NumericScale` unset (`DBNull`) for `DateTime64(N)` and `Time64(N)` columns (including their `Nullable(...)` variants). The schema table now reports the fractional-seconds precision `N` in `NumericScale`, matching how `Decimal` columns are already reported (issue #438). * Fixed mid-stream server exceptions never surfacing on the streaming read path (`ExecuteReader`/`ExecuteReaderAsync`). A query that fails after the HTTP response is committed (for example a `throwIf` partway through a large result) now raises a `ClickHouseServerException` with the real server error, instead of a bare `HttpIOException` or `EndOfStreamException` (issue #476). -* Fixed the raw / custom-FORMAT streaming surface (`ExecuteRawResultAsync` → `ClickHouseRawResult`) not surfacing in-band mid-stream server exceptions. A query that fails after the HTTP response is committed (e.g. a `throwIf` partway through a large `FORMAT CSV`/`Arrow`/`Parquet` result) now raises a `ClickHouseServerException` with the real error across all four accessors, instead of a bare `HttpIOException` ("The response ended prematurely") or a body containing the raw exception block (issue #475). +* Fixed the raw / custom-FORMAT streaming surface (`ExecuteRawResultAsync` → `ClickHouseRawResult`) not surfacing in-band mid-stream server exceptions. A query that fails after the HTTP response is committed (e.g. a `throwIf` partway through a large `FORMAT CSV`/`Arrow`/`Parquet` result) now raises a `ClickHouseServerException` with the real error across all four accessors, instead of a bare `HttpIOException` ("The response ended prematurely"). `ReadAsByteArrayAsync`/`ReadAsStringAsync` throw rather than returning a truncated body; `ReadAsStreamAsync`/`CopyToAsync` raise the exception once the end of the response is reached, so data already streamed before the failure is not retroactively filtered (issue #475). v1.3.0 --- From e7987439cf49417fafaff8e6bf1f10b3e2d9827f Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Fri, 31 Jul 2026 10:57:47 +0000 Subject: [PATCH 4/7] Serve buffered body from the raw streaming accessors after a buffered read MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On the tagged raw path, once a buffering accessor (ReadAsByteArrayAsync / ReadAsStringAsync) materializes the body into bufferedContent it has drained the underlying content stream to EOF. A subsequent ReadAsStreamAsync / CopyToAsync re-read response.Content — the now-exhausted stream — and ignored the buffer, returning an empty body. Serve those accessors from bufferedContent when present, so all four stay consistent with each other and with the untagged HttpContent path (which buffers once and re-serves). Addresses the Cursor Bugbot review on #479. --- .../ADO/MidStreamExceptionTests.cs | 24 +++++++++++++++++++ .../ADO/Readers/ClickHouseRawResult.cs | 8 +++++++ 2 files changed, 32 insertions(+) diff --git a/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs b/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs index 39fef92fc..487dc2ce7 100644 --- a/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs +++ b/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs @@ -390,4 +390,28 @@ public async Task ExecuteRawResultAsync_BufferedAccessorReadTwice_ReturnsSameBod Assert.That(Encoding.UTF8.GetString(first), Is.EqualTo("0,0\n1,2\n2,4\n")); Assert.That(second, Is.EqualTo(first)); } + + [TestCase(Accessor.Bytes, Accessor.Stream)] + [TestCase(Accessor.Bytes, Accessor.CopyTo)] + [TestCase(Accessor.String, Accessor.Stream)] + [TestCase(Accessor.String, Accessor.CopyTo)] + [FromVersion(25, 11)] + public async Task ExecuteRawResultAsync_StreamingAccessorAfterBufferedRead_ReturnsCompleteBody(Accessor buffering, Accessor streaming) + { + // The exception tag is present on every response, so a buffering accessor materializes the whole + // body into an internal buffer. A subsequent streaming accessor must serve that buffer — matching + // the untagged HttpContent path, which buffers once and re-serves it — rather than re-reading the + // now-exhausted underlying content stream and handing back an empty body. + using var command = connection.CreateCommand(); + command.CustomSettings["http_write_exception_in_output_format"] = 1; + command.CommandText = "SELECT number, number * 2 FROM system.numbers LIMIT 3 FORMAT CSV"; + + using var result = await command.ExecuteRawResultAsync(default); + + var buffered = await DrainAsync(result, buffering); + var streamed = await DrainAsync(result, streaming); + + Assert.That(Encoding.UTF8.GetString(buffered), Is.EqualTo("0,0\n1,2\n2,4\n")); + Assert.That(streamed, Is.EqualTo(buffered)); + } } diff --git a/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs b/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs index 64593bc42..c557eecdf 100644 --- a/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs +++ b/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs @@ -74,6 +74,14 @@ public Task ReadAsStreamAsync() => private async Task WrapContentStreamAsync() { + // If a buffering accessor already materialized (and validated) the whole body, serve the streaming + // accessors from that buffer too. Re-reading response.Content here would hand back the underlying + // content stream the buffering read already drained to EOF — yielding an empty body. Buffering once + // and re-serving keeps all four accessors consistent with each other and with the untagged + // HttpContent path. The buffer is only ever set after a clean read, so no marker scan is needed. + if (bufferedContent != null) + return new MemoryStream(bufferedContent, writable: false); + var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false); return new ExceptionTagAwareStream(stream, exceptionTag, throwAtEndOfStream: true); } From 7b89e2bb9670dad1766160f412f5ea04837686f3 Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Fri, 31 Jul 2026 11:37:04 +0000 Subject: [PATCH 5/7] Throw like untagged HttpContent when re-materializing a consumed raw body On the tagged raw path (ExecuteRawResultAsync -> ClickHouseRawResult) a streaming accessor (ReadAsStreamAsync/CopyToAsync) consumes the underlying single-consumption content stream. A subsequent re-materializing read (ReadAsByteArrayAsync/ReadAsStringAsync/CopyToAsync) re-read the already-drained stream and returned/cached only the bytes left after the drain -- a silently truncated body. Untagged HttpContent throws InvalidOperationException ("The stream was already consumed. It cannot be read again.") in this ordering; mirror that instead of handing back a partial result the caller cannot distinguish from a complete one. Track whether the content stream has been vended and throw the same InvalidOperationException from the re-materializing accessors once it is, without eagerly buffering (per AGENTS.md "avoid buffering entire responses"). Re-requesting the stream itself still continues reading, matching untagged HttpContent. A buffering accessor that ran to completion first still serves its cached body. Addresses the cursor[bot] review on PR #479. --- .../ADO/MidStreamExceptionTests.cs | 73 +++++++++++++++++++ .../ADO/Readers/ClickHouseRawResult.cs | 18 +++++ 2 files changed, 91 insertions(+) diff --git a/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs b/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs index 487dc2ce7..2f845e3ac 100644 --- a/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs +++ b/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs @@ -1,3 +1,4 @@ +using System; using System.IO; using System.Net; using System.Net.Http; @@ -414,4 +415,76 @@ public async Task ExecuteRawResultAsync_StreamingAccessorAfterBufferedRead_Retur Assert.That(Encoding.UTF8.GetString(buffered), Is.EqualTo("0,0\n1,2\n2,4\n")); Assert.That(streamed, Is.EqualTo(buffered)); } + + /// + /// Consumes the underlying content stream through a streaming accessor without buffering it: Stream is + /// partially drained and left open (a caller that stopped mid-read — the reported scenario); CopyTo is + /// fully drained. Either way the content stream can no longer be re-materialized from the start. + /// + private static async Task ConsumeViaStreamingAccessorAsync(ClickHouseRawResult result, Accessor consumer) + { + if (consumer == Accessor.CopyTo) + { + using var sink = new MemoryStream(); + await result.CopyToAsync(sink); + return; + } + + var stream = await result.ReadAsStreamAsync(); + var probe = new byte[4]; + Assert.That(await stream.ReadAsync(probe, 0, probe.Length), Is.GreaterThan(0)); + } + + [TestCase(Accessor.Stream, Accessor.Bytes)] + [TestCase(Accessor.Stream, Accessor.String)] + [TestCase(Accessor.Stream, Accessor.CopyTo)] + [TestCase(Accessor.CopyTo, Accessor.Bytes)] + [TestCase(Accessor.CopyTo, Accessor.String)] + [TestCase(Accessor.CopyTo, Accessor.CopyTo)] + [FromVersion(25, 11)] + public async Task ExecuteRawResultAsync_ReMaterializingAccessorAfterStreamingConsumer_ThrowsLikeUntaggedContent(Accessor consumer, Accessor rematerializer) + { + // Once a streaming accessor (ReadAsStream or CopyTo) has consumed the underlying content stream it + // cannot be re-read from the start. A read that has to re-materialize the whole body — ReadAsByteArray, + // ReadAsString or CopyTo — must then fail with the same InvalidOperationException the untagged + // HttpContent path raises, rather than silently caching/copying a truncated body the caller cannot + // tell apart from a complete one. + using var command = connection.CreateCommand(); + command.CustomSettings["http_write_exception_in_output_format"] = 1; + command.CommandText = "SELECT number, number * 2 FROM system.numbers LIMIT 3 FORMAT CSV"; + + using var result = await command.ExecuteRawResultAsync(default); + + await ConsumeViaStreamingAccessorAsync(result, consumer); + + Assert.ThrowsAsync(() => DrainAsync(result, rematerializer)); + } + + [TestCase(Accessor.Stream)] + [TestCase(Accessor.CopyTo)] + [FromVersion(25, 11)] + public async Task ExecuteRawResultAsync_StreamAccessorAfterStreamingConsumer_ContinuesWithoutThrowing(Accessor consumer) + { + // The consumed-stream guard is deliberately scoped to the re-materializing accessors: re-requesting + // the stream itself must NOT throw, matching untagged HttpContent whose second ReadAsStreamAsync hands + // back the same, now-drained stream. Pins the fix as targeted rather than a blanket "any second + // accessor throws". + using var command = connection.CreateCommand(); + command.CustomSettings["http_write_exception_in_output_format"] = 1; + command.CommandText = "SELECT number, number * 2 FROM system.numbers LIMIT 3 FORMAT CSV"; + + using var result = await command.ExecuteRawResultAsync(default); + + await ConsumeViaStreamingAccessorAsync(result, consumer); + + Assert.DoesNotThrowAsync(async () => + { + using var again = await result.ReadAsStreamAsync(); + var buffer = new byte[64]; + while (await again.ReadAsync(buffer, 0, buffer.Length) > 0) + { + // Drain whatever remains; the point is that re-reading the stream does not throw. + } + }); + } } diff --git a/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs b/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs index c557eecdf..16a5b4be1 100644 --- a/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs +++ b/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs @@ -20,6 +20,7 @@ public class ClickHouseRawResult : IDisposable private readonly HttpResponseMessage response; private readonly string exceptionTag; private byte[] bufferedContent; + private bool contentConsumed; internal ClickHouseRawResult(HttpResponseMessage response) { @@ -83,9 +84,22 @@ private async Task WrapContentStreamAsync() return new MemoryStream(bufferedContent, writable: false); var stream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false); + contentConsumed = true; return new ExceptionTagAwareStream(stream, exceptionTag, throwAtEndOfStream: true); } + // A streaming accessor (ReadAsStreamAsync/CopyToAsync) already handed out — and thereby consumed — the + // underlying content stream, which cannot be re-read from the start. An accessor that has to re-materialize + // the whole body (ReadAsByteArray/ReadAsString/CopyTo) must then fail with the same InvalidOperationException + // the untagged HttpContent path raises once its stream is consumed, rather than caching/copying only the + // bytes left after a partial drain — a truncated body the caller cannot tell apart from a complete one. + // A buffering accessor that ran to completion first leaves bufferedContent set and is served from that. + private void ThrowIfContentAlreadyConsumed() + { + if (contentConsumed && bufferedContent == null) + throw new InvalidOperationException("The stream was already consumed. It cannot be read again."); + } + /// /// Reads the response content as a byte array. /// @@ -102,6 +116,8 @@ private async Task ReadAllBytesAsync() if (bufferedContent != null) return bufferedContent; + ThrowIfContentAlreadyConsumed(); + // The wrapper is deliberately not disposed: doing so closes the response's content stream, which // HttpContent caches and hands back on the next call. Buffering the body here instead preserves // the repeat-read behaviour of HttpContent.ReadAsByteArrayAsync/ReadAsStringAsync that callers @@ -153,6 +169,8 @@ public Task CopyToAsync(Stream stream) => private async Task CopyViaWrapperAsync(Stream destination) { + ThrowIfContentAlreadyConsumed(); + // Not disposed, for the same reason as ReadAllBytesAsync: HttpContent.CopyToAsync leaves the // content stream open too, so closing it here would break any subsequent read of this result. var source = await WrapContentStreamAsync().ConfigureAwait(false); From f64fe3aadd16bbf503bfbce4c91a9a129ded142c Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Wed, 5 Aug 2026 15:40:57 +0000 Subject: [PATCH 6/7] fix(raw): detect in-band mid-stream exceptions on compressed bodies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The exception-tag scanner was only ever layered over the raw content stream, which a raw result hands over verbatim — compressed bytes included when the caller asked for a codec. The server writes its in-band exception block into the *encoded* body, so the `__exception__` marker exists only in the decoded plaintext: scanning the compressed bytes can never match it, and the caller got the truncated transport error instead of the server's own. Verified on a live 26.5 server: with `Accept-Encoding: gzip` the marker occurs 0 times in the response body as received and twice after decoding it. The four original members must stay byte-exact pass-throughs, so the scanner goes where plaintext exists instead: `ReadDecompressedStreamAsync` now layers rawStream -> decoder -> ExceptionTagAwareStream, the same order the reader's read path uses, and raises the real `ClickHouseServerException` whether or not the body was compressed (it never surfaced one before, compressed or not). That member also joins the single-consumption bookkeeping the others already do: it fails like a re-materializing accessor over an already-consumed body, decodes over the buffer a buffering member left behind rather than the stream it drained, and marks the content consumed only once it has actually handed a stream out — so an undecodable codec still leaves the body readable. The docs on the verbatim members now state the limitation and point here. Co-Authored-By: Claude Opus 4.7 --- CHANGELOG.md | 2 +- .../ClickHouseRawResultDecompressionTests.cs | 86 +++++++++++++++ .../ADO/MidStreamExceptionTests.cs | 101 ++++++++++++++++++ .../ADO/Readers/ClickHouseRawResult.cs | 92 +++++++++++++--- RELEASENOTES.md | 2 +- 5 files changed, 264 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a886d10cb..b1b7d5ea4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -55,7 +55,7 @@ Unreleased * Fixed ADO-style `@name` placeholders not working when the parameter name contains a `$`, which ClickHouse accepts in a query parameter name: `@id$x` could not be bound at all (the name was interpolated into a regex, where `$` is an end-of-input anchor), and a shorter name won over a longer one, so with only `id` defined `SELECT @id$x` was silently rewritten into a different, still valid query that aliased the value as `$x` instead of being left for the server to reject. A `$` is now part of the placeholder name, matching the server lexer (issue #516). * Fixed `JSON` columns being unreadable when a typed path name requires backtick quoting — for example ``JSON(`a b` Int64)`` or ``JSON(`a,b` Int64)``. Such a path made the whole query fail with `SerializationException: Unsupported path in JSON hint`, because the type parser split each hint on every space and did not treat backticks as quotes. Quoted path names (including ones containing spaces, commas, parentheses and escaped characters) are now parsed and unescaped correctly (issue #502). * Fixed named `Tuple` and `Nested` columns being unreadable when an element name requires backtick quoting and contains a space — for example ``Tuple(`p q` Int64, r String)`` or ``Nested(`a b` Decimal(10, 2), c String)``. Reading such a column failed with `ArgumentException: Unknown type`, because the type parser split the element declaration on its first space and so cut the quoted name in half. The element name is now skipped as a whole before the name/type separator is located. -* Fixed the raw / custom-FORMAT streaming surface (`ExecuteRawResultAsync` → `ClickHouseRawResult`) not surfacing in-band mid-stream server exceptions (issue #475). The native reader detects these via the `X-ClickHouse-Exception-Tag` header, but `ClickHouseRawResult` handed the response body straight to the caller, so a query that failed after the HTTP `200 OK` was committed (e.g. a `throwIf` partway through a large `FORMAT CSV`/`JSONEachRow`/`Arrow`/`Parquet` result) surfaced as a `System.Net.Http.HttpIOException` ("The response ended prematurely") instead of a `ClickHouseServerException`. All four accessors (`ReadAsStreamAsync`, `ReadAsByteArrayAsync`, `ReadAsStringAsync`, `CopyToAsync`) now consult the exception-tag header and raise a `ClickHouseServerException` carrying the server's message: the buffered accessors (`ReadAsByteArrayAsync`/`ReadAsStringAsync`) throw instead of returning a truncated body, while the streaming accessors (`ReadAsStreamAsync`/`CopyToAsync`) raise it once the end of the response is reached — data already streamed to the caller before the failure is still delivered as-is and is not retroactively filtered. Successful responses are unaffected. +* Fixed the raw / custom-FORMAT streaming surface (`ExecuteRawResultAsync` → `ClickHouseRawResult`) not surfacing in-band mid-stream server exceptions (issue #475). The native reader detects these via the `X-ClickHouse-Exception-Tag` header, but `ClickHouseRawResult` handed the response body straight to the caller, so a query that failed after the HTTP `200 OK` was committed (e.g. a `throwIf` partway through a large `FORMAT CSV`/`JSONEachRow`/`Arrow`/`Parquet` result) surfaced as a `System.Net.Http.HttpIOException` ("The response ended prematurely") instead of a `ClickHouseServerException`. All four accessors (`ReadAsStreamAsync`, `ReadAsByteArrayAsync`, `ReadAsStringAsync`, `CopyToAsync`) now consult the exception-tag header and raise a `ClickHouseServerException` carrying the server's message: the buffered accessors (`ReadAsByteArrayAsync`/`ReadAsStringAsync`) throw instead of returning a truncated body, while the streaming accessors (`ReadAsStreamAsync`/`CopyToAsync`) raise it once the end of the response is reached — data already streamed to the caller before the failure is still delivered as-is and is not retroactively filtered. `ReadDecompressedStreamAsync` raises it too, and — because the server writes the exception block into the *encoded* body, where only the decoded plaintext holds the marker — it is the accessor that surfaces the error on a transport-compressed response; the four verbatim accessors hand such a body over as-is, undetected, as their contract requires. Successful responses are unaffected. v1.3.0 --- diff --git a/ClickHouse.Driver.Tests/ADO/ClickHouseRawResultDecompressionTests.cs b/ClickHouse.Driver.Tests/ADO/ClickHouseRawResultDecompressionTests.cs index 31ecce85a..6185a40c3 100644 --- a/ClickHouse.Driver.Tests/ADO/ClickHouseRawResultDecompressionTests.cs +++ b/ClickHouse.Driver.Tests/ADO/ClickHouseRawResultDecompressionTests.cs @@ -285,6 +285,92 @@ public async Task RawResultDispose_DisposesTheDecoderItInserted() Assert.Throws(() => decoder.ReadByte(), "the raw result owns the decoder it created"); } + /// + /// A response carrying X-ClickHouse-Exception-Tag — sent whenever + /// http_write_exception_in_output_format is on, i.e. on every response of such a query, failing or + /// not. It engages the in-band exception scanner, and with it the single-consumption bookkeeping the + /// verbatim members already do, which the tests below extend to the decoding member. + /// + private static HttpResponseMessage CreateTaggedStreamedResponse(byte[] body, string contentEncoding) + { + var response = CreateStreamedResponse(body, contentEncoding); + response.Headers.Add("X-ClickHouse-Exception-Tag", "PU1FNUFH98"); + return response; + } + + [Test] + public async Task ReadDecompressedStreamAsync_OnATaggedResponse_AfterAPartialVerbatimRead_ThrowsLikeConsumedContent() + { + // With the scanner engaged, a verbatim read marks the content stream consumed. Decoding it from + // there would start mid-frame, so this must fail the same way the re-materializing members do — + // rather than surfacing a decoder-internal error, or worse a short body. + using var response = CreateTaggedStreamedResponse(Lz4Encoded(), "lz4"); + using var raw = new ClickHouseRawResult(response); + + (await raw.ReadAsStreamAsync()).ReadByte(); + + Assert.ThrowsAsync(() => raw.ReadDecompressedStreamAsync()); + } + + [Test] + public async Task ReadAsByteArrayAsync_OnATaggedResponse_AfterAPartialDecode_ThrowsRatherThanTruncating() + { + // The mirror image: the decoding member consumes the content stream too (and reads ahead), so a + // member that has to re-materialize the whole body afterwards must fail loudly instead of caching + // whatever bytes are left. + using var response = CreateTaggedStreamedResponse(Lz4Encoded(), "lz4"); + using var raw = new ClickHouseRawResult(response); + + Assert.That((await raw.ReadDecompressedStreamAsync()).ReadByte(), Is.EqualTo(Plaintext[0])); + + Assert.ThrowsAsync(() => raw.ReadAsByteArrayAsync()); + } + + [Test] + public async Task ReadDecompressedStreamAsync_OnATaggedResponse_AfterABufferingMember_StillSeesTheWholeBody() + { + // A buffering member on the tagged path caches the body itself — still encoded, since it hands the + // wire bytes over verbatim — so the decode must run over that buffer rather than the content stream + // it drained. + using var response = CreateTaggedStreamedResponse(Lz4Encoded(), "lz4"); + using var raw = new ClickHouseRawResult(response); + + Assert.That(await raw.ReadAsByteArrayAsync(), Is.EqualTo(Lz4Encoded())); + + using var buffer = new MemoryStream(); + await (await raw.ReadDecompressedStreamAsync()).CopyToAsync(buffer); + + Assert.That(buffer.ToArray(), Is.EqualTo(Plaintext)); + } + + [Test] + public async Task ReadDecompressedStreamAsync_OnATaggedResponse_WithUnsupportedCodec_LeavesTheBodyReadable() + { + // The undecodable-codec throw hands nothing out, so — exactly as on an untagged response — it must + // not mark the content consumed and lock the other members out of a body that is still whole. + var compressed = Lz4Encoded(); + using var response = CreateTaggedStreamedResponse(compressed, "zstd"); + using var raw = new ClickHouseRawResult(response); + + Assert.ThrowsAsync(() => raw.ReadDecompressedStreamAsync()); + + Assert.That(await raw.ReadAsByteArrayAsync(), Is.EqualTo(compressed)); + } + + [Test] + public async Task ReadDecompressedStreamAsync_OnATaggedResponse_CalledTwice_ReturnsTheSameStream() + { + // Repeat calls must hand back the same scanner: a fresh one would have observed nothing, so the + // marker recorded so far — the whole point of the wrapper — would be lost. + using var response = CreateTaggedStreamedResponse(Lz4Encoded(), "lz4"); + using var raw = new ClickHouseRawResult(response); + + var first = await raw.ReadDecompressedStreamAsync(); + var second = await raw.ReadDecompressedStreamAsync(); + + Assert.That(second, Is.SameAs(first)); + } + [Test] public async Task RawResultDispose_AfterTheCallerAlreadyDisposedTheDecoder_IsHarmless() { diff --git a/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs b/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs index 2f845e3ac..124146c21 100644 --- a/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs +++ b/ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs @@ -460,6 +460,107 @@ public async Task ExecuteRawResultAsync_ReMaterializingAccessorAfterStreamingCon Assert.ThrowsAsync(() => DrainAsync(result, rematerializer)); } + /// + /// A mid-stream failure that survives compression. The server buffers a compressed body, so it only + /// commits the 200 OK — and with it the in-band exception path — once enough output has accumulated to + /// flush; a smaller result fails pre-commit as a plain error instead, which is a different code path. + /// + private const string CompressibleMidStreamQuery = @" + SELECT toInt32(number) AS n, + throwIf(number = 1000000, 'boom mid stream') AS e + FROM system.numbers + LIMIT 2000000 + FORMAT CSV"; + + /// Drains a plaintext stream, discarding the bytes. + private static async Task DrainToEndAsync(Stream stream) + { + var buffer = new byte[64 * 1024]; + while (await stream.ReadAsync(buffer, 0, buffer.Length) > 0) + { + // The point is reaching the end of the body, where a mid-stream failure surfaces. + } + } + + [TestCase("gzip")] + [TestCase("lz4")] + [TestCase(null)] + [FromVersion(25, 11)] + public async Task ExecuteRawResultAsync_MidStreamException_DecompressedStreamSurfacesServerException(string codec) + { + // The server writes its in-band exception block into the ENCODED body, so the marker only exists in + // the decoded plaintext: the scanner must therefore sit above the decoder. Asking for a codec is what + // makes a raw body compressed at all (a raw request advertises none by default), and the null case + // pins that the same member still detects the block when there is nothing to decode. + using var streamingClient = TestUtilities.GetTestClickHouseClient(compression: false); + using var streamingConnection = streamingClient.CreateConnection(); + using var command = CreateStreamingCommand(streamingConnection); + command.AcceptEncoding = codec; + + command.CommandText = CompressibleMidStreamQuery; + + using var result = await command.ExecuteRawResultAsync(default); + + Assert.That(result.ContentEncoding, Is.EqualTo(codec), "the body must be encoded as the test asked"); + + var ex = Assert.ThrowsAsync(async () => + await DrainToEndAsync(await result.ReadDecompressedStreamAsync())); + + Assert.That(ex.Message, Does.Contain("boom mid stream")); + Assert.That(ex.ErrorCode, Is.EqualTo(395)); // FUNCTION_THROW_IF_VALUE_IS_NON_ZERO + } + + [TestCase(Accessor.Stream)] + [TestCase(Accessor.Bytes)] + [TestCase(Accessor.String)] + [TestCase(Accessor.CopyTo)] + [FromVersion(25, 11)] + public async Task ExecuteRawResultAsync_MidStreamException_OnCompressedBody_VerbatimAccessorsCannotDetectIt(Accessor accessor) + { + // Contrast case, pinning the documented boundary of the fix rather than extending it: the four + // original members hand the bytes on the wire over verbatim, and a compressed body carries the + // exception block compressed too, so no scan of it can match. The caller still gets an error — the + // truncated transport — just not the server's own; they can find the block in what they decode, or + // use ReadDecompressedStreamAsync above. + using var streamingClient = TestUtilities.GetTestClickHouseClient(compression: false); + using var streamingConnection = streamingClient.CreateConnection(); + using var command = CreateStreamingCommand(streamingConnection); + command.AcceptEncoding = "gzip"; + + command.CommandText = CompressibleMidStreamQuery; + + using var result = await command.ExecuteRawResultAsync(default); + + Assert.That(result.ContentEncoding, Is.EqualTo("gzip")); + + var ex = Assert.CatchAsync(() => DrainAsync(result, accessor)); + + Assert.That(ex, Is.Not.InstanceOf()); + Assert.That(ex, Is.InstanceOf(), "the truncated transport is what surfaces"); + } + + [TestCase("gzip")] + [TestCase("lz4")] + [FromVersion(25, 11)] + public async Task ExecuteRawResultAsync_SuccessfulCompressedQuery_DecompressedStreamReturnsCompleteBody(string codec) + { + // Contrast case. The tag header is sent on every response, so the scanner is engaged over the decoder + // for successful queries too; the decoded body must still come back complete and unmodified. + using var command = connection.CreateCommand(); + command.CustomSettings["http_write_exception_in_output_format"] = 1; + command.AcceptEncoding = codec; + command.CommandText = "SELECT number, number * 2 FROM system.numbers LIMIT 3 FORMAT CSV"; + + using var result = await command.ExecuteRawResultAsync(default); + + Assert.That(result.ContentEncoding, Is.EqualTo(codec)); + + using var sink = new MemoryStream(); + await (await result.ReadDecompressedStreamAsync()).CopyToAsync(sink); + + Assert.That(Encoding.UTF8.GetString(sink.ToArray()), Is.EqualTo("0,0\n1,2\n2,4\n")); + } + [TestCase(Accessor.Stream)] [TestCase(Accessor.CopyTo)] [FromVersion(25, 11)] diff --git a/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs b/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs index 43d1c2c5c..d3258a795 100644 --- a/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs +++ b/ClickHouse.Driver/ADO/Readers/ClickHouseRawResult.cs @@ -27,6 +27,15 @@ namespace ClickHouse.Driver.ADO; /// and buffer the whole body, so they /// can safely be followed by another read. Not safe for concurrent use. /// +/// +/// In-band exceptions and compression. A query that fails after its 200 OK has been committed +/// reports the failure inside the body (the http_write_exception_in_output_format setting), and the +/// members below raise the server's error as a rather than leaking +/// that block or a truncated body. The server writes it into the encoded body, though, so it can only +/// be found in plaintext: on a transport-compressed response — which a raw request only gets when the caller +/// asked for a codec — the four verbatim members hand the compressed bytes over undetected, as they must, +/// and is the member that surfaces the error. +/// /// public class ClickHouseRawResult : IDisposable { @@ -36,13 +45,20 @@ public class ClickHouseRawResult : IDisposable private bool contentConsumed; /// - /// The decoder inserted over the content stream, if any. - /// Kept so repeated calls hand back the same decoder rather than stacking a second one over an - /// already-partly-consumed body, and so releases it — a decoder holds pooled - /// buffers and has no finalizer to fall back on. + /// The stream vended, if any: the decoder it inserted over + /// the content stream, under the exception-tag scanner when one is engaged. Kept so repeated calls hand + /// back that same stream rather than stacking a second decoder over an already-partly-consumed body (or + /// a fresh scanner whose ring buffer has observed nothing). /// private Stream decompressedStream; + /// + /// The decoder inserted, if any — the only stream in that + /// chain that is ours to release: a decoder holds pooled buffers and has no finalizer to fall back on, + /// while the content stream below it belongs to . + /// + private Stream ownedDecoder; + internal ClickHouseRawResult(HttpResponseMessage response) { this.response = response; @@ -89,7 +105,9 @@ public string ContentEncoding /// If the server reports an in-band mid-stream exception, reading the returned stream throws a /// once the end of the body is reached. Bytes read before /// that point are returned as-is, so a caller that parses incrementally may already have consumed - /// a partial result — including the server's raw in-band exception block — before the throw. + /// a partial result — including the server's raw in-band exception block — before the throw. The + /// marker exists only in plaintext, so on a transport-compressed body this pass-through cannot find + /// it: use , or look for the block in what you decode. /// public Task ReadAsStreamAsync() => string.IsNullOrEmpty(exceptionTag) ? response.Content.ReadAsStreamAsync() : WrapContentStreamAsync(); @@ -111,7 +129,8 @@ private async Task WrapContentStreamAsync() // A streaming accessor (ReadAsStreamAsync/CopyToAsync) already handed out — and thereby consumed — the // underlying content stream, which cannot be re-read from the start. An accessor that has to re-materialize - // the whole body (ReadAsByteArray/ReadAsString/CopyTo) must then fail with the same InvalidOperationException + // the whole body (ReadAsByteArray/ReadAsString/CopyTo) — or decode it from the start + // (ReadDecompressedStream) — must then fail with the same InvalidOperationException // the untagged HttpContent path raises once its stream is consumed, rather than caching/copying only the // bytes left after a partial drain — a truncated body the caller cannot tell apart from a complete one. // A buffering accessor that ran to completion first leaves bufferedContent set and is served from that. @@ -129,13 +148,19 @@ private void ThrowIfContentAlreadyConsumed() /// br are decoded. /// /// A task that resolves to a plaintext stream over the response content. + /// + /// The server reported an in-band mid-stream exception, found once the end of the decoded body is + /// reached. Because this member yields plaintext it detects the block whether or not the body was + /// compressed — unlike the verbatim members, which can only see it in an uncompressed body. + /// /// /// The response uses a codec this client cannot decode (e.g. zstd); the message names it. /// /// /// Disposing this is always sufficient — it releases the response and /// any decoder inserted here. Disposing the returned stream is safe too, but note that with nothing to - /// decode it is the content stream, so that ends the response body. Repeated sequential calls + /// decode and no exception scanner engaged it is the content stream, so that ends the response + /// body; the decoder and the scanner both leave what is below them open. Repeated sequential calls /// return the same stream rather than stacking a decoder over a partly-consumed body; not safe for /// concurrent use. /// @@ -144,18 +169,45 @@ public async Task ReadDecompressedStreamAsync() if (decompressedStream != null) return decompressedStream; - var rawStream = await response.Content.ReadAsStreamAsync().ConfigureAwait(false); + Stream source; + if (bufferedContent != null) + { + // A buffering accessor already materialized the body — still encoded, since those accessors + // hand the wire bytes over verbatim — and drained the content stream doing so. Decode over that + // buffer, for the same reason WrapContentStreamAsync serves the streaming accessors from it. + source = new MemoryStream(bufferedContent, writable: false); + } + else + { + // Nothing buffered and the content stream already consumed: a decode would start from wherever + // the previous reader stopped and produce garbage or a truncated body, so fail exactly as the + // re-materializing accessors do. + ThrowIfContentAlreadyConsumed(); + source = await response.Content.ReadAsStreamAsync().ConfigureAwait(false); + } // Throws for a codec we cannot decode, before anything has been read — so the body is left intact // for a caller who wants to decode it themselves, through any read member. Nothing leaks either: - // rawStream is owned by `response` and released by Dispose(). - var wrapped = ResponseDecompression.Wrap(rawStream, response, leaveOpen: true); + // the content stream is owned by `response` and released by Dispose(). The consumed flag is set + // only once past this point, for the same reason: a call that threw here handed nothing out, so it + // must not lock the other members out of a body that is still whole. + var wrapped = ResponseDecompression.Wrap(source, response, leaveOpen: true); + contentConsumed = true; // Only a decoder we inserted is ours to dispose; the content stream belongs to the response. - if (!ReferenceEquals(wrapped, rawStream)) - decompressedStream = wrapped; + if (!ReferenceEquals(wrapped, source)) + ownedDecoder = wrapped; + + // The exception-tag scanner sits ABOVE the decoder, exactly as it does on the reader's read path: + // the server writes its in-band exception block into the response body, so the marker exists only + // in the decoded plaintext — a scan of the compressed bytes would never match it. This is the + // member that yields plaintext, so it is the one that can surface a mid-stream failure on a + // transport-compressed response; the verbatim accessors above can only do so when the body is not + // compressed. leaveOpen: true — the stream below belongs to the response or to ownedDecoder. + if (!string.IsNullOrEmpty(exceptionTag)) + wrapped = new ExceptionTagAwareStream(wrapped, exceptionTag, leaveOpen: true, throwAtEndOfStream: true); - return wrapped; + return decompressedStream = wrapped; } /// @@ -164,7 +216,9 @@ public async Task ReadDecompressedStreamAsync() /// A task that resolves to the response content as bytes. /// /// Throws a if the server reports an in-band mid-stream - /// exception; no truncated body is returned. The body is buffered, so repeat calls return the same bytes. + /// exception in a plaintext body; no truncated body is returned. A transport-compressed body hides the + /// marker from this verbatim read — see the remarks on the type. The body is buffered, so repeat calls + /// return the same bytes. /// public Task ReadAsByteArrayAsync() => string.IsNullOrEmpty(exceptionTag) ? response.Content.ReadAsByteArrayAsync() : ReadAllBytesAsync(); @@ -192,7 +246,9 @@ private async Task ReadAllBytesAsync() /// A task that resolves to the response content as a string. /// /// Throws a if the server reports an in-band mid-stream - /// exception; no truncated body is returned. The body is buffered, so repeat calls return the same string. + /// exception in a plaintext body; no truncated body is returned. A transport-compressed body hides the + /// marker from this verbatim read — see the remarks on the type. The body is buffered, so repeat calls + /// return the same string. /// public Task ReadAsStringAsync() => string.IsNullOrEmpty(exceptionTag) ? response.Content.ReadAsStringAsync() : ReadAllStringAsync(); @@ -218,7 +274,8 @@ private async Task ReadAllStringAsync() /// A task that completes when the copy operation is finished. /// /// Throws a if the server reports an in-band mid-stream - /// exception. The copy is streamed, so the destination may already hold the partial result — including + /// exception in a plaintext body (a transport-compressed body hides the marker from this verbatim copy — + /// see the remarks on the type). The copy is streamed, so the destination may already hold the partial result — including /// the server's raw in-band exception block — by the time the exception is raised; treat a destination /// written by a failed copy as incomplete. /// @@ -239,7 +296,8 @@ public void Dispose() { // Decoder first (it reads from the content stream), then the response that owns that stream. // Nulled so a second Dispose() does not release a decoder's pooled buffers twice. - var decoder = decompressedStream; + var decoder = ownedDecoder; + ownedDecoder = null; decompressedStream = null; decoder?.Dispose(); diff --git a/RELEASENOTES.md b/RELEASENOTES.md index 6d9b27bfc..85522e79c 100644 --- a/RELEASENOTES.md +++ b/RELEASENOTES.md @@ -55,7 +55,7 @@ Unreleased * Fixed ADO-style `@name` placeholders not working when the parameter name contains a `$`, which ClickHouse accepts in a query parameter name: `@id$x` could not be bound at all (the name was interpolated into a regex, where `$` is an end-of-input anchor), and a shorter name won over a longer one, so with only `id` defined `SELECT @id$x` was silently rewritten into a different, still valid query that aliased the value as `$x` instead of being left for the server to reject. A `$` is now part of the placeholder name, matching the server lexer (issue #516). * Fixed `JSON` columns being unreadable when a typed path name requires backtick quoting — for example ``JSON(`a b` Int64)`` or ``JSON(`a,b` Int64)``. Such a path made the whole query fail with `SerializationException: Unsupported path in JSON hint`, because the type parser split each hint on every space and did not treat backticks as quotes. Quoted path names (including ones containing spaces, commas, parentheses and escaped characters) are now parsed and unescaped correctly (issue #502). * Fixed named `Tuple` and `Nested` columns being unreadable when an element name requires backtick quoting and contains a space — for example ``Tuple(`p q` Int64, r String)`` or ``Nested(`a b` Decimal(10, 2), c String)``. Reading such a column failed with `ArgumentException: Unknown type`, because the type parser split the element declaration on its first space and so cut the quoted name in half. The element name is now skipped as a whole before the name/type separator is located. -* Fixed the raw / custom-FORMAT streaming surface (`ExecuteRawResultAsync` → `ClickHouseRawResult`) not surfacing in-band mid-stream server exceptions. A query that fails after the HTTP response is committed (e.g. a `throwIf` partway through a large `FORMAT CSV`/`Arrow`/`Parquet` result) now raises a `ClickHouseServerException` with the real error across all four accessors, instead of a bare `HttpIOException` ("The response ended prematurely"). `ReadAsByteArrayAsync`/`ReadAsStringAsync` throw rather than returning a truncated body; `ReadAsStreamAsync`/`CopyToAsync` raise the exception once the end of the response is reached, so data already streamed before the failure is not retroactively filtered (issue #475). +* Fixed the raw / custom-FORMAT streaming surface (`ExecuteRawResultAsync` → `ClickHouseRawResult`) not surfacing in-band mid-stream server exceptions. A query that fails after the HTTP response is committed (e.g. a `throwIf` partway through a large `FORMAT CSV`/`Arrow`/`Parquet` result) now raises a `ClickHouseServerException` with the real error across all four accessors, instead of a bare `HttpIOException` ("The response ended prematurely"). `ReadAsByteArrayAsync`/`ReadAsStringAsync` throw rather than returning a truncated body; `ReadAsStreamAsync`/`CopyToAsync` raise the exception once the end of the response is reached, so data already streamed before the failure is not retroactively filtered `ReadDecompressedStreamAsync` raises it as well, and is the accessor that surfaces it on a transport-compressed body (the marker only exists in the decoded plaintext, so the verbatim accessors cannot see it there) (issue #475). v1.3.0 --- From dd9ff642d537f438ce2f8b9247e4483887b0e90e Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Fri, 14 Aug 2026 16:41:00 +0000 Subject: [PATCH 7/7] test(raw): use snappy for the undecodable-codec case now that zstd decodes main gained a vendored ZstdSharp decoder, so the tagged-response unsupported-codec test could no longer reach the NotSupportedException path. Switch it to snappy, matching the untagged test next to it. --- .../ADO/ClickHouseRawResultDecompressionTests.cs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/ClickHouse.Driver.Tests/ADO/ClickHouseRawResultDecompressionTests.cs b/ClickHouse.Driver.Tests/ADO/ClickHouseRawResultDecompressionTests.cs index 766a8c18c..33852113d 100644 --- a/ClickHouse.Driver.Tests/ADO/ClickHouseRawResultDecompressionTests.cs +++ b/ClickHouse.Driver.Tests/ADO/ClickHouseRawResultDecompressionTests.cs @@ -387,8 +387,9 @@ public async Task ReadDecompressedStreamAsync_OnATaggedResponse_WithUnsupportedC { // The undecodable-codec throw hands nothing out, so — exactly as on an untagged response — it must // not mark the content consumed and lock the other members out of a body that is still whole. + // snappy, not zstd: zstd became decodable when the vendored ZstdSharp codec landed. var compressed = Lz4Encoded(); - using var response = CreateTaggedStreamedResponse(compressed, "zstd"); + using var response = CreateTaggedStreamedResponse(compressed, "snappy"); using var raw = new ClickHouseRawResult(response); Assert.ThrowsAsync(() => raw.ReadDecompressedStreamAsync());