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 3cb3479ec9 feat(storage): add model file storage and path resolution
(#6860)
3cb3479ec9 is described below
commit 3cb3479ec9201e43ca9e803da2a2a26d59a480a7
Author: Tanishq Gandhi <[email protected]>
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 <[email protected]>
Co-authored-by: Claude Opus 4.8 <[email protected]>
---
.../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/[email protected]/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/[email protected]/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/[email protected]/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/[email protected]/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/[email protected]/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/[email protected]/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/[email protected]/twitter/v1/dir/f.csv ->
dataset:///dataset-15/adeq233td/dir/f.csv
+ * /models/[email protected]/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/[email protected]/test_dataset/v2/directory/a.csv"
private val dataset1TxtFilePath =
"/datasets/[email protected]/test_dataset/v1/1.txt"
- // Legacy unprefixed form.
+ private val modelWeightsFilePath =
"/models/[email protected]/test_model/v2/weights/model.pt"
+
+ private val modelReadmeFilePath =
"/models/[email protected]/test_model/v1/README.md"
+
+ // Legacy unprefixed form — still resolvable as a dataset (backward compat).
private val unprefixedDataset1TxtFilePath =
"/[email protected]/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/[email protected]/test_dataset/v1/1.txt"
+ // A model name presented under the datasets prefix must not resolve as a
dataset.
+ private val modelNameUnderDatasetPrefix =
"/datasets/[email protected]/test_model/v1/README.md"
+
+ // A dataset name presented under the models prefix must not resolve as a
model.
+ private val datasetNameUnderModelPrefix =
"/models/[email protected]/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/[email protected]/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;
