Fix Upload Items (#682)

* Fix Upload Items

* Remove debugger

* Switch to bulk APIs

* Address TODO

Co-authored-by: Jordi Bunster <jbunster@microsoft.com>

src/Common/CosmosClient.ts

@@ -75,7 +75,10 @@ export async function getTokenFromAuthService(verb: string, resourceType: string
   }
 }
 
+let _client: Cosmos.CosmosClient;
+
 export function client(): Cosmos.CosmosClient {
+  if (_client) return _client;
   const options: Cosmos.CosmosClientOptions = {
     endpoint: endpoint() || "https://cosmos.azure.com", // CosmosClient gets upset if we pass a bad URL. This should never actually get called
     key: userContext.masterKey,
@@ -89,5 +92,6 @@ export function client(): Cosmos.CosmosClient {
   if (configContext.PROXY_PATH !== undefined) {
     (options as any).plugins = [{ on: "request", plugin: requestPlugin }];
   }
-  return new Cosmos.CosmosClient(options);
+  _client = new Cosmos.CosmosClient(options);
+  return _client;
 }

src/Common/dataAccess/bulkCreateDocument.ts

@@ -0,0 +1,36 @@
+import { JSONObject, OperationResponse } from "@azure/cosmos";
+import { CollectionBase } from "../../Contracts/ViewModels";
+import { logConsoleInfo, logConsoleProgress } from "../../Utils/NotificationConsoleUtils";
+import { client } from "../CosmosClient";
+import { handleError } from "../ErrorHandlingUtils";
+
+export const bulkCreateDocument = async (
+  collection: CollectionBase,
+  documents: JSONObject[]
+): Promise<OperationResponse[]> => {
+  const clearMessage = logConsoleProgress(
+    `Executing ${documents.length} bulk operations for container ${collection.id()}`
+  );
+
+  try {
+    const response = await client()
+      .database(collection.databaseId)
+      .container(collection.id())
+      .items.bulk(
+        documents.map((doc) => ({ operationType: "Create", resourceBody: doc })),
+        { continueOnError: true }
+      );
+
+    const successCount = response.filter((r) => r.statusCode === 201).length;
+
+    logConsoleInfo(
+      `${documents.length} operations completed for container ${collection.id()}. ${successCount} operations succeeded`
+    );
+    return response;
+  } catch (error) {
+    handleError(error, "BulkCreateDocument", `Error bulk creating items for container ${collection.id()}`);
+    throw error;
+  } finally {
+    clearMessage();
+  }
+};

src/Contracts/ViewModels.ts

@@ -15,7 +15,6 @@ import StoredProcedure from "../Explorer/Tree/StoredProcedure";
 import Trigger from "../Explorer/Tree/Trigger";
 import UserDefinedFunction from "../Explorer/Tree/UserDefinedFunction";
 import { SelfServeType } from "../SelfServe/SelfServeUtils";
-import { UploadDetails } from "../workers/upload/definitions";
 import * as DataModels from "./DataModels";
 import { SubscriptionType } from "./SubscriptionType";
 
@@ -23,6 +22,14 @@ export interface TokenProvider {
   getAuthHeader(): Promise<Headers>;
 }
 
+export interface UploadDetailsRecord {
+  fileName: string;
+  numSucceeded: number;
+  numFailed: number;
+  numThrottled: number;
+  errors: string[];
+}
+
 export interface QueryResultsMetadata {
   hasMoreResults: boolean;
   firstItemIndex: number;
@@ -174,7 +181,7 @@ export interface Collection extends CollectionBase {
 
   onDragOver(source: Collection, event: { originalEvent: DragEvent }): void;
   onDrop(source: Collection, event: { originalEvent: DragEvent }): void;
-  uploadFiles(fileList: FileList): Promise<UploadDetails>;
+  uploadFiles(fileList: FileList): Promise<{ data: UploadDetailsRecord[] }>;
 
   getLabel(): string;
   getPendingThroughputSplitNotification(): Promise<DataModels.Notification>;

src/Definitions/worker.d.ts

@@ -1,7 +0,0 @@
-declare module "worker-loader!*" {
-  class WebpackWorker extends Worker {
-    constructor();
-  }
-
-  export default WebpackWorker;
-}

src/Explorer/Panes/UploadItemsPane/index.tsx

@@ -1,9 +1,9 @@
 import { DetailsList, DetailsListLayoutMode, IColumn, SelectionMode } from "office-ui-fabric-react";
 import React, { ChangeEvent, FunctionComponent, useState } from "react";
 import { Upload } from "../../../Common/Upload";
+import { UploadDetailsRecord } from "../../../Contracts/ViewModels";
 import { userContext } from "../../../UserContext";
 import { logConsoleError } from "../../../Utils/NotificationConsoleUtils";
-import { UploadDetails, UploadDetailsRecord } from "../../../workers/upload/definitions";
 import Explorer from "../../Explorer";
 import { getErrorMessage } from "../../Tables/Utilities";
 import { GenericRightPaneComponent, GenericRightPaneProps } from "../GenericRightPaneComponent";
@@ -13,12 +13,6 @@ export interface UploadItemsPaneProps {
   closePanel: () => void;
 }
 
-interface IUploadFileData {
-  numSucceeded: number;
-  numFailed: number;
-  fileName: string;
-}
-
 const getTitle = (): string => {
   if (userContext.apiType === "Cassandra" || userContext.apiType === "Tables") {
     return "Upload Tables";
@@ -54,7 +48,7 @@ export const UploadItemsPane: FunctionComponent<UploadItemsPaneProps> = ({
     selectedCollection
       ?.uploadFiles(files)
       .then(
-        (uploadDetails: UploadDetails) => {
+        (uploadDetails) => {
           setUploadFileData(uploadDetails.data);
           setFiles(undefined);
         },
@@ -84,6 +78,7 @@ export const UploadItemsPane: FunctionComponent<UploadItemsPaneProps> = ({
     onClose: closePanel,
     onSubmit,
   };
+
   const columns: IColumn[] = [
     {
       key: "fileName",
@@ -105,12 +100,12 @@ export const UploadItemsPane: FunctionComponent<UploadItemsPaneProps> = ({
     },
   ];
 
-  const _renderItemColumn = (item: IUploadFileData, index: number, column: IColumn) => {
+  const _renderItemColumn = (item: UploadDetailsRecord, index: number, column: IColumn) => {
     switch (column.key) {
       case "status":
-        return <span>{item.numSucceeded + " items created, " + item.numFailed + " errors"}</span>;
+        return `${item.numSucceeded} created, ${item.numThrottled} throttled, ${item.numFailed} errors`;
       default:
-        return <span>{item.fileName}</span>;
+        return item.fileName;
     }
   };
 

src/Explorer/Tree/Collection.ts

@@ -1,9 +1,8 @@
 import { Resource, StoredProcedureDefinition, TriggerDefinition, UserDefinedFunctionDefinition } from "@azure/cosmos";
 import * as ko from "knockout";
 import * as _ from "underscore";
-import UploadWorker from "worker-loader!../../workers/upload";
-import { AuthType } from "../../AuthType";
 import * as Constants from "../../Common/Constants";
+import { bulkCreateDocument } from "../../Common/dataAccess/bulkCreateDocument";
 import { createDocument } from "../../Common/dataAccess/createDocument";
 import { getCollectionUsageSizeInKB } from "../../Common/dataAccess/getCollectionDataUsageSize";
 import { readCollectionOffer } from "../../Common/dataAccess/readCollectionOffer";
@@ -13,16 +12,13 @@ import { readUserDefinedFunctions } from "../../Common/dataAccess/readUserDefine
 import { getErrorMessage, getErrorStack } from "../../Common/ErrorHandlingUtils";
 import * as Logger from "../../Common/Logger";
 import { fetchPortalNotifications } from "../../Common/PortalNotifications";
-import { configContext, Platform } from "../../ConfigContext";
 import * as DataModels from "../../Contracts/DataModels";
 import * as ViewModels from "../../Contracts/ViewModels";
+import { UploadDetailsRecord } from "../../Contracts/ViewModels";
 import { Action, ActionModifiers } from "../../Shared/Telemetry/TelemetryConstants";
 import * as TelemetryProcessor from "../../Shared/Telemetry/TelemetryProcessor";
 import { userContext } from "../../UserContext";
-import * as NotificationConsoleUtils from "../../Utils/NotificationConsoleUtils";
-import { StartUploadMessageParams, UploadDetails, UploadDetailsRecord } from "../../workers/upload/definitions";
 import Explorer from "../Explorer";
-import { ConsoleDataType } from "../Menus/NotificationConsole/NotificationConsoleComponent";
 import { CassandraAPIDataClient, CassandraTableKey, CassandraTableKeys } from "../Tables/TableDataClient";
 import ConflictsTab from "../Tabs/ConflictsTab";
 import DocumentsTab from "../Tabs/DocumentsTab";
@@ -958,73 +954,6 @@ export default class Collection implements ViewModels.Collection {
     this.uploadFiles(event.originalEvent.dataTransfer.files);
   }
 
-  public uploadFiles = (fileList: FileList): Promise<UploadDetails> => {
-    // TODO: right now web worker is not working with AAD flow. Use main thread for upload for now until we have backend upload capability
-    if (configContext.platform === Platform.Hosted && userContext.authType === AuthType.AAD) {
-      return this._uploadFilesCors(fileList);
-    }
-    const documentUploader: Worker = new UploadWorker();
-    let inProgressNotificationId: string = "";
-
-    if (!fileList || fileList.length === 0) {
-      return Promise.reject("No files specified");
-    }
-
-    const onmessage = (resolve: (value: UploadDetails) => void, reject: (reason: any) => void, event: MessageEvent) => {
-      const numSuccessful: number = event.data.numUploadsSuccessful;
-      const numFailed: number = event.data.numUploadsFailed;
-      const runtimeError: string = event.data.runtimeError;
-      const uploadDetails: UploadDetails = event.data.uploadDetails;
-
-      NotificationConsoleUtils.clearInProgressMessageWithId(inProgressNotificationId);
-      documentUploader.terminate();
-      if (!!runtimeError) {
-        reject(runtimeError);
-      } else if (numSuccessful === 0) {
-        // all uploads failed
-        NotificationConsoleUtils.logConsoleError(`Failed to upload all documents to container ${this.id()}`);
-      } else if (numFailed > 0) {
-        NotificationConsoleUtils.logConsoleError(
-          `Failed to upload ${numFailed} of ${numSuccessful + numFailed} documents to container ${this.id()}`
-        );
-      } else {
-        NotificationConsoleUtils.logConsoleInfo(
-          `Successfully uploaded all ${numSuccessful} documents to container ${this.id()}`
-        );
-      }
-      this._logUploadDetailsInConsole(uploadDetails);
-      resolve(uploadDetails);
-    };
-    function onerror(reject: (reason: any) => void, event: ErrorEvent) {
-      documentUploader.terminate();
-      reject(event.error);
-    }
-
-    const uploaderMessage: StartUploadMessageParams = {
-      files: fileList,
-      documentClientParams: {
-        databaseId: this.databaseId,
-        containerId: this.id(),
-        masterKey: userContext.masterKey,
-        endpoint: userContext.endpoint,
-        accessToken: userContext.accessToken,
-        platform: configContext.platform,
-        databaseAccount: userContext.databaseAccount,
-      },
-    };
-
-    return new Promise<UploadDetails>((resolve, reject) => {
-      documentUploader.onmessage = onmessage.bind(null, resolve, reject);
-      documentUploader.onerror = onerror.bind(null, reject);
-
-      documentUploader.postMessage(uploaderMessage);
-      inProgressNotificationId = NotificationConsoleUtils.logConsoleMessage(
-        ConsoleDataType.InProgress,
-        `Uploading and creating documents in container ${this.id()}`
-      );
-    });
-  };
-
   public async getPendingThroughputSplitNotification(): Promise<DataModels.Notification> {
     if (!this.container) {
       return undefined;
@@ -1060,13 +989,13 @@ export default class Collection implements ViewModels.Collection {
     }
   }
 
-  private async _uploadFilesCors(files: FileList): Promise<UploadDetails> {
-    const data = await Promise.all(Array.from(files).map((file) => this._uploadFile(file)));
+  public async uploadFiles(files: FileList): Promise<{ data: UploadDetailsRecord[] }> {
+    const data = await Promise.all(Array.from(files).map((file) => this.uploadFile(file)));
 
     return { data };
   }
 
-  private _uploadFile(file: File): Promise<UploadDetailsRecord> {
+  private uploadFile(file: File): Promise<UploadDetailsRecord> {
     const reader = new FileReader();
     const onload = (resolve: (value: UploadDetailsRecord) => void, evt: any): void => {
       const fileData: string = evt.target.result;
@@ -1077,6 +1006,7 @@ export default class Collection implements ViewModels.Collection {
       resolve({
         fileName: file.name,
         numSucceeded: 0,
+        numThrottled: 0,
         numFailed: 1,
         errors: [(evt as any).error.message],
       });
@@ -1094,21 +1024,32 @@ export default class Collection implements ViewModels.Collection {
       fileName: fileName,
       numSucceeded: 0,
       numFailed: 0,
+      numThrottled: 0,
       errors: [],
     };
 
     try {
-      const content = JSON.parse(documentContent);
-
-      if (Array.isArray(content)) {
-        await Promise.all(
-          content.map(async (documentContent) => {
-            await createDocument(this, documentContent);
-            record.numSucceeded++;
-          })
+      const parsedContent = JSON.parse(documentContent);
+      if (Array.isArray(parsedContent)) {
+        const chunkSize = 50; // 100 is the max # of bulk operations the SDK currently accepts but usually results in throttles on 400RU collections
+        const chunkedContent = Array.from({ length: Math.ceil(parsedContent.length / chunkSize) }, (_, index) =>
+          parsedContent.slice(index * chunkSize, index * chunkSize + chunkSize)
         );
+        for (const chunk of chunkedContent) {
+          const responses = await bulkCreateDocument(this, chunk);
+          for (const response of responses) {
+            if (response.statusCode === 201) {
+              record.numSucceeded++;
+            } else if (response.statusCode === 429) {
+              record.numThrottled++;
+            } else {
+              record.numFailed++;
+              record.errors = [...record.errors, `${response.statusCode} ${response.resourceBody}`];
+            }
+          }
+        }
       } else {
-        await createDocument(this, documentContent);
+        await createDocument(this, parsedContent);
         record.numSucceeded++;
       }
 
@@ -1120,40 +1061,6 @@ export default class Collection implements ViewModels.Collection {
     }
   }
 
-  private _logUploadDetailsInConsole(uploadDetails: UploadDetails): void {
-    const uploadDetailsRecords: UploadDetailsRecord[] = uploadDetails.data;
-    const numFiles: number = uploadDetailsRecords.length;
-    const stackTraceLimit: number = 100;
-    let stackTraceCount: number = 0;
-    let currentFileIndex = 0;
-    while (stackTraceCount < stackTraceLimit && currentFileIndex < numFiles) {
-      const errors: string[] = uploadDetailsRecords[currentFileIndex].errors;
-      for (let i = 0; i < errors.length; i++) {
-        if (stackTraceCount >= stackTraceLimit) {
-          break;
-        }
-        NotificationConsoleUtils.logConsoleMessage(
-          ConsoleDataType.Error,
-          `Document creation error for container ${this.id()} - file ${
-            uploadDetailsRecords[currentFileIndex].fileName
-          }: ${errors[i]}`
-        );
-        stackTraceCount++;
-      }
-      currentFileIndex++;
-    }
-
-    uploadDetailsRecords.forEach((record: UploadDetailsRecord) => {
-      const consoleDataType: ConsoleDataType = record.numFailed > 0 ? ConsoleDataType.Error : ConsoleDataType.Info;
-      NotificationConsoleUtils.logConsoleMessage(
-        consoleDataType,
-        `Item creation summary for container ${this.id()} - file ${record.fileName}: ${
-          record.numSucceeded
-        } items created, ${record.numFailed} errors`
-      );
-    });
-  }
-
   /**
    * Top-level method that will open the correct tab type depending on account API
    */

src/workers/upload/definitions.ts

@@ -1,28 +0,0 @@
-import { DatabaseAccount } from "../../Contracts/DataModels";
-import { Platform } from "../../ConfigContext";
-
-export interface StartUploadMessageParams {
-  files: FileList;
-  documentClientParams: DocumentClientParams;
-}
-
-export interface DocumentClientParams {
-  databaseId: string;
-  containerId: string;
-  masterKey: string;
-  endpoint: string;
-  accessToken: string;
-  platform: Platform;
-  databaseAccount: DatabaseAccount;
-}
-
-export interface UploadDetailsRecord {
-  fileName: string;
-  numSucceeded: number;
-  numFailed: number;
-  errors: string[];
-}
-
-export interface UploadDetails {
-  data: UploadDetailsRecord[];
-}

src/workers/upload/index.ts

@@ -1,145 +0,0 @@
-import { DocumentClientParams, UploadDetailsRecord, UploadDetails } from "./definitions";
-import { client } from "../../Common/CosmosClient";
-import { updateConfigContext } from "../../ConfigContext";
-import { updateUserContext } from "../../UserContext";
-import { getErrorMessage } from "../../Common/ErrorHandlingUtils";
-
-let numUploadsSuccessful = 0;
-let numUploadsFailed = 0;
-let numDocuments = 0;
-let numFiles = 0;
-let numFilesProcessed = 0;
-let fileUploadDetails: { [key: string]: UploadDetailsRecord } = {};
-let databaseId: string;
-let containerId: string;
-
-onerror = (event: ProgressEvent) => {
-  postMessage(
-    {
-      numUploadsSuccessful: numUploadsSuccessful,
-      numUploadsFailed: numUploadsFailed,
-      uploadDetails: transformDetailsMap(fileUploadDetails),
-      // TODO: Typescript complains about event.error below
-      runtimeError: (event as any).error.message,
-    },
-    undefined
-  );
-};
-
-onmessage = (event: MessageEvent) => {
-  const files: FileList = event.data.files;
-  const clientParams: DocumentClientParams = event.data.documentClientParams;
-  containerId = clientParams.containerId;
-  databaseId = clientParams.databaseId;
-  updateUserContext({
-    masterKey: clientParams.masterKey,
-    endpoint: clientParams.endpoint,
-    accessToken: clientParams.accessToken,
-    databaseAccount: clientParams.databaseAccount,
-  });
-  updateConfigContext({
-    platform: clientParams.platform,
-  });
-  if (!!files && files.length > 0) {
-    numFiles = files.length;
-    for (let i = 0; i < numFiles; i++) {
-      fileUploadDetails[files[i].name] = {
-        fileName: files[i].name,
-        numSucceeded: 0,
-        numFailed: 0,
-        errors: [],
-      };
-      uploadFile(files[i]);
-    }
-  } else {
-    postMessage(
-      {
-        runtimeError: "No files specified",
-      },
-      undefined
-    );
-  }
-};
-
-function uploadFile(file: File): void {
-  const reader = new FileReader();
-  reader.onload = (evt: any): void => {
-    numFilesProcessed++;
-    const fileData: string = evt.target.result;
-    createDocumentsFromFile(file.name, fileData);
-  };
-
-  reader.onerror = (evt: ProgressEvent): void => {
-    numFilesProcessed++;
-    // TODO: Typescript complains about event.error below
-    recordUploadDetailErrorForFile(file.name, (evt as any).error.message);
-    transmitResultIfUploadComplete();
-  };
-
-  reader.readAsText(file);
-}
-
-function createDocumentsFromFile(fileName: string, documentContent: string): void {
-  try {
-    const content = JSON.parse(documentContent);
-    const triggerCreateDocument: (documentContent: any) => void = (documentContent: any) => {
-      client()
-        .database(databaseId)
-        .container(containerId)
-        .items.create(documentContent)
-        .then((savedDoc) => {
-          fileUploadDetails[fileName].numSucceeded++;
-          numUploadsSuccessful++;
-        })
-        .catch((error) => {
-          console.error(error);
-          recordUploadDetailErrorForFile(fileName, getErrorMessage(error));
-          numUploadsFailed++;
-        })
-        .finally(() => {
-          transmitResultIfUploadComplete();
-        });
-    };
-
-    if (Array.isArray(content)) {
-      numDocuments = numDocuments + content.length;
-      for (let i = 0; i < content.length; i++) {
-        triggerCreateDocument(content[i]);
-      }
-    } else {
-      numDocuments = numDocuments + 1;
-      triggerCreateDocument(content);
-    }
-  } catch (e) {
-    console.log(e);
-    recordUploadDetailErrorForFile(fileName, e.message);
-    transmitResultIfUploadComplete();
-  }
-}
-
-function transmitResultIfUploadComplete(): void {
-  if (numFilesProcessed === numFiles && numUploadsFailed + numUploadsSuccessful === numDocuments) {
-    postMessage(
-      {
-        numUploadsSuccessful: numUploadsSuccessful,
-        numUploadsFailed: numUploadsFailed,
-        uploadDetails: transformDetailsMap(fileUploadDetails),
-      },
-      undefined
-    );
-  }
-}
-
-function recordUploadDetailErrorForFile(fileName: string, error: any): void {
-  fileUploadDetails[fileName].errors.push(error);
-  fileUploadDetails[fileName].numFailed++;
-}
-
-function transformDetailsMap(map: any): UploadDetails {
-  let transformedResult: UploadDetailsRecord[] = [];
-  Object.keys(map).forEach((key: string) => {
-    transformedResult.push(map[key]);
-  });
-
-  return { data: transformedResult };
-}