fix: use external parser

Author: astandrik

package-lock.json

@@ -26,6 +26,7 @@
     "@gravity-ui/uikit": "^6.40.0",
     "@gravity-ui/websql-autocomplete": "^13.7.0",
     "@hookform/resolvers": "^3.10.0",
+    "@mjackson/multipart-parser": "^0.8.2",
     "@reduxjs/toolkit": "^2.5.0",
     "@tanstack/react-table": "^8.20.6",
     "@ydb-platform/monaco-ghost": "^0.6.1",

package.json

@@ -1,3 +1,6 @@
+// eslint-disable-next-line import/no-extraneous-dependencies
+import {parseMultipart} from '@mjackson/multipart-parser';
+
 import {
     isQueryResponseChunk,
     isSessionChunk,
@@ -12,7 +15,6 @@ import type {
     StreamingChunk,
 } from '../../types/store/streaming';
 import {BINARY_DATA_IN_PLAIN_TEXT_DISPLAY} from '../../utils/constants';
-import {MultipartStreamParser} from '../parsers/parseMultipartStream';
 import {settingsManager} from '../settings';
 
 import {BaseYdbAPI} from './base';
@@ -47,7 +49,7 @@ export class StreamingAPI extends BaseYdbAPI {
 
         let traceId: string | undefined = '';
 
-        const streamParser = new MultipartStreamParser((chunk: StreamingChunk) => {
+        const handleChunk = (chunk: StreamingChunk) => {
             if (isSessionChunk(chunk)) {
                 const sessionChunk = chunk;
                 sessionChunk.meta.trace_id = traceId;
@@ -57,7 +59,7 @@ export class StreamingAPI extends BaseYdbAPI {
             } else if (isQueryResponseChunk(chunk)) {
                 options.onQueryResponseChunk(chunk);
             }
-        });
+        };
 
         const queryParams = new URLSearchParams();
         // Add only string/number params
@@ -91,19 +93,15 @@ export class StreamingAPI extends BaseYdbAPI {
             throw new Error('Empty response body');
         }
 
-        const reader = response.body.getReader();
         traceId = response.headers.get('traceresponse')?.split('-')[1];
 
-        try {
-            for (;;) {
-                const {done, value} = await reader.read();
-                if (done) {
-                    break;
-                }
-                streamParser.processBuffer(value);
+        await parseMultipart(response.body, {boundary: 'boundary'}, async (part) => {
+            try {
+                const chunk = JSON.parse(await part.text());
+                handleChunk(chunk);
+            } catch (e) {
+                throw new Error(`Error parsing chunk: ${e}`);
             }
-        } finally {
-            reader.releaseLock();
-        }
+        });
     }
 }

src/services/api/streaming.ts

@@ -382,7 +382,9 @@ export const queryApi = api.injectEndpoints({
                                 querySettings.tracingLevel && enableTracingLevel
                                     ? TracingLevelNumber[querySettings.tracingLevel]
                                     : undefined,
-                            limit_rows: 100000000,
+                            limit_rows: isNumeric(querySettings.limitRows)
+                                ? Number(querySettings.limitRows)
+                                : undefined,
                             transaction_mode:
                                 querySettings.transactionMode === 'implicit'
                                     ? undefined