feat(file-service): add model version upload

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}"

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"

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)

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

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
       },

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": [

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"
 

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)