This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new b1c4fd29bd feat: skip completed files in resumable dataset uploads
(#5929)
b1c4fd29bd is described below
commit b1c4fd29bd2208abca1d5cc99729c75f850bd127
Author: carloea2 <[email protected]>
AuthorDate: Mon Jul 6 13:41:30 2026 -0700
feat: skip completed files in resumable dataset uploads (#5929)
### What changes were proposed in this PR?
<img width="784" height="471" alt="image"
src="https://github.com/user-attachments/assets/bb2a5b9c-a178-43a9-abf4-7c746bc9e0ba";
/>
This PR improves dataset upload retry behavior by distinguishing between
two cases that currently look similar to users:
| Case | Behavior |
| --- | --- |
| Active multipart upload session exists | Ask whether to resume or
restart the upload. |
| Matching file already exists in the dataset | Ask whether to upload
again or skip it. |
The backend adds a dataset endpoint that checks candidate upload files
against committed and staged dataset files by path and size. The
frontend uses that endpoint after checking active multipart sessions, so
failed uploads can still be resumed while completed matching files can
be skipped from the retry batch.
The user-facing copy intentionally says a file with the same path and
size exists, instead of implying byte-for-byte certainty.
### Any related issues, documentation, discussions?
Related to discussion #5744: Improve resumable upload: track completion
at the batch/session level.
### How was this PR tested?
Added backend and frontend tests covering:
| Test | Coverage |
| --- | --- |
| `DatasetResourceSpec` | Finds committed and staged files only when
path and size match. |
| `DatasetService` spec | Forces one multipart part upload to fail, then
verifies retry uploads only the missing part. |
| `FilesUploaderComponent` spec | Verifies a mixed retry batch asks
Resume/Restart for a failed multipart file and Upload/Skip for a
completed matching file. |
Commands run:
```powershell
sbt "FileService/testOnly
org.apache.texera.service.resource.DatasetResourceSpec -- -z
findExistingUploadFiles"
yarn ng run gui:test
--include=src/app/dashboard/service/user/dataset/dataset.service.spec.ts
yarn ng run gui:test
--include=src/app/dashboard/component/user/files-uploader/files-uploader.component.spec.ts
yarn eslint
src/app/dashboard/component/user/files-uploader/files-uploader.component.ts
src/app/dashboard/component/user/files-uploader/files-uploader.component.spec.ts
src/app/dashboard/component/user/files-uploader/conflicting-file-modal-content/conflicting-file-modal-content.component.ts
src/app/dashboard/service/user/dataset/dataset.service.ts
src/app/dashboard/service/user/dataset/dataset.service.spec.ts
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Me
---------
Co-authored-by: Xuan Gu <[email protected]>
---
.../texera/service/resource/DatasetResource.scala | 63 ++++++++
.../service/resource/DatasetResourceSpec.scala | 169 ++++++++++++++++++++
.../conflicting-file-modal-content.component.html | 2 +-
.../conflicting-file-modal-content.component.ts | 1 +
.../files-uploader.component.spec.ts | 173 +++++++++++++++++++++
.../files-uploader/files-uploader.component.ts | 96 ++++++++++--
.../dataset-detail.component.html | 1 +
.../service/user/dataset/dataset.service.spec.ts | 117 ++++++++++++++
.../service/user/dataset/dataset.service.ts | 11 ++
9 files changed, 622 insertions(+), 11 deletions(-)
diff --git
a/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala
b/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala
index efed479653..789f6373b9 100644
---
a/file-service/src/main/scala/org/apache/texera/service/resource/DatasetResource.scala
+++
b/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,65 @@ 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)
+ .flatMap(request => Option(request.files))
+ .getOrElse(List.empty)
+ .map { file =>
+ val originalPath = file.path
+ val path = validateAndNormalizeFilePathOrThrow(originalPath)
+ if (file.sizeBytes < 0L) throw new BadRequestException("sizeBytes
must be >= 0")
+ (path, originalPath, file.sizeBytes)
+ }
+
+ val dataset = getDatasetByID(ctx, did)
+ val committed = getLatestDatasetVersion(ctx, did)
+ .map { v =>
+ withLakeFSErrorHandling(
+ s"retrieving committed files of dataset '${dataset.getName}'"
+ ) {
+ LakeFSStorageClient
+ .retrieveObjectsOfVersion(dataset.getRepositoryName,
v.getVersionHash)
+ .map(obj => obj.getPath -> obj.getSizeBytes.longValue())
+ }
+ }
+ .getOrElse(List.empty)
+
+ val staged = withLakeFSErrorHandling(
+ s"retrieving staged files of dataset '${dataset.getName}'"
+ ) {
+
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, originalPath, size) if existing.get(path).contains(size)
=> originalPath
+ }
+ .toList
+ .distinct
+ .sorted
+
+ Response.ok(Map("filePaths" -> matches.asJava)).build()
+ }
+ }
+
@PUT
@RolesAllowed(Array("REGULAR", "ADMIN"))
@Path("/{did}/diff")
diff --git
a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
index 1730d12a0a..e4e6165241 100644
---
a/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
+++
b/file-service/src/test/scala/org/apache/texera/service/resource/DatasetResourceSpec.scala
@@ -404,6 +404,175 @@ 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 "return the original request path when matching a normalized path"
in {
+ val repoName = s"existing-upload-normalized-${System.nanoTime()}"
+ val dataset = new Dataset
+ dataset.setName(repoName)
+ dataset.setRepositoryName(repoName)
+ dataset.setDescription("existing upload normalized path 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
normalized 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 requestPath = "folder/../committed.csv"
+ val resp = datasetResource.findExistingUploadFiles(
+ dataset.getDid,
+ DatasetResource.ExistingUploadFilesRequest(
+ List(DatasetResource.ExistingUploadFile(requestPath, committed.length))
+ ),
+ sessionUser
+ )
+
+ resp.getStatus shouldEqual 200
+ mapListOfStrings(entityAsScalaMap(resp)("filePaths")) shouldEqual
List(requestPath)
+ }
+
+ it should "treat a missing files list as empty" in {
+ val repoName = s"existing-upload-empty-${System.nanoTime()}"
+ val dataset = new Dataset
+ dataset.setName(repoName)
+ dataset.setRepositoryName(repoName)
+ dataset.setDescription("existing upload empty request check")
+ dataset.setOwnerUid(ownerUser.getUid)
+ dataset.setIsPublic(true)
+ dataset.setIsDownloadable(true)
+ datasetDao.insert(dataset)
+ LakeFSStorageClient.initRepo(repoName)
+
+ val resp = datasetResource.findExistingUploadFiles(
+ dataset.getDid,
+ DatasetResource.ExistingUploadFilesRequest(null),
+ sessionUser
+ )
+
+ resp.getStatus shouldEqual 200
+ mapListOfStrings(entityAsScalaMap(resp)("filePaths")) shouldBe empty
+ }
+
+ it should "reject negative file sizes" in {
+ val ex = intercept[BadRequestException] {
+ datasetResource.findExistingUploadFiles(
+ baseDataset.getDid,
+ DatasetResource.ExistingUploadFilesRequest(
+ List(DatasetResource.ExistingUploadFile("bad-size.csv", -1L))
+ ),
+ sessionUser
+ )
+ }
+
+ ex.getMessage should include("sizeBytes")
+ }
+
+ it should "reject users without write access" in {
+ val ex = intercept[ForbiddenException] {
+ datasetResource.findExistingUploadFiles(
+ multipartDataset.getDid,
+ DatasetResource.ExistingUploadFilesRequest(
+ List(DatasetResource.ExistingUploadFile("private.csv", 1L))
+ ),
+ multipartNoWriteSessionUser
+ )
+ }
+
+ assertStatus(ex, 403)
+ }
+
+ it should "surface a LakeFS 404 as NotFoundException when checking a missing
repo" in {
+ val repoName = s"existing-upload-missing-repo-${System.nanoTime()}"
+ val dataset = new Dataset
+ dataset.setName(repoName)
+ dataset.setRepositoryName(repoName)
+ dataset.setDescription("existing upload missing repo check")
+ dataset.setOwnerUid(ownerUser.getUid)
+ dataset.setIsPublic(true)
+ dataset.setIsDownloadable(true)
+ datasetDao.insert(dataset)
+
+ val version = new DatasetVersion()
+ version.setDid(dataset.getDid)
+ version.setCreatorUid(ownerUser.getUid)
+ version.setName("v1")
+ version.setVersionHash("missing-version")
+ new DatasetVersionDao(getDSLContext.configuration()).insert(version)
+
+ val ex = intercept[NotFoundException] {
+ datasetResource.findExistingUploadFiles(
+ dataset.getDid,
+ DatasetResource.ExistingUploadFilesRequest(
+ List(DatasetResource.ExistingUploadFile("missing.csv", 1L))
+ ),
+ sessionUser
+ )
+ }
+
+ assertStatus(ex, 404)
+ }
+
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")
diff --git
a/frontend/src/app/dashboard/component/user/files-uploader/conflicting-file-modal-content/conflicting-file-modal-content.component.html
b/frontend/src/app/dashboard/component/user/files-uploader/conflicting-file-modal-content/conflicting-file-modal-content.component.html
index 1d5f8b849a..f2df36891a 100644
---
a/frontend/src/app/dashboard/component/user/files-uploader/conflicting-file-modal-content/conflicting-file-modal-content.component.html
+++
b/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>
diff --git
a/frontend/src/app/dashboard/component/user/files-uploader/conflicting-file-modal-content/conflicting-file-modal-content.component.ts
b/frontend/src/app/dashboard/component/user/files-uploader/conflicting-file-modal-content/conflicting-file-modal-content.component.ts
index b418929120..61691b59ab 100644
---
a/frontend/src/app/dashboard/component/user/files-uploader/conflicting-file-modal-content/conflicting-file-modal-content.component.ts
+++
b/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({
diff --git
a/frontend/src/app/dashboard/component/user/files-uploader/files-uploader.component.spec.ts
b/frontend/src/app/dashboard/component/user/files-uploader/files-uploader.component.spec.ts
new file mode 100644
index 0000000000..098aea82cb
--- /dev/null
+++
b/frontend/src/app/dashboard/component/user/files-uploader/files-uploader.component.spec.ts
@@ -0,0 +1,173 @@
+/**
+ * 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[];
+ let datasetService: {
+ listMultipartUploads: ReturnType<typeof vi.fn>;
+ findExistingUploadFiles: ReturnType<typeof vi.fn>;
+ };
+
+ 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;
+ datasetService = {
+ listMultipartUploads: vi.fn().mockReturnValue(of(["failed.csv"])),
+ findExistingUploadFiles: vi.fn().mockReturnValue(of(["done.csv"])),
+ };
+
+ component = new FilesUploaderComponent(
+ { error: vi.fn() } as unknown as NotificationService,
+ adminSettingsService,
+ datasetService as unknown as DatasetService,
+ modal
+ );
+ component.ownerEmail = "[email protected]";
+ 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"]);
+ });
+
+ it("asks both questions when the same file has an active upload session and
an existing match", async () => {
+ datasetService.listMultipartUploads.mockReturnValue(of(["same.csv"]));
+ datasetService.findExistingUploadFiles.mockReturnValue(of(["same.csv"]));
+ const emitted = new Promise<FileUploadItem[]>(resolve =>
component.uploadedFiles.subscribe(resolve));
+
+ component.fileDropped([droppedFile("same.csv", new File(["same"],
"same.csv"))]);
+
+ await waitUntil(() => modals.length === 1);
+ expect(modals[0].nzTitle).toBe("Conflicting File");
+ expect(modals[0].nzData.path).toBe("same.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("same.csv");
+ modals[1].nzFooter.find(button => button.label === "Upload")?.onClick();
+
+ expect((await emitted).map(item => item.name)).toEqual(["same.csv"]);
+ });
+
+ it("skips all matching files after one Skip For All choice", async () => {
+ datasetService.listMultipartUploads.mockReturnValue(of([]));
+ datasetService.findExistingUploadFiles.mockReturnValue(of(["one.csv",
"two.csv"]));
+ const emitted = new Promise<FileUploadItem[]>(resolve =>
component.uploadedFiles.subscribe(resolve));
+
+ component.fileDropped([
+ droppedFile("one.csv", new File(["one"], "one.csv")),
+ droppedFile("two.csv", new File(["two"], "two.csv")),
+ ]);
+
+ await waitUntil(() => modals.length === 1);
+ expect(modals[0].nzData.path).toBe("one.csv");
+ modals[0].nzFooter.find(button => button.label === "Skip For
All")?.onClick();
+
+ expect(await emitted).toEqual([]);
+ expect(modals).toHaveLength(1);
+ expect(component.fileUploadBannerType).toBe("info");
+ expect(component.fileUploadBannerMessage).toContain("2 matching files were
skipped.");
+ });
+
+ it("uploads all matching files after one Upload For All choice", async () =>
{
+ datasetService.listMultipartUploads.mockReturnValue(of([]));
+ datasetService.findExistingUploadFiles.mockReturnValue(of(["one.csv",
"two.csv"]));
+ const emitted = new Promise<FileUploadItem[]>(resolve =>
component.uploadedFiles.subscribe(resolve));
+
+ component.fileDropped([
+ droppedFile("one.csv", new File(["one"], "one.csv")),
+ droppedFile("two.csv", new File(["two"], "two.csv")),
+ ]);
+
+ await waitUntil(() => modals.length === 1);
+ expect(modals[0].nzData.path).toBe("one.csv");
+ modals[0].nzFooter.find(button => button.label === "Upload For
All")?.onClick();
+
+ expect((await emitted).map(item => item.name)).toEqual(["one.csv",
"two.csv"]);
+ expect(modals).toHaveLength(1);
+ expect(component.fileUploadBannerType).toBe("success");
+ });
+});
diff --git
a/frontend/src/app/dashboard/component/user/files-uploader/files-uploader.component.ts
b/frontend/src/app/dashboard/component/user/files-uploader/files-uploader.component.ts
index 6967d46b61..0a2f1c6c79 100644
---
a/frontend/src/app/dashboard/component/user/files-uploader/files-uploader.component.ts
+++
b/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) {
diff --git
a/frontend/src/app/dashboard/component/user/user-dataset/user-dataset-explorer/dataset-detail.component.html
b/frontend/src/app/dashboard/component/user/user-dataset/user-dataset-explorer/dataset-detail.component.html
index 75dbbaa545..4369cd76ba 100644
---
a/frontend/src/app/dashboard/component/user/user-dataset/user-dataset-explorer/dataset-detail.component.html
+++
b/frontend/src/app/dashboard/component/user/user-dataset/user-dataset-explorer/dataset-detail.component.html
@@ -338,6 +338,7 @@
nzActive="true"
nzHeader="Create New Version">
<texera-user-files-uploader
+ [did]="did"
[ownerEmail]="ownerEmail"
[datasetName]="datasetName"
(uploadedFiles)="onNewUploadFilesChanged($event)">
diff --git
a/frontend/src/app/dashboard/service/user/dataset/dataset.service.spec.ts
b/frontend/src/app/dashboard/service/user/dataset/dataset.service.spec.ts
index d930e50835..229ac32c5e 100644
--- a/frontend/src/app/dashboard/service/user/dataset/dataset.service.spec.ts
+++ b/frontend/src/app/dashboard/service/user/dataset/dataset.service.spec.ts
@@ -74,6 +74,48 @@ const SAMPLE_FILE_NODES: DatasetFileNode[] = [
{ name: "root", type: "directory", parentDir: "", children: [] as
DatasetFileNode[] } as DatasetFileNode,
];
+class FakeXMLHttpRequest {
+ static instances: FakeXMLHttpRequest[] = [];
+
+ readonly upload = {
+ addEventListener: vi.fn(),
+ };
+ status = 0;
+ url = "";
+ private listeners = new Map<string, EventListener[]>();
+
+ open(_method: string, url: string): void {
+ this.url = url;
+ }
+
+ setRequestHeader(): void {}
+
+ send(): void {
+ FakeXMLHttpRequest.instances.push(this);
+ }
+
+ abort(): void {}
+
+ addEventListener(type: string, listener: EventListener): void {
+ this.listeners.set(type, [...(this.listeners.get(type) ?? []), listener]);
+ }
+
+ respond(status: number): void {
+ this.status = status;
+ this.emit("load");
+ }
+
+ fail(): void {
+ this.emit("error");
+ }
+
+ private emit(type: string): void {
+ for (const listener of this.listeners.get(type) ?? []) {
+ listener(new Event(type));
+ }
+ }
+}
+
describe("DatasetService", () => {
let service: DatasetService;
let http: HttpTestingController;
@@ -89,6 +131,7 @@ describe("DatasetService", () => {
});
afterEach(() => {
+ vi.unstubAllGlobals();
http.verify();
});
@@ -222,6 +265,80 @@ describe("DatasetService", () => {
// ─── finalizeMultipartUpload (abort vs finish) ────────────────────────────
+ it("multipartUpload resumes a failed upload by sending only missing parts",
async () => {
+ vi.stubGlobal("XMLHttpRequest", FakeXMLHttpRequest);
+ FakeXMLHttpRequest.instances = [];
+ const file = new File(["abcdefgh"], "resume.txt");
+ const firstProgress: string[] = [];
+
+ const firstAttempt = new Promise<unknown>(resolve => {
+ service.multipartUpload("[email protected]", "ds", "resume.txt", file, 4, 1,
false).subscribe({
+ next: progress => firstProgress.push(progress.status),
+ error: (error: unknown): void => {
+ resolve(error);
+ },
+ complete: () => resolve(undefined),
+ });
+ });
+
+ http
+ .expectOne(r => r.url === `${API}/${DATASET_BASE_URL}/multipart-upload`
&& r.params.get("type") === "init")
+ .flush({ missingParts: [1, 2], completedPartsCount: 0 });
+
+ expect(FakeXMLHttpRequest.instances[0].url).toContain("partNumber=1");
+ FakeXMLHttpRequest.instances[0].respond(204);
+ expect(FakeXMLHttpRequest.instances[1].url).toContain("partNumber=2");
+ FakeXMLHttpRequest.instances[1].fail();
+
+ expect(await firstAttempt).toBeInstanceOf(Error);
+ expect(firstProgress).toContain("failed");
+
+ FakeXMLHttpRequest.instances = [];
+ const secondProgress: Array<{ percentage: number; status: string }> = [];
+ const secondAttempt = new Promise<void>((resolve, reject) => {
+ service.multipartUpload("[email protected]", "ds", "resume.txt", file, 4, 1,
false).subscribe({
+ next: progress => secondProgress.push({ percentage:
progress.percentage, status: progress.status }),
+ error: (error: unknown): void => {
+ reject(error);
+ },
+ complete: resolve,
+ });
+ });
+
+ http
+ .expectOne(r => r.url === `${API}/${DATASET_BASE_URL}/multipart-upload`
&& r.params.get("type") === "init")
+ .flush({ missingParts: [2], completedPartsCount: 1 });
+
+ expect(secondProgress[0]).toEqual({ percentage: 50, status: "initializing"
});
+ expect(
+ FakeXMLHttpRequest.instances.map(xhr => new URL(xhr.url,
"http://localhost";).searchParams.get("partNumber"))
+ ).toEqual(["2"]);
+ FakeXMLHttpRequest.instances[0].respond(204);
+
+ http
+ .expectOne(r => r.url === `${API}/${DATASET_BASE_URL}/multipart-upload`
&& r.params.get("type") === "finish")
+ .flush({});
+
+ await secondAttempt;
+ expect(secondProgress.at(-1)).toEqual({ percentage: 100, status:
"finished" });
+ });
+
+ it("findExistingUploadFiles posts path and size candidates", async () => {
+ const pending = firstValueFrom(service.findExistingUploadFiles(7, [{ path:
"a.csv", sizeBytes: 12 }]));
+ const req =
http.expectOne(`${API}/${DATASET_BASE_URL}/7/existing-upload-files`);
+ expect(req.request.method).toBe("POST");
+ expect(req.request.body).toEqual({ files: [{ path: "a.csv", sizeBytes: 12
}] });
+ req.flush({ filePaths: ["a.csv"] });
+ expect(await pending).toEqual(["a.csv"]);
+ });
+
+ it("findExistingUploadFiles tolerates a null payload", async () => {
+ const pending = firstValueFrom(service.findExistingUploadFiles(7, [{ path:
"a.csv", sizeBytes: 12 }]));
+ const req =
http.expectOne(`${API}/${DATASET_BASE_URL}/7/existing-upload-files`);
+ req.flush(null);
+ expect(await pending).toEqual([]);
+ });
+
it("finalizeMultipartUpload routes through type=finish when not aborting",
() => {
service.finalizeMultipartUpload("[email protected]", "ds", "f",
false).subscribe();
const req = http.expectOne(r => r.url ===
`${API}/${DATASET_BASE_URL}/multipart-upload`);
diff --git a/frontend/src/app/dashboard/service/user/dataset/dataset.service.ts
b/frontend/src/app/dashboard/service/user/dataset/dataset.service.ts
index aab1f6567a..6ec6a25aba 100644
--- a/frontend/src/app/dashboard/service/user/dataset/dataset.service.ts
+++ b/frontend/src/app/dashboard/service/user/dataset/dataset.service.ts
@@ -420,6 +420,17 @@ export class DatasetService {
.pipe(map(res => res?.filePaths ?? []));
}
+ public findExistingUploadFiles(did: number, files: { path: string;
sizeBytes: number }[]): Observable<string[]> {
+ return this.http
+ .post<{ filePaths: string[] }>(
+
`${AppSettings.getApiEndpoint()}/${DATASET_BASE_URL}/${did}/existing-upload-files`,
+ {
+ files,
+ }
+ )
+ .pipe(map(res => res?.filePaths ?? []));
+ }
+
public finalizeMultipartUpload(
ownerEmail: string,
datasetName: string,
