fix(output): throw server error for streamed exceptions

src/Format/JsonEachRow.php

@@ -6,6 +6,7 @@
 
 use JsonException;
 use Psr\Http\Message\StreamInterface;
+use SimPod\ClickHouseClient\Exception\ServerError;
 use SimPod\ClickHouseClient\Output\Output;
 
 /**
@@ -14,7 +15,7 @@
  */
 final readonly class JsonEachRow implements Format
 {
-    /** @throws JsonException */
+    /** @throws JsonException|ServerError */
     public static function output(string|StreamInterface $contents): Output
     {
         /** @var \SimPod\ClickHouseClient\Output\JsonEachRow<T> $output */

src/Output/JsonEachRow.php

@@ -7,9 +7,12 @@
 use Generator;
 use JsonException;
 use Psr\Http\Message\StreamInterface;
+use SimPod\ClickHouseClient\Exception\ServerError;
 
 use function explode;
+use function implode;
 use function json_decode;
+use function rtrim;
 use function strpos;
 use function substr;
 
@@ -24,7 +27,7 @@
     /** @phpstan-var Generator<int, T> */
     public Generator $data;
 
-    /** @throws JsonException */
+    /** @throws JsonException|ServerError */
     public function __construct(string|StreamInterface $contentsJson)
     {
         $lines = $contentsJson instanceof StreamInterface
@@ -40,20 +43,40 @@ public function __construct(string|StreamInterface $contentsJson)
      *
      * @return Generator<int, T>
      *
-     * @throws JsonException
+     * @throws JsonException|ServerError
      */
     private static function decodeLines(iterable $lines): Generator
     {
+        $streamedExceptionLines = [];
+
         foreach ($lines as $line) {
+            $line = rtrim($line, "\r");
+
+            if ($streamedExceptionLines !== []) {
+                $streamedExceptionLines[] = $line;
+
+                continue;
+            }
+
             if ($line === '') {
                 continue;
             }
 
+            if ($line === '__exception__') {
+                $streamedExceptionLines[] = $line;
+
+                continue;
+            }
+
             /** @phpstan-var T $row */
             $row = json_decode($line, true, flags: JSON_THROW_ON_ERROR);
 
             yield $row;
         }
+
+        if ($streamedExceptionLines !== []) {
+            throw ServerError::fromResponseContent(implode("\n", $streamedExceptionLines), 200);
+        }
     }
 
     /** @return Generator<string> */

tests/Output/JsonEachRowTest.php

@@ -5,12 +5,17 @@
 namespace SimPod\ClickHouseClient\Tests\Output;
 
 use GuzzleHttp\Psr7\Utils;
+use InvalidArgumentException;
 use PHPUnit\Framework\Attributes\CoversClass;
+use PHPUnit\Framework\Attributes\DataProvider;
+use Psr\Http\Message\StreamInterface;
+use SimPod\ClickHouseClient\Exception\ServerError;
 use SimPod\ClickHouseClient\Output\JsonEachRow;
 use SimPod\ClickHouseClient\Tests\TestCaseBase;
 
 use function iterator_to_array;
 use function str_repeat;
+use function str_replace;
 
 #[CoversClass(JsonEachRow::class)]
 final class JsonEachRowTest extends TestCaseBase
@@ -61,4 +66,36 @@ public function testStreamLineCanSpanMultipleReads(): void
             iterator_to_array($format->data, preserve_keys: false),
         );
     }
+
+    #[DataProvider('provideStreamedExceptionContents')]
+    public function testStreamedExceptionFrameThrowsServerError(string|StreamInterface $contents): void
+    {
+        $format = new JsonEachRow($contents);
+
+        $this->expectException(ServerError::class);
+        $this->expectExceptionCode(60);
+        $this->expectExceptionMessage('UNKNOWN_TABLE');
+
+        iterator_to_array($format->data, preserve_keys: false);
+    }
+
+    /**
+     * @return iterable<array{string|StreamInterface}>
+     *
+     * @throws InvalidArgumentException
+     */
+    public static function provideStreamedExceptionContents(): iterable
+    {
+        $contents = <<<'JSON'
+{"number":"0"}
+__exception__
+abcdefghijklmnop
+Code: 60. DB::Exception: Table default.missing does not exist. (UNKNOWN_TABLE)
+JSON;
+
+        yield 'string' => [$contents];
+        yield 'stream' => [Utils::streamFor($contents)];
+        yield 'CRLF string' => [str_replace("\n", "\r\n", $contents)];
+        yield 'CRLF stream' => [Utils::streamFor(str_replace("\n", "\r\n", $contents))];
+    }
 }