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 5a063f60470..9665abbe670 100644 --- a/amber/src/main/python/pytexera/storage/dataset_file_document.py +++ b/amber/src/main/python/pytexera/storage/dataset_file_document.py @@ -22,6 +22,8 @@ from requests.adapters import HTTPAdapter from urllib3.util.retry import Retry +from .resource_type import ResourceType + class DatasetFileDocument: # (connect, read) timeout and retry settings for the file-service GETs below. @@ -56,20 +58,23 @@ def __init__(self, file_path: str): Parses the file path into dataset metadata. :param file_path: - Expected format - "/ownerEmail/datasetName/versionName/fileRelativePath" - Example: "/bob@texera.com/twitterDataset/v1/california/irvine/tw1.csv" + Expected format - + "/datasets/ownerEmail/datasetName/versionName/fileRelativePath" + Example: + "/datasets/bob@texera.com/twitterDataset/v1/california/tw1.csv" """ parts = file_path.strip("/").split("/") - if len(parts) < 4: + + if len(parts) < 5 or parts[0] != ResourceType.DATASETS.value: raise ValueError( - "Invalid file path format. " - "Expected: /ownerEmail/datasetName/versionName/fileRelativePath" + "Invalid file path format. Expected: " + "/datasets/ownerEmail/datasetName/versionName/fileRelativePath" ) - self.owner_email = parts[0] - self.dataset_name = parts[1] - self.version_name = parts[2] - self.file_relative_path = "/".join(parts[3:]) + self.owner_email = parts[1] + self.dataset_name = parts[2] + self.version_name = parts[3] + self.file_relative_path = "/".join(parts[4:]) self.jwt_token = os.getenv("USER_JWT_TOKEN") self.presign_endpoint = os.getenv("FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT") @@ -90,6 +95,7 @@ def get_presigned_url(self) -> str: """ headers = {"Authorization": f"Bearer {self.jwt_token}"} encoded_file_path = urllib.parse.quote( + f"/{ResourceType.DATASETS.value}" f"/{self.owner_email}" f"/{self.dataset_name}" f"/{self.version_name}" diff --git a/amber/src/main/python/pytexera/storage/resource_type.py b/amber/src/main/python/pytexera/storage/resource_type.py new file mode 100644 index 00000000000..596fb5d7453 --- /dev/null +++ b/amber/src/main/python/pytexera/storage/resource_type.py @@ -0,0 +1,24 @@ +# 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. + +from enum import Enum + + +class ResourceType(str, Enum): + """The leading segment of a logical file path, identifying the resource kind""" + + DATASETS = "datasets" 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 36882fe27ee..fba9b2e1d22 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 @@ -44,51 +44,61 @@ def make_response(status_code: int, body=None, content: bytes = b""): class TestDatasetFileDocumentInit: - def test_parses_minimal_four_part_path(self, auth_env): - doc = DatasetFileDocument("/bob@x.com/ds/v1/file.csv") + def test_parses_prefixed_path(self, auth_env): + doc = DatasetFileDocument("/datasets/bob@x.com/ds/v1/file.csv") assert doc.owner_email == "bob@x.com" assert doc.dataset_name == "ds" assert doc.version_name == "v1" assert doc.file_relative_path == "file.csv" def test_joins_nested_relative_path_back_with_slashes(self, auth_env): - doc = DatasetFileDocument("/bob@x.com/ds/v1/a/b/c/file.csv") + doc = DatasetFileDocument("/datasets/bob@x.com/ds/v1/a/b/c/file.csv") assert doc.file_relative_path == "a/b/c/file.csv" def test_strips_leading_and_trailing_slashes_before_parsing(self, auth_env): - doc = DatasetFileDocument("///bob@x.com/ds/v1/file.csv///") + doc = DatasetFileDocument("///datasets/bob@x.com/ds/v1/file.csv///") assert doc.owner_email == "bob@x.com" assert doc.file_relative_path == "file.csv" - def test_rejects_path_with_fewer_than_four_segments(self, auth_env): + def test_rejects_unprefixed_path(self, auth_env): + # Without the datasets prefix the path is not a dataset path. with pytest.raises(ValueError, match="Invalid file path format"): - DatasetFileDocument("/bob@x.com/ds/v1") + DatasetFileDocument("/bob@x.com/ds/v1/file.csv") + + def test_rejects_non_datasets_prefix(self, auth_env): + # Any other leading segment (e.g. "models") is not the datasets prefix. + with pytest.raises(ValueError, match="Invalid file path format"): + DatasetFileDocument("/models/bob@x.com/ds/v1/file.csv") + + def test_rejects_path_with_too_few_segments(self, auth_env): + with pytest.raises(ValueError, match="Invalid file path format"): + DatasetFileDocument("/datasets/bob@x.com/ds/v1") 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) with pytest.raises(ValueError, match="JWT token is required"): - DatasetFileDocument("/bob@x.com/ds/v1/file.csv") + DatasetFileDocument("/datasets/bob@x.com/ds/v1/file.csv") def test_treats_empty_jwt_as_missing(self, monkeypatch): # An empty string is falsy and should be rejected just like an unset var. monkeypatch.setenv("USER_JWT_TOKEN", "") with pytest.raises(ValueError, match="JWT token is required"): - DatasetFileDocument("/bob@x.com/ds/v1/file.csv") + DatasetFileDocument("/datasets/bob@x.com/ds/v1/file.csv") 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) - doc = DatasetFileDocument("/bob@x.com/ds/v1/file.csv") + doc = DatasetFileDocument("/datasets/bob@x.com/ds/v1/file.csv") assert doc.presign_endpoint == DEFAULT_ENDPOINT def test_uses_explicit_endpoint_from_environment(self, auth_env): - doc = DatasetFileDocument("/bob@x.com/ds/v1/file.csv") + doc = DatasetFileDocument("/datasets/bob@x.com/ds/v1/file.csv") assert doc.presign_endpoint == CUSTOM_ENDPOINT class TestGetPresignedUrl: - def _make_doc(self, monkeypatch, path="/bob@x.com/ds/v1/file.csv"): + def _make_doc(self, monkeypatch, path="/datasets/bob@x.com/ds/v1/file.csv"): monkeypatch.setenv("USER_JWT_TOKEN", "test-jwt-token") monkeypatch.setenv("FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT", CUSTOM_ENDPOINT) return DatasetFileDocument(path) @@ -116,7 +126,9 @@ def test_sends_bearer_authorization_header_with_jwt(self, monkeypatch): def test_url_encodes_filepath_query_parameter(self, monkeypatch): # urllib.parse.quote keeps "/" as safe by default, but encodes "@" # and " " — pin both pieces so the contract is explicit. - doc = self._make_doc(monkeypatch, path="/bob@x.com/ds/v1/data file.csv") + doc = self._make_doc( + monkeypatch, path="/datasets/bob@x.com/ds/v1/data file.csv" + ) with patch( "pytexera.storage.dataset_file_document.requests.Session.get" ) as mock_get: @@ -128,6 +140,17 @@ def test_url_encodes_filepath_query_parameter(self, monkeypatch): assert "bob%40x.com" in file_path assert file_path.startswith("/") + def test_sends_datasets_prefixed_filepath(self, monkeypatch): + # The reconstructed filePath sent to the file-service carries the "datasets" prefix. + doc = self._make_doc(monkeypatch, path="/datasets/bob@x.com/ds/v1/file.csv") + with patch( + "pytexera.storage.dataset_file_document.requests.Session.get" + ) as mock_get: + mock_get.return_value = make_response(200, body={"presignedUrl": "u"}) + doc.get_presigned_url() + _, kwargs = mock_get.call_args + assert kwargs["params"]["filePath"].startswith("/datasets/") + def test_calls_configured_endpoint(self, monkeypatch): doc = self._make_doc(monkeypatch) with patch( @@ -192,7 +215,7 @@ 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) - return DatasetFileDocument("/bob@x.com/ds/v1/file.csv") + return DatasetFileDocument("/datasets/bob@x.com/ds/v1/file.csv") def test_returns_bytesio_with_downloaded_content(self, monkeypatch): doc = self._make_doc(monkeypatch) @@ -246,7 +269,7 @@ 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) - return DatasetFileDocument("/bob@x.com/ds/v1/file.csv") + return DatasetFileDocument("/datasets/bob@x.com/ds/v1/file.csv") def test_presigned_url_request_passes_request_timeout(self, monkeypatch): doc = self._make_doc(monkeypatch) diff --git a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala index c01489aa865..e9c92d4ac5e 100644 --- a/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala +++ b/amber/src/test/scala/org/apache/texera/web/resource/dashboard/user/workflow/WorkflowExecutionsResourceSpec.scala @@ -666,7 +666,7 @@ class WorkflowExecutionsResourceSpec val content = """{ | "operators": [ - | {"operatorID": "scanA", "operatorProperties": {"fileName": "/owner@example.com/LockedDS/v1/data.csv"}}, + | {"operatorID": "scanA", "operatorProperties": {"fileName": "/datasets/owner@example.com/LockedDS/v1/data.csv"}}, | {"operatorID": "downstreamB", "operatorProperties": {}} | ], | "links": [ @@ -725,7 +725,7 @@ class WorkflowExecutionsResourceSpec val content = """{ | "operators": [ - | {"operatorID": "scan", "operatorProperties": {"fileName": "/test@example.com/MyDS/v1/data.csv"}} + | {"operatorID": "scan", "operatorProperties": {"fileName": "/datasets/test@example.com/MyDS/v1/data.csv"}} | ], | "links": [] |}""".stripMargin diff --git a/bin/single-node/examples/workflows/[Example] Data Exploration on Movies Dataset.json b/bin/single-node/examples/workflows/[Example] Data Exploration on Movies Dataset.json index 3e9f99c613f..cdc9a0e8708 100644 --- a/bin/single-node/examples/workflows/[Example] Data Exploration on Movies Dataset.json +++ b/bin/single-node/examples/workflows/[Example] Data Exploration on Movies Dataset.json @@ -8,7 +8,7 @@ "fileEncoding": "UTF_8", "customDelimiter": ",", "hasHeader": true, - "fileName": "/texera/popular-movies-of-imdb/v1/TMDb_updated.csv", + "fileName": "/datasets/texera/popular-movies-of-imdb/v1/TMDb_updated.csv", "offset": 0, "limit": 1000 }, diff --git a/bin/single-node/examples/workflows/[Example] Machine Learning on Iris Dataset.json b/bin/single-node/examples/workflows/[Example] Machine Learning on Iris Dataset.json index d36d35dae56..7e9a5d11fd4 100644 --- a/bin/single-node/examples/workflows/[Example] Machine Learning on Iris Dataset.json +++ b/bin/single-node/examples/workflows/[Example] Machine Learning on Iris Dataset.json @@ -8,7 +8,7 @@ "fileEncoding": "UTF_8", "customDelimiter": ",", "hasHeader": true, - "fileName": "/texera/iris-species/v1/Iris.csv" + "fileName": "/datasets/texera/iris-species/v1/Iris.csv" }, "inputPorts": [], "outputPorts": [ 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 a335ddeff6c..a79fc6cef20 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 @@ -36,6 +36,8 @@ object EnvironmentalVariable { * FileService related endpoint */ val ENV_FILE_SERVICE_GET_PRESIGNED_URL_ENDPOINT = "FILE_SERVICE_GET_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 a26340e79cc..a020dfdbb48 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( @@ -164,6 +168,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 (_, _, _, resourceType) = decodeURI(uri) val storageKey = sanitizeURIPath(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 c8a407df993..9265ec9f723 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 @@ -39,6 +46,7 @@ import scala.util.{Success, Try} 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 +59,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, datasetResolveFunc, modelResolveFunc) // Try each resolver function in sequence resolvers @@ -76,53 +84,102 @@ object FileResolver { } /** - * Parses a dataset file path and extracts its components. - * Expected format: /ownerEmail/datasetName/versionName/fileRelativePath + * Parses a versioned-resource file path of the form + * //ownerEmail/resourceName/versionName/fileRelativePath and extracts its components. * + * @param resourceType the resource-type segment the path must start with * @param fileName The file path to parse - * @return Some((ownerEmail, datasetName, versionName, fileRelativePath)) if valid, None otherwise + * @return Some((ownerEmail, resourceName, versionName, fileRelativePath)) if valid, None otherwise */ - private def parseDatasetFilePath( + private def parsePrefixedPath( + resourceType: ResourceType.Value, fileName: String ): Option[(String, String, String, Array[String])] = { val filePath = Paths.get(fileName) val pathSegments = (0 until filePath.getNameCount).map(filePath.getName(_).toString).toArray - if (pathSegments.length < 4) { + if (pathSegments.length < 5 || pathSegments(0) != resourceType.toString) { return None } - val ownerEmail = pathSegments(0) - val datasetName = pathSegments(1) - val versionName = pathSegments(2) - val fileRelativePathSegments = pathSegments.drop(3) + val ownerEmail = pathSegments(1) + val resourceName = pathSegments(2) + val versionName = pathSegments(3) + val fileRelativePathSegments = pathSegments.drop(4) - Some((ownerEmail, datasetName, versionName, fileRelativePathSegments)) + Some((ownerEmail, resourceName, versionName, fileRelativePathSegments)) } + private def parseDatasetFilePath( + fileName: String + ): Option[(String, String, String, Array[String])] = + parsePrefixedPath(ResourceType.Datasets, fileName) + /** - * 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: /ownerEmail/datasetName/versionName/fileRelativePath - * e.g. /bob@texera.com/twitterDataset/v1/california/irvine/tw1.csv - * The output dataset URI format is: {DATASET_FILE_URI_SCHEME}:///{repositoryName}/{versionHash}/fileRelativePath - * e.g. {DATASET_FILE_URI_SCHEME}:///dataset-15/adeq233td/some/dir/file.txt - * - * @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 + * @throws java.io.FileNotFoundException if the path is not a valid `resourceType` 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 resolveVersionedFile( + scheme: String, + resourceType: ResourceType.Value, + fileName: String, + notFoundMessage: String + )(lookup: (String, String, String) => (String, String)): URI = { + val (ownerEmail, resourceName, versionName, fileRelativePathSegments) = + parsePrefixedPath(resourceType, fileName).getOrElse( + throw new FileNotFoundException(notFoundMessage) ) + val (repositoryName, versionHash) = lookup(ownerEmail, resourceName, versionName) + 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) = + // Convert each segment of fileRelativePath to an encoded String + val encodedFileRelativePath = fileRelativePath + .iterator() + .asScala + .map { segment => + URLEncoder.encode(segment.toString, StandardCharsets.UTF_8) + } + .toArray + + // Prepend repositoryName and versionHash to the encoded path segments + val allPathSegments = Array(repositoryName, versionHash) ++ 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) + + try { + new URI(scheme, "", encodedPath, null) + } catch { + case _: Exception => + throw new FileNotFoundException(notFoundMessage) + } + } + + /** + * Attempts to resolve a dataset file path to a URI. + * + * The fileName format should be: /datasets/ownerEmail/datasetName/versionName/fileRelativePath + * e.g. /datasets/bob@texera.com/twitterDataset/v1/california/irvine/tw1.csv + * The output dataset URI format is: {DATASET_FILE_URI_SCHEME}:///{repositoryName}/{versionHash}/fileRelativePath + * e.g. {DATASET_FILE_URI_SCHEME}:///dataset-15/adeq233td/some/dir/file.txt + * + * @param fileName the name of the file to attempt resolving as a DatasetFileDocument + * @throws java.io.FileNotFoundException if the dataset file does not exist or cannot be created + */ + private def datasetResolveFunc(fileName: String): URI = + resolveVersionedFile( + DATASET_FILE_URI_SCHEME, + ResourceType.Datasets, + fileName, + s"Dataset file $fileName not found." + ) { (ownerEmail, datasetName, versionName) => + // fetch the dataset and version from DB to get the repository name and version hash withTransaction( SqlServer .getInstance() @@ -138,6 +195,11 @@ object FileResolver { .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.") + } + // fetch the dataset version from DB val datasetVersion = ctx .selectFrom(DATASET_VERSION) @@ -145,38 +207,65 @@ object FileResolver { .and(DATASET_VERSION.NAME.eq(versionName)) .fetchOneInto(classOf[DatasetVersion]) - if (dataset == null || datasetVersion == null) { + if (datasetVersion == null) { throw new FileNotFoundException(s"Dataset file $fileName not found.") } - (dataset, datasetVersion) + (dataset.getRepositoryName, datasetVersion.getVersionHash) } + } - // Convert each segment of fileRelativePath to an encoded String - val encodedFileRelativePath = fileRelativePath - .iterator() - .asScala - .map { segment => - URLEncoder.encode(segment.toString, StandardCharsets.UTF_8) - } - .toArray + /** + * Attempts to resolve a model file path to a URI. + * + * The fileName format should be: /models/ownerEmail/modelName/versionName/fileRelativePath + * e.g. /models/bob@texera.com/resnet/v1/weights/model.pt + * The output model URI format is: {MODEL_FILE_URI_SCHEME}:///{repositoryName}/{versionHash}/fileRelativePath + * e.g. {MODEL_FILE_URI_SCHEME}:///model-15/adeq233td/weights/model.pt + * + * @param fileName the name of the file to attempt resolving as a ModelFileDocument + * @throws java.io.FileNotFoundException if the model file does not exist or cannot be created + */ + private def modelResolveFunc(fileName: String): URI = + resolveVersionedFile( + MODEL_FILE_URI_SCHEME, + ResourceType.Models, + fileName, + s"Model file $fileName not found." + ) { (ownerEmail, modelName, versionName) => + // fetch the model and version from DB to get the repository name and version hash + withTransaction( + SqlServer + .getInstance() + .createDSLContext() + ) { ctx => + // fetch the model from DB + 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]) - // Prepend dataset name and versionHash to the encoded path segments - val allPathSegments = Array( - dataset.getRepositoryName, - datasetVersion.getVersionHash - ) ++ encodedFileRelativePath + // fail early if the model does not exist (before dereferencing it below) + if (model == null) { + throw new FileNotFoundException(s"Model file $fileName not found.") + } - // Build the format /{repositoryName}/{versionHash}/{fileRelativePath}, both Linux and Windows use forward slash as the splitter - val uriSplitter = "/" - val encodedPath = uriSplitter + allPathSegments.mkString(uriSplitter) + // fetch the model version from DB + val modelVersion = ctx + .selectFrom(MODEL_VERSION) + .where(MODEL_VERSION.MID.eq(model.getMid)) + .and(MODEL_VERSION.NAME.eq(versionName)) + .fetchOneInto(classOf[ModelVersion]) - try { - new URI(DATASET_FILE_URI_SCHEME, "", encodedPath, null) - } catch { - case e: Exception => - throw new FileNotFoundException(s"Dataset file $fileName not found.") + 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. @@ -194,11 +283,8 @@ object FileResolver { } /** - * Parses a dataset file path to extract owner email and dataset name. - * Expected format: /ownerEmail/datasetName/versionName/fileRelativePath - * - * @param path The file path from operator properties - * @return Some((ownerEmail, datasetName)) if path is valid, None otherwise + * Extracts the owner email and dataset name from a dataset logical path, + * or None if it is not a well-formed dataset path. */ def parseDatasetOwnerAndName(path: String): Option[(String, String)] = { if (path == null) { 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 new file mode 100644 index 00000000000..03c1ce4c01f --- /dev/null +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/ResourceType.scala @@ -0,0 +1,31 @@ +/* + * 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. + */ + +package org.apache.texera.amber.core.storage + +/** + * The leading segment of a logical file path, identifying which resource kind (and thus which + * backing table) a path belongs to. + * + * Path shape: //ownerEmail/resourceName/versionName/fileRelativePath + */ +object ResourceType extends Enumeration { + val Datasets: Value = Value("datasets") + val Models: Value = Value("models") +} 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 6d8f917c7f4..044db2013b9 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,28 +19,14 @@ 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 @@ -52,122 +38,12 @@ object DatasetFileDocument { } 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/LakeFSFileDocument.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocument.scala new file mode 100644 index 00000000000..044437889e9 --- /dev/null +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/LakeFSFileDocument.scala @@ -0,0 +1,175 @@ +/* + * 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. + */ + +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.LakeFSFileDocument.userJwtToken +import org.apache.texera.amber.core.storage.util.LakeFSStorageClient + +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 + +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 +} + +/** + * 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 OnVersionedFileResource + 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 (repositoryName, versionHash, fileRelativePath) + val repositoryName = segments(0) + 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, versionHash, fileRelativePath) + } + + // Extract components from URI using the utility function + protected val (repositoryName, versionHash, fileRelativePath) = parseUri(uri) + + protected 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"$presignEndpoint?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 + } + } + + override def clear(): Unit = { + // remove the temporary file, if one was materialized + tempFile match { + case Some(file) => Files.delete(file.toPath) + case None => // Do nothing + } + tempFile = None + } + + override def getRepositoryName(): String = repositoryName + + 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/ModelFileDocument.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/ModelFileDocument.scala new file mode 100644 index 00000000000..e67ec92992b --- /dev/null +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/model/ModelFileDocument.scala @@ -0,0 +1,43 @@ +/* + * 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. + */ + +package org.apache.texera.amber.core.storage.model + +import org.apache.texera.common.config.EnvironmentalVariable +import org.apache.texera.amber.core.storage.model.ModelFileDocument.fileServiceGetModelPresignURLEndpoint + +import java.net.URI + +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 6c19ee1002b..cbb76ee4f59 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 ad8b5580c53..2578d497380 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 /** @@ -135,6 +137,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 { @@ -165,6 +179,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 c16d417a73c..72285984971 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,11 +142,11 @@ 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 = @@ -159,9 +159,11 @@ class DatasetFileDocumentSpec extends AnyFlatSpec with Matchers { 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 62daa811d35..a5f441265a8 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} @@ -77,11 +89,59 @@ 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 = "/test_user@test.com/test_dataset/v2/directory/a.csv" + private val datasetACsvFilePath = "/datasets/test_user@test.com/test_dataset/v2/directory/a.csv" + + private val dataset1TxtFilePath = "/datasets/test_user@test.com/test_dataset/v1/1.txt" + + private val modelWeightsFilePath = "/models/test_user@test.com/test_model/v2/weights/model.pt" + + private val modelReadmeFilePath = "/models/test_user@test.com/test_model/v1/README.md" - private val dataset1TxtFilePath = "/test_user@test.com/test_dataset/v1/1.txt" + // Unprefixed form (no resource-type segment); no longer resolvable as a dataset. + private val unprefixedDataset1TxtFilePath = "/test_user@test.com/test_dataset/v1/1.txt" + + // An unknown resource-type prefix belongs to neither datasets nor models. + private val unknownPrefixFilePath = "/notaprefix/test_user@test.com/test_dataset/v1/1.txt" + + // A model name presented under the datasets prefix must not resolve as a dataset. + private val modelNameUnderDatasetPrefix = "/datasets/test_user@test.com/test_model/v1/README.md" + + // A dataset name presented under the models prefix must not resolve as a model. + private val datasetNameUnderModelPrefix = "/models/test_user@test.com/test_dataset/v1/1.txt" override protected def beforeAll(): Unit = { initializeDBAndReplaceDSLContext() @@ -98,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 { @@ -118,6 +187,48 @@ 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 "not resolve an unprefixed dataset path (the datasets prefix is required)" in { + // Without the datasets prefix the path is not a dataset path; it falls through to the + // local-file resolver and fails. + assertThrows[FileNotFoundException] { + FileResolver.resolve(unprefixedDataset1TxtFilePath) + } + } + + "FileResolver" should "not resolve a path whose prefix is neither datasets nor models" in { + assertThrows[FileNotFoundException] { + FileResolver.resolve(unknownPrefixFilePath) + } + } + + "FileResolver" should "keep datasets and models isolated by prefix" in { + // a real dataset name under the models prefix is not a model, and vice-versa + assertThrows[FileNotFoundException] { + FileResolver.resolve(datasetNameUnderModelPrefix) + } + assertThrows[FileNotFoundException] { + FileResolver.resolve(modelNameUnderDatasetPrefix) + } + } + + "FileResolver" should "throw not found exception when a prefixed path has too few segments" in { + assertThrows[FileNotFoundException] { + FileResolver.resolve("/datasets/test_user@test.com/test_dataset") + } + } + "FileResolver" should "throw not found exception" in { assertThrows[FileNotFoundException] { FileResolver.resolve("some/random/path") @@ -143,18 +254,22 @@ class FileResolverSpec "parseDatasetOwnerAndName" should "extract owner email and dataset name from a valid path" in { assert( - FileResolver.parseDatasetOwnerAndName("/test_user@test.com/test_dataset/v1/1.txt") + FileResolver.parseDatasetOwnerAndName("/datasets/test_user@test.com/test_dataset/v1/1.txt") == Some(("test_user@test.com", "test_dataset")) ) // extra segments beyond the file-relative path are ignored assert( - FileResolver.parseDatasetOwnerAndName("/owner@x.com/ds/v2/directory/nested/a.csv") + FileResolver.parseDatasetOwnerAndName("/datasets/owner@x.com/ds/v2/directory/nested/a.csv") == Some(("owner@x.com", "ds")) ) } - it should "return None when the path has fewer than four segments" in { - assert(FileResolver.parseDatasetOwnerAndName("/owner@x.com/ds/v1").isEmpty) + it should "return None for an unprefixed path (the datasets prefix is required)" in { + assert(FileResolver.parseDatasetOwnerAndName(unprefixedDataset1TxtFilePath).isEmpty) + } + + it should "return None when the prefixed path has too few segments" in { + assert(FileResolver.parseDatasetOwnerAndName("/datasets/owner@x.com/ds").isEmpty) assert(FileResolver.parseDatasetOwnerAndName("owner/dataset").isEmpty) } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/dataset/FileListerSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/dataset/FileListerSourceOpExec.scala index c58da8ca847..2e4bad27324 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/dataset/FileListerSourceOpExec.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/dataset/FileListerSourceOpExec.scala @@ -20,6 +20,7 @@ package org.apache.texera.amber.operator.source.dataset import org.apache.texera.amber.core.executor.SourceOperatorExecutor +import org.apache.texera.amber.core.storage.ResourceType import org.apache.texera.amber.core.storage.util.LakeFSStorageClient import org.apache.texera.amber.core.tuple.TupleLike import org.apache.texera.amber.util.JSONUtils.objectMapper @@ -28,13 +29,34 @@ 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.User.USER +object FileListerSourceOpExec { + + /** + * Parses a dataset version path (/datasets/ownerEmail/datasetName/versionName) into its + * (ownerEmail, datasetName, versionName) components. + * + * @throws IllegalArgumentException if the path is not a well-formed dataset version path + */ + private[dataset] def parseDatasetVersionPath( + datasetVersionPath: String + ): (String, String, String) = { + val segments = datasetVersionPath.split("/").filter(_.nonEmpty) + require( + segments.length >= 4 && segments.head == ResourceType.Datasets.toString, + s"Invalid dataset version path '$datasetVersionPath'; " + + "expected /datasets/ownerEmail/datasetName/versionName" + ) + (segments(1), segments(2), segments(3)) + } +} + class FileListerSourceOpExec private[dataset] (descString: String) extends SourceOperatorExecutor { private val desc: FileListerSourceOpDesc = objectMapper.readValue(descString, classOf[FileListerSourceOpDesc]) override def produceTuple(): Iterator[TupleLike] = { - val Seq(_, ownerEmail, datasetName, versionName, _*) = - desc.datasetVersionPath.split("/").toSeq + val (ownerEmail, datasetName, versionName) = + FileListerSourceOpExec.parseDatasetVersionPath(desc.datasetVersionPath) val (repositoryName, versionHash) = SqlServer diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/dataset/FileListerSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/dataset/FileListerSourceOpExecSpec.scala new file mode 100644 index 00000000000..b5b5611c991 --- /dev/null +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/dataset/FileListerSourceOpExecSpec.scala @@ -0,0 +1,67 @@ +/* + * 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. + */ + +package org.apache.texera.amber.operator.source.dataset + +import org.scalatest.flatspec.AnyFlatSpec + +class FileListerSourceOpExecSpec extends AnyFlatSpec { + + "parseDatasetVersionPath" should "extract components from a datasets-prefixed path" in { + val (owner, name, version) = + FileListerSourceOpExec.parseDatasetVersionPath("/datasets/bob@texera.com/twitterDataset/v1") + assert(owner == "bob@texera.com") + assert(name == "twitterDataset") + assert(version == "v1") + } + + it should "work when the owner segment is a username without an '@'" in { + val (owner, name, version) = + FileListerSourceOpExec.parseDatasetVersionPath("/datasets/texera/test-ds/v1") + assert(owner == "texera") + assert(name == "test-ds") + assert(version == "v1") + } + + it should "ignore trailing slashes and extra segments" in { + val (owner, name, version) = + FileListerSourceOpExec.parseDatasetVersionPath("/datasets/alice/ds/v2/extra/") + assert(owner == "alice") + assert(name == "ds") + assert(version == "v2") + } + + it should "reject an unprefixed path (the datasets prefix is required)" in { + assertThrows[IllegalArgumentException] { + FileListerSourceOpExec.parseDatasetVersionPath("/alice/ds/v1") + } + } + + it should "reject a path whose prefix is not the datasets prefix" in { + assertThrows[IllegalArgumentException] { + FileListerSourceOpExec.parseDatasetVersionPath("/models/alice/ds/v1") + } + } + + it should "reject a path with too few segments" in { + assertThrows[IllegalArgumentException] { + FileListerSourceOpExec.parseDatasetVersionPath("/datasets/alice/ds") + } + } +} 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 c376e4ce041..c27117c17ee 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 @@ -25,9 +25,9 @@ import jakarta.annotation.security.{PermitAll, RolesAllowed} import jakarta.ws.rs._ import jakarta.ws.rs.core._ import org.apache.texera.common.config.StorageConfig -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} +import org.apache.texera.amber.core.storage.{DocumentFactory, FileResolver, ResourceType} import org.apache.texera.auth.SessionUser import org.apache.texera.dao.SiteSettings import org.apache.texera.dao.SqlServer @@ -1264,15 +1264,25 @@ class DatasetResource extends LazyLogging { throw new NotFoundException(ERR_DATASET_VERSION_NOT_FOUND_MESSAGE) ) - val ownerNode = DatasetFileNode + val datasetsNode = DatasetFileNode .fromLakeFSRepositoryCommittedObjects( Map( - (user.getEmail, dataset.getName, latestVersion.getName) -> LakeFSStorageClient + ( + getOwner(ctx, did).getEmail, + dataset.getName, + latestVersion.getName + ) -> LakeFSStorageClient .retrieveObjectsOfVersion(dataset.getRepositoryName, latestVersion.getVersionHash) ) ) .head + val ownerNode = datasetsNode.getChildren.headOption.getOrElse( + throw new IllegalStateException( + s"Dataset file tree for ${dataset.getName} is missing its owner node" + ) + ) + DashboardDatasetVersion( latestVersion, ownerNode.children.get @@ -1499,7 +1509,7 @@ class DatasetResource extends LazyLogging { val datasetName = dataset.dataset.getName val repositoryName = dataset.dataset.getRepositoryName - val ownerFileNode = DatasetFileNode + val datasetsNode = DatasetFileNode .fromLakeFSRepositoryCommittedObjects( Map( (dataset.ownerEmail, datasetName, datasetVersion.getName) -> LakeFSStorageClient @@ -1508,6 +1518,12 @@ class DatasetResource extends LazyLogging { ) .head + val ownerFileNode = datasetsNode.getChildren.headOption.getOrElse( + throw new IllegalStateException( + s"Dataset file tree for $datasetName is missing its owner node" + ) + ) + DatasetVersionRootFileNodesResponse( ownerFileNode.children.get .find(_.getName == datasetName) @@ -1518,7 +1534,7 @@ class DatasetResource extends LazyLogging { .head .children .get, - DatasetFileNode.calculateTotalSize(List(ownerFileNode)) + DatasetFileNode.calculateTotalSize(List(datasetsNode)) ) } @@ -1588,7 +1604,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 @@ -2226,9 +2243,11 @@ class DatasetResource extends LazyLogging { val owner = getOwner(ctx, did) val document = DocumentFactory .openReadonlyDocument( - FileResolver.resolve(s"${owner.getEmail}/${dataset.getName}/$normalized") + FileResolver.resolve( + s"${ResourceType.Datasets}/${owner.getEmail}/${dataset.getName}/$normalized" + ) ) - .asInstanceOf[OnDataset] + .asInstanceOf[OnVersionedFileResource] val fileSize = withLakeFSErrorHandling(s"reading the size of cover image '$normalized'") { LakeFSStorageClient.getFileSize( @@ -2280,11 +2299,12 @@ class DatasetResource extends LazyLogging { ) val owner = getOwner(ctx, did) - val fullPath = s"${owner.getEmail}/${dataset.getName}/$coverImage" + val fullPath = + s"${ResourceType.Datasets}/${owner.getEmail}/${dataset.getName}/$coverImage" val document = DocumentFactory .openReadonlyDocument(FileResolver.resolve(fullPath)) - .asInstanceOf[OnDataset] + .asInstanceOf[OnVersionedFileResource] val presignedUrl = withLakeFSErrorHandling( s"generating a presigned URL for cover image '$coverImage'" @@ -2329,11 +2349,12 @@ class DatasetResource extends LazyLogging { Response.ok(Map("url" -> null)).build() case Some(coverImage) => val owner = getOwner(ctx, did) - val fullPath = s"${owner.getEmail}/${dataset.getName}/$coverImage" + val fullPath = + s"${ResourceType.Datasets}/${owner.getEmail}/${dataset.getName}/$coverImage" 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/file-service/src/main/scala/org/apache/texera/service/type/dataset/DatasetFileNode.scala b/file-service/src/main/scala/org/apache/texera/service/type/dataset/DatasetFileNode.scala index 7c91d30b94a..6cb76f13bf6 100644 --- a/file-service/src/main/scala/org/apache/texera/service/type/dataset/DatasetFileNode.scala +++ b/file-service/src/main/scala/org/apache/texera/service/type/dataset/DatasetFileNode.scala @@ -20,15 +20,15 @@ package org.apache.texera.service.`type` import io.lakefs.clients.sdk.model.ObjectStats +import org.apache.texera.amber.core.storage.ResourceType import org.apache.texera.amber.core.storage.util.dataset.PhysicalFileNode import java.util import scala.collection.mutable // DatasetFileNode represents a unique file in dataset, its full path is in the format of: -// /ownerEmail/datasetName/versionName/fileRelativePath -// e.g. /bob@texera.com/twitterDataset/v1/california/irvine/tw1.csv -// ownerName is bob@texera.com; datasetName is twitterDataset, versionName is v1, fileRelativePath is california/irvine/tw1.csv +// /datasets/ownerEmail/datasetName/versionName/fileRelativePath +// e.g. /datasets/bob@texera.com/twitterDataset/v1/california/irvine/tw1.csv class DatasetFileNode( val name: String, // direct name of this node val nodeType: String, // "file" or "directory" @@ -81,6 +81,11 @@ object DatasetFileNode { ): List[DatasetFileNode] = { val rootNode = new DatasetFileNode("/", "directory", null, "") + // Root the tree at the datasets prefix node (a directory node named "datasets"). + val datasetsNode = + new DatasetFileNode(ResourceType.Datasets.toString, "directory", rootNode, "") + rootNode.children = Some(List(datasetsNode)) + // Owner level nodes map val ownerNodes = mutable.Map[String, DatasetFileNode]() @@ -88,8 +93,8 @@ object DatasetFileNode { case ((ownerEmail, datasetName, versionName), objects) => val ownerNode = ownerNodes.getOrElseUpdate( ownerEmail, { - val newNode = new DatasetFileNode(ownerEmail, "directory", rootNode, ownerEmail) - rootNode.children = Some(rootNode.getChildren :+ newNode) + val newNode = new DatasetFileNode(ownerEmail, "directory", datasetsNode, ownerEmail) + datasetsNode.children = Some(datasetsNode.getChildren :+ newNode) newNode } ) @@ -153,14 +158,19 @@ object DatasetFileNode { map: Map[(String, String, String), List[PhysicalFileNode]] ): List[DatasetFileNode] = { val rootNode = new DatasetFileNode("/", "directory", null, "") + + val datasetsNode = + new DatasetFileNode(ResourceType.Datasets.toString, "directory", rootNode, "") + rootNode.children = Some(List(datasetsNode)) + val ownerNodes = mutable.Map[String, DatasetFileNode]() map.foreach { case ((ownerEmail, datasetName, versionName), physicalNodes) => val ownerNode = ownerNodes.getOrElseUpdate( ownerEmail, { - val newNode = new DatasetFileNode(ownerEmail, "directory", rootNode, ownerEmail) - rootNode.children = Some(rootNode.getChildren :+ newNode) + val newNode = new DatasetFileNode(ownerEmail, "directory", datasetsNode, ownerEmail) + datasetsNode.children = Some(datasetsNode.getChildren :+ newNode) newNode } ) diff --git a/file-service/src/test/scala/org/apache/texera/service/type/dataset/DatasetFileNodeSpec.scala b/file-service/src/test/scala/org/apache/texera/service/type/dataset/DatasetFileNodeSpec.scala index 93e08a9afc1..c47675dd308 100644 --- a/file-service/src/test/scala/org/apache/texera/service/type/dataset/DatasetFileNodeSpec.scala +++ b/file-service/src/test/scala/org/apache/texera/service/type/dataset/DatasetFileNodeSpec.scala @@ -105,10 +105,13 @@ class DatasetFileNodeSpec extends AnyFlatSpec with Matchers { Map(("bob@texera.com", "twitter", "v1") -> objects) ) - // One owner root. + // The tree is rooted at a single "datasets" prefix node; owners nest under it. roots should have size 1 - val ownerNode = roots.head - ownerNode.getName shouldBe "bob@texera.com" + val datasetsNode = roots.head + datasetsNode.getName shouldBe "datasets" + datasetsNode.getNodeType shouldBe "directory" + + val ownerNode = datasetsNode.getChildren.find(_.getName == "bob@texera.com").get ownerNode.getNodeType shouldBe "directory" val datasetNode = ownerNode.getChildren.find(_.getName == "twitter").get @@ -126,7 +129,7 @@ class DatasetFileNodeSpec extends AnyFlatSpec with Matchers { val file1 = bDir.getChildren.find(_.getName == "1.csv").get file1.getNodeType shouldBe "file" file1.getSize shouldBe Some(2L) - file1.getFilePath shouldBe "/bob@texera.com/twitter/v1/b/1.csv" + file1.getFilePath shouldBe "/datasets/bob@texera.com/twitter/v1/b/1.csv" // Total size equals the sum of the three files. DatasetFileNode.calculateTotalSize(roots) shouldBe 6L @@ -152,7 +155,10 @@ class DatasetFileNodeSpec extends AnyFlatSpec with Matchers { ) roots should have size 1 - val versionNode = roots.head.getChildren + val versionNode = roots.head.getChildren // "datasets" prefix node + .find(_.getName == "alice@texera.com") + .get + .getChildren .find(_.getName == "ds") .get .getChildren @@ -165,7 +171,7 @@ class DatasetFileNodeSpec extends AnyFlatSpec with Matchers { val leaf = dirNode.getChildren.find(_.getName == "a.csv").get leaf.getNodeType shouldBe "file" leaf.getSize shouldBe Some(5L) - leaf.getFilePath shouldBe "/alice@texera.com/ds/v1/dir1/a.csv" + leaf.getFilePath shouldBe "/datasets/alice@texera.com/ds/v1/dir1/a.csv" DatasetFileNode.calculateTotalSize(roots) shouldBe 5L } finally { diff --git a/frontend/src/app/common/type/dataset-file.spec.ts b/frontend/src/app/common/type/dataset-file.spec.ts deleted file mode 100644 index 76acdfe9144..00000000000 --- a/frontend/src/app/common/type/dataset-file.spec.ts +++ /dev/null @@ -1,85 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, - * software distributed under the License is distributed on an - * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY - * KIND, either express or implied. See the License for the - * specific language governing permissions and limitations - * under the License. - */ - -import { DatasetFile, parseDatasetFileToFilePath, parseFilePathToDatasetFile } from "./dataset-file"; - -describe("parseFilePathToDatasetFile", () => { - it("parses owner, dataset, version, and single-segment relative path", () => { - const result = parseFilePathToDatasetFile("/bob@texera.com/twitterDataset/v1/tw1.csv"); - expect(result).toEqual({ - ownerEmail: "bob@texera.com", - datasetName: "twitterDataset", - versionName: "v1", - fileRelativePath: "tw1.csv", - }); - }); - - it("joins remaining segments into a nested relative path", () => { - const result = parseFilePathToDatasetFile("/bob@texera.com/twitterDataset/v1/california/irvine/tw1.csv"); - expect(result.ownerEmail).toBe("bob@texera.com"); - expect(result.datasetName).toBe("twitterDataset"); - expect(result.versionName).toBe("v1"); - expect(result.fileRelativePath).toBe("california/irvine/tw1.csv"); - }); - - it("ignores empty segments from leading, trailing, and duplicate slashes", () => { - const result = parseFilePathToDatasetFile("//bob@texera.com//twitterDataset/v1/dir//file.csv/"); - expect(result).toEqual({ - ownerEmail: "bob@texera.com", - datasetName: "twitterDataset", - versionName: "v1", - fileRelativePath: "dir/file.csv", - }); - }); - - it("throws when there are fewer than four path segments", () => { - expect(() => parseFilePathToDatasetFile("/bob@texera.com/twitterDataset/v1")).toThrow("Invalid file path format"); - expect(() => parseFilePathToDatasetFile("")).toThrow("Invalid file path format"); - expect(() => parseFilePathToDatasetFile("/just/three/parts")).toThrow("Invalid file path format"); - }); -}); - -describe("parseDatasetFileToFilePath", () => { - it("assembles a slash-delimited path with a leading slash", () => { - const datasetFile: DatasetFile = { - ownerEmail: "bob@texera.com", - datasetName: "twitterDataset", - versionName: "v1", - fileRelativePath: "california/irvine/tw1.csv", - }; - expect(parseDatasetFileToFilePath(datasetFile)).toBe("/bob@texera.com/twitterDataset/v1/california/irvine/tw1.csv"); - }); -}); - -describe("dataset-file round trips", () => { - it("path -> DatasetFile -> path is stable for a canonical path", () => { - const path = "/bob@texera.com/twitterDataset/v1/california/irvine/tw1.csv"; - expect(parseDatasetFileToFilePath(parseFilePathToDatasetFile(path))).toBe(path); - }); - - it("DatasetFile -> path -> DatasetFile is stable for a canonical object", () => { - const datasetFile: DatasetFile = { - ownerEmail: "alice@texera.com", - datasetName: "sensorData", - versionName: "v42", - fileRelativePath: "2026/reading.json", - }; - expect(parseFilePathToDatasetFile(parseDatasetFileToFilePath(datasetFile))).toEqual(datasetFile); - }); -}); diff --git a/frontend/src/app/common/type/dataset-file.ts b/frontend/src/app/common/type/dataset-file.ts deleted file mode 100644 index 5fe561d3720..00000000000 --- a/frontend/src/app/common/type/dataset-file.ts +++ /dev/null @@ -1,60 +0,0 @@ -/** - * 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. - */ - -// user given filePath is /ownerEmail/datasetName/versionName/fileRelativePath -// e.g. /bob@texera.com/twitterDataset/v1/california/irvine/tw1.csv -export interface DatasetFile { - ownerEmail: string; - datasetName: string; - versionName: string; - fileRelativePath: string; -} - -/** - * Parses a file path string to a DatasetFile interface. - * @param filePath - The file path string to parse. - * @returns The parsed DatasetFile object. - */ -export function parseFilePathToDatasetFile(filePath: string): DatasetFile { - const parts = filePath.split("/").filter(part => part.length > 0); - - if (parts.length < 4) { - throw new Error("Invalid file path format"); - } - - const [ownerEmail, datasetName, versionName, ...fileRelativePathParts] = parts; - const fileRelativePath = fileRelativePathParts.join("/"); - - return { - ownerEmail, - datasetName, - versionName, - fileRelativePath, - }; -} - -/** - * Converts a DatasetFile object to a file path string. - * @param datasetFile - The DatasetFile object to convert. - * @returns The file path string. - */ -export function parseDatasetFileToFilePath(datasetFile: DatasetFile): string { - const { ownerEmail, datasetName, versionName, fileRelativePath } = datasetFile; - return `/${ownerEmail}/${datasetName}/${versionName}/${fileRelativePath}`; -} diff --git a/frontend/src/app/common/type/datasetVersionFileTree.spec.ts b/frontend/src/app/common/type/datasetVersionFileTree.spec.ts index 76aa8f286d3..64231dd1564 100644 --- a/frontend/src/app/common/type/datasetVersionFileTree.spec.ts +++ b/frontend/src/app/common/type/datasetVersionFileTree.spec.ts @@ -9,12 +9,11 @@ * * 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. + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. */ import { @@ -38,27 +37,40 @@ describe("getFullPathFromDatasetFileNode", () => { }); describe("getRelativePathFromDatasetFileNode", () => { - it("strips the first three path segments", () => { - const node: DatasetFileNode = { name: "file.csv", type: "file", parentDir: "/owner/dataset/v1" }; - // full path is /owner/dataset/v1/file.csv -> segments [owner, dataset, v1, file.csv] - expect(getRelativePathFromDatasetFileNode(node)).toBe("file.csv"); + it("strips the datasets/owner/dataset/version prefix (4 segments)", () => { + const node: DatasetFileNode = { + name: "tw1.csv", + type: "file", + parentDir: "/datasets/bob@texera.com/twitterDataset/v1/california/irvine", + }; + expect(getRelativePathFromDatasetFileNode(node)).toBe("california/irvine/tw1.csv"); }); - it("preserves nested relative segments beyond the first three", () => { - const node: DatasetFileNode = { name: "f.txt", type: "file", parentDir: "/owner/dataset/v1/sub/dir" }; - expect(getRelativePathFromDatasetFileNode(node)).toBe("sub/dir/f.txt"); + it("returns the bare file name for a file at the version root", () => { + const node: DatasetFileNode = { + name: "readme.txt", + type: "file", + parentDir: "/datasets/bob@texera.com/twitterDataset/v1", + }; + expect(getRelativePathFromDatasetFileNode(node)).toBe("readme.txt"); }); - it("returns an empty string when there are three or fewer segments", () => { - const node: DatasetFileNode = { name: "v1", type: "directory", parentDir: "/owner/dataset" }; - // full path /owner/dataset/v1 -> exactly 3 segments -> no relative path + it("returns an empty string when there is no path below the version", () => { + const node: DatasetFileNode = { + name: "v1", + type: "directory", + parentDir: "/datasets/bob@texera.com/twitterDataset", + }; expect(getRelativePathFromDatasetFileNode(node)).toBe(""); }); it("ignores empty segments from duplicate slashes when counting", () => { - const node: DatasetFileNode = { name: "file.csv", type: "file", parentDir: "/owner//dataset/v1" }; - // empty segment between the duplicate slashes is filtered out, leaving 4 real segments - expect(getRelativePathFromDatasetFileNode(node)).toBe("file.csv"); + const node: DatasetFileNode = { + name: "f.csv", + type: "file", + parentDir: "/datasets/bob@texera.com/twitterDataset//v1/sub", + }; + expect(getRelativePathFromDatasetFileNode(node)).toBe("sub/f.csv"); }); }); diff --git a/frontend/src/app/common/type/datasetVersionFileTree.ts b/frontend/src/app/common/type/datasetVersionFileTree.ts index 8d1686998ca..98f898d4432 100644 --- a/frontend/src/app/common/type/datasetVersionFileTree.ts +++ b/frontend/src/app/common/type/datasetVersionFileTree.ts @@ -31,19 +31,20 @@ export function getFullPathFromDatasetFileNode(node: DatasetFileNode): string { } /** - * Returns the relative path of a DatasetFileNode by stripping the first three segments. + * Returns the relative path of a DatasetFileNode by stripping the first four segments + * (datasets/ownerEmail/datasetName/versionName). * @param node The DatasetFileNode whose relative path is needed. - * @returns The relative path (without the first three segments and without a leading slash). + * @returns The relative path (without the first four segments and without a leading slash). */ export function getRelativePathFromDatasetFileNode(node: DatasetFileNode): string { const fullPath = getFullPathFromDatasetFileNode(node); // Get the full path const pathSegments = fullPath.split("/").filter(segment => segment.length > 0); // Split and remove empty segments - if (pathSegments.length <= 3) { - return ""; // If there are 3 or fewer segments, return an empty string (no relative path exists) + if (pathSegments.length <= 4) { + return ""; // If there are 4 or fewer segments, return an empty string (no relative path exists) } - return pathSegments.slice(3).join("/"); // Join remaining segments as the relative path + return pathSegments.slice(4).join("/"); // Join remaining segments as the relative path } export function getPathsUnderOrEqualDatasetFileNode(node: DatasetFileNode): string[] { diff --git a/frontend/src/app/common/type/resource-type.ts b/frontend/src/app/common/type/resource-type.ts new file mode 100644 index 00000000000..a129c917e27 --- /dev/null +++ b/frontend/src/app/common/type/resource-type.ts @@ -0,0 +1,24 @@ +/** + * 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. + */ + +/** + * The leading segment of a logical file path, identifying the resource kind (e.g. /datasets/...). + */ +export enum ResourceType { + Datasets = "datasets", +} diff --git a/frontend/src/app/dashboard/component/user/user-dataset/user-dataset-explorer/user-dataset-version-filetree/user-dataset-version-filetree.component.spec.ts b/frontend/src/app/dashboard/component/user/user-dataset/user-dataset-explorer/user-dataset-version-filetree/user-dataset-version-filetree.component.spec.ts index 0abb9626218..673c1d39bda 100644 --- a/frontend/src/app/dashboard/component/user/user-dataset/user-dataset-explorer/user-dataset-version-filetree/user-dataset-version-filetree.component.spec.ts +++ b/frontend/src/app/dashboard/component/user/user-dataset/user-dataset-explorer/user-dataset-version-filetree/user-dataset-version-filetree.component.spec.ts @@ -167,9 +167,9 @@ describe("UserDatasetVersionFiletreeComponent", () => { const emitted: string[] = []; component.setCoverImage.subscribe((path: string) => emitted.push(path)); - // parentDir has exactly the three stripped segments (owner/dataset/version), + // parentDir has exactly the four stripped segments (datasets/owner/dataset/version), // so the relative path is just the file name. - component.onSetCover({ name: "photo.png", type: "file", parentDir: "/owner/dataset/v1" }); + component.onSetCover({ name: "photo.png", type: "file", parentDir: "/datasets/owner/dataset/v1" }); expect(emitted).toEqual(["photo.png"]); }); diff --git a/frontend/src/app/workspace/component/dataset-selection-modal/dataset-selection-modal.component.spec.ts b/frontend/src/app/workspace/component/dataset-selection-modal/dataset-selection-modal.component.spec.ts index bf3bfdb1334..d1483b71ad5 100644 --- a/frontend/src/app/workspace/component/dataset-selection-modal/dataset-selection-modal.component.spec.ts +++ b/frontend/src/app/workspace/component/dataset-selection-modal/dataset-selection-modal.component.spec.ts @@ -108,7 +108,7 @@ describe("DatasetSelectionModalComponent", () => { it("ngOnInit initializes selectedDataset and selectedVersion from data.selectedPath", () => { modalData.fileMode = true; - modalData.selectedPath = `/${OWNER}/myds/v1`; + modalData.selectedPath = `/datasets/${OWNER}/myds/v1`; build(); @@ -141,7 +141,7 @@ describe("DatasetSelectionModalComponent", () => { expect(datasetService.retrieveDatasetVersionFileTree).toHaveBeenCalledWith(10, 100); expect(component.fileTree).toEqual([fileNode]); - expect(component.selectedPath).toBe(`/${OWNER}/myds/v1`); + expect(component.selectedPath).toBe(`/datasets/${OWNER}/myds/v1`); }); it("onFileSelected sets selectedPath to the node's full path in file mode", () => { diff --git a/frontend/src/app/workspace/component/dataset-selection-modal/dataset-selection-modal.component.ts b/frontend/src/app/workspace/component/dataset-selection-modal/dataset-selection-modal.component.ts index fb289b02904..5ee5b9115e8 100644 --- a/frontend/src/app/workspace/component/dataset-selection-modal/dataset-selection-modal.component.ts +++ b/frontend/src/app/workspace/component/dataset-selection-modal/dataset-selection-modal.component.ts @@ -22,6 +22,7 @@ import { NZ_MODAL_DATA, NzModalRef } from "ng-zorro-antd/modal"; import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy"; import { DatasetFileNode, getFullPathFromDatasetFileNode } from "../../../common/type/datasetVersionFileTree"; import { DatasetVersion } from "../../../common/type/dataset"; +import { ResourceType } from "../../../common/type/resource-type"; import { DashboardDataset } from "../../../dashboard/type/dashboard-dataset.interface"; import { DatasetService } from "../../../dashboard/service/user/dataset/dataset.service"; import { NzRowDirective, NzColDirective } from "ng-zorro-antd/grid"; @@ -83,7 +84,8 @@ export class DatasetSelectionModalComponent implements OnInit { this.datasets = datasets; const selectedPath = this.data.selectedPath; if (selectedPath) { - const [ownerEmail, datasetName, versionName] = selectedPath.split("/").filter(part => part.length > 0); + // Stored paths always carry the resource-type prefix; skip it so that owner/dataset/version line up. + const [, ownerEmail, datasetName, versionName] = selectedPath.split("/").filter(part => part.length > 0); this.selectedDataset = this.datasets.find( dataset => dataset.ownerEmail === ownerEmail && dataset.dataset.name === datasetName ); @@ -118,7 +120,7 @@ export class DatasetSelectionModalComponent implements OnInit { this.fileTree = data.fileNodes; }); if (!this.data.fileMode) { - this.selectedPath = `/${this.selectedDataset.ownerEmail}/${this.selectedDataset.dataset.name}/${this.selectedVersion.name}`; + this.selectedPath = `/${ResourceType.Datasets}/${this.selectedDataset.ownerEmail}/${this.selectedDataset.dataset.name}/${this.selectedVersion.name}`; } } } diff --git a/sql/changelog.xml b/sql/changelog.xml index 469868f8958..ac90daed46a 100644 --- a/sql/changelog.xml +++ b/sql/changelog.xml @@ -53,6 +53,16 @@ + + + + + + + + + +