SNOW-1926267: Fix promise rejecting for file upload errors

lib/connection/result/row_stream.js

@@ -65,7 +65,7 @@ function RowStream(statement, context, options) {
     } else if (context.isFetchingResult) {
       // if we're still fetching the result, wait for the operation to complete
       context.on('statement-complete', init);
-    } else if (context.result || isStatementErrorFatal(context.resultError)) {
+    } else if (context.result || isStatementErrorFatal(context)) {
       // if we have a result or a fatal error, call init() in the next tick of
       // the event loop
       process.nextTick(init);
@@ -295,12 +295,16 @@ Util.inherits(RowStream, Readable);
 /**
  * Determines if a statement error is fatal.
  *
- * @param {Error} error
- *
  * @returns {Boolean}
+ * @param context
  */
-function isStatementErrorFatal(error) {
-  return Errors.isOperationFailedError(error) && error.sqlState;
+function isStatementErrorFatal(context) {
+  const error = context.resultError;
+  return (Errors.isOperationFailedError(error) && error.sqlState) || isFileUploadError(error, context.type);
+}
+
+function isFileUploadError(error, contextType) {
+  return error && contextType === 'FILE_PRE_EXEC';
 }
 
 /**

lib/connection/statement.js

@@ -813,26 +813,7 @@ function FileStatementPreExec(
    * @param {Object} body
    */
   context.onStatementRequestSucc = async function (body) {
-    context.fileMetadata = body;
-
-    const fta = new FileTransferAgent(context);
-    await fta.execute();
-
-    // build a result from the response
-    const result = fta.result();
-
-    // init result and meta
-    body.data.rowset = result.rowset;
-    body.data.returned = body.data.rowset.length;
-    body.data.rowtype = result.rowtype;
-    body.data.parameters = [];
-
-    context.result = new Result({
-      response: body,
-      statement: this,
-      services: context.services,
-      connectionConfig: context.connectionConfig
-    });
+    await executeFileTransferRequest(context, body, this);
   };
 
   /**
@@ -858,6 +839,36 @@ function FileStatementPreExec(
   sendRequestPreExec(context, context.onStatementRequestComp);
 }
 
+async function executeFileTransferRequest(context, body, statement, fileTransferAgent) {
+  context.fileMetadata = body;
+
+  const fta = fileTransferAgent ?? new FileTransferAgent(context);   await fta.execute();
+
+  try {
+    // build a result from the response
+    const result = fta.result();
+
+    // init result and meta
+    body.data = {
+      rowset: result.rowset,
+      returned: result.rowset.length,
+      rowtype: result.rowtype,
+      parameters: [],
+    };
+
+    context.result = new Result({
+      response: body,
+      statement: statement,
+      services: context.services,
+      connectionConfig: context.connectionConfig
+    });
+  } catch (error) {
+    context.resultError = error;
+  }
+}
+
+exports.executeFileTransferRequest = executeFileTransferRequest;
+
 Util.inherits(FileStatementPreExec, BaseStatement);
 
 /**

lib/file_transfer_agent/file_transfer_agent.js

@@ -213,6 +213,13 @@ function FileTransferAgent(context) {
 
       if (results) {
         for (const meta of results) {
+          if (meta['resultStatus'] === 'ERROR') {
+            errorDetails = meta['errorDetails'];
+            if (!errorDetails) {
+              errorDetails = `Unknown error during PUT of file: ${meta['srcFilePath']}`;
+            }
+            throw new Error(errorDetails);
+          }
           if (meta['srcCompressionType']) {
             srcCompressionType = meta['srcCompressionType']['name'];
           } else {
@@ -227,8 +234,8 @@ function FileTransferAgent(context) {
 
           errorDetails = meta['errorDetails'];
 
-          srcFileSize = meta['srcFileSize'].toString();
-          dstFileSize = meta['dstFileSize'].toString();
+          srcFileSize = meta['srcFileSize'];
+          dstFileSize = meta['dstFileSize'];
 
           rowset.push([
             meta['srcFileName'],
@@ -334,6 +341,9 @@ function FileTransferAgent(context) {
         continue;
       }
       results.push(result);
+      if (result['resultStatus'] === resultStatus.ERROR) {
+        break;
+      }
       index += 1;
       if (INJECT_WAIT_IN_PUT > 0) {
         await new Promise(resolve => setTimeout(resolve, INJECT_WAIT_IN_PUT));

test/unit/connection/statement_test.js

@@ -411,3 +411,20 @@ describe('Statement.fetchResult()', function () {
     it(testCase.name, createItCallback(testCase));
   }
 });
+
+it('Statement file transfer error', async function () {
+  const mockFta = {
+    execute: async function () {
+      return null;
+    },
+    result: function () {
+      throw new Error('some file transfer error');
+    }
+  };
+  const context = {};
+  const body = {
+    'data': {},
+  };
+  await Statement.executeFileTransferRequest(context, body, null, mockFta);
+  assert.strictEqual(context.resultError.message, 'some file transfer error');
+});