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 <[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;
