[PECO-1541] Optimize UC Volume ingestion

Signed-off-by: Levko Kravets <levko.ne@gmail.com>

Author: kravets-levko

lib/DBSQLSession.ts

@@ -1,5 +1,6 @@
 import * as fs from 'fs';
 import * as path from 'path';
+import { pipeline } from 'stream';
 import { stringify, NIL, parse } from 'uuid';
 import fetch, { HeadersInit } from 'node-fetch';
 import {
@@ -271,8 +272,23 @@ export default class DBSQLSession implements IDBSQLSession {
     if (!response.ok) {
       throw new StagingError(`HTTP error ${response.status} ${response.statusText}`);
     }
-    const buffer = await response.arrayBuffer();
-    fs.writeFileSync(localFile, Buffer.from(buffer));
+
+    return new Promise((resolve, reject) => {
+      try {
+        const fileStream = fs.createWriteStream(localFile);
+        // `pipeline` will do all the dirty job for us, including error handling and closing all the streams properly
+        // Also, we use callback-style `pipeline` because Promise-style one is not available in Node 14
+        pipeline(response.body, fileStream, (error) => {
+          if (error) {
+            reject(error);
+          } else {
+            resolve();
+          }
+        });
+      } catch (error) {
+        reject(error);
+      }
+    });
   }
 
   private async handleStagingRemove(presignedUrl: string, headers: HeadersInit): Promise<void> {
@@ -301,8 +317,19 @@ export default class DBSQLSession implements IDBSQLSession {
     const connectionProvider = await this.context.getConnectionProvider();
     const agent = await connectionProvider.getAgent();
 
-    const data = fs.readFileSync(localFile);
-    const response = await fetch(presignedUrl, { method: 'PUT', headers, agent, body: data });
+    const fileStream = fs.createReadStream(localFile);
+    const fileInfo = fs.statSync(localFile, { bigint: true });
+
+    const response = await fetch(presignedUrl, {
+      method: 'PUT',
+      headers: {
+        ...headers,
+        // This header is required by server
+        'Content-Length': fileInfo.size.toString(),
+      },
+      agent,
+      body: fileStream,
+    });
     if (!response.ok) {
       throw new StagingError(`HTTP error ${response.status} ${response.statusText}`);
     }