You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/08/14 17:28:06 UTC
(texera) branch gh-readonly-queue/main/pr-6860-490e57f2cc384581401f56aa9eafb5cd7af03d7d created (now 3cb3479ec9)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-6860-490e57f2cc384581401f56aa9eafb5cd7af03d7d
in repository https://gitbox.apache.org/repos/asf/texera.git
at 3cb3479ec9 feat(storage): add model file storage and path resolution (#6860)
This branch includes the following new commits:
new 3cb3479ec9 feat(storage): add model file storage and path resolution (#6860)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: feat(storage): add model file storage and path resolution (#6860)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6860-490e57f2cc384581401f56aa9eafb5cd7af03d7d
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 3cb3479ec9201e43ca9e803da2a2a26d59a480a7
Author: Tanishq Gandhi <56...@users.noreply.github.com>
AuthorDate: Fri Aug 14 17:27:42 2026 +0000
feat(storage): add model file storage and path resolution (#6860)
### What changes were proposed in this PR?
Adds ML **models** as a first-class versioned-file resource — a sibling
of datasets that reuses the existing LakeFS + object-storage engine. A
model version is stored as one commit in a per-model repository
(`model-{mid}`); a logical path
`/models/<owner>/<name>/<version>/<file>` resolves to a stable physical
URI `model:///<repositoryName>/<versionHash>/<file>` that can be opened
to stream the file's bytes.
Rather than duplicate the dataset path, the shared logic is
**generalized** , then models are added on top:
### Any related issues, documentation, discussions?
Closes #6497.
Part of the "Supporting ML models" umbrella #6494. Stacked on #6495
(#6502) — the diff carries its commits until #6502 merges.
### How was this PR tested?
Unit tests (`sbt "WorkflowCore/testOnly *FileResolverSpec
*DocumentFactorySpec *DatasetFileDocumentSpec"`):
model path resolution, required-prefix enforcement, dataset↔model
cross-prefix isolation (both directions), and `DocumentFactory` routing
the `model` scheme to a `ModelFileDocument`. Also verified end-to-end
against a live Postgres: a seeded `model`/`model_version` row resolves
to the expected `model:///…` URI and opens as a `ModelFileDocument`.
## Notes
- The model **REST/upload API** and the `/api/model/presign-download`
endpoint land in #6498. No code produces a `/models/…` path yet; the
JWT/file-service read branch is inherited from the shared base and falls
back to a direct LakeFS fetch, so model reads work without that endpoint
today.
- The remaining model tables (access control, upload sessions, likes,
view count) are added in later PRs alongside the logic that uses them.
- Frontend/Python `ResourceType` still expose only `Datasets`; `Models`
is added there when the UI/UDF work lands.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 4.8)
---------
Co-authored-by: ali <al...@gmail.com>
Co-authored-by: Claude Opus 4.8 <no...@anthropic.com>
---
.../pytexera/storage/dataset_file_document.py | 4 +-
.../pytexera/storage/test_dataset_file_document.py | 24 +-
.../workflow-compiling-service-deployment.yaml | 2 +-
...workflow-computing-unit-manager-deployment.yaml | 2 +-
bin/local-dev/main.sh | 2 +-
bin/single-node/.env | 2 +-
.../common/config/EnvironmentalVariable.scala | 5 +-
.../amber/core/storage/DocumentFactory.scala | 7 +-
.../texera/amber/core/storage/FileResolver.scala | 260 +++++++++++++++------
.../texera/amber/core/storage/ResourceType.scala | 13 +-
.../core/storage/model/DatasetFileDocument.scala | 148 +-----------
...FileDocument.scala => LakeFSFileDocument.scala} | 63 ++---
.../{OnDataset.scala => ModelFileDocument.scala} | 23 +-
...Dataset.scala => OnVersionedFileResource.scala} | 5 +-
.../amber/core/storage/DocumentFactorySpec.scala | 23 +-
.../storage/model/DatasetFileDocumentSpec.scala | 18 +-
.../texera/amber/storage/FileResolverSpec.scala | 93 +++++++-
.../resource/ComputingUnitManagingResource.scala | 4 +-
.../texera/service/resource/DatasetResource.scala | 11 +-
sql/changelog.xml | 5 +
sql/texera_ddl.sql | 33 ++-
sql/updates/37.sql | 57 +++++
22 files changed, 507 insertions(+), 297 deletions(-)
diff --git a/amber/src/main/python/pytexera/storage/dataset_file_document.py b/amber/src/main/python/pytexera/storage/dataset_file_document.py
index 4446da5ee5..b8d778d342 100644
--- a/amber/src/main/python/pytexera/storage/dataset_file_document.py
+++ b/amber/src/main/python/pytexera/storage/dataset_file_document.py
@@ -89,7 +89,9 @@ class DatasetFileDocument:
raise invalid_format
self.jwt_token = os.getenv("USER_JWT_TOKEN")
- self.presign_endpoint = os.getenv("FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT")
+ self.presign_endpoint = os.getenv(
+ "FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT"
+ )
if not self.jwt_token:
raise ValueError(
diff --git a/amber/src/test/python/pytexera/storage/test_dataset_file_document.py b/amber/src/test/python/pytexera/storage/test_dataset_file_document.py
index 760e6234be..bef134bc74 100644
--- a/amber/src/test/python/pytexera/storage/test_dataset_file_document.py
+++ b/amber/src/test/python/pytexera/storage/test_dataset_file_document.py
@@ -31,7 +31,9 @@ CUSTOM_ENDPOINT = "https://example.test/api/presign"
def auth_env(monkeypatch):
"""Provide a JWT and pinned presign endpoint for the duration of one test."""
monkeypatch.setenv("USER_JWT_TOKEN", "test-jwt-token")
- monkeypatch.setenv("FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT", CUSTOM_ENDPOINT)
+ monkeypatch.setenv(
+ "FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT", CUSTOM_ENDPOINT
+ )
def make_response(status_code: int, body=None, content: bytes = b""):
@@ -86,7 +88,9 @@ class TestDatasetFileDocumentInit:
def test_requires_jwt_token_in_environment(self, monkeypatch):
monkeypatch.delenv("USER_JWT_TOKEN", raising=False)
- monkeypatch.setenv("FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT", CUSTOM_ENDPOINT)
+ monkeypatch.setenv(
+ "FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT", CUSTOM_ENDPOINT
+ )
with pytest.raises(ValueError, match="JWT token is required"):
DatasetFileDocument("/datasets/bob@x.com/ds/v1/file.csv")
@@ -98,7 +102,9 @@ class TestDatasetFileDocumentInit:
def test_falls_back_to_default_endpoint_when_env_missing(self, monkeypatch):
monkeypatch.setenv("USER_JWT_TOKEN", "tok")
- monkeypatch.delenv("FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT", raising=False)
+ monkeypatch.delenv(
+ "FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT", raising=False
+ )
doc = DatasetFileDocument("/datasets/bob@x.com/ds/v1/file.csv")
assert doc.presign_endpoint == DEFAULT_ENDPOINT
@@ -110,7 +116,9 @@ class TestDatasetFileDocumentInit:
class TestGetPresignedUrl:
def _make_doc(self, monkeypatch, path="/datasets/bob@x.com/ds/v1/file.csv"):
monkeypatch.setenv("USER_JWT_TOKEN", "test-jwt-token")
- monkeypatch.setenv("FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT", CUSTOM_ENDPOINT)
+ monkeypatch.setenv(
+ "FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT", CUSTOM_ENDPOINT
+ )
return DatasetFileDocument(path)
def test_returns_presigned_url_field_from_json_body(self, monkeypatch):
@@ -224,7 +232,9 @@ class TestGetPresignedUrl:
class TestReadFile:
def _make_doc(self, monkeypatch):
monkeypatch.setenv("USER_JWT_TOKEN", "test-jwt-token")
- monkeypatch.setenv("FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT", CUSTOM_ENDPOINT)
+ monkeypatch.setenv(
+ "FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT", CUSTOM_ENDPOINT
+ )
return DatasetFileDocument("/datasets/bob@x.com/ds/v1/file.csv")
def test_returns_bytesio_with_downloaded_content(self, monkeypatch):
@@ -278,7 +288,9 @@ class TestReadFile:
class TestTimeoutsAndRetries:
def _make_doc(self, monkeypatch):
monkeypatch.setenv("USER_JWT_TOKEN", "test-jwt-token")
- monkeypatch.setenv("FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT", CUSTOM_ENDPOINT)
+ monkeypatch.setenv(
+ "FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT", CUSTOM_ENDPOINT
+ )
return DatasetFileDocument("/datasets/bob@x.com/ds/v1/file.csv")
def test_presigned_url_request_passes_request_timeout(self, monkeypatch):
diff --git a/bin/k8s/templates/base/workflow-compiling-service/workflow-compiling-service-deployment.yaml b/bin/k8s/templates/base/workflow-compiling-service/workflow-compiling-service-deployment.yaml
index 50a0a04e1b..2061e9e53e 100644
--- a/bin/k8s/templates/base/workflow-compiling-service/workflow-compiling-service-deployment.yaml
+++ b/bin/k8s/templates/base/workflow-compiling-service/workflow-compiling-service-deployment.yaml
@@ -40,7 +40,7 @@ spec:
- containerPort: {{ .Values.workflowCompilingService.service.port }}
env:
# FileService Access
- - name: FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT
+ - name: FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT
value: http://{{ .Values.fileService.name }}-svc:9092/api/dataset/presign-download
# LakeFS Access
- name: STORAGE_LAKEFS_ENDPOINT
diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml
index a911845041..ea61b242d1 100644
--- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml
+++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml
@@ -75,7 +75,7 @@ spec:
name: {{ .Release.Name }}-postgresql
key: postgres-password
# FileService Access
- - name: FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT
+ - name: FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT
value: http://{{ .Values.fileService.name }}-svc:9092/api/dataset/presign-download
- name: FILE_SERVICE_UPLOAD_ONE_FILE_TO_DATASET_ENDPOINT
value: http://{{ .Values.fileService.name }}-svc:9092/api/dataset/did/upload
diff --git a/bin/local-dev/main.sh b/bin/local-dev/main.sh
index 7ef7a20cb6..08e8cd2495 100755
--- a/bin/local-dev/main.sh
+++ b/bin/local-dev/main.sh
@@ -707,7 +707,7 @@ export STORAGE_LAKEFS_AUTH_API_SECRET="${STORAGE_LAKEFS_AUTH_API_SECRET:-random_
export TEXERA_DASHBOARD_SERVICE_ENDPOINT="${TEXERA_DASHBOARD_SERVICE_ENDPOINT:-http://localhost:8080}"
export WORKFLOW_COMPILING_SERVICE_ENDPOINT="${WORKFLOW_COMPILING_SERVICE_ENDPOINT:-http://localhost:9090}"
export WORKFLOW_EXECUTION_SERVICE_ENDPOINT="${WORKFLOW_EXECUTION_SERVICE_ENDPOINT:-http://localhost:8085}"
-export FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT="${FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT:-http://localhost:9092/api/dataset/presign-download}"
+export FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT="${FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT:-http://localhost:9092/api/dataset/presign-download}"
export FILE_SERVICE_UPLOAD_ONE_FILE_TO_DATASET_ENDPOINT="${FILE_SERVICE_UPLOAD_ONE_FILE_TO_DATASET_ENDPOINT:-http://localhost:9092/api/dataset/did/upload}"
export LITELLM_BASE_URL="${LITELLM_BASE_URL:-http://localhost:4000}"
export LITELLM_MASTER_KEY="${LITELLM_MASTER_KEY:-sk-texera-internal-do-not-share}"
diff --git a/bin/single-node/.env b/bin/single-node/.env
index 555e14db7d..dc8877275d 100644
--- a/bin/single-node/.env
+++ b/bin/single-node/.env
@@ -81,7 +81,7 @@ STORAGE_ICEBERG_CATALOG_POSTGRES_USERNAME=texera
STORAGE_ICEBERG_CATALOG_POSTGRES_PASSWORD=password
# File service endpoints
-FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT=http://file-service:9092/api/dataset/presign-download
+FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT=http://file-service:9092/api/dataset/presign-download
FILE_SERVICE_UPLOAD_ONE_FILE_TO_DATASET_ENDPOINT=http://file-service:9092/api/dataset/did/upload
# Toggles the texera agent panel; set false to hide it in the GUI.
diff --git a/common/config/src/main/scala/org/apache/texera/common/config/EnvironmentalVariable.scala b/common/config/src/main/scala/org/apache/texera/common/config/EnvironmentalVariable.scala
index a335ddeff6..b0cc3028f9 100644
--- a/common/config/src/main/scala/org/apache/texera/common/config/EnvironmentalVariable.scala
+++ b/common/config/src/main/scala/org/apache/texera/common/config/EnvironmentalVariable.scala
@@ -35,7 +35,10 @@ object EnvironmentalVariable {
/**
* FileService related endpoint
*/
- val ENV_FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT = "FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT"
+ val ENV_FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT =
+ "FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT"
+ val ENV_FILE_SERVICE_GET_MODEL_PRESIGNED_URL_ENDPOINT =
+ "FILE_SERVICE_GET_MODEL_PRESIGNED_URL_ENDPOINT"
val ENV_FILE_SERVICE_UPLOAD_ONE_FILE_TO_DATASET_ENDPOINT =
"FILE_SERVICE_UPLOAD_ONE_FILE_TO_DATASET_ENDPOINT"
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala
index 16fa5c07cb..dfc63dc6a0 100644
--- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/DocumentFactory.scala
@@ -20,7 +20,10 @@
package org.apache.texera.amber.core.storage
import org.apache.texera.common.config.StorageConfig
-import org.apache.texera.amber.core.storage.FileResolver.DATASET_FILE_URI_SCHEME
+import org.apache.texera.amber.core.storage.FileResolver.{
+ DATASET_FILE_URI_SCHEME,
+ MODEL_FILE_URI_SCHEME
+}
import org.apache.texera.amber.core.storage.VFSResourceType._
import org.apache.texera.amber.core.storage.VFSURIFactory.{VFS_FILE_URI_SCHEME, decodeURI}
import org.apache.texera.amber.core.storage.model._
@@ -58,6 +61,7 @@ object DocumentFactory {
def openReadonlyDocument(fileUri: URI): ReadonlyVirtualDocument[_] = {
fileUri.getScheme match {
case DATASET_FILE_URI_SCHEME => new DatasetFileDocument(fileUri)
+ case MODEL_FILE_URI_SCHEME => new ModelFileDocument(fileUri)
case "file" => new ReadonlyLocalFileDocument(fileUri)
case unsupportedScheme =>
throw new UnsupportedOperationException(
@@ -181,6 +185,7 @@ object DocumentFactory {
def openDocument(uri: URI): (VirtualDocument[_], Option[Schema]) = {
uri.getScheme match {
case DATASET_FILE_URI_SCHEME => (new DatasetFileDocument(uri), None)
+ case MODEL_FILE_URI_SCHEME => (new ModelFileDocument(uri), None)
case VFS_FILE_URI_SCHEME =>
val IcebergLocation(warehouse, namespace, storageKey) = resolveIcebergLocation(uri)
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/FileResolver.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/FileResolver.scala
index ac7f8d71f6..1a2e4fcf2a 100644
--- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/FileResolver.scala
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/FileResolver.scala
@@ -24,8 +24,15 @@ import org.apache.texera.dao.SqlServer
import org.apache.texera.dao.SqlServer.withTransaction
import org.apache.texera.dao.jooq.generated.tables.Dataset.DATASET
import org.apache.texera.dao.jooq.generated.tables.DatasetVersion.DATASET_VERSION
+import org.apache.texera.dao.jooq.generated.tables.Model.MODEL
+import org.apache.texera.dao.jooq.generated.tables.ModelVersion.MODEL_VERSION
import org.apache.texera.dao.jooq.generated.tables.User.USER
-import org.apache.texera.dao.jooq.generated.tables.pojos.{Dataset, DatasetVersion}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{
+ Dataset,
+ DatasetVersion,
+ Model,
+ ModelVersion
+}
import java.net.{URI, URLEncoder}
import java.nio.charset.StandardCharsets
@@ -34,11 +41,13 @@ import scala.jdk.CollectionConverters.IteratorHasAsScala
import scala.util.{Success, Try}
/**
- * Unified object for resolving both VFS resources and local/dataset files.
+ * Unified object for resolving local files and versioned-resource (dataset / model) logical
+ * paths to physical URIs.
*/
object FileResolver {
val DATASET_FILE_URI_SCHEME = "dataset"
+ val MODEL_FILE_URI_SCHEME = "model"
/**
* Resolves a given fileName to either a file on the local file system or a dataset file.
@@ -51,7 +60,7 @@ object FileResolver {
if (isFileResolved(fileName)) {
return new URI(fileName)
}
- val resolvers: Seq[String => URI] = Seq(localResolveFunc, datasetResolveFunc)
+ val resolvers: Seq[String => URI] = Seq(localResolveFunc, versionedResourceResolveFunc)
// Try each resolver function in sequence
resolvers
@@ -76,106 +85,203 @@ object FileResolver {
}
/**
- * Parses a dataset logical path into its components, or None if it is not a well-formed dataset path.
- * Expected format: /datasets/ownerEmail/datasetName/versionName/fileRelativePath
- * The legacy unprefixed format is also accepted.
+ * Parses a versioned-resource file path into its components, or None if it is not well-formed.
+ *
+ * Two accepted forms:
+ * - Prefixed: /<prefix>/ownerEmail/resourceName/versionName/fileRelativePath (>= 5 segments)
+ * where the leading segment names a known [[ResourceType]] (dataset, model, …), so a single
+ * caller can dispatch to the right backing table.
+ * - Legacy unprefixed (datasets only, backward compat): /ownerEmail/datasetName/versionName/
+ * fileRelativePath (>= 4 segments), resolved as a dataset. Models are new and always require
+ * the /models/ prefix.
*
- * @param fileName The file path to parse
- * @return Some((ownerEmail, datasetName, versionName, fileRelativePath)) if valid, None otherwise
+ * @param fileName the file path to parse
+ * @return Some((resourceType, ownerEmail, resourceName, versionName, fileRelativePath)) if valid,
+ * None otherwise
*/
- private def parseDatasetFilePath(
+ private def parsePrefixedPath(
fileName: String
- ): Option[(String, String, String, Array[String])] = {
+ ): Option[(ResourceType.Value, String, String, String, Array[String])] = {
val filePath = Paths.get(fileName)
val pathSegments = (0 until filePath.getNameCount).map(filePath.getName(_).toString).toArray
- // TODO(datasets-prefix): require the prefix once all stored paths are migrated (36.sql).
- if (pathSegments.headOption.exists(ResourceType.isValidPrefix)) {
- if (pathSegments.length < 5) None
- else Some((pathSegments(1), pathSegments(2), pathSegments(3), pathSegments.drop(4)))
- } else if (pathSegments.length >= 4) {
- Some((pathSegments(0), pathSegments(1), pathSegments(2), pathSegments.drop(3)))
- } else None
+ pathSegments.headOption.flatMap(ResourceType.fromPrefix) match {
+ case Some(resourceType) =>
+ // Prefixed: /<prefix>/ownerEmail/resourceName/versionName/<file> (>= 5 segments).
+ if (pathSegments.length < 5) None
+ else
+ Some(
+ (resourceType, pathSegments(1), pathSegments(2), pathSegments(3), pathSegments.drop(4))
+ )
+ case None =>
+ // Legacy unprefixed dataset path (backward compat): /ownerEmail/datasetName/versionName/<file>.
+ // TODO(datasets-prefix): require the prefix once all stored paths are migrated (36.sql).
+ if (pathSegments.length >= 4)
+ Some(
+ (
+ ResourceType.Datasets,
+ pathSegments(0),
+ pathSegments(1),
+ pathSegments(2),
+ pathSegments.drop(3)
+ )
+ )
+ else None
+ }
}
/**
- * Attempts to resolve a given fileName to a URI.
+ * Resolves a versioned-resource logical path to its physical `scheme:///` URI.
*
- * The fileName format should be: /datasets/ownerEmail/datasetName/versionName/fileRelativePath
- * e.g. /datasets/bob@texera.com/twitterDataset/v1/california/irvine/tw1.csv
- * The output dataset URI format is: {DATASET_FILE_URI_SCHEME}:///{repositoryName}/{versionHash}/fileRelativePath
- * e.g. {DATASET_FILE_URI_SCHEME}:///dataset-15/adeq233td/some/dir/file.txt
+ * The leading prefix selects the resource kind; only the per-type DB lookup and URI scheme
+ * differ, so adding a new versioned resource is a single new `case` below plus a lookup helper.
+ * An unprefixed path is resolved as a legacy dataset path (see [[parsePrefixedPath]]).
*
- * @param fileName the name of the file to attempt resolving as a DatasetFileDocument
- * @return Either[String, DatasetFileDocument] - Right(document) if creation succeeds
- * @throws java.io.FileNotFoundException if the dataset file does not exist or cannot be created
+ * Input: /<prefix>/ownerEmail/resourceName/versionName/fileRelativePath (or legacy unprefixed)
+ * Output: {scheme}:///{repositoryName}/{versionHash}/fileRelativePath
+ * e.g. /datasets/bob@x.com/twitter/v1/dir/f.csv -> dataset:///dataset-15/adeq233td/dir/f.csv
+ * /models/bob@x.com/resnet/v1/weights/m.pt -> model:///model-15/adeq233td/weights/m.pt
+ *
+ * @throws java.io.FileNotFoundException if the path is not a valid versioned-resource path, the
+ * resource/version does not exist, or the URI is malformed
*/
- private def datasetResolveFunc(fileName: String): URI = {
- val (ownerEmail, datasetName, versionName, fileRelativePathSegments) =
- parseDatasetFilePath(fileName).getOrElse(
- throw new FileNotFoundException(s"Dataset file $fileName not found.")
+ private def versionedResourceResolveFunc(fileName: String): URI = {
+ val (resourceType, ownerEmail, resourceName, versionName, fileRelativePathSegments) =
+ parsePrefixedPath(fileName).getOrElse(
+ throw new FileNotFoundException(s"Versioned-resource file $fileName not found.")
)
+ val (scheme, repositoryName, versionHash) = resourceType match {
+ case ResourceType.Datasets =>
+ val (repo, hash) = lookupDataset(ownerEmail, resourceName, versionName, fileName)
+ (DATASET_FILE_URI_SCHEME, repo, hash)
+ case ResourceType.Models =>
+ val (repo, hash) = lookupModel(ownerEmail, resourceName, versionName, fileName)
+ (MODEL_FILE_URI_SCHEME, repo, hash)
+ case other =>
+ throw new FileNotFoundException(s"Unsupported resource type $other for file $fileName.")
+ }
+
+ buildVersionedFileURI(scheme, repositoryName, versionHash, fileRelativePathSegments, fileName)
+ }
+
+ /**
+ * Builds the physical URI {scheme}:///{repositoryName}/{versionHash}/{fileRelativePath},
+ * URL-encoding each file-relative-path segment. Uses forward slash on both Linux and Windows.
+ */
+ private def buildVersionedFileURI(
+ scheme: String,
+ repositoryName: String,
+ versionHash: String,
+ fileRelativePathSegments: Array[String],
+ fileName: String
+ ): URI = {
val fileRelativePath =
Paths.get(fileRelativePathSegments.head, fileRelativePathSegments.tail: _*)
- // fetch the dataset and version from DB to get dataset ID and version hash
- val (dataset, datasetVersion) =
- withTransaction(
- SqlServer
- .getInstance()
- .createDSLContext()
- ) { ctx =>
- // fetch the dataset from DB
- val dataset = ctx
- .select(DATASET.fields: _*)
- .from(DATASET)
- .leftJoin(USER)
- .on(USER.UID.eq(DATASET.OWNER_UID))
- .where(USER.EMAIL.eq(ownerEmail))
- .and(DATASET.NAME.eq(datasetName))
- .fetchOneInto(classOf[Dataset])
-
- // fetch the dataset version from DB
- val datasetVersion = ctx
- .selectFrom(DATASET_VERSION)
- .where(DATASET_VERSION.DID.eq(dataset.getDid))
- .and(DATASET_VERSION.NAME.eq(versionName))
- .fetchOneInto(classOf[DatasetVersion])
-
- if (dataset == null || datasetVersion == null) {
- throw new FileNotFoundException(s"Dataset file $fileName not found.")
- }
- (dataset, datasetVersion)
- }
-
- // Convert each segment of fileRelativePath to an encoded String
val encodedFileRelativePath = fileRelativePath
.iterator()
.asScala
- .map { segment =>
- URLEncoder.encode(segment.toString, StandardCharsets.UTF_8)
- }
+ .map(segment => URLEncoder.encode(segment.toString, StandardCharsets.UTF_8))
.toArray
- // Prepend dataset name and versionHash to the encoded path segments
- val allPathSegments = Array(
- dataset.getRepositoryName,
- datasetVersion.getVersionHash
- ) ++ encodedFileRelativePath
-
- // Build the format /{repositoryName}/{versionHash}/{fileRelativePath}, both Linux and Windows use forward slash as the splitter
- val uriSplitter = "/"
- val encodedPath = uriSplitter + allPathSegments.mkString(uriSplitter)
+ val allPathSegments = Array(repositoryName, versionHash) ++ encodedFileRelativePath
+ val encodedPath = "/" + allPathSegments.mkString("/")
try {
- new URI(DATASET_FILE_URI_SCHEME, "", encodedPath, null)
+ new URI(scheme, "", encodedPath, null)
} catch {
- case e: Exception =>
- throw new FileNotFoundException(s"Dataset file $fileName not found.")
+ case _: Exception =>
+ throw new FileNotFoundException(s"Versioned-resource file $fileName not found.")
}
}
+ /**
+ * Looks up a dataset + version by owner email / name / version name, returning its
+ * (repositoryName, versionHash).
+ *
+ * @throws java.io.FileNotFoundException if the dataset or version does not exist
+ */
+ private def lookupDataset(
+ ownerEmail: String,
+ datasetName: String,
+ versionName: String,
+ fileName: String
+ ): (String, String) =
+ withTransaction(
+ SqlServer
+ .getInstance()
+ .createDSLContext()
+ ) { ctx =>
+ val dataset = ctx
+ .select(DATASET.fields: _*)
+ .from(DATASET)
+ .leftJoin(USER)
+ .on(USER.UID.eq(DATASET.OWNER_UID))
+ .where(USER.EMAIL.eq(ownerEmail))
+ .and(DATASET.NAME.eq(datasetName))
+ .fetchOneInto(classOf[Dataset])
+
+ // fail early if the dataset does not exist (before dereferencing it below)
+ if (dataset == null) {
+ throw new FileNotFoundException(s"Dataset file $fileName not found.")
+ }
+
+ val datasetVersion = ctx
+ .selectFrom(DATASET_VERSION)
+ .where(DATASET_VERSION.DID.eq(dataset.getDid))
+ .and(DATASET_VERSION.NAME.eq(versionName))
+ .fetchOneInto(classOf[DatasetVersion])
+
+ if (datasetVersion == null) {
+ throw new FileNotFoundException(s"Dataset file $fileName not found.")
+ }
+ (dataset.getRepositoryName, datasetVersion.getVersionHash)
+ }
+
+ /**
+ * Looks up a model + version by owner email / name / version name, returning its
+ * (repositoryName, versionHash). Mirrors [[lookupDataset]] against the model tables.
+ *
+ * @throws java.io.FileNotFoundException if the model or version does not exist
+ */
+ private def lookupModel(
+ ownerEmail: String,
+ modelName: String,
+ versionName: String,
+ fileName: String
+ ): (String, String) =
+ withTransaction(
+ SqlServer
+ .getInstance()
+ .createDSLContext()
+ ) { ctx =>
+ val model = ctx
+ .select(MODEL.fields: _*)
+ .from(MODEL)
+ .leftJoin(USER)
+ .on(USER.UID.eq(MODEL.OWNER_UID))
+ .where(USER.EMAIL.eq(ownerEmail))
+ .and(MODEL.NAME.eq(modelName))
+ .fetchOneInto(classOf[Model])
+
+ // fail early if the model does not exist (before dereferencing it below)
+ if (model == null) {
+ throw new FileNotFoundException(s"Model file $fileName not found.")
+ }
+
+ val modelVersion = ctx
+ .selectFrom(MODEL_VERSION)
+ .where(MODEL_VERSION.MID.eq(model.getMid))
+ .and(MODEL_VERSION.NAME.eq(versionName))
+ .fetchOneInto(classOf[ModelVersion])
+
+ if (modelVersion == null) {
+ throw new FileNotFoundException(s"Model file $fileName not found.")
+ }
+ (model.getRepositoryName, modelVersion.getVersionHash)
+ }
+
/**
* Checks if a given file path has a valid scheme.
*
@@ -199,8 +305,8 @@ object FileResolver {
if (path == null) {
return None
}
- parseDatasetFilePath(path).map {
- case (ownerEmail, datasetName, _, _) => (ownerEmail, datasetName)
+ parsePrefixedPath(path).collect {
+ case (ResourceType.Datasets, ownerEmail, datasetName, _, _) => (ownerEmail, datasetName)
}
}
}
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/ResourceType.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/ResourceType.scala
index cb3db537d5..087dc6ef87 100644
--- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/ResourceType.scala
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/ResourceType.scala
@@ -27,6 +27,17 @@ package org.apache.texera.amber.core.storage
*/
object ResourceType extends Enumeration {
val Datasets: Value = Value("datasets")
+ val Models: Value = Value("models")
- def isValidPrefix(segment: String): Boolean = values.exists(_.toString == segment)
+ /**
+ * Returns the resource type named by the given path segment, or None if it is not a known
+ * resource type.
+ */
+ def fromPrefix(segment: String): Option[Value] = values.find(_.toString == segment)
+
+ /**
+ * Returns true if the given path segment names a known resource type.
+ * Used to validate the leading prefix of a logical path.
+ */
+ def isValidPrefix(segment: String): Boolean = fromPrefix(segment).isDefined
}
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocument.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocument.scala
index 6d8f917c7f..f2862c87ee 100644
--- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocument.scala
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocument.scala
@@ -19,155 +19,31 @@
package org.apache.texera.amber.core.storage.model
-import com.typesafe.scalalogging.LazyLogging
import org.apache.texera.common.config.EnvironmentalVariable
-import org.apache.texera.amber.core.storage.model.DatasetFileDocument.{
- fileServiceGetPresignURLEndpoint,
- userJwtToken
-}
-import org.apache.texera.amber.core.storage.util.LakeFSStorageClient
+import org.apache.texera.amber.core.storage.model.DatasetFileDocument.fileServiceGetPresignURLEndpoint
import org.apache.texera.amber.core.storage.util.dataset.GitVersionControlLocalFileStorage
-import java.io.{File, FileOutputStream, InputStream}
-import java.net._
-import java.nio.charset.StandardCharsets
-import java.nio.file.{Files, Path, Paths}
-import scala.jdk.CollectionConverters.IteratorHasAsScala
+import java.net.URI
+import java.nio.file.Path
object DatasetFileDocument {
- // Since requests need to be sent to the FileService in order to read the file, we store USER_JWT_TOKEN in the environment vars
- // This variable should be NON-EMPTY in the dynamic-computing-unit architecture, i.e. each user-created computing unit should store user's jwt token.
- // In the local development or other architectures, this token can be empty.
- lazy val userJwtToken: String =
- sys.env.getOrElse(EnvironmentalVariable.ENV_USER_JWT_TOKEN, "").trim
-
// The endpoint of getting presigned url from the file service, also stored in the environment vars.
lazy val fileServiceGetPresignURLEndpoint: String =
sys.env
.getOrElse(
- EnvironmentalVariable.ENV_FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT,
+ EnvironmentalVariable.ENV_FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT,
"http://localhost:9092/api/dataset/presign-download"
)
.trim
}
private[storage] class DatasetFileDocument(uri: URI)
- extends VirtualDocument[Nothing]
- with OnDataset
- with LazyLogging {
- // Utility function to parse and decode URI segments into individual components
- private def parseUri(uri: URI): (String, String, Path) = {
- val segments = Paths.get(uri.getPath).iterator().asScala.map(_.toString).toArray
- if (segments.length < 3)
- throw new IllegalArgumentException("URI format is incorrect")
-
- // parse uri to dataset components
- val repositoryName = segments(0)
- val datasetVersionHash = URLDecoder.decode(segments(1), StandardCharsets.UTF_8)
- val decodedRelativeSegments =
- segments.drop(2).map(part => URLDecoder.decode(part, StandardCharsets.UTF_8))
- val fileRelativePath = Paths.get(decodedRelativeSegments.head, decodedRelativeSegments.tail: _*)
-
- (repositoryName, datasetVersionHash, fileRelativePath)
- }
-
- // Extract components from URI using the utility function
- private val (repositoryName, datasetVersionHash, fileRelativePath) = parseUri(uri)
-
- private var tempFile: Option[File] = None
-
- override def getURI: URI = uri
-
- override def asInputStream(): InputStream = {
-
- def fallbackToLakeFS(exception: Throwable): InputStream = {
- logger.warn(s"${exception.getMessage}. Falling back to LakeFS direct file fetch.", exception)
- val file = LakeFSStorageClient.getFileFromRepo(
- getRepositoryName(),
- getVersionHash(),
- getFileRelativePath()
- )
- Files.newInputStream(file.toPath)
- }
-
- if (userJwtToken.isEmpty) {
- try {
- val presignUrl = LakeFSStorageClient.getFilePresignedUrl(
- getRepositoryName(),
- getVersionHash(),
- getFileRelativePath()
- )
- new URL(presignUrl).openStream()
- } catch {
- case e: Exception =>
- fallbackToLakeFS(e)
- }
- } else {
- val presignRequestUrl =
- s"$fileServiceGetPresignURLEndpoint?repositoryName=${getRepositoryName()}&commitHash=${getVersionHash()}&filePath=${URLEncoder
- .encode(getFileRelativePath(), StandardCharsets.UTF_8.name())}"
-
- val connection = new URL(presignRequestUrl).openConnection().asInstanceOf[HttpURLConnection]
- connection.setRequestMethod("GET")
- connection.setRequestProperty("Authorization", s"Bearer $userJwtToken")
-
- try {
- if (connection.getResponseCode != HttpURLConnection.HTTP_OK) {
- throw new RuntimeException(
- s"Failed to retrieve presigned URL: HTTP ${connection.getResponseCode}"
- )
- }
-
- // Read response body as a string
- val responseBody =
- new String(connection.getInputStream.readAllBytes(), StandardCharsets.UTF_8)
-
- // Extract presigned URL from JSON response
- val presignedUrl = responseBody
- .split("\"presignedUrl\"\\s*:\\s*\"")(1)
- .split("\"")(0)
-
- new URL(presignedUrl).openStream()
- } catch {
- case e: Exception =>
- fallbackToLakeFS(e)
- } finally {
- connection.disconnect()
- }
- }
- }
-
- override def asFile(): File = {
- tempFile match {
- case Some(file) => file
- case None =>
- val tempFilePath = Files.createTempFile("versionedFile", ".tmp")
- val tempFileStream = new FileOutputStream(tempFilePath.toFile)
- val inputStream = asInputStream()
-
- val buffer = new Array[Byte](1024)
-
- // Create an iterator to repeatedly call inputStream.read, and direct buffered data to file
- Iterator
- .continually(inputStream.read(buffer))
- .takeWhile(_ != -1)
- .foreach(tempFileStream.write(buffer, 0, _))
-
- inputStream.close()
- tempFileStream.close()
-
- val file = tempFilePath.toFile
- tempFile = Some(file)
- file
- }
- }
+ extends LakeFSFileDocument(uri, fileServiceGetPresignURLEndpoint) {
override def clear(): Unit = {
- // first remove the temporary file
- tempFile match {
- case Some(file) => Files.delete(file.toPath)
- case None => // Do nothing
- }
+ // first remove the temporary file (handled by the shared base)
+ super.clear()
+
lazy val datasetsRootPath =
Path
.of(sys.env.getOrElse("TEXERA_HOME", "."))
@@ -179,16 +55,10 @@ private[storage] class DatasetFileDocument(uri: URI)
datasetsRootPath.resolve(did.toString)
}
- // then remove the dataset file
+ // then remove the dataset file from the local git-backed storage
GitVersionControlLocalFileStorage.removeFileFromRepo(
getDatasetPath(0),
getDatasetPath(0).resolve(fileRelativePath)
)
}
-
- override def getRepositoryName(): String = repositoryName
-
- override def getVersionHash(): String = datasetVersionHash
-
- override def getFileRelativePath(): String = fileRelativePath.toString
}
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocument.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocument.scala
similarity index 76%
copy from common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocument.scala
copy to common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocument.scala
index 6d8f917c7f..044437889e 100644
--- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocument.scala
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocument.scala
@@ -21,12 +21,8 @@ package org.apache.texera.amber.core.storage.model
import com.typesafe.scalalogging.LazyLogging
import org.apache.texera.common.config.EnvironmentalVariable
-import org.apache.texera.amber.core.storage.model.DatasetFileDocument.{
- fileServiceGetPresignURLEndpoint,
- userJwtToken
-}
+import org.apache.texera.amber.core.storage.model.LakeFSFileDocument.userJwtToken
import org.apache.texera.amber.core.storage.util.LakeFSStorageClient
-import org.apache.texera.amber.core.storage.util.dataset.GitVersionControlLocalFileStorage
import java.io.{File, FileOutputStream, InputStream}
import java.net._
@@ -34,26 +30,26 @@ import java.nio.charset.StandardCharsets
import java.nio.file.{Files, Path, Paths}
import scala.jdk.CollectionConverters.IteratorHasAsScala
-object DatasetFileDocument {
+object LakeFSFileDocument {
// Since requests need to be sent to the FileService in order to read the file, we store USER_JWT_TOKEN in the environment vars
// This variable should be NON-EMPTY in the dynamic-computing-unit architecture, i.e. each user-created computing unit should store user's jwt token.
// In the local development or other architectures, this token can be empty.
lazy val userJwtToken: String =
sys.env.getOrElse(EnvironmentalVariable.ENV_USER_JWT_TOKEN, "").trim
-
- // The endpoint of getting presigned url from the file service, also stored in the environment vars.
- lazy val fileServiceGetPresignURLEndpoint: String =
- sys.env
- .getOrElse(
- EnvironmentalVariable.ENV_FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT,
- "http://localhost:9092/api/dataset/presign-download"
- )
- .trim
}
-private[storage] class DatasetFileDocument(uri: URI)
+/**
+ * A read-only document over a single file stored in a LakeFS repository, addressed by the URI
+ * {scheme}:///{repositoryName}/{versionHash}/{fileRelativePath}. This is the shared behavior
+ * for every versioned-file resource (datasets, models, …): the file bytes are fetched via a
+ * presigned URL, falling back to a direct LakeFS fetch.
+ *
+ * @param uri the resolved {scheme}:///{repositoryName}/{versionHash}/{file} URI
+ * @param presignEndpoint the file-service presign-download endpoint for this resource kind
+ */
+private[storage] abstract class LakeFSFileDocument(uri: URI, presignEndpoint: String)
extends VirtualDocument[Nothing]
- with OnDataset
+ with OnVersionedFileResource
with LazyLogging {
// Utility function to parse and decode URI segments into individual components
private def parseUri(uri: URI): (String, String, Path) = {
@@ -61,20 +57,20 @@ private[storage] class DatasetFileDocument(uri: URI)
if (segments.length < 3)
throw new IllegalArgumentException("URI format is incorrect")
- // parse uri to dataset components
+ // parse uri to (repositoryName, versionHash, fileRelativePath)
val repositoryName = segments(0)
- val datasetVersionHash = URLDecoder.decode(segments(1), StandardCharsets.UTF_8)
+ val versionHash = URLDecoder.decode(segments(1), StandardCharsets.UTF_8)
val decodedRelativeSegments =
segments.drop(2).map(part => URLDecoder.decode(part, StandardCharsets.UTF_8))
val fileRelativePath = Paths.get(decodedRelativeSegments.head, decodedRelativeSegments.tail: _*)
- (repositoryName, datasetVersionHash, fileRelativePath)
+ (repositoryName, versionHash, fileRelativePath)
}
// Extract components from URI using the utility function
- private val (repositoryName, datasetVersionHash, fileRelativePath) = parseUri(uri)
+ protected val (repositoryName, versionHash, fileRelativePath) = parseUri(uri)
- private var tempFile: Option[File] = None
+ protected var tempFile: Option[File] = None
override def getURI: URI = uri
@@ -104,7 +100,7 @@ private[storage] class DatasetFileDocument(uri: URI)
}
} else {
val presignRequestUrl =
- s"$fileServiceGetPresignURLEndpoint?repositoryName=${getRepositoryName()}&commitHash=${getVersionHash()}&filePath=${URLEncoder
+ s"$presignEndpoint?repositoryName=${getRepositoryName()}&commitHash=${getVersionHash()}&filePath=${URLEncoder
.encode(getFileRelativePath(), StandardCharsets.UTF_8.name())}"
val connection = new URL(presignRequestUrl).openConnection().asInstanceOf[HttpURLConnection]
@@ -163,32 +159,17 @@ private[storage] class DatasetFileDocument(uri: URI)
}
override def clear(): Unit = {
- // first remove the temporary file
+ // remove the temporary file, if one was materialized
tempFile match {
case Some(file) => Files.delete(file.toPath)
case None => // Do nothing
}
- lazy val datasetsRootPath =
- Path
- .of(sys.env.getOrElse("TEXERA_HOME", "."))
- .resolve("amber")
- .resolve("user-resources")
- .resolve("datasets")
-
- def getDatasetPath(did: Integer): Path = {
- datasetsRootPath.resolve(did.toString)
- }
-
- // then remove the dataset file
- GitVersionControlLocalFileStorage.removeFileFromRepo(
- getDatasetPath(0),
- getDatasetPath(0).resolve(fileRelativePath)
- )
+ tempFile = None
}
override def getRepositoryName(): String = repositoryName
- override def getVersionHash(): String = datasetVersionHash
+ override def getVersionHash(): String = versionHash
override def getFileRelativePath(): String = fileRelativePath.toString
}
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/OnDataset.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/ModelFileDocument.scala
similarity index 50%
copy from common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/OnDataset.scala
copy to common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/ModelFileDocument.scala
index 6c19ee1002..e67ec92992 100644
--- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/OnDataset.scala
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/ModelFileDocument.scala
@@ -19,10 +19,25 @@
package org.apache.texera.amber.core.storage.model
-trait OnDataset {
- def getRepositoryName(): String
+import org.apache.texera.common.config.EnvironmentalVariable
+import org.apache.texera.amber.core.storage.model.ModelFileDocument.fileServiceGetModelPresignURLEndpoint
- def getVersionHash(): String
+import java.net.URI
- def getFileRelativePath(): String
+object ModelFileDocument {
+ // The endpoint of getting a presigned url for a model file from the file service.
+ lazy val fileServiceGetModelPresignURLEndpoint: String =
+ sys.env
+ .getOrElse(
+ EnvironmentalVariable.ENV_FILE_SERVICE_GET_MODEL_PRESIGNED_URL_ENDPOINT,
+ "http://localhost:9092/api/model/presign-download"
+ )
+ .trim
}
+
+/**
+ * A read-only document over a single file in a model's LakeFS repository (`model-{mid}`),
+ * addressed by a `model:///{repositoryName}/{versionHash}/{fileRelativePath}` URI.
+ */
+private[storage] class ModelFileDocument(uri: URI)
+ extends LakeFSFileDocument(uri, fileServiceGetModelPresignURLEndpoint)
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/OnDataset.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/OnVersionedFileResource.scala
similarity index 90%
rename from common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/OnDataset.scala
rename to common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/OnVersionedFileResource.scala
index 6c19ee1002..cbb76ee4f5 100644
--- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/OnDataset.scala
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/OnVersionedFileResource.scala
@@ -19,7 +19,10 @@
package org.apache.texera.amber.core.storage.model
-trait OnDataset {
+/**
+ * A document backed by a versioned file in a LakeFS repository.
+ */
+trait OnVersionedFileResource {
def getRepositoryName(): String
def getVersionHash(): String
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala
index ecb5f5c54f..eb18e87762 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/DocumentFactorySpec.scala
@@ -21,6 +21,8 @@ package org.apache.texera.amber.core.storage
import org.apache.texera.amber.core.storage.model.{
DatasetFileDocument,
+ ModelFileDocument,
+ OnVersionedFileResource,
ReadonlyLocalFileDocument,
VirtualDocument
}
@@ -38,7 +40,7 @@ import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
import java.net.URI
-import java.nio.file.Files
+import java.nio.file.{Files, Paths}
import java.util.UUID
/**
@@ -139,6 +141,18 @@ class DocumentFactorySpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
doc.getURI shouldBe datasetUri
}
+ it should "return a ModelFileDocument for the model scheme and parse its URI components" in {
+ val modelUri = new URI(s"model:///model-1/$versionHash/weights/model.pt")
+ val doc = DocumentFactory.openReadonlyDocument(modelUri)
+ doc shouldBe a[ModelFileDocument]
+ doc.getURI shouldBe modelUri
+
+ val resource = doc.asInstanceOf[OnVersionedFileResource]
+ resource.getRepositoryName() shouldBe "model-1"
+ resource.getVersionHash() shouldBe versionHash
+ resource.getFileRelativePath() shouldBe Paths.get("weights", "model.pt").toString
+ }
+
it should "return a ReadonlyLocalFileDocument for the file scheme" in {
val tempFile = Files.createTempFile("doc-factory-spec", ".txt")
try {
@@ -169,6 +183,13 @@ class DocumentFactorySpec extends AnyFlatSpec with Matchers with BeforeAndAfterA
schemaOpt shouldBe None
}
+ it should "return a ModelFileDocument and no schema for the model scheme" in {
+ val modelUri = new URI(s"model:///model-1/$versionHash/weights/model.pt")
+ val (doc, schemaOpt) = DocumentFactory.openDocument(modelUri)
+ doc shouldBe a[ModelFileDocument]
+ schemaOpt shouldBe None
+ }
+
it should "reject an unsupported scheme" in {
val thrown = intercept[UnsupportedOperationException] {
DocumentFactory.openDocument(new URI("ftp://host/path"))
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocumentSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocumentSpec.scala
index c16d417a73..5a313c2763 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocumentSpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/storage/model/DatasetFileDocumentSpec.scala
@@ -142,26 +142,28 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers {
}
}
- // The companion object resolves the file-service endpoint and the user JWT
- // token from environment variables, falling back to a trimmed default. These
- // lazy vals are read whenever asInputStream needs to fetch a file; assert their
- // fallback behavior without requiring a live FileService or LakeFS. The checks
- // are guarded so they hold regardless of whether the env overrides are present.
+ // The companion object resolves the file-service endpoint from environment
+ // variables, falling back to a trimmed default. This lazy val is read whenever
+ // asInputStream needs to fetch a file; assert its fallback behavior without
+ // requiring a live FileService or LakeFS. The check is guarded so it holds
+ // regardless of whether the env override is present.
"DatasetFileDocument companion" should
"expose the default presigned-URL endpoint when the env override is absent" in {
val expected =
sys.env
.getOrElse(
- EnvironmentalVariable.ENV_FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT,
+ EnvironmentalVariable.ENV_FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT,
"http://localhost:9092/api/dataset/presign-download"
)
.trim
DatasetFileDocument.fileServiceGetPresignURLEndpoint shouldBe expected
}
- it should "expose a trimmed user JWT token defaulting to empty" in {
+ // The user JWT token is shared by every LakeFS-backed document, so it now lives on
+ // the LakeFSFileDocument base object rather than on DatasetFileDocument.
+ "LakeFSFileDocument companion" should "expose a trimmed user JWT token defaulting to empty" in {
val expected =
sys.env.getOrElse(EnvironmentalVariable.ENV_USER_JWT_TOKEN, "").trim
- DatasetFileDocument.userJwtToken shouldBe expected
+ LakeFSFileDocument.userJwtToken shouldBe expected
}
}
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/FileResolverSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/FileResolverSpec.scala
index e6533c7f1b..e414371222 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/FileResolverSpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/FileResolverSpec.scala
@@ -23,8 +23,20 @@ import org.apache.texera.amber.core.storage.FileResolver
import org.apache.commons.vfs2.FileNotFoundException
import org.apache.texera.dao.MockTexeraDB
import org.apache.texera.dao.jooq.generated.enums.UserRoleEnum
-import org.apache.texera.dao.jooq.generated.tables.daos.{DatasetDao, DatasetVersionDao, UserDao}
-import org.apache.texera.dao.jooq.generated.tables.pojos.{Dataset, DatasetVersion, User}
+import org.apache.texera.dao.jooq.generated.tables.daos.{
+ DatasetDao,
+ DatasetVersionDao,
+ ModelDao,
+ ModelVersionDao,
+ UserDao
+}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{
+ Dataset,
+ DatasetVersion,
+ Model,
+ ModelVersion,
+ User
+}
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach}
@@ -76,19 +88,61 @@ class FileResolverSpec
datasetVersion
}
+ private val testModel: Model = {
+ val model = new Model
+ model.setMid(Integer.valueOf(1))
+ model.setName("test_model")
+ model.setRepositoryName("model-1")
+ model.setDescription("model for test")
+ model.setIsPublic(true)
+ model.setIsDownloadable(true)
+ model.setOwnerUid(Integer.valueOf(1))
+ model
+ }
+
+ private val testModelVersion1: ModelVersion = {
+ val modelVersion = new ModelVersion
+ modelVersion.setMid(Integer.valueOf(1))
+ modelVersion.setName("v1")
+ modelVersion.setMvid(Integer.valueOf(1))
+ modelVersion.setCreatorUid(Integer.valueOf(1))
+ modelVersion.setVersionHash("a1b2c3d4e5f60718293a4b5c6d7e8f9001122334")
+ modelVersion
+ }
+
+ private val testModelVersion2: ModelVersion = {
+ val modelVersion = new ModelVersion
+ modelVersion.setMid(Integer.valueOf(1))
+ modelVersion.setName("v2")
+ modelVersion.setMvid(Integer.valueOf(2))
+ modelVersion.setCreatorUid(Integer.valueOf(1))
+ modelVersion.setVersionHash("998877665544332211ffeeddccbbaa0099887766")
+ modelVersion
+ }
+
private val localCsvFilePath = "common/workflow-core/src/test/resources/country_sales_small.csv"
private val datasetACsvFilePath = "/datasets/test_user@test.com/test_dataset/v2/directory/a.csv"
private val dataset1TxtFilePath = "/datasets/test_user@test.com/test_dataset/v1/1.txt"
- // Legacy unprefixed form.
+ private val modelWeightsFilePath = "/models/test_user@test.com/test_model/v2/weights/model.pt"
+
+ private val modelReadmeFilePath = "/models/test_user@test.com/test_model/v1/README.md"
+
+ // Legacy unprefixed form — still resolvable as a dataset (backward compat).
private val unprefixedDataset1TxtFilePath = "/test_user@test.com/test_dataset/v1/1.txt"
- // Leading segment names no known dataset owner.
+ // Leading segment names no known dataset owner, so this is not resolvable.
private val unknownResourceTypeFilePath =
"/notAResourceType/test_user@test.com/test_dataset/v1/1.txt"
+ // A model name presented under the datasets prefix must not resolve as a dataset.
+ private val modelNameUnderDatasetPrefix = "/datasets/test_user@test.com/test_model/v1/README.md"
+
+ // A dataset name presented under the models prefix must not resolve as a model.
+ private val datasetNameUnderModelPrefix = "/models/test_user@test.com/test_dataset/v1/1.txt"
+
override protected def beforeAll(): Unit = {
initializeDBAndReplaceDSLContext()
@@ -104,6 +158,15 @@ class FileResolverSpec
val datasetVersionDao = new DatasetVersionDao(getDSLContext.configuration())
datasetVersionDao.insert(testDatasetVersion1)
datasetVersionDao.insert(testDatasetVersion2)
+
+ // add test model
+ val modelDao = new ModelDao(getDSLContext.configuration())
+ modelDao.insert(testModel)
+
+ // add test model versions
+ val modelVersionDao = new ModelVersionDao(getDSLContext.configuration())
+ modelVersionDao.insert(testModelVersion1)
+ modelVersionDao.insert(testModelVersion2)
}
"FileResolver" should "resolve local file correctly" in {
@@ -124,6 +187,18 @@ class FileResolverSpec
)
}
+ "FileResolver" should "resolve model file correctly" in {
+ val modelWeightsUri = FileResolver.resolve(modelWeightsFilePath)
+ val modelReadmeUri = FileResolver.resolve(modelReadmeFilePath)
+
+ assert(
+ modelWeightsUri.toString == f"${FileResolver.MODEL_FILE_URI_SCHEME}:///${testModel.getRepositoryName}/${testModelVersion2.getVersionHash}/weights/model.pt"
+ )
+ assert(
+ modelReadmeUri.toString == f"${FileResolver.MODEL_FILE_URI_SCHEME}:///${testModel.getRepositoryName}/${testModelVersion1.getVersionHash}/README.md"
+ )
+ }
+
"FileResolver" should "still resolve a legacy unprefixed dataset path" in {
val dataset1TxtUri = FileResolver.resolve(unprefixedDataset1TxtFilePath)
@@ -138,6 +213,16 @@ class FileResolverSpec
}
}
+ "FileResolver" should "keep datasets and models isolated by prefix" in {
+ // a real dataset name under the models prefix is not a model, and vice-versa
+ assertThrows[FileNotFoundException] {
+ FileResolver.resolve(datasetNameUnderModelPrefix)
+ }
+ assertThrows[FileNotFoundException] {
+ FileResolver.resolve(modelNameUnderDatasetPrefix)
+ }
+ }
+
"FileResolver" should "throw not found exception when a prefixed path has too few segments" in {
assertThrows[FileNotFoundException] {
FileResolver.resolve("/datasets/test_user@test.com/test_dataset")
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala
index effd9a4499..3a249d296e 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/resource/ComputingUnitManagingResource.scala
@@ -109,8 +109,8 @@ object ComputingUnitManagingResource {
EnvironmentalVariable.ENV_S3_REGION -> StorageConfig.s3Region,
EnvironmentalVariable.ENV_S3_AUTH_USERNAME -> StorageConfig.s3Username,
EnvironmentalVariable.ENV_S3_AUTH_PASSWORD -> StorageConfig.s3Password,
- EnvironmentalVariable.ENV_FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT -> EnvironmentalVariable
- .get(EnvironmentalVariable.ENV_FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT)
+ EnvironmentalVariable.ENV_FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT -> EnvironmentalVariable
+ .get(EnvironmentalVariable.ENV_FILE_SERVICE_GET_DATASET_PRESIGNED_URL_ENDPOINT)
.get,
EnvironmentalVariable.ENV_FILE_SERVICE_UPLOAD_ONE_FILE_TO_DATASET_ENDPOINT -> EnvironmentalVariable
.get(EnvironmentalVariable.ENV_FILE_SERVICE_UPLOAD_ONE_FILE_TO_DATASET_ENDPOINT)
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 50af933699..75391149c6 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
@@ -26,7 +26,7 @@ import jakarta.ws.rs._
import jakarta.ws.rs.core._
import org.apache.texera.common.config.StorageConfig
import org.apache.texera.common.util.EmailUtil
-import org.apache.texera.amber.core.storage.model.OnDataset
+import org.apache.texera.amber.core.storage.model.OnVersionedFileResource
import org.apache.texera.amber.core.storage.util.LakeFSStorageClient
import org.apache.texera.amber.core.storage.{DocumentFactory, FileResolver, ResourceType}
import org.apache.texera.auth.SessionUser
@@ -1764,7 +1764,8 @@ class DatasetResource extends LazyLogging {
// Case 3: Neither repositoryName nor commitHash are provided, resolve normally
val response = withTransaction(context) { ctx =>
val fileUri = FileResolver.resolve(decodedPathStr)
- val document = DocumentFactory.openReadonlyDocument(fileUri).asInstanceOf[OnDataset]
+ val document =
+ DocumentFactory.openReadonlyDocument(fileUri).asInstanceOf[OnVersionedFileResource]
val datasetDao = new DatasetDao(ctx.configuration())
val datasets =
datasetDao.fetchByRepositoryName(document.getRepositoryName()).asScala.toList
@@ -2406,7 +2407,7 @@ class DatasetResource extends LazyLogging {
logicalPath(resourceType, owner.getEmail, dataset.getName, normalized)
)
)
- .asInstanceOf[OnDataset]
+ .asInstanceOf[OnVersionedFileResource]
val fileSize = withLakeFSErrorHandling(s"reading the size of cover image '$normalized'") {
LakeFSStorageClient.getFileSize(
@@ -2463,7 +2464,7 @@ class DatasetResource extends LazyLogging {
val document = DocumentFactory
.openReadonlyDocument(FileResolver.resolve(fullPath))
- .asInstanceOf[OnDataset]
+ .asInstanceOf[OnVersionedFileResource]
val presignedUrl = withLakeFSErrorHandling(
s"generating a presigned URL for cover image '$coverImage'"
@@ -2513,7 +2514,7 @@ class DatasetResource extends LazyLogging {
val document = DocumentFactory
.openReadonlyDocument(FileResolver.resolve(fullPath))
- .asInstanceOf[OnDataset]
+ .asInstanceOf[OnVersionedFileResource]
val presignedUrl = withLakeFSErrorHandling(
s"generating a presigned URL for cover image '$coverImage'"
diff --git a/sql/changelog.xml b/sql/changelog.xml
index a945b6bcb6..d3a7d55228 100644
--- a/sql/changelog.xml
+++ b/sql/changelog.xml
@@ -94,6 +94,11 @@
<sqlFile path="sql/updates/36.sql"/>
</changeSet>
+ <!-- Add model metadata tables -->
+ <changeSet id="37" author="tanishqgandhi1908">
+ <sqlFile path="sql/updates/37.sql"/>
+ </changeSet>
+
<!-- example changeSet
<changeSet id="1" author="author">
<sqlFile path="sql/updates/1.sql"/>
diff --git a/sql/texera_ddl.sql b/sql/texera_ddl.sql
index aedfd6a848..b8ebe3caf6 100644
--- a/sql/texera_ddl.sql
+++ b/sql/texera_ddl.sql
@@ -62,10 +62,11 @@ DROP TABLE IF EXISTS workflow_of_project CASCADE;
DROP TABLE IF EXISTS workflow_executions CASCADE;
DROP TABLE IF EXISTS dataset_upload_session CASCADE;
DROP TABLE IF EXISTS dataset_upload_session_part CASCADE;
-
DROP TABLE IF EXISTS dataset CASCADE;
DROP TABLE IF EXISTS dataset_user_access CASCADE;
DROP TABLE IF EXISTS dataset_version CASCADE;
+DROP TABLE IF EXISTS model_version CASCADE;
+DROP TABLE IF EXISTS model CASCADE;
DROP TABLE IF EXISTS dataset_contributor CASCADE;
DROP TABLE IF EXISTS public_project CASCADE;
DROP TABLE IF EXISTS project_user_access CASCADE;
@@ -418,6 +419,36 @@ CREATE TABLE IF NOT EXISTS dataset_upload_session_part
ON DELETE CASCADE
);
+-- ML models
+CREATE TABLE IF NOT EXISTS model
+(
+ mid SERIAL PRIMARY KEY,
+ owner_uid INT NOT NULL,
+ name VARCHAR(128) NOT NULL,
+ repository_name VARCHAR(128),
+ is_public BOOLEAN NOT NULL DEFAULT TRUE,
+ is_downloadable BOOLEAN NOT NULL DEFAULT TRUE,
+ description TEXT NOT NULL,
+ creation_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ cover_image varchar(255),
+ framework VARCHAR(32),
+ format VARCHAR(32),
+ FOREIGN KEY (owner_uid) REFERENCES "user"(uid) ON DELETE CASCADE,
+ UNIQUE (owner_uid, name)
+ );
+
+-- model_version
+CREATE TABLE IF NOT EXISTS model_version
+(
+ mvid SERIAL PRIMARY KEY,
+ mid INT NOT NULL,
+ creator_uid INT NOT NULL,
+ name VARCHAR(128) NOT NULL,
+ version_hash VARCHAR(64) NOT NULL,
+ creation_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ FOREIGN KEY (mid) REFERENCES model(mid) ON DELETE CASCADE
+ );
+
-- operator_executions (modified to match MySQL: no separate primary key; added console_messages_uri)
CREATE TABLE IF NOT EXISTS operator_executions
(
diff --git a/sql/updates/37.sql b/sql/updates/37.sql
new file mode 100644
index 0000000000..5492387a34
--- /dev/null
+++ b/sql/updates/37.sql
@@ -0,0 +1,57 @@
+/*
+ * 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.
+ */
+
+\c texera_db
+
+SET search_path TO texera_db;
+
+BEGIN;
+
+-- Introduce ML models as a first-class resource (own primary key `mid`, own LakeFS repo namespace)
+-- and add model-specific attributes (framework, format).
+
+CREATE TABLE IF NOT EXISTS model
+(
+ mid SERIAL PRIMARY KEY,
+ owner_uid INT NOT NULL,
+ name VARCHAR(128) NOT NULL,
+ repository_name VARCHAR(128),
+ is_public BOOLEAN NOT NULL DEFAULT TRUE,
+ is_downloadable BOOLEAN NOT NULL DEFAULT TRUE,
+ description TEXT NOT NULL,
+ creation_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ cover_image varchar(255),
+ framework VARCHAR(32),
+ format VARCHAR(32),
+ FOREIGN KEY (owner_uid) REFERENCES "user"(uid) ON DELETE CASCADE,
+ UNIQUE (owner_uid, name)
+);
+
+CREATE TABLE IF NOT EXISTS model_version
+(
+ mvid SERIAL PRIMARY KEY,
+ mid INT NOT NULL,
+ creator_uid INT NOT NULL,
+ name VARCHAR(128) NOT NULL,
+ version_hash VARCHAR(64) NOT NULL,
+ creation_time TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+ FOREIGN KEY (mid) REFERENCES model(mid) ON DELETE CASCADE
+);
+
+COMMIT;