Fix ClickHouseRawResult: surface in-band mid-stream server exceptions on the raw streaming path

CHANGELOG.md

@@ -28,6 +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.
 
 v1.3.0
 ---

ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs

@@ -263,3 +263,146 @@ public async Task WithoutExceptionTagHeader_AndExceptionMarkerInStream_ShouldNot
         Assert.That(ex, Is.TypeOf<EndOfStreamException>());
     }
 }
+
+/// <summary>
+/// Tests that the raw / custom-FORMAT streaming surface (<see cref="ClickHouseRawResult"/>) surfaces an
+/// in-band mid-stream server exception as a <see cref="ClickHouseServerException"/> across every accessor,
+/// using mock HTTP responses (no ClickHouse server required).
+/// </summary>
+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: <rows>\r\n__exception__\r\n<token>\r\n<message>\n<size> <token>\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<byte[]> 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<ClickHouseServerException>(() => 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));
+    }
+}
+
+/// <summary>
+/// Integration test for <see cref="ClickHouseRawResult"/> mid-stream exception detection against a real server.
+/// </summary>
+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<ClickHouseServerException>(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"));
+    }
+}

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<ClickHouseServerException>(() => 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<ClickHouseServerException>(() =>
+        {
+            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);
+    }
+
+    /// <summary>Stream that yields its content, then throws an IOException at end-of-stream (like a dropped HTTP connection).</summary>
+    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);
+        }
+    }
 }