feat: skip completed dataset uploads

Author: carloea2

file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala

@@ -202,6 +202,10 @@ object DatasetResource {
       sizeBytes: Option[Long] // Size of the changed file (None for directories)
   )
 
+  case class ExistingUploadFile(path: String, sizeBytes: Long)
+
+  case class ExistingUploadFilesRequest(files: List[ExistingUploadFile])
+
   case class DatasetDescriptionModification(did: Integer, description: String)
 
   case class DatasetNameModification(did: Integer, name: String)
@@ -1030,6 +1034,55 @@ class DatasetResource extends LazyLogging {
     }
   }
 
+  @POST
+  @RolesAllowed(Array("REGULAR", "ADMIN"))
+  @Path("/{did}/existing-upload-files")
+  @Consumes(Array(MediaType.APPLICATION_JSON))
+  def findExistingUploadFiles(
+      @PathParam("did") did: Integer,
+      request: ExistingUploadFilesRequest,
+      @Auth user: SessionUser
+  ): Response = {
+    val uid = user.getUid
+    withTransaction(context) { ctx =>
+      if (!userHasWriteAccess(ctx, did, uid)) {
+        throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_DATASET_MESSAGE)
+      }
+
+      val requested = Option(request)
+        .map(_.files)
+        .getOrElse(List.empty)
+        .map { file =>
+          val path = validateAndNormalizeFilePathOrThrow(file.path)
+          if (file.sizeBytes < 0L) throw new BadRequestException("sizeBytes must be >= 0")
+          path -> file.sizeBytes
+        }
+        .toMap
+
+      val dataset = getDatasetByID(ctx, did)
+      val committed = getLatestDatasetVersion(ctx, did)
+        .map(v =>
+          LakeFSStorageClient
+            .retrieveObjectsOfVersion(dataset.getRepositoryName, v.getVersionHash)
+            .map(obj => obj.getPath -> obj.getSizeBytes.longValue())
+        )
+        .getOrElse(List.empty)
+
+      val staged = LakeFSStorageClient
+        .retrieveUncommittedObjects(dataset.getRepositoryName)
+        .filterNot(diff => Option(diff.getType).exists(_.getValue.equalsIgnoreCase("removed")))
+        .flatMap(diff => Option(diff.getSizeBytes).map(size => diff.getPath -> size.longValue()))
+
+      val existing = (committed ++ staged).toMap
+      val matches = requested
+        .collect { case (path, size) if existing.get(path).contains(size) => path }
+        .toList
+        .sorted
+
+      Response.ok(Map("filePaths" -> matches.asJava)).build()
+    }
+  }
+
   @PUT
   @RolesAllowed(Array("REGULAR", "ADMIN"))
   @Path("/{did}/diff")

file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala

@@ -404,6 +404,55 @@ class DatasetResourceSpec
     dashboardDataset.size should be >= 0L
   }
 
+  "findExistingUploadFiles" should "match committed and staged files by path and size" in {
+    val repoName = s"existing-upload-${System.nanoTime()}"
+    val dataset = new Dataset
+    dataset.setName(repoName)
+    dataset.setRepositoryName(repoName)
+    dataset.setDescription("existing upload checks")
+    dataset.setOwnerUid(ownerUser.getUid)
+    dataset.setIsPublic(true)
+    dataset.setIsDownloadable(true)
+    datasetDao.insert(dataset)
+    LakeFSStorageClient.initRepo(repoName)
+
+    val committed = "committed".getBytes(StandardCharsets.UTF_8)
+    LakeFSStorageClient.writeFileToRepo(
+      repoName,
+      "committed.csv",
+      new ByteArrayInputStream(committed)
+    )
+    val commit = LakeFSStorageClient.createCommit(repoName, "main", "commit existing file")
+    val version = new DatasetVersion()
+    version.setDid(dataset.getDid)
+    version.setCreatorUid(ownerUser.getUid)
+    version.setName("v1")
+    version.setVersionHash(commit.getId)
+    new DatasetVersionDao(getDSLContext.configuration()).insert(version)
+
+    val staged = "staged".getBytes(StandardCharsets.UTF_8)
+    LakeFSStorageClient.writeFileToRepo(repoName, "staged.csv", new ByteArrayInputStream(staged))
+
+    val resp = datasetResource.findExistingUploadFiles(
+      dataset.getDid,
+      DatasetResource.ExistingUploadFilesRequest(
+        List(
+          DatasetResource.ExistingUploadFile("committed.csv", committed.length),
+          DatasetResource.ExistingUploadFile("staged.csv", staged.length),
+          DatasetResource.ExistingUploadFile("wrong-size.csv", staged.length + 1),
+          DatasetResource.ExistingUploadFile("missing.csv", 1L)
+        )
+      ),
+      sessionUser
+    )
+
+    resp.getStatus shouldEqual 200
+    mapListOfStrings(entityAsScalaMap(resp)("filePaths")) should contain theSameElementsAs List(
+      "committed.csv",
+      "staged.csv"
+    )
+  }
+
   it should "surface a LakeFS 404 as NotFoundException when the dataset repo is missing" in {
     val dataset = new Dataset
     dataset.setName("get-ds-no-repo")

frontend/src/app/dashboard/component/user/files-uploader/conflicting-file-modal-content/conflicting-file-modal-content.component.html

@@ -29,7 +29,7 @@
       <div><b>Path:</b> {{ data.path }}</div>
       <div><b>Size:</b> {{ data.size }}</div>
 
-      <div class="hint">An upload session already exists for this path.</div>
+      <div class="hint">{{ data.hint || "An upload session already exists for this path." }}</div>
     </div>
   </body>
 </html>

frontend/src/app/dashboard/component/user/files-uploader/conflicting-file-modal-content/conflicting-file-modal-content.component.ts

@@ -24,6 +24,7 @@ export interface ConflictingFileModalData {
   fileName: string;
   path: string;
   size: string;
+  hint?: string;
 }
 
 @Component({

frontend/src/app/dashboard/component/user/files-uploader/files-uploader.component.spec.ts

@@ -0,0 +1,110 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { of } from "rxjs";
+import { NgxFileDropEntry } from "ngx-file-drop";
+import { NzModalService } from "ng-zorro-antd/modal";
+import { AdminSettingsService } from "../../../service/admin/settings/admin-settings.service";
+import { DatasetService } from "../../../service/user/dataset/dataset.service";
+import { NotificationService } from "../../../../common/service/notification/notification.service";
+import { FileUploadItem } from "../../../type/dashboard-file.interface";
+import { FilesUploaderComponent } from "./files-uploader.component";
+
+interface CapturedModal {
+  nzTitle: string;
+  nzData: {
+    path: string;
+    hint?: string;
+  };
+  nzFooter: Array<{
+    label: string;
+    onClick: () => void;
+  }>;
+}
+
+const waitUntil = async (condition: () => boolean): Promise<void> => {
+  for (let i = 0; i < 20; i++) {
+    if (condition()) return;
+    await new Promise(resolve => setTimeout(resolve, 0));
+  }
+  throw new Error("condition was not met");
+};
+
+const droppedFile = (relativePath: string, file: File): NgxFileDropEntry =>
+  ({
+    relativePath,
+    fileEntry: {
+      isFile: true,
+      file: (success: (file: File) => void): void => success(file),
+    },
+  }) as unknown as NgxFileDropEntry;
+
+describe("FilesUploaderComponent", () => {
+  let component: FilesUploaderComponent;
+  let modals: CapturedModal[];
+
+  beforeEach(() => {
+    modals = [];
+    const modal = {
+      create: vi.fn(config => {
+        modals.push(config as CapturedModal);
+        return { destroy: vi.fn() };
+      }),
+    } as unknown as NzModalService;
+    const adminSettingsService = {
+      getSetting: vi.fn().mockReturnValue(of("20")),
+    } as unknown as AdminSettingsService;
+    const datasetService = {
+      listMultipartUploads: vi.fn().mockReturnValue(of(["failed.csv"])),
+      findExistingUploadFiles: vi.fn().mockReturnValue(of(["done.csv"])),
+    } as unknown as DatasetService;
+
+    component = new FilesUploaderComponent(
+      { error: vi.fn() } as unknown as NotificationService,
+      adminSettingsService,
+      datasetService,
+      modal
+    );
+    component.ownerEmail = "owner@example.com";
+    component.datasetName = "dataset";
+    component.did = 7;
+  });
+
+  it("asks to resume failed multipart files and skip completed matching files in one retry batch", async () => {
+    const emitted = new Promise<FileUploadItem[]>(resolve => component.uploadedFiles.subscribe(resolve));
+
+    component.fileDropped([
+      droppedFile("failed.csv", new File(["half"], "failed.csv")),
+      droppedFile("done.csv", new File(["done"], "done.csv")),
+    ]);
+
+    await waitUntil(() => modals.length === 1);
+    expect(modals[0].nzTitle).toBe("Conflicting File");
+    expect(modals[0].nzData.path).toBe("failed.csv");
+    modals[0].nzFooter.find(button => button.label === "Resume")?.onClick();
+
+    await waitUntil(() => modals.length === 2);
+    expect(modals[1].nzTitle).toBe("Matching File Found");
+    expect(modals[1].nzData.path).toBe("done.csv");
+    expect(modals[1].nzData.hint).toContain("same path and size");
+    modals[1].nzFooter.find(button => button.label === "Skip")?.onClick();
+
+    expect((await emitted).map(item => item.name)).toEqual(["failed.csv"]);
+  });
+});

frontend/src/app/dashboard/component/user/files-uploader/files-uploader.component.ts

@@ -65,6 +65,7 @@ export class FilesUploaderComponent {
    */
   @Input() ownerEmail: string = "";
   @Input() datasetName: string = "";
+  @Input() did: number | undefined;
 
   @Output() uploadedFiles = new EventEmitter<FileUploadItem[]>();
 
@@ -149,6 +150,41 @@ export class FilesUploaderComponent {
     });
   }
 
+  private askUploadOrSkip(
+    item: FileUploadItem,
+    showForAll: boolean
+  ): Promise<"upload" | "uploadAll" | "skip" | "skipAll"> {
+    return new Promise(resolve => {
+      const fileName = item.name.split("/").pop() || item.name;
+      let ref: NzModalRef;
+      const button = (label: string, choice: "upload" | "uploadAll" | "skip" | "skipAll", type?: "primary") => ({
+        label,
+        type,
+        onClick: () => {
+          resolve(choice);
+          ref.destroy();
+        },
+      });
+      ref = this.modal.create<ConflictingFileModalContentComponent, ConflictingFileModalData>({
+        nzTitle: "Matching File Found",
+        nzMaskClosable: false,
+        nzClosable: false,
+        nzContent: ConflictingFileModalContentComponent,
+        nzData: {
+          fileName,
+          path: item.name,
+          size: formatSize(item.file.size),
+          hint: "A file with the same path and size exists in this dataset. Skip only if you expect it is the same file.",
+        },
+        nzFooter: [
+          ...(showForAll ? [button("Upload For All", "uploadAll"), button("Skip For All", "skipAll")] : []),
+          button("Upload", "upload"),
+          button("Skip", "skip", "primary"),
+        ],
+      });
+    });
+  }
+
   private async resolveConflicts(items: FileUploadItem[], activePaths: string[]): Promise<FileUploadItem[]> {
     const active = new Set(activePaths ?? []);
     const isConflict = (p: string) => active.has(p) || active.has(encodeURIComponent(p));
@@ -201,6 +237,27 @@ export class FilesUploaderComponent {
     return out;
   }
 
+  private async resolveExistingFiles(items: FileUploadItem[], existingPaths: string[]): Promise<FileUploadItem[]> {
+    const existing = new Set(existingPaths ?? []);
+    const showForAll = items.length > 1;
+    let mode: "ask" | "uploadAll" | "skipAll" = "ask";
+    const out: FileUploadItem[] = [];
+
+    for (const item of items) {
+      if (!existing.has(item.name)) {
+        out.push(item);
+      } else if (mode === "uploadAll") {
+        out.push(item);
+      } else if (mode === "ask") {
+        const choice = await this.askUploadOrSkip(item, showForAll);
+        if (choice === "upload" || choice === "uploadAll") out.push(item);
+        if (choice === "uploadAll" || choice === "skipAll") mode = choice;
+      }
+    }
+
+    return out;
+  }
+
   hideBanner(): void {
     this.fileUploadingFinished = false;
   }
@@ -254,20 +311,39 @@ export class FilesUploaderComponent {
       .then(async results => {
         const { ownerEmail, datasetName } = this.getOwnerAndName();
 
-        const activePathsPromise =
-          ownerEmail && datasetName
-            ? firstValueFrom(this.datasetService.listMultipartUploads(ownerEmail, datasetName)).catch(() => [])
-            : [];
-
-        const activePaths = await activePathsPromise;
         const successfulUploads = results
           .filter((r): r is PromiseFulfilledResult<FileUploadItem | null> => r.status === "fulfilled")
           .map(r_1 => r_1.value)
           .filter((item): item is FileUploadItem => item !== null);
-        const filteredUploads = await this.resolveConflicts(successfulUploads, activePaths);
-        if (filteredUploads.length > 0) {
-          const msg = `${filteredUploads.length} file${filteredUploads.length > 1 ? "s" : ""} selected successfully!`;
-          this.showFileUploadBanner("success", msg);
+
+        const activePathsPromise: Promise<string[]> =
+          ownerEmail && datasetName
+            ? firstValueFrom(this.datasetService.listMultipartUploads(ownerEmail, datasetName)).catch(() => [])
+            : Promise.resolve([]);
+        const existingPathsPromise: Promise<string[]> = this.did
+          ? firstValueFrom(
+              this.datasetService.findExistingUploadFiles(
+                this.did,
+                successfulUploads.map(item => ({ path: item.name, sizeBytes: item.file.size }))
+              )
+            ).catch(() => [])
+          : Promise.resolve([]);
+
+        const [activePaths, existingPaths] = await Promise.all([activePathsPromise, existingPathsPromise]);
+        const resumableUploads = await this.resolveConflicts(successfulUploads, activePaths);
+        const filteredUploads = await this.resolveExistingFiles(resumableUploads, existingPaths);
+        const skippedCount = resumableUploads.length - filteredUploads.length;
+        if (filteredUploads.length > 0 || skippedCount > 0) {
+          const messages = [];
+          if (filteredUploads.length > 0) {
+            messages.push(
+              `${filteredUploads.length} file${filteredUploads.length > 1 ? "s" : ""} selected successfully!`
+            );
+          }
+          if (skippedCount > 0) {
+            messages.push(`${skippedCount} matching file${skippedCount > 1 ? "s were" : " was"} skipped.`);
+          }
+          this.showFileUploadBanner(skippedCount > 0 ? "info" : "success", messages.join(" "));
         }
         const failedCount = results.length - successfulUploads.length;
         if (failedCount > 0) {

frontend/src/app/dashboard/component/user/user-dataset/user-dataset-explorer/dataset-detail.component.html

@@ -338,6 +338,7 @@ <h6 style="font-weight: lighter; font-size: 0.9em">Choose a Version:</h6>
           nzActive="true"
           nzHeader="Create New Version">
           <texera-user-files-uploader
+            [did]="did"
             [ownerEmail]="ownerEmail"
             [datasetName]="datasetName"
             (uploadedFiles)="onNewUploadFilesChanged($event)">