fix: FIR-36189 nest js script stuck when running a query with limit 58 and above (#115)

Author: stepansergeevitch

src/connection/base.ts

@@ -7,6 +7,7 @@ import {
 import { Statement } from "../statement";
 import { generateUserAgent } from "../common/util";
 import { ConnectionError, CompositeError } from "../common/errors";
+import JSONbig from "json-bigint";
 
 const defaultQuerySettings = {
  output_format: OutputFormat.COMPACT
@@ -206,14 +207,14 @@ export abstract class Connection {
 
  try {
  const response = await request.ready();
+ const text = await response.text();
  await this.processHeaders(response.headers);
- await this.throwErrorIfErrorBody(response);
- const statement = new Statement(this.context, {
+ await this.throwErrorIfErrorBody(text, response);
+ return new Statement(this.context, {
  query: formattedQuery,
- request,
+ text,
  executeQueryOptions
  });
- return statement;
  } catch (error) {
  // In case it was a set query, remove set parameter if query fails
  if (setKey.length > 0) {
@@ -225,14 +226,13 @@ export abstract class Connection {
  }
  }
 
- private async throwErrorIfErrorBody(response: Response) {
+ private async throwErrorIfErrorBody(text: string, response: Response) {
  // Hack, but looks like this is a limitation of the fetch API
  // In order to read the body here and elesewhere, we need to clone the response
  // since body can only be read once
- const clonedResponse = response.clone();
  let json;
  try {
- json = await clonedResponse.json();
+ json = JSONbig.parse(text);
  } catch (error) {
  // If we can't parse the JSON, we'll have to ignore it
  if (this.hasJsonContent(response)) {

src/statement/index.ts

@@ -13,47 +13,50 @@ import {
  normalizeResponse,
  getNormalizedStatistics
 } from "./normalizeResponse";
+import { CompositeError } from "../common/errors";
 
 export class Statement {
  private context: Context;
  private query: string;
  private executeQueryOptions: ExecuteQueryOptions;
 
- private request: { ready: () => Promise<any>; abort: () => void };
+ private text;
  private rowStream: RowStream;
 
  constructor(
  context: Context,
  {
  query,
- request,
+ text,
  executeQueryOptions
  }: {
  query: string;
- request: { ready: () => Promise<any>; abort: () => void };
+ text: string;
  executeQueryOptions: ExecuteQueryOptions;
  }
  ) {
  this.context = context;
- this.request = request;
+ this.text = text;
  this.query = query;
  this.executeQueryOptions = executeQueryOptions;
  this.rowStream = new RowStream();
  }
 
  private parseResponse(response: string) {
  const parsed = JSONbig.parse(response);
- const { data, meta, statistics } = parsed;
+ const { data, meta, statistics, errors = undefined } = parsed;
  return {
  data,
  meta,
- statistics
+ statistics,
+ errors
  };
  }
 
  private handleParseResponse(response: string) {
+ let errors, json;
  try {
- return this.parseResponse(response);
+ ({ errors, ...json } = this.parseResponse(response));
  } catch (error) {
  const isData = isDataQuery(this.query);
  if (isData || (response.length && !isData)) {
@@ -65,111 +68,37 @@ export class Statement {
  statistics: null
  };
  }
+ if (errors !== undefined) throw new CompositeError(errors);
+ return json;
  }
 
  async streamResult(options?: StreamOptions) {
- const response = await this.request.ready();
- const jsonStream = new JSONStream({
- emitter: this.rowStream,
- options,
- executeQueryOptions: this.executeQueryOptions
- });
-
- let resolveMetadata: (metadata: Meta[]) => void;
- let rejectMetadata: (reason?: any) => void;
-
- let resolveStatistics: (statistics: Statistics) => void;
- let rejectStatistics: (reason?: any) => void;
-
- const metadataPromise = new Promise<Meta[]>((resolve, reject) => {
- resolveMetadata = resolve;
- rejectMetadata = reject;
- });
-
- const statisticsPromise = new Promise<Statistics>((resolve, reject) => {
- resolveStatistics = resolve;
- rejectStatistics = reject;
- });
-
- let str = Buffer.alloc(0);
-
- this.rowStream.on("metadata", (metadata: Meta[]) => {
- resolveMetadata(metadata);
- });
-
- this.rowStream.on("statistics", (statistics: Statistics) => {
- resolveStatistics(statistics);
- });
-
- const errorHandler = (error: any) => {
- this.rowStream.emit("error", error);
- this.rowStream.push(null);
- rejectStatistics(error);
- rejectMetadata(error);
- };
-
- response.body.on("error", errorHandler);
-
- response.body.on("data", (chunk: Buffer) => {
- // content type should be application/json?
- // maybe in the future it will change
- const contentType = response.headers.get("content-type");
-
- if (chunk.lastIndexOf("\n") !== -1 && str) {
- // store in buffer anything after
- const newLinePosition = chunk.lastIndexOf("\n");
- const rest = chunk.slice(newLinePosition + 1);
-
- const lines = Buffer.concat([str, chunk.slice(0, newLinePosition)])
- .toString("utf8")
- .split("\n");
- try {
- for (const line of lines) {
- jsonStream.processLine(line);
- }
- } catch (error) {
- errorHandler(error);
- return;
- }
-
- const result = jsonStream.getResult(0);
- // for now only supports single statement sql
- for (const row of result.rows) {
- this.rowStream.push(row);
- }
+ // Streaming is not supported right now in Firebolt
+ // This is a placeholder for future implementation
+ const parsed = this.handleParseResponse(this.text);
+ const normalized = normalizeResponse(parsed, this.executeQueryOptions);
 
- result.rows = [];
- str = rest;
- }
- });
+ const { data, meta, statistics } = normalized;
 
- response.body.on("end", () => {
- try {
- const result = jsonStream.getResult(0);
- const statistics = getNormalizedStatistics(result.statistics);
- this.rowStream.emit("statistics", statistics);
- this.rowStream.push(null);
- } catch (error) {
- errorHandler(error);
- }
- });
+ for (const row of data) {
+ this.rowStream.push(row);
+ }
+ this.rowStream.push(null);
+ this.rowStream.end();
 
  return {
  data: this.rowStream,
- meta: metadataPromise,
- statistics: statisticsPromise
+ meta: Promise.resolve(meta),
+ statistics: Promise.resolve(statistics)
  };
  }
 
- async fetchResult() {
- const response = await this.request.ready();
- const text = await response.text();
-
- if (this.executeQueryOptions?.settings?.async_execution) {
- return JSONbig.parse(text);
- }
-
- const parsed = this.handleParseResponse(text);
+ async fetchResult(): Promise<{
+ data: any;
+ meta: Meta[];
+ statistics: Statistics | null;
+ }> {
+ const parsed = this.handleParseResponse(this.text);
  const normalized = normalizeResponse(parsed, this.executeQueryOptions);
 
  const { data, meta, statistics } = normalized;

test/integration/v1/index.test.ts

@@ -39,19 +39,6 @@ describe("integration test", () => {
  const row = data[0];
  expect(row).toMatchObject({ "?column?": 1 });
  });
- it("async query", async () => {
- const firebolt = Firebolt({
- apiEndpoint: process.env.FIREBOLT_API_ENDPOINT as string
- });
-
- const connection = await firebolt.connect(connectionParams);
- const statement = await connection.execute("SELECT 1", {
- settings: { async_execution: true },
- response: { normalizeData: false }
- });
- const result = await statement.fetchResult();
- expect(result.query_id).toBeTruthy();
- });
  it("returns Date type", async () => {
  const firebolt = Firebolt({
  apiEndpoint: process.env.FIREBOLT_API_ENDPOINT as string
@@ -179,7 +166,8 @@ describe("integration test", () => {
  await firebolt.testConnection(connectionParams);
  expect(true).toBeTruthy();
  });
- it("custom parser", async () => {
+ // Since streaming is currently disabled, custom parser is not supported
+ it.skip("custom parser", async () => {
  const firebolt = Firebolt({
  apiEndpoint: process.env.FIREBOLT_API_ENDPOINT as string
  });

test/integration/v2/index.test.ts

@@ -268,7 +268,8 @@ describe("integration test", () => {
  await firebolt.testConnection(connectionParams);
  expect(true).toBeTruthy();
  });
- it("custom parser", async () => {
+ // Since streaming is currently disabled, custom parser is not supported
+ it.skip("custom parser", async () => {
  const firebolt = Firebolt({
  apiEndpoint: process.env.FIREBOLT_API_ENDPOINT as string
  });

test/unit/connection.test.ts

@@ -114,7 +114,7 @@ describe("Connection", () => {
  resetServerHandlers(server);
  });
 
- it("throws an error when error body is present", async () => {
+ it("throws an error when error json is present", async () => {
  server.use(
  rest.post(`https://some_engine.com`, async (req, res, ctx) => {
  const body = await req.text();
@@ -166,7 +166,8 @@ describe("Connection", () => {
 INFO: SYNTAX_ERROR - Unexpected character at {"failingLine":42,"startOffset":120,"endOffset":135}`
  );
  });
- it("throws an error when error body is present", async () => {
+
+ it("throws an error when error json 2 is present", async () => {
  server.use(
  rest.post(`https://some_engine.com`, async (req, res, ctx) => {
  const body = await req.text();
@@ -524,4 +525,33 @@ INFO: SYNTAX_ERROR - Unexpected character at {"failingLine":42,"startOffset":120
  expect(searchParamsUsed.get("param")).toEqual("value");
  expect(searchParamsUsed2.get("param")).toEqual(null);
  });
+
+ it("handles large response bodies correctly", async () => {
+ const largeResponse = {
+ data: Array(1000).fill({ one: 1, two: 2, three: 3 })
+ };
+ server.use(
+ rest.post(`https://some_engine.com`, async (req, res, ctx) => {
+ return res(ctx.json(largeResponse));
+ })
+ );
+
+ const connectionParams: ConnectionOptions = {
+ auth: {
+ client_id: "dummy",
+ client_secret: "dummy"
+ },
+ database: "dummy",
+ engineName: "dummy",
+ account: "my_account"
+ };
+ const firebolt = Firebolt({
+ apiEndpoint
+ });
+
+ const connection = await firebolt.connect(connectionParams);
+ const statement = await connection.execute("SELECT 1");
+ const { data } = await statement.fetchResult();
+ expect(data.length).toEqual(1000);
+ }, 10000);
 });