From 640b541059b9df84e8e2402113858d7f130c3af1 Mon Sep 17 00:00:00 2001 From: Simon Podlipsky Date: Fri, 10 Jul 2026 13:37:26 +0300 Subject: [PATCH 1/3] fix(output): throw server error for streamed exceptions --- src/Output/JsonEachRow.php | 24 ++++++++++++++++++++++-- tests/Output/JsonEachRowTest.php | 29 +++++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 2 deletions(-) diff --git a/src/Output/JsonEachRow.php b/src/Output/JsonEachRow.php index 36403d5b..0c4ae4c1 100644 --- a/src/Output/JsonEachRow.php +++ b/src/Output/JsonEachRow.php @@ -7,8 +7,10 @@ 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 strpos; use function substr; @@ -24,7 +26,7 @@ /** @phpstan-var Generator */ public Generator $data; - /** @throws JsonException */ + /** @throws JsonException|ServerError */ public function __construct(string|StreamInterface $contentsJson) { $lines = $contentsJson instanceof StreamInterface @@ -40,20 +42,38 @@ public function __construct(string|StreamInterface $contentsJson) * * @return Generator * - * @throws JsonException + * @throws JsonException|ServerError */ private static function decodeLines(iterable $lines): Generator { + $streamedExceptionLines = []; + foreach ($lines as $line) { + 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 */ diff --git a/tests/Output/JsonEachRowTest.php b/tests/Output/JsonEachRowTest.php index d9af256c..1354f6d6 100644 --- a/tests/Output/JsonEachRowTest.php +++ b/tests/Output/JsonEachRowTest.php @@ -6,6 +6,9 @@ use GuzzleHttp\Psr7\Utils; 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; @@ -61,4 +64,30 @@ 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 */ + 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)]; + } } From 8ee28f5f74d66c3236aa65a0f5970897321d28af Mon Sep 17 00:00:00 2001 From: Simon Podlipsky Date: Fri, 10 Jul 2026 13:43:56 +0300 Subject: [PATCH 2/3] fix(output): document streamed exception throws --- src/Format/JsonEachRow.php | 3 ++- tests/Output/JsonEachRowTest.php | 7 ++++++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/src/Format/JsonEachRow.php b/src/Format/JsonEachRow.php index 13444419..a18fa779 100644 --- a/src/Format/JsonEachRow.php +++ b/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 $output */ diff --git a/tests/Output/JsonEachRowTest.php b/tests/Output/JsonEachRowTest.php index 1354f6d6..96aeef6d 100644 --- a/tests/Output/JsonEachRowTest.php +++ b/tests/Output/JsonEachRowTest.php @@ -5,6 +5,7 @@ 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; @@ -77,7 +78,11 @@ public function testStreamedExceptionFrameThrowsServerError(string|StreamInterfa iterator_to_array($format->data, preserve_keys: false); } - /** @return iterable */ + /** + * @return iterable + * + * @throws InvalidArgumentException + */ public static function provideStreamedExceptionContents(): iterable { $contents = <<<'JSON' From e65943c7160f227f16ba9f65b17f34247d2f997c Mon Sep 17 00:00:00 2001 From: Simon Podlipsky Date: Sat, 11 Jul 2026 08:38:49 +0300 Subject: [PATCH 3/3] fix(output): normalize JSONEachRow line endings --- src/Output/JsonEachRow.php | 3 +++ tests/Output/JsonEachRowTest.php | 3 +++ 2 files changed, 6 insertions(+) diff --git a/src/Output/JsonEachRow.php b/src/Output/JsonEachRow.php index 0c4ae4c1..1c1d20e9 100644 --- a/src/Output/JsonEachRow.php +++ b/src/Output/JsonEachRow.php @@ -12,6 +12,7 @@ use function explode; use function implode; use function json_decode; +use function rtrim; use function strpos; use function substr; @@ -49,6 +50,8 @@ private static function decodeLines(iterable $lines): Generator $streamedExceptionLines = []; foreach ($lines as $line) { + $line = rtrim($line, "\r"); + if ($streamedExceptionLines !== []) { $streamedExceptionLines[] = $line; diff --git a/tests/Output/JsonEachRowTest.php b/tests/Output/JsonEachRowTest.php index 96aeef6d..8a4c52ee 100644 --- a/tests/Output/JsonEachRowTest.php +++ b/tests/Output/JsonEachRowTest.php @@ -15,6 +15,7 @@ use function iterator_to_array; use function str_repeat; +use function str_replace; #[CoversClass(JsonEachRow::class)] final class JsonEachRowTest extends TestCaseBase @@ -94,5 +95,7 @@ public static function provideStreamedExceptionContents(): iterable 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))]; } }