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") 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
 ---

ClickHouse.Driver.Tests/ADO/MidStreamExceptionTests.cs

@@ -1,3 +1,4 @@
+using System;
 using System.IO;
 using System.Net;
 using System.Net.Http;
@@ -263,3 +264,227 @@ public async Task WithoutExceptionTagHeader_AndExceptionMarkerInStream_ShouldNot
         Assert.That(ex, Is.TypeOf<EndOfStreamException>());
     }
 }
+
+/// <summary>
+/// Integration tests for <see cref="ClickHouseRawResult"/> mid-stream exception detection against a real
+/// server, covering every accessor of the raw / custom-FORMAT streaming surface.
+/// </summary>
+public class ClickHouseRawResultMidStreamTests : AbstractConnectionTestFixture
+{
+    public enum Accessor
+    {
+        Stream,
+        Bytes,
+        String,
+        CopyTo,
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    private static async Task<byte[]> DrainAsync(ClickHouseRawResult result, Accessor accessor)
+    {
+        switch (accessor)
+        {
+            case Accessor.Bytes:
+                return await result.ReadAsByteArrayAsync();
+            case Accessor.String:
+                return Encoding.UTF8.GetBytes(await result.ReadAsStringAsync());
+            case Accessor.CopyTo:
+            {
+                using var sink = new MemoryStream();
+                await result.CopyToAsync(sink);
+                return sink.ToArray();
+            }
+            default: // Stream, drained through the array ReadAsync overload
+            {
+                using var stream = await result.ReadAsStreamAsync();
+                using var sink = new MemoryStream();
+                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)]
+    [FromVersion(25, 11)]
+    public async Task ExecuteRawResultAsync_MidStreamException_SurfacesServerException(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 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";
+
+        using var result = await command.ExecuteRawResultAsync(default);
+
+        var ex = Assert.ThrowsAsync<ClickHouseServerException>(() => 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)]
+    [FromVersion(25, 11)]
+    public async Task ExecuteRawResultAsync_SuccessfulQuery_ReturnsCompleteBody(Accessor accessor)
+    {
+        // 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";
+
+        using var result = await command.ExecuteRawResultAsync(default);
+
+        var body = await DrainAsync(result, accessor);
+
+        Assert.That(Encoding.UTF8.GetString(body), Is.EqualTo("0,0\n1,2\n2,4\n"));
+    }
+
+    [TestCase(Accessor.Bytes)]
+    [TestCase(Accessor.String)]
+    [FromVersion(25, 11)]
+    public async Task ExecuteRawResultAsync_BufferedAccessorReadTwice_ReturnsSameBody(Accessor accessor)
+    {
+        // 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);
+
+        var first = await DrainAsync(result, accessor);
+        var second = await DrainAsync(result, accessor);
+
+        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));
+    }
+
+    /// <summary>
+    /// 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.
+    /// </summary>
+    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<InvalidOperationException>(() => 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.
+            }
+        });
+    }
+}

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);
+        }
+    }
 }