fix(clickhouse): classify streamed limit failures and keep canonical log fields

Spread caller-supplied logFields before the canonical ones so a caller
cannot overwrite error, query, params, or queryId in a failure log.

queryFastStream classifies quota failures the same way the buffered query
paths do; a limit hit partway through a stream is still the caller asking
for too much.

The supplemental EXPLAIN queries carry the originating TSQL too.

Author: ericallam

internal-packages/clickhouse/src/client/client.ts

@@ -332,12 +332,12 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
 
         if (clickhouseError) {
           const errorLogFields = {
+            ...req.logFields,
             name: req.name,
             error: clickhouseError,
             query: req.query,
             params,
             queryId,
-            ...req.logFields,
           };
 
           if (isClickhouseQuotaError(clickhouseError)) {
@@ -623,13 +623,19 @@ export class ClickhouseClient implements ClickhouseReader, ClickhouseWriter {
 
         span.setAttributes({ "clickhouse.rows": rowCount });
       } catch (error) {
-        self.logger.error("Error streaming clickhouse", {
+        const errorLogFields = {
           name: req.name,
           error,
           query: req.query,
           params,
           queryId,
-        });
+        };
+
+        if (error instanceof Error && isClickhouseQuotaError(error)) {
+          self.logger.warn("Streamed query exceeded a ClickHouse limit", errorLogFields);
+        } else {
+          self.logger.error("Error streaming clickhouse", errorLogFields);
+        }
 
         if (error instanceof Error) {
           recordClickhouseError(span, error);

internal-packages/clickhouse/src/client/tsql.ts

@@ -242,6 +242,7 @@ export async function executeTSQL<TOut extends z.ZodSchema>(
             params: z.record(z.any()),
             schema: z.object({ explain: z.string() }),
             settings: options.clickhouseSettings,
+            logFields: { tsql: options.query },
           });
 
           const [additionalError, additionalResult] = await additionalQueryFn(params);