aicam commented on code in PR #6872:
URL: https://github.com/apache/texera/pull/6872#discussion_r3855419473
##########
file-service/src/main/scala/org/apache/texera/service/resource/ModelResource.scala:
##########
@@ -408,4 +460,341 @@ class ModelResource extends LazyLogging {
): DashboardModel = {
withTransaction(context)(ctx => getDashboardModel(ctx, mid, None))
}
+
+ //
===========================================================================
+ // Versioning
+ //
===========================================================================
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/create")
+ @Consumes(Array(MediaType.TEXT_PLAIN))
+ def createModelVersion(
+ versionName: String,
+ @PathParam("mid") mid: Integer,
+ @Auth user: SessionUser
+ ): DashboardModelVersion = {
+ val uid = user.getUid
+ withTransaction(context) { ctx =>
+ if (!userHasWriteAccess(ctx, mid, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_MODEL_MESSAGE)
+ }
+
+ val model = getModelByID(ctx, mid)
+ val modelName = model.getName
+ val repositoryName = model.getRepositoryName
+
+ // Check if there are any changes in LakeFS before creating a new version
+ val diffs = withLakeFSErrorHandling {
+ LakeFSStorageClient.retrieveUncommittedObjects(repoName =
repositoryName)
+ }
+
+ if (diffs.isEmpty) {
+ throw new WebApplicationException(
+ "No changes detected in model. Version creation aborted.",
+ Response.Status.BAD_REQUEST
+ )
+ }
+
+ // Generate a new version name
+ val versionCount = ctx
+ .selectCount()
+ .from(MODEL_VERSION)
+ .where(MODEL_VERSION.MID.eq(mid))
+ .fetchOne(0, classOf[Int])
+
+ val sanitizedVersionName =
Option(versionName).filter(_.nonEmpty).getOrElse("")
+ val newVersionName = if (sanitizedVersionName.isEmpty) {
+ s"v${versionCount + 1}"
+ } else {
+ s"v${versionCount + 1} - $sanitizedVersionName"
Review Comment:
The version name goes straight into `model_version.name`, which is
`VARCHAR(128)`, with no length check — and the LakeFS commit at line 514 lands
*before* the insert at 536.
Reproduced on `eafd888` by posting a 200-char name:
```
POST /model/8/version/create --data 'release-candidate-...' -> 500
server log: PSQLException: value too long for type character varying(128)
lakefs commits: 7e2c221cdbd3 Created model version: v3 -
release-candidate-release-...
model_version rows for mid=8: was 2, still 2
POST /model/8/version/create --data 'retry-with-a-sane-name' -> 400 No
changes detected in model.
```
The transaction rolls back the row, nothing rolls back the commit, and the
retry now fails the `diffs.isEmpty` guard. The staged file is committed,
belongs to no version, and can never be versioned again. Validating or
truncating the name before `createCommit` avoids it.
##########
file-service/src/main/scala/org/apache/texera/service/resource/ModelResource.scala:
##########
@@ -408,4 +460,341 @@ class ModelResource extends LazyLogging {
): DashboardModel = {
withTransaction(context)(ctx => getDashboardModel(ctx, mid, None))
}
+
+ //
===========================================================================
+ // Versioning
+ //
===========================================================================
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/create")
+ @Consumes(Array(MediaType.TEXT_PLAIN))
+ def createModelVersion(
+ versionName: String,
+ @PathParam("mid") mid: Integer,
+ @Auth user: SessionUser
+ ): DashboardModelVersion = {
+ val uid = user.getUid
+ withTransaction(context) { ctx =>
+ if (!userHasWriteAccess(ctx, mid, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_MODEL_MESSAGE)
+ }
+
+ val model = getModelByID(ctx, mid)
+ val modelName = model.getName
+ val repositoryName = model.getRepositoryName
+
+ // Check if there are any changes in LakeFS before creating a new version
+ val diffs = withLakeFSErrorHandling {
+ LakeFSStorageClient.retrieveUncommittedObjects(repoName =
repositoryName)
+ }
+
+ if (diffs.isEmpty) {
+ throw new WebApplicationException(
+ "No changes detected in model. Version creation aborted.",
+ Response.Status.BAD_REQUEST
+ )
+ }
+
+ // Generate a new version name
+ val versionCount = ctx
+ .selectCount()
+ .from(MODEL_VERSION)
+ .where(MODEL_VERSION.MID.eq(mid))
+ .fetchOne(0, classOf[Int])
+
+ val sanitizedVersionName =
Option(versionName).filter(_.nonEmpty).getOrElse("")
+ val newVersionName = if (sanitizedVersionName.isEmpty) {
+ s"v${versionCount + 1}"
+ } else {
+ s"v${versionCount + 1} - $sanitizedVersionName"
+ }
+
+ // Create a commit in LakeFS
+ val commit = withLakeFSErrorHandling {
+ LakeFSStorageClient.createCommit(
+ repoName = repositoryName,
+ branch = "main",
+ commitMessage = s"Created model version: $newVersionName"
+ )
+ }
+
+ if (commit == null || commit.getId == null) {
+ throw new WebApplicationException(
+ "Failed to create commit in LakeFS. Version creation aborted.",
+ Response.Status.INTERNAL_SERVER_ERROR
+ )
+ }
+
+ // Create a new model version entry in the database
+ val modelVersion = new ModelVersion()
+ modelVersion.setMid(mid)
+ modelVersion.setCreatorUid(uid)
+ modelVersion.setName(newVersionName)
+ modelVersion.setVersionHash(commit.getId) // Store LakeFS version hash
+
+ val insertedVersion = ctx
+ .insertInto(MODEL_VERSION)
+ .set(ctx.newRecord(MODEL_VERSION, modelVersion))
+ .returning()
+ .fetchOne()
+ .into(classOf[ModelVersion])
+
+ // Retrieve committed file structure
+ val fileNodes = withLakeFSErrorHandling {
+ LakeFSStorageClient.retrieveObjectsOfVersion(repositoryName,
commit.getId)
+ }
+
+ DashboardModelVersion(
+ insertedVersion,
+ LakeFSFileNode
+ .fromLakeFSRepositoryCommittedObjects(
+ ResourceType.Model,
+ Map((user.getEmail, modelName, newVersionName) -> fileNodes)
+ )
+ )
+ }
+ }
+
+ @GET
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/list")
+ def getModelVersionList(
+ @PathParam("mid") mid: Integer,
+ @Auth user: SessionUser
+ ): List[ModelVersion] = {
+ val uid = user.getUid
+ withTransaction(context)(ctx => {
+ val model = getModelByID(ctx, mid)
+ if (!userHasReadAccess(ctx, model.getMid, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_MODEL_MESSAGE)
+ }
+ fetchModelVersions(ctx, model.getMid)
+ })
+ }
+
+ @GET
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/latest")
+ def retrieveLatestModelVersion(
+ @PathParam("mid") mid: Integer,
+ @Auth user: SessionUser
+ ): DashboardModelVersion = {
+ val uid = user.getUid
+ withTransaction(context)(ctx => {
+ if (!userHasReadAccess(ctx, mid, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_MODEL_MESSAGE)
+ }
+ val model = getModelByID(ctx, mid)
+ val latestVersion = getLatestModelVersion(ctx, mid).getOrElse(
+ throw new NotFoundException(ERR_MODEL_VERSION_NOT_FOUND_MESSAGE)
+ )
+ DashboardModelVersion(latestVersion, versionRootFileNodes(ctx, mid,
latestVersion))
+ })
+ }
+
+ @GET
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/{mvid}/rootFileNodes")
+ def retrieveModelVersionRootFileNodes(
+ @PathParam("mid") mid: Integer,
+ @PathParam("mvid") mvid: Integer,
+ @Auth user: SessionUser
+ ): ModelVersionRootFileNodesResponse = {
+ val uid = user.getUid
+ withTransaction(context)(ctx => fetchModelVersionRootFileNodes(ctx, mid,
mvid, Some(uid)))
+ }
+
+ //
===========================================================================
+ // File upload (one-shot + session-based multipart)
+ //
===========================================================================
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/upload")
+ @Consumes(Array(MediaType.APPLICATION_OCTET_STREAM))
+ def uploadOneFileToModel(
+ @PathParam("mid") mid: Integer,
+ @QueryParam("filePath") encodedFilePath: String,
+ @QueryParam("message") message: String,
+ fileStream: InputStream,
+ @Context headers: HttpHeaders,
+ @Auth user: SessionUser
+ ): Response = {
+ ResourceUploadService.uploadOneFile(
+ ResourceStorage.Model,
+ mid,
+ encodedFilePath,
+ fileStream,
+ headers,
+ user.getUid
+ )
+ }
+
+ @DELETE
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/file")
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ def deleteModelFile(
+ @PathParam("mid") mid: Integer,
+ @QueryParam("filePath") encodedFilePath: String,
+ @Auth user: SessionUser
+ ): Response = {
+ ResourceUploadService.deleteStagedFile(
+ ResourceStorage.Model,
+ mid,
+ encodedFilePath,
+ user.getUid
+ )
+ }
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/multipart-upload")
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ def multipartUpload(
+ @QueryParam("type") operationType: String,
+ @QueryParam("ownerEmail") ownerEmail: String,
+ @QueryParam("modelName") modelName: String,
+ @QueryParam("filePath") filePath: String,
+ @QueryParam("fileSizeBytes") fileSizeBytes: Optional[java.lang.Long],
+ @QueryParam("partSizeBytes") partSizeBytes: Optional[java.lang.Long],
+ @QueryParam("restart") restart: Optional[java.lang.Boolean],
+ @Auth user: SessionUser
+ ): Response = {
+ val uid = user.getUid
+ val model: Model = getModelBy(ownerEmail, modelName)
+
+ operationType.toLowerCase match {
Review Comment:
`operationType` is an optional query param, so omitting it NPEs instead of
returning the 400 that line 675 intends:
```
POST /model/multipart-upload?ownerEmail=texera&modelName=resnet50 -> 500
java.lang.NullPointerException: Cannot invoke "String.toLowerCase()" because
"operationType" is null
```
Two small things while you're here: `getModelBy` on line 666 runs before the
type is validated, so an invalid request still costs a DB round-trip; and line
675's message lists `'init', 'finish', 'abort'` but `list` is handled on line
669 as well.
##########
file-service/src/main/scala/org/apache/texera/service/resource/ModelResource.scala:
##########
@@ -50,6 +63,9 @@ object ModelResource {
.getInstance()
.createDSLContext()
+ private def singleFileUploadMaxBytes(defaultMiB: Long = 20L): Long =
Review Comment:
This is a verbatim copy of the private helper in `ResourceUploadService`,
and nothing in this file calls it — it's the only user of the new
`SiteSettings` import. Worth noting `POST /model/{mid}/upload` enforces no size
limit at all (the multipart path enforces its own in
`initUpload`/`finishUpload`), which looks like what this was meant for. Either
wire it up or drop it with the import.
##########
file-service/src/main/scala/org/apache/texera/service/resource/ResourceUploadService.scala:
##########
@@ -139,6 +171,49 @@ object ResourceUploadService {
private def singleFileUploadMaxBytes(defaultMiB: Long = 20L): Long =
SiteSettings.getLong("single_file_upload_max_size_mib", defaultMiB) *
1024L * 1024L
+ /**
+ * Builds the file nodes of one committed version, plus the version's total
size.
+ *
+ * The tree is rooted at the resource-type prefix, so the paths it yields
resolve against
+ * the right table when they are handed back to `FileResolver`.
+ */
+ def versionRootFileNodes(
Review Comment:
Nice extraction. Since the point of it is one shared implementation, the two
inlined copies in `DatasetResource` are worth migrating in the same pass —
`fetchDatasetVersionRootFileNodes` (`DatasetResource.scala:1234`) and
`retrieveLatestDatasetVersion` (1033-1063) still build this tree by hand.
Otherwise there are three copies where the comment says there is one.
##########
file-service/src/main/scala/org/apache/texera/service/resource/ModelResource.scala:
##########
@@ -408,4 +460,341 @@ class ModelResource extends LazyLogging {
): DashboardModel = {
withTransaction(context)(ctx => getDashboardModel(ctx, mid, None))
}
+
+ //
===========================================================================
+ // Versioning
+ //
===========================================================================
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/create")
+ @Consumes(Array(MediaType.TEXT_PLAIN))
+ def createModelVersion(
+ versionName: String,
+ @PathParam("mid") mid: Integer,
+ @Auth user: SessionUser
+ ): DashboardModelVersion = {
+ val uid = user.getUid
+ withTransaction(context) { ctx =>
+ if (!userHasWriteAccess(ctx, mid, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_MODEL_MESSAGE)
+ }
+
+ val model = getModelByID(ctx, mid)
+ val modelName = model.getName
+ val repositoryName = model.getRepositoryName
+
+ // Check if there are any changes in LakeFS before creating a new version
+ val diffs = withLakeFSErrorHandling {
+ LakeFSStorageClient.retrieveUncommittedObjects(repoName =
repositoryName)
+ }
+
+ if (diffs.isEmpty) {
+ throw new WebApplicationException(
+ "No changes detected in model. Version creation aborted.",
+ Response.Status.BAD_REQUEST
+ )
+ }
+
+ // Generate a new version name
+ val versionCount = ctx
+ .selectCount()
+ .from(MODEL_VERSION)
+ .where(MODEL_VERSION.MID.eq(mid))
+ .fetchOne(0, classOf[Int])
+
+ val sanitizedVersionName =
Option(versionName).filter(_.nonEmpty).getOrElse("")
+ val newVersionName = if (sanitizedVersionName.isEmpty) {
+ s"v${versionCount + 1}"
+ } else {
+ s"v${versionCount + 1} - $sanitizedVersionName"
+ }
+
+ // Create a commit in LakeFS
+ val commit = withLakeFSErrorHandling {
+ LakeFSStorageClient.createCommit(
+ repoName = repositoryName,
+ branch = "main",
+ commitMessage = s"Created model version: $newVersionName"
+ )
+ }
+
+ if (commit == null || commit.getId == null) {
+ throw new WebApplicationException(
+ "Failed to create commit in LakeFS. Version creation aborted.",
+ Response.Status.INTERNAL_SERVER_ERROR
+ )
+ }
+
+ // Create a new model version entry in the database
+ val modelVersion = new ModelVersion()
+ modelVersion.setMid(mid)
+ modelVersion.setCreatorUid(uid)
+ modelVersion.setName(newVersionName)
+ modelVersion.setVersionHash(commit.getId) // Store LakeFS version hash
+
+ val insertedVersion = ctx
+ .insertInto(MODEL_VERSION)
+ .set(ctx.newRecord(MODEL_VERSION, modelVersion))
+ .returning()
+ .fetchOne()
+ .into(classOf[ModelVersion])
+
+ // Retrieve committed file structure
+ val fileNodes = withLakeFSErrorHandling {
+ LakeFSStorageClient.retrieveObjectsOfVersion(repositoryName,
commit.getId)
+ }
+
+ DashboardModelVersion(
+ insertedVersion,
+ LakeFSFileNode
+ .fromLakeFSRepositoryCommittedObjects(
+ ResourceType.Model,
+ Map((user.getEmail, modelName, newVersionName) -> fileNodes)
+ )
+ )
+ }
+ }
+
+ @GET
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/list")
+ def getModelVersionList(
+ @PathParam("mid") mid: Integer,
+ @Auth user: SessionUser
+ ): List[ModelVersion] = {
+ val uid = user.getUid
+ withTransaction(context)(ctx => {
+ val model = getModelByID(ctx, mid)
+ if (!userHasReadAccess(ctx, model.getMid, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_MODEL_MESSAGE)
+ }
+ fetchModelVersions(ctx, model.getMid)
+ })
+ }
+
+ @GET
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/latest")
+ def retrieveLatestModelVersion(
+ @PathParam("mid") mid: Integer,
+ @Auth user: SessionUser
+ ): DashboardModelVersion = {
+ val uid = user.getUid
+ withTransaction(context)(ctx => {
+ if (!userHasReadAccess(ctx, mid, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_MODEL_MESSAGE)
+ }
+ val model = getModelByID(ctx, mid)
+ val latestVersion = getLatestModelVersion(ctx, mid).getOrElse(
+ throw new NotFoundException(ERR_MODEL_VERSION_NOT_FOUND_MESSAGE)
+ )
+ DashboardModelVersion(latestVersion, versionRootFileNodes(ctx, mid,
latestVersion))
+ })
+ }
+
+ @GET
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/{mvid}/rootFileNodes")
+ def retrieveModelVersionRootFileNodes(
+ @PathParam("mid") mid: Integer,
+ @PathParam("mvid") mvid: Integer,
+ @Auth user: SessionUser
+ ): ModelVersionRootFileNodesResponse = {
+ val uid = user.getUid
+ withTransaction(context)(ctx => fetchModelVersionRootFileNodes(ctx, mid,
mvid, Some(uid)))
+ }
+
+ //
===========================================================================
+ // File upload (one-shot + session-based multipart)
+ //
===========================================================================
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/upload")
+ @Consumes(Array(MediaType.APPLICATION_OCTET_STREAM))
+ def uploadOneFileToModel(
+ @PathParam("mid") mid: Integer,
+ @QueryParam("filePath") encodedFilePath: String,
+ @QueryParam("message") message: String,
+ fileStream: InputStream,
+ @Context headers: HttpHeaders,
+ @Auth user: SessionUser
+ ): Response = {
+ ResourceUploadService.uploadOneFile(
+ ResourceStorage.Model,
+ mid,
+ encodedFilePath,
+ fileStream,
+ headers,
+ user.getUid
+ )
+ }
+
+ @DELETE
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/file")
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ def deleteModelFile(
+ @PathParam("mid") mid: Integer,
+ @QueryParam("filePath") encodedFilePath: String,
+ @Auth user: SessionUser
+ ): Response = {
+ ResourceUploadService.deleteStagedFile(
+ ResourceStorage.Model,
+ mid,
+ encodedFilePath,
+ user.getUid
+ )
+ }
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/multipart-upload")
+ @Consumes(Array(MediaType.APPLICATION_JSON))
+ def multipartUpload(
+ @QueryParam("type") operationType: String,
+ @QueryParam("ownerEmail") ownerEmail: String,
+ @QueryParam("modelName") modelName: String,
+ @QueryParam("filePath") filePath: String,
+ @QueryParam("fileSizeBytes") fileSizeBytes: Optional[java.lang.Long],
+ @QueryParam("partSizeBytes") partSizeBytes: Optional[java.lang.Long],
+ @QueryParam("restart") restart: Optional[java.lang.Boolean],
+ @Auth user: SessionUser
+ ): Response = {
+ val uid = user.getUid
+ val model: Model = getModelBy(ownerEmail, modelName)
+
+ operationType.toLowerCase match {
+ case "list" => listMultipartUploads(model.getMid, uid)
+ case "init" =>
+ initMultipartUpload(model.getMid, filePath, fileSizeBytes,
partSizeBytes, restart, uid)
+ case "finish" => finishMultipartUpload(model.getMid, filePath, uid)
+ case "abort" => abortMultipartUpload(model.getMid, filePath, uid)
+ case _ =>
+ throw new BadRequestException("Invalid type parameter. Use 'init',
'finish', or 'abort'.")
+ }
+ }
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Consumes(Array(MediaType.APPLICATION_OCTET_STREAM))
+ @Path("/multipart-upload/part")
+ def uploadPart(
+ @QueryParam("ownerEmail") modelOwnerEmail: String,
+ @QueryParam("modelName") modelName: String,
+ @QueryParam("filePath") encodedFilePath: String,
+ @QueryParam("partNumber") partNumber: Int,
+ partStream: InputStream,
+ @Context headers: HttpHeaders,
+ @Auth user: SessionUser
+ ): Response = {
+ val model = getModelBy(modelOwnerEmail, modelName)
+ ResourceUploadService.uploadPart(
+ ResourceStorage.Model,
+ model.getMid,
+ user.getUid,
+ encodedFilePath,
+ partNumber,
+ partStream,
+ headers
+ )
+ }
+
+ //
===========================================================================
+ // Private helpers
+ //
===========================================================================
+
+ private def fetchModelVersions(ctx: DSLContext, mid: Integer):
List[ModelVersion] = {
+ ctx
+ .selectFrom(MODEL_VERSION)
+ .where(MODEL_VERSION.MID.eq(mid))
+ .orderBy(MODEL_VERSION.CREATION_TIME.desc())
+ .fetchInto(classOf[ModelVersion])
+ .asScala
+ .toList
+ }
+
+ /**
+ * Builds the file-tree children of a single model version, drilling into
the
+ * owner/model/version nesting produced by LakeFSFileNode.
+ */
+ private def versionRootFileNodes(
+ ctx: DSLContext,
+ mid: Integer,
+ modelVersion: ModelVersion
+ ): List[LakeFSFileNode] = {
+ val model = getModelByID(ctx, mid)
+ ResourceUploadService
+ .versionRootFileNodes(
+ ResourceType.Model,
+ getOwner(ctx, mid).getEmail,
+ model.getName,
+ modelVersion.getName,
+ model.getRepositoryName,
+ modelVersion.getVersionHash
+ )
+ ._1
+ }
+
+ private def fetchModelVersionRootFileNodes(
+ ctx: DSLContext,
+ mid: Integer,
+ mvid: Integer,
+ uid: Option[Integer]
+ ): ModelVersionRootFileNodesResponse = {
+ val model = getDashboardModel(ctx, mid, uid)
+ val modelVersion = getModelVersionByID(ctx, mvid)
Review Comment:
`getModelVersionByID(ctx, mvid)` isn't constrained to `mid`, but the access
check above ran against `mid` — so a version of another model can be resolved
through this one's repository:
```
GET /model/8/version/17/rootFileNodes (mid of resnet50, mvid of
bert-base) -> 500
server log: io.lakefs.clients.sdk.ApiException: Not Found (404) from
listObjects
```
Two things: the 500 should be a 404, and by then the other model's version
name has already been spliced into the path being built. Adding
`.and(MODEL_VERSION.MID.eq(mid))` to the lookup covers both.
##########
file-service/src/main/scala/org/apache/texera/service/resource/ModelResource.scala:
##########
@@ -408,4 +460,341 @@ class ModelResource extends LazyLogging {
): DashboardModel = {
withTransaction(context)(ctx => getDashboardModel(ctx, mid, None))
}
+
+ //
===========================================================================
+ // Versioning
+ //
===========================================================================
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/create")
+ @Consumes(Array(MediaType.TEXT_PLAIN))
+ def createModelVersion(
+ versionName: String,
+ @PathParam("mid") mid: Integer,
+ @Auth user: SessionUser
+ ): DashboardModelVersion = {
+ val uid = user.getUid
+ withTransaction(context) { ctx =>
+ if (!userHasWriteAccess(ctx, mid, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_MODEL_MESSAGE)
+ }
+
+ val model = getModelByID(ctx, mid)
+ val modelName = model.getName
+ val repositoryName = model.getRepositoryName
+
+ // Check if there are any changes in LakeFS before creating a new version
+ val diffs = withLakeFSErrorHandling {
+ LakeFSStorageClient.retrieveUncommittedObjects(repoName =
repositoryName)
+ }
+
+ if (diffs.isEmpty) {
+ throw new WebApplicationException(
+ "No changes detected in model. Version creation aborted.",
+ Response.Status.BAD_REQUEST
+ )
+ }
+
+ // Generate a new version name
+ val versionCount = ctx
+ .selectCount()
+ .from(MODEL_VERSION)
+ .where(MODEL_VERSION.MID.eq(mid))
+ .fetchOne(0, classOf[Int])
+
+ val sanitizedVersionName =
Option(versionName).filter(_.nonEmpty).getOrElse("")
+ val newVersionName = if (sanitizedVersionName.isEmpty) {
+ s"v${versionCount + 1}"
+ } else {
+ s"v${versionCount + 1} - $sanitizedVersionName"
+ }
+
+ // Create a commit in LakeFS
+ val commit = withLakeFSErrorHandling {
+ LakeFSStorageClient.createCommit(
+ repoName = repositoryName,
+ branch = "main",
+ commitMessage = s"Created model version: $newVersionName"
+ )
+ }
+
+ if (commit == null || commit.getId == null) {
+ throw new WebApplicationException(
+ "Failed to create commit in LakeFS. Version creation aborted.",
+ Response.Status.INTERNAL_SERVER_ERROR
+ )
+ }
+
+ // Create a new model version entry in the database
+ val modelVersion = new ModelVersion()
+ modelVersion.setMid(mid)
+ modelVersion.setCreatorUid(uid)
+ modelVersion.setName(newVersionName)
+ modelVersion.setVersionHash(commit.getId) // Store LakeFS version hash
+
+ val insertedVersion = ctx
+ .insertInto(MODEL_VERSION)
+ .set(ctx.newRecord(MODEL_VERSION, modelVersion))
+ .returning()
+ .fetchOne()
+ .into(classOf[ModelVersion])
+
+ // Retrieve committed file structure
+ val fileNodes = withLakeFSErrorHandling {
+ LakeFSStorageClient.retrieveObjectsOfVersion(repositoryName,
commit.getId)
+ }
+
+ DashboardModelVersion(
+ insertedVersion,
+ LakeFSFileNode
+ .fromLakeFSRepositoryCommittedObjects(
+ ResourceType.Model,
+ Map((user.getEmail, modelName, newVersionName) -> fileNodes)
+ )
+ )
+ }
+ }
+
+ @GET
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/list")
+ def getModelVersionList(
+ @PathParam("mid") mid: Integer,
+ @Auth user: SessionUser
+ ): List[ModelVersion] = {
+ val uid = user.getUid
+ withTransaction(context)(ctx => {
+ val model = getModelByID(ctx, mid)
+ if (!userHasReadAccess(ctx, model.getMid, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_MODEL_MESSAGE)
+ }
+ fetchModelVersions(ctx, model.getMid)
+ })
+ }
+
+ @GET
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/latest")
+ def retrieveLatestModelVersion(
+ @PathParam("mid") mid: Integer,
+ @Auth user: SessionUser
+ ): DashboardModelVersion = {
+ val uid = user.getUid
+ withTransaction(context)(ctx => {
+ if (!userHasReadAccess(ctx, mid, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_MODEL_MESSAGE)
+ }
+ val model = getModelByID(ctx, mid)
Review Comment:
`model` is never used — `versionRootFileNodes` on the next line re-fetches
the model and the owner itself, so this is three redundant queries per call. It
can't serve as an existence check either, since `userHasReadAccess` above
already returns false for a nonexistent `mid`.
##########
file-service/src/main/scala/org/apache/texera/service/resource/ModelResource.scala:
##########
@@ -408,4 +460,341 @@ class ModelResource extends LazyLogging {
): DashboardModel = {
withTransaction(context)(ctx => getDashboardModel(ctx, mid, None))
}
+
+ //
===========================================================================
+ // Versioning
+ //
===========================================================================
+
+ @POST
+ @RolesAllowed(Array("REGULAR", "ADMIN"))
+ @Path("/{mid}/version/create")
+ @Consumes(Array(MediaType.TEXT_PLAIN))
+ def createModelVersion(
+ versionName: String,
+ @PathParam("mid") mid: Integer,
+ @Auth user: SessionUser
+ ): DashboardModelVersion = {
+ val uid = user.getUid
+ withTransaction(context) { ctx =>
+ if (!userHasWriteAccess(ctx, mid, uid)) {
+ throw new ForbiddenException(ERR_USER_HAS_NO_ACCESS_TO_MODEL_MESSAGE)
+ }
+
+ val model = getModelByID(ctx, mid)
+ val modelName = model.getName
+ val repositoryName = model.getRepositoryName
+
+ // Check if there are any changes in LakeFS before creating a new version
+ val diffs = withLakeFSErrorHandling {
+ LakeFSStorageClient.retrieveUncommittedObjects(repoName =
repositoryName)
+ }
+
+ if (diffs.isEmpty) {
+ throw new WebApplicationException(
+ "No changes detected in model. Version creation aborted.",
+ Response.Status.BAD_REQUEST
+ )
+ }
+
+ // Generate a new version name
+ val versionCount = ctx
Review Comment:
Nit / defensive: the name comes from a `count(*)` with no lock on the model
row, and `model_version` has no `UNIQUE (mid, name)`. Two creates interleaving
between this read and the insert would both produce `v2`, and
`FileResolver.lookupModel` does `MODEL_VERSION.NAME.eq(versionName) …
fetchOneInto`, which would then throw `TooManyRowsException` for every file
lookup in that model.
I could not trigger it — 8 rounds of 3 concurrent creates produced no
duplicate, because LakeFS rejects the second empty commit and serialises them
in practice. So this is theoretical, but a `UNIQUE (mid, name)` in `41.sql`
would close it for free.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
