You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/09/21 21:54:24 UTC
(texera) 01/02: feat(LakeFS-mount): name a model or dataset from a Python UDF parameter (#6897)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 60a37b1e8495b7d16b1458560ae48d0834f7c513
Author: ali risheh <al...@gmail.com>
AuthorDate: Mon Sep 21 21:27:45 2026 +0000
feat(LakeFS-mount): name a model or dataset from a Python UDF parameter (#6897)
### What changes were proposed in this PR?
Lets a Python UDF name a model or dataset version as a parameter and
read it from a local directory. This is the last step of mounting
(#6606).
```python
self.model_dir = self.UiParameter("MODEL", AttributeType.STRING, value=Resource.MODEL).value
```
- pytexera gains `Resource` (`MODEL`, `DATASET`), accepted as `value=`
on string parameters only.
- The editor recognizes that declaration, and the parameter row offers a
model or dataset browser instead of a text box. The chosen version is
stored as its path. The model browser is new.
- Resolution happens when the execution starts, not on compile: the
operator carries an execution-time binding that resolves each version
path to its repository and commit, then gives the UDF the mount
directory in the path's place.
- The region scheduler collects the distinct repositories its operators
name and mounts them once, before any worker gets its code. A
reconfiguration mounts the same way.
- The engine asks access-control-service for each mount with the
computing unit's own JWT, and #6896 authorizes the request. The engine
then waits for the mount to appear in the pod. The manager passes the
service's address to the pod, taken from the chart.
The request carries the identity of the computing unit's owner. A WRITE
sharee can therefore mount only what the owner can also read; a version
only the sharee can read is refused with 403.
### Any related issues, documentation, discussions?
Closes #6895. Part of #6606. Follows #6866 and #6896, both merged.
### How was this PR tested?
- Frontend unit tests: the parser, parameter sync, the resource value
editor, the model browser, and the model service.
- pytexera unit tests: `Resource` declarations.
- Scala unit tests:
- The binding: a dataset and a model resolved against a test database,
and a missing version refused.
- The UDF descriptors.
- The region scheduler: it mounts before initializing workers, fails the
region before any worker starts when a mount is refused, and does not
mount when nothing is named.
- Reconfiguration.
- End to end on minikube: a source UDF with a dataset parameter and a
model parameter, both picked in the property panel. The run mounted both
through access-control-service and the node mounter, and the UDF read a
file from each directory.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_01KCY3USGKcgmf8CbN44trJ5
---------
Co-authored-by: Claude Opus 5 (1M context) <no...@anthropic.com>
---
access-control-service/LICENSE-binary | 33 +++-
access-control-service/build.sbt | 6 +-
.../service/util/ComputingUnitNodeLocator.scala | 86 +++-------
.../util/ComputingUnitNodeLocatorSpec.scala | 19 ++-
amber/src/main/python/pytexera/__init__.py | 2 +
amber/src/main/python/pytexera/udf/udf_operator.py | 33 ++++
.../scheduling/RegionExecutionManager.scala | 32 +++-
.../service/ExecutionReconfigurationService.scala | 10 +-
.../test/python/pytexera/udf/test_udf_operator.py | 41 ++++-
.../scheduling/RegionExecutionManagerSpec.scala | 60 ++++++-
.../ExecutionReconfigurationServiceSpec.scala | 23 ++-
...workflow-computing-unit-manager-deployment.yaml | 7 +
bin/k8s/values.yaml | 2 +
build.sbt | 2 +
.../common/config/EnvironmentalVariable.scala | 1 +
.../texera/amber/core/storage/FileResolver.scala | 66 +++++++-
.../core/storage/RepositoryMountManager.scala | 182 ++++++++++++++++++++
.../amber/core/workflow/ExecutionTimeBinding.scala | 39 +++++
.../texera/amber/core/workflow/PhysicalOp.scala | 26 ++-
.../amber/core/workflow/PhysicalOpSpec.scala | 20 +++
.../texera/amber/storage/FileResolverSpec.scala | 50 ++++++
.../amber/storage/RepositoryMountManagerSpec.scala | 118 +++++++++++++
.../python/DualInputPortsPythonUDFOpDescV2.scala | 2 +
.../operator/udf/python/PythonUDFOpDescV2.scala | 2 +
.../udf/python/PythonUdfUiParameterSupport.scala | 80 ++++++++-
.../amber/operator/udf/python/UiUDFParameter.scala | 9 +
.../python/source/PythonUDFSourceOpDescV2.scala | 2 +
.../DualInputPortsPythonUDFOpDescV2Spec.scala | 17 ++
.../udf/python/PythonUDFOpDescV2Spec.scala | 39 +++++
.../python/PythonUdfUiParameterSupportSpec.scala | 188 +++++++++++++++++++++
.../operator/udf/python/UiUDFParameterSpec.scala | 10 ++
.../source/PythonUDFSourceOpDescV2Spec.scala | 17 ++
.../texera/service/util/KubernetesClient.scala | 35 +++-
.../texera/service/util/KubernetesClientSpec.scala | 20 ++-
frontend/src/app/common/formly/formly-config.ts | 2 +
frontend/src/app/common/type/resource-type.ts | 1 +
.../service/user/model/model.service.spec.ts | 7 +
.../dashboard/service/user/model/model.service.ts | 6 +
.../model-selection-modal.component.html | 69 ++++++++
.../model-selection-modal.component.scss | 27 ++-
.../model-selection-modal.component.spec.ts | 143 ++++++++++++++++
.../model-selection-modal.component.ts | 129 ++++++++++++++
.../resource-value-selector.component.html | 32 ++++
.../resource-value-selector.component.scss | 29 ++--
.../resource-value-selector.component.spec.ts | 83 +++++++++
.../resource-value-selector.component.ts | 118 +++++++++++++
.../ui-udf-parameters.component.spec.ts | 87 +++++++++-
.../ui-udf-parameters.component.ts | 58 ++++++-
.../ui-udf-parameters-parser.service.spec.ts | 40 +++++
.../ui-udf-parameters-parser.service.ts | 31 +++-
.../ui-udf-parameters-sync.service.spec.ts | 28 +++
.../code-editor/ui-udf-parameters-sync.service.ts | 9 +-
52 files changed, 2023 insertions(+), 155 deletions(-)
diff --git a/access-control-service/LICENSE-binary b/access-control-service/LICENSE-binary
index 78f1df46a9..4f40d12738 100644
--- a/access-control-service/LICENSE-binary
+++ b/access-control-service/LICENSE-binary
@@ -224,10 +224,10 @@ Scala/Java jars:
- com.fasterxml.jackson.core.jackson-annotations-2.18.8.jar
- com.fasterxml.jackson.core.jackson-core-2.18.8.jar
- com.fasterxml.jackson.core.jackson-databind-2.18.8.jar
- - com.fasterxml.jackson.dataformat.jackson-dataformat-yaml-2.16.1.jar
+ - com.fasterxml.jackson.dataformat.jackson-dataformat-yaml-2.17.0.jar
- com.fasterxml.jackson.datatype.jackson-datatype-guava-2.16.1.jar
- com.fasterxml.jackson.datatype.jackson-datatype-jdk8-2.16.1.jar
- - com.fasterxml.jackson.datatype.jackson-datatype-jsr310-2.16.1.jar
+ - com.fasterxml.jackson.datatype.jackson-datatype-jsr310-2.17.0.jar
- com.fasterxml.jackson.jakarta.rs.jackson-jakarta-rs-base-2.16.1.jar
- com.fasterxml.jackson.jakarta.rs.jackson-jakarta-rs-json-provider-2.16.1.jar
- com.fasterxml.jackson.module.jackson-module-blackbird-2.16.1.jar
@@ -274,6 +274,32 @@ Scala/Java jars:
- io.dropwizard.metrics.metrics-json-4.2.25.jar
- io.dropwizard.metrics.metrics-jvm-4.2.25.jar
- io.dropwizard.metrics.metrics-logback-4.2.25.jar
+ - io.fabric8.kubernetes-client-6.12.1.jar
+ - io.fabric8.kubernetes-client-api-6.12.1.jar
+ - io.fabric8.kubernetes-httpclient-jdk-6.12.1.jar
+ - io.fabric8.kubernetes-model-admissionregistration-6.12.1.jar
+ - io.fabric8.kubernetes-model-apiextensions-6.12.1.jar
+ - io.fabric8.kubernetes-model-apps-6.12.1.jar
+ - io.fabric8.kubernetes-model-autoscaling-6.12.1.jar
+ - io.fabric8.kubernetes-model-batch-6.12.1.jar
+ - io.fabric8.kubernetes-model-certificates-6.12.1.jar
+ - io.fabric8.kubernetes-model-common-6.12.1.jar
+ - io.fabric8.kubernetes-model-coordination-6.12.1.jar
+ - io.fabric8.kubernetes-model-core-6.12.1.jar
+ - io.fabric8.kubernetes-model-discovery-6.12.1.jar
+ - io.fabric8.kubernetes-model-events-6.12.1.jar
+ - io.fabric8.kubernetes-model-extensions-6.12.1.jar
+ - io.fabric8.kubernetes-model-flowcontrol-6.12.1.jar
+ - io.fabric8.kubernetes-model-gatewayapi-6.12.1.jar
+ - io.fabric8.kubernetes-model-metrics-6.12.1.jar
+ - io.fabric8.kubernetes-model-networking-6.12.1.jar
+ - io.fabric8.kubernetes-model-node-6.12.1.jar
+ - io.fabric8.kubernetes-model-policy-6.12.1.jar
+ - io.fabric8.kubernetes-model-rbac-6.12.1.jar
+ - io.fabric8.kubernetes-model-resource-6.12.1.jar
+ - io.fabric8.kubernetes-model-scheduling-6.12.1.jar
+ - io.fabric8.kubernetes-model-storageclass-6.12.1.jar
+ - io.fabric8.zjsonpatch-0.3.0.jar
- io.r2dbc.r2dbc-spi-1.0.0.RELEASE.jar
- jakarta.inject.jakarta.inject-api-2.0.1.jar
- jakarta.validation.jakarta.validation-api-3.0.2.jar
@@ -300,6 +326,7 @@ Scala/Java jars:
- org.scala-lang.scala-reflect-2.13.18.jar
- org.slf4j.jcl-over-slf4j-2.0.12.jar
- org.slf4j.log4j-over-slf4j-2.0.12.jar
+ - org.snakeyaml.snakeyaml-engine-2.7.jar
- org.yaml.snakeyaml-2.2.jar
--------------------------------------------------------------------------------
@@ -327,7 +354,7 @@ Scala/Java jars:
- net.sourceforge.argparse4j.argparse4j-0.9.0.jar
- org.checkerframework.checker-qual-3.52.0.jar
- org.slf4j.jul-to-slf4j-2.0.12.jar
- - org.slf4j.slf4j-api-2.0.12.jar
+ - org.slf4j.slf4j-api-2.0.13.jar
--------------------------------------------------------------------------------
Dependencies under the BSD 3-Clause License
diff --git a/access-control-service/build.sbt b/access-control-service/build.sbt
index 93af38e442..e524ddd3ce 100644
--- a/access-control-service/build.sbt
+++ b/access-control-service/build.sbt
@@ -82,5 +82,9 @@ libraryDependencies ++= Seq(
libraryDependencies ++= Seq(
"io.dropwizard" % "dropwizard-core" % dropwizardVersion,
"io.dropwizard" % "dropwizard-auth" % dropwizardVersion, // Dropwizard Authentication module
- "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.18.8"
+ "com.fasterxml.jackson.module" %% "jackson-module-scala" % "2.18.8",
+ // Reads the pod a mount is bound for. The okhttp backend is swapped for the JDK one:
+ // it drops three jars and okhttp 3.x, which is end of life.
+ "io.fabric8" % "kubernetes-client" % "6.12.1" exclude ("io.fabric8", "kubernetes-httpclient-okhttp"),
+ "io.fabric8" % "kubernetes-httpclient-jdk" % "6.12.1"
)
\ No newline at end of file
diff --git a/access-control-service/src/main/scala/org/apache/texera/service/util/ComputingUnitNodeLocator.scala b/access-control-service/src/main/scala/org/apache/texera/service/util/ComputingUnitNodeLocator.scala
index 237d6cb2c0..92686d9ea8 100644
--- a/access-control-service/src/main/scala/org/apache/texera/service/util/ComputingUnitNodeLocator.scala
+++ b/access-control-service/src/main/scala/org/apache/texera/service/util/ComputingUnitNodeLocator.scala
@@ -19,20 +19,11 @@
package org.apache.texera.service.util
-import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
import com.typesafe.scalalogging.LazyLogging
+import io.fabric8.kubernetes.api.model.Pod
+import io.fabric8.kubernetes.client.{KubernetesClientBuilder, KubernetesClientException}
import org.apache.texera.common.config.KubernetesConfig
-import java.io.FileInputStream
-import java.net.URI
-import java.net.http.{HttpClient, HttpRequest, HttpResponse}
-import java.nio.file.{Files, Paths}
-import java.security.KeyStore
-import java.security.cert.CertificateFactory
-import java.time.Duration
-import javax.net.ssl.{SSLContext, TrustManagerFactory}
-import scala.jdk.CollectionConverters._
-
/**
* Finds the node a computing unit's pod runs on, so a mount can be sent to that node's
* mounter.
@@ -41,73 +32,36 @@ import scala.jdk.CollectionConverters._
* hand anything that can reach this service the ability to aim requests at any node's
* privileged mounter.
*/
-class ComputingUnitNodeLocator(fetchPod: String => Option[JsonNode]) extends LazyLogging {
+class ComputingUnitNodeLocator(fetchPod: String => Option[Pod]) extends LazyLogging {
def nodeIpOf(cuid: Int): Option[String] = {
val podName = s"${KubernetesConfig.computeUnitPodNamePrefix}-$cuid"
- fetchPod(podName).map(_.at("/status/hostIP").asText("")).filter(_.nonEmpty)
+ // Every field is nullable until the pod is scheduled, which reads as "no node yet".
+ fetchPod(podName)
+ .flatMap(pod => Option(pod.getStatus))
+ .flatMap(status => Option(status.getHostIP))
+ .filter(_.nonEmpty)
}
}
object ComputingUnitNodeLocator extends ComputingUnitNodeLocator(InClusterKubernetesApi.getPod)
-// One read of one field, so the API is called directly rather than through a Kubernetes
-// client library and its transitive dependencies.
-private[util] object InClusterKubernetesApi extends LazyLogging {
-
- private val serviceAccountDir = "/var/run/secrets/kubernetes.io/serviceaccount"
- private val mapper = new ObjectMapper()
+private[util] object InClusterKubernetesApi {
- private lazy val client: HttpClient =
- HttpClient
- .newBuilder()
- .connectTimeout(Duration.ofSeconds(5))
- .sslContext(clusterSslContext)
- .build()
-
- /** Trusts only the cluster CA, so this talks to the API server and nothing else. */
- private def clusterSslContext: SSLContext = {
- val certificates = {
- val stream = new FileInputStream(s"$serviceAccountDir/ca.crt")
- try CertificateFactory.getInstance("X.509").generateCertificates(stream).asScala.toList
- finally stream.close()
- }
- val keyStore = KeyStore.getInstance(KeyStore.getDefaultType)
- keyStore.load(null, null)
- certificates.zipWithIndex.foreach {
- case (certificate, index) => keyStore.setCertificateEntry(s"cluster-ca-$index", certificate)
- }
- val trustManagerFactory =
- TrustManagerFactory.getInstance(TrustManagerFactory.getDefaultAlgorithm)
- trustManagerFactory.init(keyStore)
- val context = SSLContext.getInstance("TLS")
- context.init(null, trustManagerFactory.getTrustManagers, null)
- context
- }
+ // In-cluster config: the client reads the API address, the cluster CA and the projected
+ // service-account token itself, and refreshes that token as the kubelet rotates it.
+ private lazy val client = new KubernetesClientBuilder().build()
- def getPod(podName: String): Option[JsonNode] = {
- val host = sys.env.getOrElse("KUBERNETES_SERVICE_HOST", "kubernetes.default.svc")
- val port = sys.env.getOrElse("KUBERNETES_SERVICE_PORT", "443")
+ def getPod(podName: String): Option[Pod] = {
val namespace = KubernetesConfig.computeUnitPoolNamespace
- val token = Files.readString(Paths.get(s"$serviceAccountDir/token")).trim
-
- val request = HttpRequest
- .newBuilder()
- .uri(URI.create(s"https://$host:$port/api/v1/namespaces/$namespace/pods/$podName"))
- .header("Authorization", s"Bearer $token")
- .timeout(Duration.ofSeconds(10))
- .GET()
- .build()
-
- val response = client.send(request, HttpResponse.BodyHandlers.ofString())
- response.statusCode() match {
- case 200 => Some(mapper.readTree(response.body()))
- case 404 => None
- case other =>
- // Distinguished from a missing pod: a missing RBAC rule or an unreachable API
- // server must not read as "the computing unit is not running".
+ try Option(client.pods().inNamespace(namespace).withName(podName).get())
+ catch {
+ case e: KubernetesClientException =>
+ // Distinguished from a missing pod, which reads as None: a missing RBAC rule or an
+ // unreachable API server must not read as "the computing unit is not running".
throw new IllegalStateException(
- s"cannot read pod $podName in namespace $namespace: HTTP $other ${response.body()}"
+ s"cannot read pod $podName in namespace $namespace: ${e.getMessage}",
+ e
)
}
}
diff --git a/access-control-service/src/test/scala/org/apache/texera/service/util/ComputingUnitNodeLocatorSpec.scala b/access-control-service/src/test/scala/org/apache/texera/service/util/ComputingUnitNodeLocatorSpec.scala
index 33f3a6a034..a24f5d8a4e 100644
--- a/access-control-service/src/test/scala/org/apache/texera/service/util/ComputingUnitNodeLocatorSpec.scala
+++ b/access-control-service/src/test/scala/org/apache/texera/service/util/ComputingUnitNodeLocatorSpec.scala
@@ -19,7 +19,7 @@
package org.apache.texera.service.util
-import com.fasterxml.jackson.databind.{JsonNode, ObjectMapper}
+import io.fabric8.kubernetes.api.model.{Pod, PodBuilder}
import org.apache.texera.common.config.KubernetesConfig
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
@@ -28,11 +28,14 @@ import scala.collection.mutable
class ComputingUnitNodeLocatorSpec extends AnyFlatSpec with Matchers {
- private val mapper = new ObjectMapper()
- private def pod(json: String): JsonNode = mapper.readTree(json)
+ private def pod(hostIp: Option[String]): Pod = {
+ val status = new PodBuilder().withNewStatus().withPhase("Pending")
+ hostIp.foreach(status.withHostIP)
+ status.endStatus().build()
+ }
private def locator(
- pods: Map[String, JsonNode],
+ pods: Map[String, Pod],
asked: mutable.Buffer[String] = mutable.Buffer()
): ComputingUnitNodeLocator =
new ComputingUnitNodeLocator(name => { asked += name; pods.get(name) })
@@ -40,8 +43,7 @@ class ComputingUnitNodeLocatorSpec extends AnyFlatSpec with Matchers {
private val podName = s"${KubernetesConfig.computeUnitPodNamePrefix}-7"
"nodeIpOf" should "return the host IP of the computing unit's pod" in {
- locator(Map(podName -> pod("""{"status":{"hostIP":"10.0.0.4"}}"""))).nodeIpOf(7) shouldBe
- Some("10.0.0.4")
+ locator(Map(podName -> pod(Some("10.0.0.4")))).nodeIpOf(7) shouldBe Some("10.0.0.4")
}
it should "ask for the pod named by the configured prefix and the cuid" in {
@@ -55,8 +57,9 @@ class ComputingUnitNodeLocatorSpec extends AnyFlatSpec with Matchers {
}
it should "return None while the pod is not scheduled yet" in {
- locator(Map(podName -> pod("""{"status":{"phase":"Pending"}}"""))).nodeIpOf(7) shouldBe None
- locator(Map(podName -> pod("""{"status":{"hostIP":""}}"""))).nodeIpOf(7) shouldBe None
+ locator(Map(podName -> pod(None))).nodeIpOf(7) shouldBe None
+ locator(Map(podName -> pod(Some("")))).nodeIpOf(7) shouldBe None
+ locator(Map(podName -> new PodBuilder().build())).nodeIpOf(7) shouldBe None
}
it should "propagate a lookup failure rather than reporting the unit as unscheduled" in {
diff --git a/amber/src/main/python/pytexera/__init__.py b/amber/src/main/python/pytexera/__init__.py
index 4c5db21491..4e64b753ef 100644
--- a/amber/src/main/python/pytexera/__init__.py
+++ b/amber/src/main/python/pytexera/__init__.py
@@ -25,6 +25,7 @@ from .storage.dataset_file_document import DatasetFileDocument
from .storage.large_binary_input_stream import LargeBinaryInputStream
from .storage.large_binary_output_stream import LargeBinaryOutputStream
from .udf.udf_operator import (
+ Resource,
UDFOperatorV2,
UDFTableOperator,
UDFBatchOperator,
@@ -60,4 +61,5 @@ __all__ = [
"Dict",
"Any",
"AttributeType",
+ "Resource",
]
diff --git a/amber/src/main/python/pytexera/udf/udf_operator.py b/amber/src/main/python/pytexera/udf/udf_operator.py
index 4727c9399a..a3ed76a4a3 100644
--- a/amber/src/main/python/pytexera/udf/udf_operator.py
+++ b/amber/src/main/python/pytexera/udf/udf_operator.py
@@ -17,6 +17,7 @@
from abc import abstractmethod
from dataclasses import dataclass
+from enum import Enum
import functools
from typing import Any, Dict, Iterator, Optional, Set, Union
@@ -25,11 +26,25 @@ from core.models.schema.attribute_type import AttributeType, FROM_STRING_PARSER_
from loguru import logger
+class Resource(Enum):
+ """A Texera resource whose version a UiParameter names, instead of free text.
+
+ model_dir = self.UiParameter("MODEL", AttributeType.STRING, value=Resource.MODEL).value
+
+ The property panel offers that resource's browser, and the value the UDF receives
+ is the local directory the chosen version is mounted at.
+ """
+
+ MODEL = "model"
+ DATASET = "dataset"
+
+
@dataclass(frozen=True)
class _UiParameterValue:
name: str
type: AttributeType
value: Any
+ resource: Optional[Resource] = None
class _UiParameterSupport:
@@ -104,6 +119,9 @@ class _UiParameterSupport:
"""
Return the current UI parameter value parsed as attr_type.
+ Pass value=Resource.MODEL or value=Resource.DATASET to have it name a version
+ of that resource; the parameter must then be a string.
+
Re-reading the same name with the same type is idempotent. Reusing a
name with a different type is rejected because the parsed value would be
ambiguous.
@@ -113,6 +131,8 @@ class _UiParameterSupport:
raise TypeError("UiParameter.type was provided multiple times.")
attr_type = kwargs.pop("type")
+ resource = kwargs.pop("value", None)
+
if kwargs:
unexpected_arguments = ", ".join(sorted(kwargs))
raise TypeError(
@@ -128,6 +148,18 @@ class _UiParameterSupport:
f"UiParameter.type must be an AttributeType, got {attr_type!r}."
)
+ if resource is not None:
+ if not isinstance(resource, Resource):
+ raise TypeError(
+ f"UiParameter.value must be a Resource, got {resource!r}."
+ )
+ # What the UDF receives is a directory path.
+ if attr_type is not AttributeType.STRING:
+ raise TypeError(
+ f"UiParameter '{name}' names a {resource.name.lower()}, so its type "
+ f"must be AttributeType.STRING, not {attr_type.name}."
+ )
+
self._ensure_ui_parameter_state()
existing_type = self._ui_parameter_name_types.get(name)
if existing_type is not None and existing_type != attr_type:
@@ -151,6 +183,7 @@ class _UiParameterSupport:
name=name,
type=attr_type,
value=self._parse(raw_value, attr_type),
+ resource=resource,
)
@staticmethod
diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
index 2f19887cc3..c7371f8c47 100644
--- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
+++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManager.scala
@@ -20,9 +20,9 @@
package org.apache.texera.amber.engine.architecture.scheduling
import org.apache.pekko.pattern.gracefulStop
-import com.twitter.util.{Future, JavaTimer, Return, Throw, Timer}
+import com.twitter.util.{Future, JavaTimer, Promise, Return, Throw, Timer}
import org.apache.texera.amber.core.state.State
-import org.apache.texera.amber.core.storage.{DocumentFactory, VFSURIFactory}
+import org.apache.texera.amber.core.storage.{DocumentFactory, RepositoryMountManager, VFSURIFactory}
import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity
import org.apache.texera.amber.core.workflow.{GlobalPortIdentity, PhysicalLink, PhysicalOp}
import org.apache.texera.amber.engine.architecture.common.{
@@ -116,7 +116,9 @@ class RegionExecutionManager(
// Loop bookkeeping addresses (Loop Start logical op id -> its input
// port's BASE materialization URI), shipped to every worker in
// InitializeExecutorRequest. See WorkflowExecutionManager.loopStartPortUris.
- loopStartPortUris: Map[String, String] = Map.empty
+ loopStartPortUris: Map[String, String] = Map.empty,
+ // Mounts the repositories a region's operators name; a parameter so a test can observe it.
+ ensureMounted: Set[String] => Unit = RepositoryMountManager.ensureAllMounted
) extends AmberLogging {
initRegionExecution()
@@ -131,6 +133,9 @@ class RegionExecutionManager(
Unexecuted
)
private val terminationFutureRef: AtomicReference[Future[Unit]] = new AtomicReference(null)
+ // Both launch phases pass through launchPhaseExecutionInternal; the second must await the
+ // first's mount rather than repeat it or race past it, hence a shared future, not a flag.
+ private val mountFutureRef: AtomicReference[Future[Unit]] = new AtomicReference(null)
/**
* Sync the status of `RegionExecution` and transition this manager's phase to `Completed` only when the
@@ -352,6 +357,23 @@ class RegionExecutionManager(
)
}
+ /**
+ * Mount every repository this region's operators name before any of them receives its code,
+ * which already refers to the mount paths. It happens here rather than in a worker because
+ * workers are created before either phase launches. Operators carry no mount code: they only
+ * name what they need, through `PhysicalOp.mountLocators`.
+ */
+ private def mountRegionRepositories(): Future[Unit] = {
+ val mount = Promise[Unit]()
+ if (mountFutureRef.compareAndSet(null, mount)) {
+ mount.become(Future {
+ val locators = region.getOperators.flatMap(_.mountLocators)
+ if (locators.nonEmpty) ensureMounted(locators)
+ })
+ }
+ mountFutureRef.get
+ }
+
/**
* Unified logic for launching either of the two phases asynchronously.
*/
@@ -383,6 +405,7 @@ class RegionExecutionManager(
)
)
Future(())
+ .flatMap(_ => mountRegionRepositories())
.flatMap(_ => initExecutors(operatorsToRun, resourceConfig))
.flatMap(_ => assignPortsLogic())
.flatMap(_ => connectChannelsLogic())
@@ -446,11 +469,12 @@ class RegionExecutionManager(
operators
.flatMap(physicalOp => {
val workerConfigs = resourceConfig.operatorConfigs(physicalOp.id).workerConfigs
+ val opExecInitInfo = physicalOp.executableOpExecInitInfo
workerConfigs.map(_.workerId).map { workerId =>
asyncRPCClient.workerInterface.initializeExecutor(
InitializeExecutorRequest(
workerConfigs.length,
- physicalOp.opExecInitInfo,
+ opExecInitInfo,
physicalOp.isSourceOperator,
loopStartPortUris
),
diff --git a/amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala b/amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala
index 2147231abb..d1da4b4935 100644
--- a/amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala
+++ b/amber/src/main/scala/org/apache/texera/web/service/ExecutionReconfigurationService.scala
@@ -19,6 +19,7 @@
package org.apache.texera.web.service
+import org.apache.texera.amber.core.storage.RepositoryMountManager
import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity
import org.apache.texera.amber.engine.architecture.coordinator.{UpdateExecutorCompleted, Workflow}
import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{
@@ -90,7 +91,10 @@ class ExecutionReconfigurationService(
val reconfigurationId = UUID.randomUUID().toString
val updateExecutorRequests = reconfigurations.map {
- case (op, _) => UpdateExecutorRequest(op.id, op.opExecInitInfo)
+ case (op, _) =>
+ // An edited UDF may now name a repository, and its code already refers to the mount.
+ ensureMounted(op.mountLocators)
+ UpdateExecutorRequest(op.id, op.executableOpExecInitInfo)
}
dispatch(
WorkflowReconfigureRequest(
@@ -105,6 +109,10 @@ class ExecutionReconfigurationService(
)
}
+ // Seam for unit testing without a computing unit to mount into.
+ protected def ensureMounted(locators: Set[String]): Unit =
+ RepositoryMountManager.ensureAllMounted(locators)
+
// Seam for unit testing the dispatch path without spinning up an AmberClient.
protected def dispatch(request: WorkflowReconfigureRequest): Unit = {
client.coordinatorInterface.reconfigureWorkflow(request, ())
diff --git a/amber/src/test/python/pytexera/udf/test_udf_operator.py b/amber/src/test/python/pytexera/udf/test_udf_operator.py
index 1d93eba5e4..86cd2a88c5 100644
--- a/amber/src/test/python/pytexera/udf/test_udf_operator.py
+++ b/amber/src/test/python/pytexera/udf/test_udf_operator.py
@@ -22,7 +22,7 @@ import pandas
import pytest
import pytexera.udf.udf_operator as udf_operator
-from pytexera import AttributeType, Tuple, TupleLike, UDFOperatorV2
+from pytexera import AttributeType, Resource, Tuple, TupleLike, UDFOperatorV2
from pytexera import UDFBatchOperator, UDFSourceOperator, UDFTableOperator
from pytexera.udf.udf_operator import _UiParameterSupport
@@ -48,6 +48,19 @@ class InjectedParametersOperator(UDFOperatorV2):
yield tuple_
+class ResourceParameterOperator(UDFOperatorV2):
+ def _texera_injected_ui_parameters(self):
+ return {"MODEL": "/mnt/texera-mounts/model-1/abc123"}
+
+ def open(self):
+ self.model_parameter = self.UiParameter(
+ "MODEL", AttributeType.STRING, value=Resource.MODEL
+ )
+
+ def process_tuple(self, tuple_: Tuple, port: int) -> Iterator[Optional[TupleLike]]:
+ yield tuple_
+
+
class ConflictingParameterOperator(UDFOperatorV2):
def _texera_injected_ui_parameters(self):
return {"duplicate": "1"}
@@ -349,12 +362,36 @@ class TestUiParameterSupport:
with pytest.raises(TypeError, match="provided multiple times"):
operator.UiParameter("count", AttributeType.INT, type=AttributeType.INT)
with pytest.raises(TypeError, match="unexpected keyword argument"):
- operator.UiParameter("count", AttributeType.INT, value="1")
+ operator.UiParameter("count", AttributeType.INT, default="1")
with pytest.raises(TypeError, match="UiParameter.type is required"):
operator.UiParameter("count")
with pytest.raises(TypeError, match="must be an AttributeType"):
operator.UiParameter("count", object())
+ def test_resource_parameter_receives_its_mount_directory(self):
+ operator = ResourceParameterOperator()
+
+ operator.open()
+
+ assert operator.model_parameter.value == "/mnt/texera-mounts/model-1/abc123"
+ assert operator.model_parameter.resource is Resource.MODEL
+
+ def test_resource_parameter_rejects_anything_but_a_resource(self):
+ operator = MissingParameterOperator()
+
+ with pytest.raises(TypeError, match="must be a Resource"):
+ operator.UiParameter("MODEL", AttributeType.STRING, value="model")
+
+ def test_resource_parameter_must_be_a_string(self):
+ operator = MissingParameterOperator()
+
+ # What the UDF receives is a directory path.
+ with pytest.raises(TypeError, match="must be AttributeType.STRING"):
+ operator.UiParameter("MODEL", AttributeType.INT, value=Resource.DATASET)
+
+ def test_resource_kinds_match_the_frontend_and_backend(self):
+ assert {resource.value for resource in Resource} == {"model", "dataset"}
+
def test_super_open_applies_injected_values_once(self):
operator = SuperOpenParameterOperator()
diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerSpec.scala
index 147a8e153a..afd3836a68 100644
--- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/scheduling/RegionExecutionManagerSpec.scala
@@ -22,9 +22,11 @@ package org.apache.texera.amber.engine.architecture.scheduling
import com.twitter.util.{Future, JavaTimer, Time, Timer}
import org.apache.pekko.actor.ActorSystem
import org.apache.pekko.testkit.TestKit
+import org.apache.texera.amber.core.executor.{OpExecInitInfo, OpExecWithClassName}
import org.apache.texera.amber.core.storage.VFSURIFactory
import org.apache.texera.amber.core.virtualidentity.{ActorVirtualIdentity, ChannelIdentity}
import org.apache.texera.amber.core.workflow.{
+ ExecutionTimeBinding,
GlobalPortIdentity,
OutputPort,
PhysicalOp,
@@ -53,6 +55,7 @@ import org.scalatest.flatspec.AnyFlatSpecLike
import java.util.concurrent.{CountDownLatch, TimeUnit}
import java.util.concurrent.atomic
+import scala.collection.mutable
/**
* Tests the real region-execution lifecycle around synchronous region kill.
@@ -104,6 +107,55 @@ class RegionExecutionManagerSpec
assertControlChannelsAreRemoved(fixture)
}
+ it should "mount the region's repositories before initializing any executor" in {
+ val binding = new ExecutionTimeBinding {
+ override def opExecInitInfo: OpExecInitInfo = OpExecWithClassName("bound")
+ override def mountLocators: Set[String] = Set("dataset-1:abc123")
+ }
+ // What had been sent to workers at the moment the mount was asked for.
+ val mounts = mutable.ArrayBuffer[(Set[String], Seq[String])]()
+ var probe: CoordinatorRpcProbe = null
+ val fixture = createSingleRegionFixture(
+ endWorkerResponse = _ => None,
+ physicalOp = createSourceOp("test-op").withExecutionTimeBinding(Some(binding)),
+ ensureMounted = locators => mounts += ((locators, probe.methodTrace))
+ )
+ probe = fixture.rpcProbe
+
+ launchRegion(fixture.manager)
+
+ assert(mounts.map(_._1) == Seq(Set("dataset-1:abc123")))
+ assert(!mounts.head._2.contains(InitializeExecutor))
+ assert(fixture.rpcProbe.methodTrace.contains(InitializeExecutor))
+ }
+
+ it should "fail the region without initializing any executor when a mount is refused" in {
+ val binding = new ExecutionTimeBinding {
+ override def opExecInitInfo: OpExecInitInfo = OpExecWithClassName("bound")
+ override def mountLocators: Set[String] = Set("dataset-1:abc123")
+ }
+ val fixture = createSingleRegionFixture(
+ endWorkerResponse = _ => None,
+ physicalOp = createSourceOp("test-op").withExecutionTimeBinding(Some(binding)),
+ ensureMounted = _ => throw new RuntimeException("the mount request was refused: HTTP 403")
+ )
+
+ val failure = intercept[RuntimeException](launchRegion(fixture.manager))
+
+ assert(failure.getMessage.contains("HTTP 403"))
+ assert(!fixture.rpcProbe.methodTrace.contains(InitializeExecutor))
+ }
+
+ it should "not mount anything for a region whose operators name nothing" in {
+ val mounts = mutable.ArrayBuffer[Set[String]]()
+ val fixture =
+ createSingleRegionFixture(endWorkerResponse = _ => None, ensureMounted = mounts += _)
+
+ launchRegion(fixture.manager)
+
+ assert(mounts.isEmpty)
+ }
+
it should "retry EndWorker failures and delay gracefulStop until a retry succeeds" in {
val attempts = new atomic.AtomicInteger(0)
// The first EndWorker is sent on the test thread; the retry is sent later from the manager's
@@ -380,9 +432,10 @@ class RegionExecutionManagerSpec
endWorkerResponse: WorkerRpcCall => Option[ControlReturn],
maxTerminationAttempts: Int = RegionExecutionManager.DefaultMaxTerminationAttempts,
killRetryBaseBackoffMs: Long = RegionExecutionManager.DefaultKillRetryBaseBackoffMs,
- killRetryTimer: Timer = new JavaTimer(true)
+ killRetryTimer: Timer = new JavaTimer(true),
+ physicalOp: PhysicalOp = createSourceOp("test-op"),
+ ensureMounted: Set[String] => Unit = _ => ()
): SingleRegionFixture = {
- val physicalOp = createSourceOp("test-op")
val workerId = createWorkerId(physicalOp)
val region = createSingleWorkerRegion(1, physicalOp, workerId)
@@ -410,7 +463,8 @@ class RegionExecutionManagerSpec
coordinator.actorRefService,
maxTerminationAttempts,
killRetryBaseBackoffMs,
- killRetryTimer
+ killRetryTimer,
+ ensureMounted = ensureMounted
)
SingleRegionFixture(
diff --git a/amber/src/test/scala/org/apache/texera/web/service/ExecutionReconfigurationServiceSpec.scala b/amber/src/test/scala/org/apache/texera/web/service/ExecutionReconfigurationServiceSpec.scala
index dfc3569385..fc60af0c35 100644
--- a/amber/src/test/scala/org/apache/texera/web/service/ExecutionReconfigurationServiceSpec.scala
+++ b/amber/src/test/scala/org/apache/texera/web/service/ExecutionReconfigurationServiceSpec.scala
@@ -23,7 +23,7 @@ import com.twitter.util.Future
import io.reactivex.rxjava3.disposables.Disposable
import org.apache.pekko.actor.ActorSystem
import org.apache.pekko.testkit.TestKit
-import org.apache.texera.amber.core.executor.OpExecWithClassName
+import org.apache.texera.amber.core.executor.{OpExecInitInfo, OpExecWithClassName}
import org.apache.texera.amber.core.virtualidentity.{
ActorVirtualIdentity,
ExecutionIdentity,
@@ -32,6 +32,7 @@ import org.apache.texera.amber.core.virtualidentity.{
WorkflowIdentity
}
import org.apache.texera.amber.core.workflow.{
+ ExecutionTimeBinding,
InputPort,
OutputPort,
PhysicalOp,
@@ -228,8 +229,10 @@ class ExecutionReconfigurationServiceSpec
private class RecordingService(stateStore: ExecutionStateStore)
extends ExecutionReconfigurationService(client = null, stateStore, workflow = null) {
val captured: mutable.ArrayBuffer[WorkflowReconfigureRequest] = mutable.ArrayBuffer.empty
+ val mounted: mutable.ArrayBuffer[Set[String]] = mutable.ArrayBuffer.empty
override protected def dispatch(request: WorkflowReconfigureRequest): Unit =
captured += request
+ override protected def ensureMounted(locators: Set[String]): Unit = mounted += locators
override protected def registerWorkerCompletionCallback(): Unit = ()
override protected def registerCompletionDiffHandler(): Unit = ()
}
@@ -481,6 +484,24 @@ class ExecutionReconfigurationServiceSpec
state.completedReconfigurations shouldBe empty
}
+ it should "mount what an edited operator names and send its bound code" in {
+ val stateStore = new ExecutionStateStore()
+ val service = new RecordingService(stateStore)
+ val bound = OpExecWithClassName("bound.Class", "")
+ val op = mkPhysicalOp("op-1").withExecutionTimeBinding(Some(new ExecutionTimeBinding {
+ override def opExecInitInfo: OpExecInitInfo = bound
+ override def mountLocators: Set[String] = Set("dataset-1:abc123")
+ }))
+ stateStore.reconfigurationStore.updateState(_ =>
+ ExecutionReconfigurationStore(unscheduledReconfigurations = List((op, None)))
+ )
+
+ service.performReconfigurationOnResume()
+
+ service.mounted shouldBe Seq(Set("dataset-1:abc123"))
+ service.captured.head.reconfiguration.map(_.newExecInitInfo) shouldBe Seq(bound)
+ }
+
it should "dispatch one request carrying every pending reconfiguration and reset the store" in {
val stateStore = new ExecutionStateStore()
val service = new RecordingService(stateStore)
diff --git a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml
index e758a51594..f529fe3653 100644
--- a/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml
+++ b/bin/k8s/templates/base/workflow-computing-unit-manager/workflow-computing-unit-manager-deployment.yaml
@@ -83,6 +83,13 @@ spec:
{{- if .Values.mounter.enabled }}
- name: KUBERNETES_MOUNTER_HOST_ROOT
value: "{{ .Values.mounter.hostMountRoot }}"
+ # These two are passed straight on to each computing-unit pod, which the chart
+ # cannot reach itself: the pod runs in the pool namespace and is created at
+ # runtime, so the address is qualified rather than a bare service name.
+ - name: TEXERA_MOUNT_IN_POD_ROOT
+ value: "{{ .Values.mounter.podMountRoot }}"
+ - name: ACCESS_CONTROL_SERVICE_URL
+ value: http://{{ .Values.accessControlService.name }}-svc.{{ .Release.Namespace }}.svc.cluster.local:{{ .Values.accessControlService.service.port }}
{{- end }}
# TexeraDB Access
- name: STORAGE_JDBC_URL
diff --git a/bin/k8s/values.yaml b/bin/k8s/values.yaml
index 4f24269b39..5b4a2bc1d7 100644
--- a/bin/k8s/values.yaml
+++ b/bin/k8s/values.yaml
@@ -420,6 +420,8 @@ mounter:
imagePullPolicy: IfNotPresent
port: 8100
hostMountRoot: /var/lib/texera-mounts
+ # Where a computing-unit pod sees its mounts.
+ podMountRoot: /mnt/texera-mounts
serviceAccountName: texera-mounter-service-account
# The mounter authenticates its callers: a caller presents a service-account token
# projected for the mounter's audience, which the mounter verifies with TokenReview and
diff --git a/build.sbt b/build.sbt
index cb16bda952..d89f7a7508 100644
--- a/build.sbt
+++ b/build.sbt
@@ -234,6 +234,8 @@ lazy val FileService = (project in file("file-service"))
)
lazy val WorkflowOperator = (project in file("common/workflow-operator")).settings(commonModuleSettingsWithVendored).dependsOn(WorkflowCore)
+ .configs(Test)
+ .dependsOn(DAO % "test->test") // MockTexeraDB, to force a UDF's execution-time binding
lazy val WorkflowCompiler = (project in file("common/workflow-compiler"))
.settings(commonModuleSettings)
.configs(Test)
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 82c563b165..e1006c3fb5 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
@@ -62,6 +62,7 @@ object EnvironmentalVariable {
*/
val ENV_CU_ID = "TEXERA_CU_ID"
val ENV_MOUNT_IN_POD_ROOT = "TEXERA_MOUNT_IN_POD_ROOT"
+ val ENV_ACCESS_CONTROL_SERVICE_URL = "ACCESS_CONTROL_SERVICE_URL"
// JDBC
val ENV_JDBC_URL = "STORAGE_JDBC_URL"
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 95127e234d..f43db07198 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
@@ -102,21 +102,22 @@ object FileResolver {
private def parsePrefixedPath(
fileName: String
): Option[(ResourceType.Value, String, String, String, Array[String])] = {
- val filePath = Paths.get(fileName)
+ if (Option(fileName).forall(_.trim.isEmpty)) return None
+ val filePath = Paths.get(fileName.trim)
val pathSegments = (0 until filePath.getNameCount).map(filePath.getName(_).toString).toArray
pathSegments.headOption.flatMap(ResourceType.fromPrefix) match {
case Some(resourceType) =>
- // Prefixed: /<prefix>/ownerEmail/resourceName/versionName/<file> (>= 5 segments).
- if (pathSegments.length < 5) None
+ // Prefixed: /<prefix>/ownerEmail/resourceName/versionName, then the file path if any.
+ if (pathSegments.length < 4) None
else
Some(
(resourceType, pathSegments(1), pathSegments(2), pathSegments(3), pathSegments.drop(4))
)
case None =>
- // Legacy unprefixed dataset path (backward compat): /ownerEmail/datasetName/versionName/<file>.
+ // Legacy unprefixed dataset path (backward compat): /ownerEmail/datasetName/versionName.
// TODO(dataset-prefix): require the prefix once all stored paths are migrated (36.sql).
- if (pathSegments.length >= 4)
+ if (pathSegments.length >= 3)
Some(
(
ResourceType.Dataset,
@@ -147,9 +148,11 @@ object FileResolver {
*/
private def versionedResourceResolveFunc(fileName: String): URI = {
val (resourceType, ownerEmail, resourceName, versionName, fileRelativePathSegments) =
- parsePrefixedPath(fileName).getOrElse(
- throw new FileNotFoundException(s"Versioned-resource file $fileName not found.")
- )
+ parsePrefixedPath(fileName)
+ .filter { case (_, _, _, _, file) => file.nonEmpty }
+ .getOrElse(
+ throw new FileNotFoundException(s"Versioned-resource file $fileName not found.")
+ )
val (scheme, repositoryName, versionHash) = resourceType match {
case ResourceType.Dataset =>
@@ -282,6 +285,50 @@ object FileResolver {
(model.getRepositoryName, modelVersion.getVersionHash)
}
+ /**
+ * Resolves a version path to the repository and commit backing it — the pair that addresses
+ * a mount, a mount being a whole repository pinned to one commit.
+ *
+ * Resolves only; the mount authority checks read access before mounting.
+ */
+ def resolveRepositoryVersion(versionPath: String): (String, String) = {
+ // Nothing may follow the version: a longer path names a file, a different request.
+ val (resourceType, ownerEmail, resourceName, versionName) =
+ parsePrefixedPath(versionPath) match {
+ case Some((resourceType, ownerEmail, resourceName, versionName, file)) if file.isEmpty =>
+ (resourceType, ownerEmail, resourceName, versionName)
+ case _ =>
+ throw new FileNotFoundException(
+ s"Versioned-resource version path '$versionPath' is invalid; expected " +
+ s"/<${ResourceType.values.mkString("|")}>/ownerEmail/resourceName/versionName."
+ )
+ }
+
+ try {
+ resourceType match {
+ case ResourceType.Dataset =>
+ lookupDataset(ownerEmail, resourceName, versionName, versionPath)
+ case ResourceType.Model =>
+ lookupModel(ownerEmail, resourceName, versionName, versionPath)
+ case other =>
+ throw new FileNotFoundException(
+ s"Unsupported resource type $other for version path $versionPath."
+ )
+ }
+ } catch {
+ case e: Throwable => throw unwrapNotFound(e)
+ }
+ }
+
+ // withTransaction rethrows whatever the block threw wrapped in a DataAccessException, which
+ // would reach the caller as an opaque database error instead of "no such version".
+ private def unwrapNotFound(e: Throwable): Throwable =
+ Iterator
+ .iterate(e)(_.getCause)
+ .takeWhile(_ != null)
+ .collectFirst { case notFound: FileNotFoundException => notFound }
+ .getOrElse(e)
+
/**
* Checks if a given file path has a valid scheme.
*
@@ -306,7 +353,8 @@ object FileResolver {
return None
}
parsePrefixedPath(path).collect {
- case (ResourceType.Dataset, ownerEmail, datasetName, _, _) => (ownerEmail, datasetName)
+ case (ResourceType.Dataset, ownerEmail, datasetName, _, file) if file.nonEmpty =>
+ (ownerEmail, datasetName)
}
}
}
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/RepositoryMountManager.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/RepositoryMountManager.scala
new file mode 100644
index 0000000000..d24b267bf9
--- /dev/null
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/storage/RepositoryMountManager.scala
@@ -0,0 +1,182 @@
+/*
+ * 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
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.typesafe.scalalogging.LazyLogging
+import org.apache.texera.common.config.EnvironmentalVariable
+
+import java.net.{HttpURLConnection, URI}
+import java.nio.charset.StandardCharsets
+import java.nio.file.{Files, Path, Paths}
+import scala.io.Source
+import scala.util.Using
+
+/**
+ * Makes a versioned LakeFS repository readable inside this computing unit's pod.
+ *
+ * The mount is not performed here — this pod runs user code and is unprivileged. It asks the
+ * mount authority, presenting the pod's own user JWT, and the mount reaches this pod through
+ * Kubernetes mount propagation.
+ *
+ * A repository is addressed by a locator "<repositoryName>:<commitHash>". A commit is
+ * immutable, so one mount serves every worker and execution for the life of the pod.
+ */
+class RepositoryMountManager(
+ env: String => Option[String],
+ post: (String, String, String) => Unit,
+ isMounted: Path => Boolean,
+ mountTimeoutMs: Long
+) extends LazyLogging {
+
+ private val mapper = new ObjectMapper()
+
+ private def inPodMountRoot: Path =
+ Paths.get(
+ env(EnvironmentalVariable.ENV_MOUNT_IN_POD_ROOT)
+ .map(_.trim)
+ .filter(_.nonEmpty)
+ .getOrElse(
+ throw new IllegalStateException(
+ s"${EnvironmentalVariable.ENV_MOUNT_IN_POD_ROOT} is not set in this computing unit."
+ )
+ )
+ )
+
+ private def parseLocator(locator: String): (String, String) =
+ Option(locator).getOrElse("").split(":", 2) match {
+ case Array(repository, commit) if repository.nonEmpty && commit.nonEmpty =>
+ (repository, commit)
+ case _ =>
+ throw new IllegalArgumentException(
+ s"Invalid mount locator '$locator'; expected <repositoryName>:<commitHash>."
+ )
+ }
+
+ /**
+ * Where `locator` is, or will be, readable in this pod. Pure — it neither mounts nor checks
+ * anything, because an operator is given this path while its execution is being set up,
+ * before the region is launched and the mount actually happens.
+ */
+ def mountPointOf(locator: String): Path = {
+ val (repository, commit) = parseLocator(locator)
+ inPodMountRoot.resolve(repository).resolve(commit)
+ }
+
+ /**
+ * Mount every locator, deduplicated, in one place rather than once per worker.
+ *
+ * Warning: a FUSE mount is visible only in the mount namespace of the process that receives
+ * it, which is fine while a region's controller and workers share one pod. Were workers ever
+ * to become separate pods, this step — and only this step — would have to be dispatched to
+ * each of them.
+ */
+ def ensureAllMounted(locators: Set[String]): Unit = locators.foreach(ensureMounted)
+
+ /** Mount `locator` if it is not already mounted, and return its in-pod path. */
+ def ensureMounted(locator: String): Path =
+ synchronized {
+ val (repository, commit) = parseLocator(locator)
+ val mountPoint = mountPointOf(locator)
+
+ if (isMounted(mountPoint)) {
+ logger.info(s"$locator is already mounted at $mountPoint")
+ return mountPoint
+ }
+
+ val Seq(accessControlService, cuid, jwt) = Seq(
+ EnvironmentalVariable.ENV_ACCESS_CONTROL_SERVICE_URL,
+ EnvironmentalVariable.ENV_CU_ID,
+ EnvironmentalVariable.ENV_USER_JWT_TOKEN
+ ).map(name =>
+ env(name).map(_.trim).filter(_.nonEmpty).getOrElse {
+ throw new IllegalStateException(s"$name is not set in this computing unit.")
+ }
+ )
+
+ val body = mapper.createObjectNode()
+ body.put("repositoryName", repository)
+ body.put("commitHash", commit)
+ logger.info(s"requesting a mount of $locator for computing unit $cuid")
+ post(s"$accessControlService/api/mounts/$cuid", body.toString, jwt)
+
+ // A successful response is not yet a readable directory: GeeseFS is daemonized, and the
+ // mount still has to propagate in.
+ val deadline = System.currentTimeMillis() + mountTimeoutMs
+ while (!isMounted(mountPoint)) {
+ if (System.currentTimeMillis() > deadline) {
+ throw new RuntimeException(
+ s"$locator did not appear as a mount at $mountPoint within ${mountTimeoutMs}ms."
+ )
+ }
+ Thread.sleep(200)
+ }
+ logger.info(s"$locator mounted at $mountPoint")
+ mountPoint
+ }
+}
+
+object RepositoryMountManager
+ extends RepositoryMountManager(
+ name => sys.env.get(name),
+ InPodMount.postJson,
+ InPodMount.isFuseMounted,
+ 35000
+ )
+
+private object InPodMount {
+
+ def postJson(url: String, body: String, jwt: String): Unit = {
+ val connection = URI.create(url).toURL.openConnection().asInstanceOf[HttpURLConnection]
+ connection.setRequestMethod("POST")
+ connection.setRequestProperty("Content-Type", "application/json")
+ connection.setRequestProperty("Authorization", s"Bearer $jwt")
+ connection.setDoOutput(true)
+ connection.setConnectTimeout(10000)
+ connection.setReadTimeout(40000)
+ try {
+ Using(connection.getOutputStream)(_.write(body.getBytes(StandardCharsets.UTF_8)))
+ val code = connection.getResponseCode
+ if (code < 200 || code >= 300) {
+ val error = Option(connection.getErrorStream)
+ .map(stream => new String(stream.readAllBytes(), StandardCharsets.UTF_8))
+ .getOrElse("")
+ throw new RuntimeException(s"the mount request was refused: HTTP $code $error")
+ }
+ } finally {
+ connection.disconnect()
+ }
+ }
+
+ // /proc/mounts is the kernel's read-only view of this pod's mount table: the kernel adds the
+ // entry when a propagated GeeseFS mount arrives and removes it on unmount, so nothing writes it.
+ def isFuseMounted(mountPoint: Path): Boolean = {
+ if (!Files.exists(mountPoint)) {
+ return false
+ }
+ val target = mountPoint.toAbsolutePath.toString
+ Using(Source.fromFile("/proc/mounts")) { source =>
+ source.getLines().exists { line =>
+ val fields = line.split(" ")
+ fields.length > 2 && fields(1) == target && fields(2).startsWith("fuse")
+ }
+ }.getOrElse(false)
+ }
+}
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/ExecutionTimeBinding.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/ExecutionTimeBinding.scala
new file mode 100644
index 0000000000..49e10b55e4
--- /dev/null
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/ExecutionTimeBinding.scala
@@ -0,0 +1,39 @@
+/*
+ * 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.workflow
+
+import org.apache.texera.amber.core.executor.OpExecInitInfo
+
+/**
+ * Setup an operator leaves until its execution starts.
+ *
+ * Compilation runs again on every edit of a workflow, so anything expensive there is paid far
+ * more often than the workflow is run; and some answers — a directory inside the computing unit
+ * that will run the operator — only exist at run time. Implementations must be idempotent: the
+ * engine may ask more than once, and a region that re-executes asks again.
+ */
+trait ExecutionTimeBinding {
+
+ /** Executor initialization info with every late-bound value in place. */
+ def opExecInitInfo: OpExecInitInfo
+
+ /** Repositories, as "<repositoryName>:<commitHash>", to mount before the operator runs. */
+ def mountLocators: Set[String]
+}
diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalOp.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalOp.scala
index c4dc2307ee..9c57359f11 100644
--- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalOp.scala
+++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/workflow/PhysicalOp.scala
@@ -168,7 +168,8 @@ object PhysicalOp {
"outputPorts", // same reason with above
"propagateSchema", // function type, so ignore it
"locationPreference", // ignore it for the deserialization
- "partitionRequirement" // ignore it for deserialization
+ "partitionRequirement", // ignore it for deserialization
+ "executionTimeBinding" // deferred work rather than state; see ExecutionTimeBinding
)
)
case class PhysicalOp(
@@ -215,7 +216,9 @@ case class PhysicalOp(
// hint for number of workers
suggestedWorkerNum: Option[Int] = None,
// name of the PVE to execute within
- pveName: String = ""
+ pveName: String = "",
+ // setup postponed until the execution starts; None when there is nothing to defer
+ executionTimeBinding: Option[ExecutionTimeBinding] = None
) extends LazyLogging {
// all the "dependee" links are also blocking
@@ -227,6 +230,21 @@ case class PhysicalOp(
.toList
.distinct
+ /**
+ * The executor info to actually run with. It differs from `opExecInitInfo` only for an
+ * operator that deferred part of its setup: that one is the compile-time view, enough to
+ * propagate schemas, and this one is complete. Forcing it can do real work, so ask for it
+ * when the execution starts — never on the editing path.
+ */
+ @JsonIgnore
+ def executableOpExecInitInfo: OpExecInitInfo =
+ executionTimeBinding.map(_.opExecInitInfo).getOrElse(opExecInitInfo)
+
+ /** Repositories to mount before this operator runs. Forces the binding, like the above. */
+ @JsonIgnore
+ def mountLocators: Set[String] =
+ executionTimeBinding.map(_.mountLocators).getOrElse(Set.empty)
+
/**
* Helper functions related to compile-time operations
*/
@@ -394,6 +412,10 @@ case class PhysicalOp(
this.copy(pveName = name)
}
+ def withExecutionTimeBinding(binding: Option[ExecutionTimeBinding]): PhysicalOp = {
+ this.copy(executionTimeBinding = binding)
+ }
+
/**
* creates a copy with an additional input link specified on an input port
*/
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/PhysicalOpSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/PhysicalOpSpec.scala
index 1e8d247e45..3cfbbdad4f 100644
--- a/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/PhysicalOpSpec.scala
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/core/workflow/PhysicalOpSpec.scala
@@ -163,6 +163,26 @@ class PhysicalOpSpec extends AnyFlatSpec {
assert(newOp("a").withPveName("my-pve").pveName == "my-pve")
}
+ "PhysicalOp.executableOpExecInitInfo" should
+ "be the compile-time info when the operator deferred nothing" in {
+ val op = newOp("a")
+ assert(op.executionTimeBinding.isEmpty)
+ assert(op.executableOpExecInitInfo == op.opExecInitInfo)
+ assert(op.mountLocators.isEmpty)
+ }
+
+ it should "come from the execution-time binding when there is one" in {
+ val bound = OpExecWithCode("bound", "python")
+ val binding = new ExecutionTimeBinding {
+ override def opExecInitInfo: OpExecInitInfo = bound
+ override def mountLocators: Set[String] = Set("dataset-1:abc123")
+ }
+ val op = newOp("a").withExecutionTimeBinding(Some(binding))
+ assert(op.opExecInitInfo == OpExecInitInfo.Empty)
+ assert(op.executableOpExecInitInfo == bound)
+ assert(op.mountLocators == Set("dataset-1:abc123"))
+ }
+
// ----- propagateSchema -----
"PhysicalOp.propagateSchema" should "wait for all input schemas before running propagation" in {
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 2649a3d860..93b92c1ee2 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
@@ -280,6 +280,56 @@ class FileResolverSpec
assert(FileResolver.parseDatasetOwnerAndName(null).isEmpty)
}
+ "resolveRepositoryVersion" should "resolve a dataset version path to its repository and commit" in {
+ assert(
+ FileResolver.resolveRepositoryVersion("/dataset/test_user@test.com/test_dataset/v2")
+ == ((testDataset.getRepositoryName, testDatasetVersion2.getVersionHash))
+ )
+ }
+
+ it should "resolve a model version path to its repository and commit" in {
+ assert(
+ FileResolver.resolveRepositoryVersion("/model/test_user@test.com/test_model/v1")
+ == ((testModel.getRepositoryName, testModelVersion1.getVersionHash))
+ )
+ }
+
+ it should "resolve a legacy unprefixed dataset version path" in {
+ assert(
+ FileResolver.resolveRepositoryVersion("/test_user@test.com/test_dataset/v1")
+ == ((testDataset.getRepositoryName, testDatasetVersion1.getVersionHash))
+ )
+ }
+
+ it should "reject a path that names a file rather than a version" in {
+ assertThrows[FileNotFoundException] {
+ FileResolver.resolveRepositoryVersion(dataset1TxtFilePath)
+ }
+ }
+
+ it should "reject a path that stops short of a version" in {
+ assertThrows[FileNotFoundException] {
+ FileResolver.resolveRepositoryVersion("/dataset/test_user@test.com/test_dataset")
+ }
+ }
+
+ it should "reject an empty or null path" in {
+ assertThrows[FileNotFoundException](FileResolver.resolveRepositoryVersion(""))
+ assertThrows[FileNotFoundException](FileResolver.resolveRepositoryVersion(null))
+ }
+
+ it should "reject a version that does not exist" in {
+ assertThrows[FileNotFoundException] {
+ FileResolver.resolveRepositoryVersion("/dataset/test_user@test.com/test_dataset/v99")
+ }
+ }
+
+ it should "not resolve a model under the dataset prefix" in {
+ assertThrows[FileNotFoundException] {
+ FileResolver.resolveRepositoryVersion("/dataset/test_user@test.com/test_model/v1")
+ }
+ }
+
override protected def afterAll(): Unit = {
closeConnectionPool()
}
diff --git a/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/RepositoryMountManagerSpec.scala b/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/RepositoryMountManagerSpec.scala
new file mode 100644
index 0000000000..46273d96ac
--- /dev/null
+++ b/common/workflow-core/src/test/scala/org/apache/texera/amber/storage/RepositoryMountManagerSpec.scala
@@ -0,0 +1,118 @@
+/*
+ * 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.storage
+
+import org.apache.texera.amber.core.storage.RepositoryMountManager
+import org.apache.texera.common.config.EnvironmentalVariable
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+import java.nio.file.{Path, Paths}
+import scala.collection.mutable
+
+class RepositoryMountManagerSpec extends AnyFlatSpec with Matchers {
+
+ private val locator = "dataset-1:abc123"
+ private val inPodPath = Paths.get("/mnt/texera-mounts/dataset-1/abc123")
+
+ private val podEnv = Map(
+ EnvironmentalVariable.ENV_MOUNT_IN_POD_ROOT -> "/mnt/texera-mounts",
+ EnvironmentalVariable.ENV_ACCESS_CONTROL_SERVICE_URL -> "http://access-control-service-svc:9096",
+ EnvironmentalVariable.ENV_CU_ID -> "7",
+ EnvironmentalVariable.ENV_USER_JWT_TOKEN -> "user-jwt"
+ )
+
+ private class Fixture(
+ env: Map[String, String] = podEnv,
+ mountedAfterRequest: Boolean = true,
+ alreadyMounted: Boolean = false,
+ timeoutMs: Long = 400
+ ) {
+ val requests: mutable.Buffer[(String, String, String)] = mutable.Buffer()
+ private var mounted = alreadyMounted
+
+ val manager: RepositoryMountManager = new RepositoryMountManager(
+ env.get,
+ (url, body, jwt) => {
+ requests += ((url, body, jwt))
+ if (mountedAfterRequest) mounted = true
+ },
+ (_: Path) => mounted,
+ timeoutMs
+ )
+ }
+
+ "mountPointOf" should "place a locator under the pod's mount root" in {
+ new Fixture().manager.mountPointOf(locator) shouldBe inPodPath
+ }
+
+ it should "reject a locator that is not <repositoryName>:<commitHash>" in {
+ val manager = new Fixture().manager
+ Seq("dataset-1", "", ":abc", "dataset-1:", null).foreach { bad =>
+ an[IllegalArgumentException] should be thrownBy manager.mountPointOf(bad)
+ }
+ }
+
+ "ensureMounted" should "ask the authority and return the in-pod path" in {
+ val fixture = new Fixture()
+ fixture.manager.ensureMounted(locator) shouldBe inPodPath
+
+ val (url, body, jwt) = fixture.requests.head
+ // The authority is addressed per computing unit; the pod's own cuid is a claim it checks.
+ url shouldBe "http://access-control-service-svc:9096/api/mounts/7"
+ body should include(""""repositoryName":"dataset-1"""")
+ body should include(""""commitHash":"abc123"""")
+ // The pod's user token: the authority re-checks that user, and GeeseFS presents it on
+ // every read afterwards.
+ jwt shouldBe "user-jwt"
+ }
+
+ it should "not ask again for something already mounted" in {
+ val fixture = new Fixture(alreadyMounted = true)
+ fixture.manager.ensureMounted(locator) shouldBe inPodPath
+ fixture.requests shouldBe empty
+ }
+
+ it should "ask once per distinct locator" in {
+ val fixture = new Fixture()
+ fixture.manager.ensureAllMounted(Set(locator, locator))
+ fixture.requests should have size 1
+ }
+
+ it should "give up if the mount never propagates into the pod" in {
+ val fixture = new Fixture(mountedAfterRequest = false)
+ val failure = the[RuntimeException] thrownBy fixture.manager.ensureMounted(locator)
+ failure.getMessage should include("did not appear as a mount")
+ }
+
+ it should "say which variable is missing rather than fail obscurely" in {
+ Seq(
+ EnvironmentalVariable.ENV_MOUNT_IN_POD_ROOT,
+ EnvironmentalVariable.ENV_ACCESS_CONTROL_SERVICE_URL,
+ EnvironmentalVariable.ENV_CU_ID,
+ EnvironmentalVariable.ENV_USER_JWT_TOKEN
+ ).foreach { missing =>
+ val fixture = new Fixture(env = podEnv - missing)
+ val failure = the[IllegalStateException] thrownBy fixture.manager.ensureMounted(locator)
+ failure.getMessage should include(missing)
+ fixture.requests shouldBe empty
+ }
+ }
+}
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2.scala
index 07fec7af92..14a35faabe 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2.scala
@@ -37,6 +37,7 @@ class DualInputPortsPythonUDFOpDescV2 extends LogicalOp with PythonUdfUiParamete
"# \n" +
"# Define UiParameter inside open() of ProcessTupleOperator, ProcessBatchOperator, or ProcessTableOperator.\n" +
"# Example: self.count = self.UiParameter(\"count\", AttributeType.INT).value\n" +
+ "# Add value=Resource.MODEL or Resource.DATASET to pick a version; the value is its mount directory.\n" +
"# See the Python UDF operator documentation for supported types and behavior.\n" +
"# \n" +
"# from pytexera import *\n" +
@@ -159,6 +160,7 @@ class DualInputPortsPythonUDFOpDescV2 extends LogicalOp with PythonUdfUiParamete
})
)
.withPveName(pveName)
+ .withExecutionTimeBinding(executionBinding(code))
}
override def operatorInfo: OperatorInfo =
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUDFOpDescV2.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUDFOpDescV2.scala
index 01610979ec..0239ca9145 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUDFOpDescV2.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUDFOpDescV2.scala
@@ -39,6 +39,7 @@ class PythonUDFOpDescV2 extends LogicalOp with PythonUdfUiParameterSupport {
"# \n" +
"# Define UiParameter inside open() of ProcessTupleOperator, ProcessBatchOperator, or ProcessTableOperator.\n" +
"# Example: self.count = self.UiParameter(\"count\", AttributeType.INT).value\n" +
+ "# Add value=Resource.MODEL or Resource.DATASET to pick a version; the value is its mount directory.\n" +
"# See the Python UDF operator documentation for supported types and behavior.\n" +
"# \n" +
"# from pytexera import *\n" +
@@ -169,6 +170,7 @@ class PythonUDFOpDescV2 extends LogicalOp with PythonUdfUiParameterSupport {
.withIsOneToManyOp(true)
.withPropagateSchema(SchemaPropagationFunc(propagateSchema))
.withPveName(pveName)
+ .withExecutionTimeBinding(executionBinding(code))
}
override def operatorInfo: OperatorInfo = {
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupport.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupport.scala
index 97548c344b..4849e1480e 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupport.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupport.scala
@@ -21,6 +21,10 @@ package org.apache.texera.amber.operator.udf.python
import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
+import org.apache.texera.amber.core.executor.{OpExecInitInfo, OpExecWithCode}
+import org.apache.texera.amber.core.storage.{FileResolver, RepositoryMountManager}
+import org.apache.texera.amber.core.workflow.ExecutionTimeBinding
+import org.apache.texera.amber.operator.udf.python.PythonUdfUiParameterSupport._
/** Shared serialized UI-parameter property and code-injection hook for Python UDF descriptors. */
trait PythonUdfUiParameterSupport {
@@ -32,6 +36,80 @@ trait PythonUdfUiParameterSupport {
)
var uiParameters: List[UiUDFParameter] = List()
- protected final def injectUiParameters(code: String): String =
+ /**
+ * The code to compile against. A parameter naming a resource keeps the version path the user
+ * chose; the execution-time binding replaces it with the mount path before any worker sees
+ * it, so editing a workflow never pays to resolve one.
+ */
+ protected final def injectUiParameters(code: String): String = {
+ resourceParameters(uiParameters).foreach(selectedVersionPath)
PythonUdfUiParameterInjector.inject(code, uiParameters)
+ }
+
+ /** None when no parameter names a resource: the compiled code is then the code to run. */
+ protected final def executionBinding(code: String): Option[ExecutionTimeBinding] =
+ Option.when(resourceParameters(uiParameters).nonEmpty)(
+ new ResourceParameterBinding(code, uiParameters, mounts)
+ )
+
+ // The pod's mounts; overridable so a spec can supply the mount root the chart would.
+ protected def mounts: RepositoryMountManager = RepositoryMountManager
+}
+
+object PythonUdfUiParameterSupport {
+
+ // The kinds of resource a parameter's value can name; keep in sync with pytexera's Resource.
+ private val ResourceInputTypes: Set[String] = Set("model", "dataset")
+
+ private def isResource(parameter: UiUDFParameter): Boolean =
+ parameter != null && ResourceInputTypes.contains(Option(parameter.inputType).getOrElse(""))
+
+ private def resourceParameters(parameters: List[UiUDFParameter]): List[UiUDFParameter] =
+ Option(parameters).getOrElse(Nil).filter(isResource)
+
+ // Checked on every compile, so an unchosen version shows in the editor rather than at run time.
+ private def selectedVersionPath(parameter: UiUDFParameter): String =
+ Option(parameter.value)
+ .map(_.trim)
+ .filter(_.nonEmpty)
+ .getOrElse(
+ throw new RuntimeException(
+ s"No ${parameter.inputType} selected for the parameter " +
+ s"'${Option(parameter.attribute).map(_.getName).getOrElse("")}'."
+ )
+ )
+
+ /**
+ * Resolves each resource parameter to the directory its version is mounted at, yielding both
+ * the code the workers run and the repositories behind those directories. Memoized, since
+ * resolving costs a database round trip per parameter.
+ */
+ private class ResourceParameterBinding(
+ code: String,
+ parameters: List[UiUDFParameter],
+ mounts: RepositoryMountManager
+ ) extends ExecutionTimeBinding {
+
+ private lazy val resolved: (List[UiUDFParameter], Set[String]) = {
+ val bound = parameters.map { parameter =>
+ if (!isResource(parameter)) (parameter, None)
+ else {
+ val (repositoryName, commitHash) =
+ FileResolver.resolveRepositoryVersion(selectedVersionPath(parameter))
+ val locator = s"$repositoryName:$commitHash"
+ val mounted = new UiUDFParameter
+ mounted.attribute = parameter.attribute
+ mounted.inputType = parameter.inputType
+ mounted.value = mounts.mountPointOf(locator).toString
+ (mounted, Some(locator))
+ }
+ }
+ (bound.map(_._1), bound.flatMap(_._2).toSet)
+ }
+
+ override lazy val opExecInitInfo: OpExecInitInfo =
+ OpExecWithCode(PythonUdfUiParameterInjector.inject(code, resolved._1), "python")
+
+ override def mountLocators: Set[String] = resolved._2
+ }
}
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/UiUDFParameter.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/UiUDFParameter.scala
index b18b9a181d..926df53d01 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/UiUDFParameter.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/UiUDFParameter.scala
@@ -44,4 +44,13 @@ class UiUDFParameter {
@JsonProperty()
@JsonSchemaTitle("Value")
var value: EncodableString = ""
+
+ /**
+ * The Texera resource the value names — "model" or "dataset" — or empty for free text.
+ * Inferred from the UDF's source (`value=Resource.MODEL`), like `attribute`. A resource value
+ * is a version path, turned into the directory that version is mounted at when the run starts.
+ */
+ @JsonProperty()
+ @JsonSchemaTitle("Input type")
+ var inputType: String = ""
}
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/source/PythonUDFSourceOpDescV2.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/source/PythonUDFSourceOpDescV2.scala
index 6ddfa23e01..5932c6643c 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/source/PythonUDFSourceOpDescV2.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/source/PythonUDFSourceOpDescV2.scala
@@ -35,6 +35,7 @@ class PythonUDFSourceOpDescV2 extends SourceOperatorDescriptor with PythonUdfUiP
required = true,
defaultValue = "# Define UiParameter inside GenerateOperator.open().\n" +
"# Example: self.count = self.UiParameter(\"count\", AttributeType.INT).value\n" +
+ "# Add value=Resource.MODEL or Resource.DATASET to pick a version; the value is its mount directory.\n" +
"# See the Python UDF operator documentation for supported types and behavior.\n" +
"# \n" +
"# from pytexera import *\n" +
@@ -104,6 +105,7 @@ class PythonUDFSourceOpDescV2 extends SourceOperatorDescriptor with PythonUdfUiP
)
.withLocationPreference(Option.empty)
.withPveName(pveName)
+ .withExecutionTimeBinding(executionBinding(code))
if (workers > 1) {
physicalOp
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2Spec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2Spec.scala
index 74caa27379..3b4f6c3207 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2Spec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/DualInputPortsPythonUDFOpDescV2Spec.scala
@@ -112,6 +112,23 @@ class DualInputPortsPythonUDFOpDescV2Spec extends AnyFlatSpec with Matchers {
}
}
+ it should "leave a resource parameter to the execution" in {
+ val d = new DualInputPortsPythonUDFOpDescV2
+ d.code = """from pytexera import *
+ |
+ |class ProcessTupleOperator(UDFOperatorV2):
+ | def process_tuple(self, tuple_, port):
+ | yield tuple_
+ |""".stripMargin
+ val resource = new UiUDFParameter
+ resource.attribute = new Attribute("DS", AttributeType.STRING)
+ resource.inputType = "dataset"
+ resource.value = "/dataset/owner@x.com/ds/v1"
+ d.uiParameters = List(resource)
+
+ d.getPhysicalOp(workflowId, executionId).executionTimeBinding should not be empty
+ }
+
it should "reject a blank virtual-environment name when the default env is disabled" in {
val d = new DualInputPortsPythonUDFOpDescV2
d.defaultEnv = false
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/PythonUDFOpDescV2Spec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/PythonUDFOpDescV2Spec.scala
index a36dee6107..bc0f22ff1c 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/PythonUDFOpDescV2Spec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/PythonUDFOpDescV2Spec.scala
@@ -34,6 +34,22 @@ class PythonUDFOpDescV2Spec extends AnyFlatSpec with Matchers {
private val workflowId = WorkflowIdentity(1L)
private val executionId = ExecutionIdentity(1L)
+ private def resourceParameter(name: String, inputType: String, value: String): UiUDFParameter = {
+ val parameter = new UiUDFParameter
+ parameter.attribute = new Attribute(name, AttributeType.STRING)
+ parameter.inputType = inputType
+ parameter.value = value
+ parameter
+ }
+
+ private val tupleUdfCode =
+ """from pytexera import *
+ |
+ |class ProcessTupleOperator(UDFOperatorV2):
+ | def process_tuple(self, tuple_, port):
+ | yield tuple_
+ |""".stripMargin
+
private def uiParameter(name: String, value: String): UiUDFParameter = {
val parameter = new UiUDFParameter
parameter.attribute = new Attribute(name, AttributeType.INTEGER)
@@ -99,6 +115,29 @@ class PythonUDFOpDescV2Spec extends AnyFlatSpec with Matchers {
}
}
+ it should "defer nothing when no parameter names a resource" in {
+ val d = new PythonUDFOpDescV2
+ d.code = tupleUdfCode
+ d.uiParameters = List(uiParameter("count", "7"))
+ d.getPhysicalOp(workflowId, executionId).executionTimeBinding shouldBe None
+ }
+
+ it should "leave a resource parameter to the execution, compiling without resolving it" in {
+ val d = new PythonUDFOpDescV2
+ d.code = tupleUdfCode
+ d.uiParameters = List(resourceParameter("DS", "dataset", "/dataset/owner@x.com/ds/v1"))
+ // Resolving needs a database, which this suite has none of, so compiling must not try.
+ d.getPhysicalOp(workflowId, executionId).executionTimeBinding should not be empty
+ }
+
+ it should "refuse a resource parameter whose version was never chosen" in {
+ val d = new PythonUDFOpDescV2
+ d.code = tupleUdfCode
+ d.uiParameters = List(resourceParameter("DS", "dataset", " "))
+ val failure = the[RuntimeException] thrownBy d.getPhysicalOp(workflowId, executionId)
+ failure.getMessage should include("No dataset selected for the parameter 'DS'")
+ }
+
it should "reject a blank virtual-environment name when the default env is disabled" in {
val d = new PythonUDFOpDescV2
d.defaultEnv = false
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupportSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupportSpec.scala
new file mode 100644
index 0000000000..0d39cbc6e8
--- /dev/null
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupportSpec.scala
@@ -0,0 +1,188 @@
+/*
+ * 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.udf.python
+
+import org.apache.commons.vfs2.FileNotFoundException
+import org.apache.texera.amber.core.executor.OpExecWithCode
+import org.apache.texera.amber.core.storage.RepositoryMountManager
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeType}
+import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity}
+import org.apache.texera.common.config.EnvironmentalVariable
+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,
+ ModelDao,
+ ModelVersionDao,
+ UserDao
+}
+import org.apache.texera.dao.jooq.generated.tables.pojos.{
+ Dataset,
+ DatasetVersion,
+ Model,
+ ModelVersion,
+ User
+}
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/** Forces a Python UDF's execution-time binding against real dataset and model rows. */
+class PythonUdfUiParameterSupportSpec
+ extends AnyFlatSpec
+ with Matchers
+ with BeforeAndAfterAll
+ with MockTexeraDB {
+
+ private val owner: User = {
+ val user = new User
+ user.setUid(1)
+ user.setName("owner")
+ user.setEmail("owner@test.com")
+ user.setRole(UserRoleEnum.ADMIN)
+ user
+ }
+
+ private val dataset: Dataset = {
+ val dataset = new Dataset
+ dataset.setDid(1)
+ dataset.setName("sales")
+ dataset.setRepositoryName("dataset-1")
+ dataset.setDescription("")
+ dataset.setIsPublic(false)
+ dataset.setOwnerUid(owner.getUid)
+ dataset
+ }
+
+ private val datasetVersion: DatasetVersion = {
+ val version = new DatasetVersion
+ version.setDvid(1)
+ version.setDid(dataset.getDid)
+ version.setName("v1")
+ version.setCreatorUid(owner.getUid)
+ version.setVersionHash("d4t4")
+ version
+ }
+
+ private val model: Model = {
+ val model = new Model
+ model.setMid(1)
+ model.setName("iris")
+ model.setRepositoryName("model-1")
+ model.setDescription("")
+ model.setIsPublic(false)
+ model.setIsDownloadable(true)
+ model.setOwnerUid(owner.getUid)
+ model
+ }
+
+ private val modelVersion: ModelVersion = {
+ val version = new ModelVersion
+ version.setMvid(1)
+ version.setMid(model.getMid)
+ version.setName("v2")
+ version.setCreatorUid(owner.getUid)
+ version.setVersionHash("m0d3l")
+ version
+ }
+
+ override protected def beforeAll(): Unit = {
+ initializeDBAndReplaceDSLContext()
+ val config = getDSLContext.configuration()
+ new UserDao(config).insert(owner)
+ new DatasetDao(config).insert(dataset)
+ new DatasetVersionDao(config).insert(datasetVersion)
+ new ModelDao(config).insert(model)
+ new ModelVersionDao(config).insert(modelVersion)
+ }
+
+ override protected def afterAll(): Unit = closeConnectionPool()
+
+ private val code =
+ """from pytexera import *
+ |
+ |class ProcessTupleOperator(UDFOperatorV2):
+ | def process_tuple(self, tuple_, port):
+ | yield tuple_
+ |""".stripMargin
+
+ private def parameter(
+ name: String,
+ attributeType: AttributeType,
+ inputType: String,
+ value: String
+ ): UiUDFParameter = {
+ val parameter = new UiUDFParameter
+ parameter.attribute = new Attribute(name, attributeType)
+ parameter.inputType = inputType
+ parameter.value = value
+ parameter
+ }
+
+ // The in-pod mount manager, given the root the chart passes to each computing-unit pod.
+ private val podMounts = new RepositoryMountManager(
+ Map(EnvironmentalVariable.ENV_MOUNT_IN_POD_ROOT -> "/mnt/texera-mounts").get,
+ (_, _, _) => (),
+ _ => false,
+ 0
+ )
+
+ private def udfWith(parameters: UiUDFParameter*): PythonUDFOpDescV2 = {
+ val udf = new PythonUDFOpDescV2 {
+ override protected def mounts: RepositoryMountManager = podMounts
+ }
+ udf.code = code
+ udf.uiParameters = parameters.toList
+ udf
+ }
+
+ "A Python UDF's execution-time binding" should
+ "hand workers the mount directory of every resource it names" in {
+ val count = parameter("count", AttributeType.INTEGER, "", "7")
+ val op = udfWith(
+ parameter("DATA", AttributeType.STRING, "dataset", "/dataset/owner@test.com/sales/v1"),
+ parameter("MODEL", AttributeType.STRING, "model", "/model/owner@test.com/iris/v2"),
+ count
+ ).getPhysicalOp(WorkflowIdentity(1L), ExecutionIdentity(1L))
+
+ op.mountLocators shouldBe Set("dataset-1:d4t4", "model-1:m0d3l")
+ val bound = PythonUdfUiParameterInjector.inject(
+ code,
+ List(
+ parameter("DATA", AttributeType.STRING, "dataset", "/mnt/texera-mounts/dataset-1/d4t4"),
+ parameter("MODEL", AttributeType.STRING, "model", "/mnt/texera-mounts/model-1/m0d3l"),
+ count
+ )
+ )
+ op.executableOpExecInitInfo shouldBe OpExecWithCode(bound, "python")
+ // The compile-time view still names the versions; only the run sees the mounts.
+ op.opExecInitInfo should not be op.executableOpExecInitInfo
+ }
+
+ it should "report a version that no longer exists when the run starts, not while editing" in {
+ val op = udfWith(
+ parameter("DATA", AttributeType.STRING, "dataset", "/dataset/owner@test.com/sales/v9")
+ ).getPhysicalOp(WorkflowIdentity(1L), ExecutionIdentity(1L))
+
+ val failure = the[FileNotFoundException] thrownBy op.mountLocators
+ failure.getMessage should include("/dataset/owner@test.com/sales/v9")
+ }
+}
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/UiUDFParameterSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/UiUDFParameterSpec.scala
index 1bb09e04f0..2578a5b643 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/UiUDFParameterSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/UiUDFParameterSpec.scala
@@ -42,6 +42,16 @@ class UiUDFParameterSpec extends AnyFlatSpec with Matchers {
restored.value shouldBe "x"
}
+ it should "name no resource by default, and round-trip the one it names" in {
+ (new UiUDFParameter).inputType shouldBe ""
+ val p = new UiUDFParameter
+ p.attribute = new Attribute("MODEL", AttributeType.STRING)
+ p.inputType = "model"
+ objectMapper
+ .readValue(objectMapper.writeValueAsString(p), classOf[UiUDFParameter])
+ .inputType shouldBe "model"
+ }
+
it should "default value to the empty string when the JSON omits it" in {
val json = """{"attribute":{"attributeName":"col","attributeType":"string"}}"""
val restored = objectMapper.readValue(json, classOf[UiUDFParameter])
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/source/PythonUDFSourceOpDescV2Spec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/source/PythonUDFSourceOpDescV2Spec.scala
index cda363511b..dcb648e282 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/source/PythonUDFSourceOpDescV2Spec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/source/PythonUDFSourceOpDescV2Spec.scala
@@ -96,6 +96,23 @@ class PythonUDFSourceOpDescV2Spec extends AnyFlatSpec with Matchers {
}
}
+ it should "leave a resource parameter to the execution" in {
+ val d = new PythonUDFSourceOpDescV2
+ d.code = """from pytexera import *
+ |
+ |class GenerateOperator(UDFSourceOperator):
+ | def produce(self):
+ | yield
+ |""".stripMargin
+ val resource = new UiUDFParameter
+ resource.attribute = new Attribute("DS", AttributeType.STRING)
+ resource.inputType = "dataset"
+ resource.value = "/dataset/owner@x.com/ds/v1"
+ d.uiParameters = List(resource)
+
+ d.getPhysicalOp(workflowId, executionId).executionTimeBinding should not be empty
+ }
+
it should "reject a blank virtual-environment name when the default env is disabled" in {
val d = new PythonUDFSourceOpDescV2
d.code = "yield"
diff --git a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala
index 856a423484..76f2abc7bd 100644
--- a/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala
+++ b/computing-unit-managing-service/src/main/scala/org/apache/texera/service/util/KubernetesClient.scala
@@ -37,7 +37,11 @@ class KubernetesClient(
// A constructor parameter rather than a direct KubernetesConfig read, so the spec can
// build a pod both ways: the mount contract below is security- and scheduling-sensitive
// and needs asserting on, but the default must stay the PodSecurity-safe one.
- mountingEnabled: Boolean = KubernetesConfig.mounterEnabled
+ mountingEnabled: Boolean = KubernetesConfig.mounterEnabled,
+ // By-name: only a deployment that opted into mounting has to supply these.
+ accessControlServiceUrl: => String =
+ MountingEnv.required(EnvironmentalVariable.ENV_ACCESS_CONTROL_SERVICE_URL),
+ podMountRoot: => String = MountingEnv.required(EnvironmentalVariable.ENV_MOUNT_IN_POD_ROOT)
) {
private val namespace: String = KubernetesConfig.computeUnitPoolNamespace
@@ -140,11 +144,11 @@ class KubernetesClient(
new EnvVarBuilder().withName(key).withValue(value.toString).build()
}.toList
- // Which CU this is, and where its propagated mounts show up. The pod is deliberately
- // not given the mounter's address: only an authenticated platform caller may request a
- // mount, so the address would be of no use to code running here except to probe the
- // node's privileged mounter.
- val inPodMountRoot = "/mnt/texera-mounts"
+ // Which CU this is, where its propagated mounts show up, and who to ask for one. The
+ // pod is deliberately not given the mounter's address: only an authenticated platform
+ // caller may request a mount, so the address would be of no use to code running here
+ // except to probe the node's privileged mounter.
+ lazy val inPodMountRoot = podMountRoot
val mounterEnv =
if (!mountingEnabled) Nil
else
@@ -156,6 +160,10 @@ class KubernetesClient(
new EnvVarBuilder()
.withName(EnvironmentalVariable.ENV_MOUNT_IN_POD_ROOT)
.withValue(inPodMountRoot)
+ .build(),
+ new EnvVarBuilder()
+ .withName(EnvironmentalVariable.ENV_ACCESS_CONTROL_SERVICE_URL)
+ .withValue(accessControlServiceUrl)
.build()
)
@@ -291,5 +299,18 @@ object KubernetesClient
new KubernetesClientBuilder().build(),
// Passed explicitly: a companion object extending its companion class may not rely on
// the class's default constructor arguments.
- KubernetesConfig.mounterEnabled
+ KubernetesConfig.mounterEnabled,
+ MountingEnv.required(EnvironmentalVariable.ENV_ACCESS_CONTROL_SERVICE_URL),
+ MountingEnv.required(EnvironmentalVariable.ENV_MOUNT_IN_POD_ROOT)
)
+
+private object MountingEnv {
+
+ /** Passed straight through to the pod, so the chart is the only place these are written. */
+ def required(variable: String): String =
+ EnvironmentalVariable
+ .get(variable)
+ .getOrElse(
+ throw new IllegalStateException(s"Mounting is enabled but $variable is unset.")
+ )
+}
diff --git a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
index 5e6743c953..8931cb8b51 100644
--- a/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
+++ b/computing-unit-managing-service/src/test/scala/org/apache/texera/service/util/KubernetesClientSpec.scala
@@ -65,6 +65,10 @@ import scala.jdk.CollectionConverters._
// resolution) is covered by ComputingUnitHelpersSpec.
class KubernetesClientSpec extends AnyFlatSpec with Matchers {
+ private val testAccessControlServiceUrl =
+ "http://access-control-service-svc.texera.svc.cluster.local:9096"
+ private val testPodMountRoot = "/mnt/test-mounts"
+
private val namespace: String = KubernetesConfig.computeUnitPoolNamespace
// getVolumes/getVolumeMounts are null rather than empty when nothing was added.
@@ -303,7 +307,12 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers {
when(namespaceable.inNamespace(namespace)).thenReturn(resource)
when(resource.create()).thenReturn(null)
- new KubernetesClient(client, mountingEnabled = true)
+ new KubernetesClient(
+ client,
+ mountingEnabled = true,
+ testAccessControlServiceUrl,
+ testPodMountRoot
+ )
.createPod(5, "2", "4Gi", "1", Map("UID" -> 9, "MODE" -> "batch"))
verify(client).resource(captor.capture())
@@ -321,14 +330,19 @@ class KubernetesClientSpec extends AnyFlatSpec with Matchers {
// makes and must not be able to propagate one back out to the node.
val mount = container.getVolumeMounts.asScala.find(_.getName == "texera-mounts")
mount shouldBe defined
- mount.get.getMountPath shouldBe "/mnt/texera-mounts"
+ mount.get.getMountPath shouldBe testPodMountRoot
mount.get.getMountPropagation shouldBe "HostToContainer"
// The pod is told which CU it is and where its mounts appear -- and deliberately not
// how to reach the mounter.
val env = container.getEnv.asScala.map(e => e.getName -> e.getValue).toMap
env should contain("TEXERA_CU_ID" -> "5")
- env should contain("TEXERA_MOUNT_IN_POD_ROOT" -> "/mnt/texera-mounts")
+ env should contain("TEXERA_MOUNT_IN_POD_ROOT" -> testPodMountRoot)
+ // Who the pod asks for a mount. Without it the engine cannot request one, and the pod
+ // has no other way to reach a mounter -- which is the point.
+ env should contain(
+ "ACCESS_CONTROL_SERVICE_URL" -> testAccessControlServiceUrl
+ )
// Still unprivileged: the whole point of mounting out of pod.
val privileged = Option(container.getSecurityContext).flatMap(c => Option(c.getPrivileged))
diff --git a/frontend/src/app/common/formly/formly-config.ts b/frontend/src/app/common/formly/formly-config.ts
index 2b42d1239b..788ed6a784 100644
--- a/frontend/src/app/common/formly/formly-config.ts
+++ b/frontend/src/app/common/formly/formly-config.ts
@@ -31,6 +31,7 @@ import { EditableLabelWrapperComponent } from "./editable-label-wrapper/editable
import { FormlyRepeatDndComponent } from "./repeat-dnd/repeat-dnd.component";
import { UiUdfParametersComponent } from "../../workspace/component/ui-udf-parameters/ui-udf-parameters.component";
import { DatasetVersionSelectorComponent } from "../../workspace/component/dataset-version-selector/dataset-version-selector.component";
+import { ResourceValueSelectorComponent } from "../../workspace/component/resource-value-selector/resource-value-selector.component";
import { HuggingFaceImageUploadComponent } from "../../workspace/component/hugging-face-image-upload/hugging-face-image-upload.component";
import { HuggingFaceComponent } from "../../workspace/component/hugging-face/hugging-face.component";
import { HuggingFaceAudioUploadComponent } from "../../workspace/component/hugging-face-audio-upload/hugging-face-audio-upload.component";
@@ -90,6 +91,7 @@ export const TEXERA_FORMLY_CONFIG = {
{ name: "huggingface-image-upload", component: HuggingFaceImageUploadComponent, wrappers: ["form-field"] },
{ name: "repeat-section-dnd", component: FormlyRepeatDndComponent },
{ name: "ui-udf-parameters", component: UiUdfParametersComponent, wrappers: ["form-field"] },
+ { name: "resourcevalue", component: ResourceValueSelectorComponent },
],
wrappers: [
{ name: "preset-wrapper", component: PresetWrapperComponent },
diff --git a/frontend/src/app/common/type/resource-type.ts b/frontend/src/app/common/type/resource-type.ts
index 22c4a8592a..acf0bf5791 100644
--- a/frontend/src/app/common/type/resource-type.ts
+++ b/frontend/src/app/common/type/resource-type.ts
@@ -21,4 +21,5 @@
*/
export enum ResourceType {
Dataset = "dataset",
+ Model = "model",
}
diff --git a/frontend/src/app/dashboard/service/user/model/model.service.spec.ts b/frontend/src/app/dashboard/service/user/model/model.service.spec.ts
index f1e07ec099..1ad183a1bd 100644
--- a/frontend/src/app/dashboard/service/user/model/model.service.spec.ts
+++ b/frontend/src/app/dashboard/service/user/model/model.service.spec.ts
@@ -106,6 +106,13 @@ describe("ModelService", () => {
http.verify();
});
+ it("lists the models the caller owns or has been granted", async () => {
+ const pending = firstValueFrom(service.retrieveAccessibleModels());
+ const req = http.expectOne(`${API}/model/list`);
+ expect(req.request.method).toBe("GET");
+ req.flush([]);
+ expect(await pending).toEqual([]);
+ });
it("posts a create request under the model-shaped field names", () => {
// The backend reads modelName/isModelPublic; the dataset spellings would be silently dropped.
service.createModel(newModel({ isPublic: true, isDownloadable: true })).subscribe();
diff --git a/frontend/src/app/dashboard/service/user/model/model.service.ts b/frontend/src/app/dashboard/service/user/model/model.service.ts
index 11fa1d4c7d..d721253818 100644
--- a/frontend/src/app/dashboard/service/user/model/model.service.ts
+++ b/frontend/src/app/dashboard/service/user/model/model.service.ts
@@ -37,6 +37,7 @@ export const MODEL_UPDATE_PUBLICITY_URL = "update/publicity";
export const MODEL_UPDATE_DOWNLOADABLE_URL = "update/downloadable";
export const MODEL_UPDATE_COVER_URL = "update/cover";
export const MODEL_GET_OWNERS_URL = MODEL_BASE_URL + "/user-model-owners";
+export const MODEL_LIST_URL = MODEL_BASE_URL + "/list";
export const MODEL_VERSION_BASE_URL = "version";
export const MODEL_VERSION_RETRIEVE_LIST_URL = MODEL_VERSION_BASE_URL + "/list";
@@ -144,6 +145,11 @@ export class ModelService {
);
}
+ /** The models the caller owns or has been granted. */
+ public retrieveAccessibleModels(): Observable<DashboardModel[]> {
+ return this.http.get<DashboardModel[]>(`${AppSettings.getApiEndpoint()}/${MODEL_LIST_URL}`);
+ }
+
/** A model's versions, newest first; an anonymous caller gets the public-only endpoint. */
public retrieveModelVersionList(mid: number, isLogin: boolean = true): Observable<ModelVersion[]> {
const listUrl = isLogin ? MODEL_VERSION_RETRIEVE_LIST_URL : MODEL_PUBLIC_VERSION_RETRIEVE_LIST_URL;
diff --git a/frontend/src/app/workspace/component/model-selection-modal/model-selection-modal.component.html b/frontend/src/app/workspace/component/model-selection-modal/model-selection-modal.component.html
new file mode 100644
index 0000000000..9643d7bb71
--- /dev/null
+++ b/frontend/src/app/workspace/component/model-selection-modal/model-selection-modal.component.html
@@ -0,0 +1,69 @@
+<!--
+ 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.
+-->
+<br />
+<div
+ nz-row
+ nzGutter="8">
+ <nz-select
+ nz-col
+ nzFlex="auto"
+ nzShowSearch
+ [nzFilterOption]="modelFilterOption"
+ nzPlaceHolder="Select model"
+ [(ngModel)]="selectedModel"
+ (ngModelChange)="onModelChange()">
+ <nz-option
+ *ngFor="let model of models"
+ [nzValue]="model"
+ [nzLabel]="model.model.name"
+ [nzCustomContent]="true">
+ <div class="model-row">
+ <span class="model-name"> #{{ model.model.mid }} {{ model.model.name }} </span>
+ <span class="access-level"> {{ model.isOwner ? 'OWNER' : model.accessPrivilege }} </span>
+ </div>
+ </nz-option>
+ </nz-select>
+
+ <nz-select
+ nz-col
+ nzFlex="130px"
+ nzShowSearch
+ nzPlaceHolder="Select version"
+ [(ngModel)]="selectedVersion"
+ (ngModelChange)="onVersionChange()">
+ <nz-option
+ *ngFor="let version of modelVersions"
+ [nzValue]="version"
+ [nzLabel]="version.name">
+ </nz-option>
+ </nz-select>
+</div>
+<br />
+<texera-user-dataset-version-filetree
+ [isExpandAllAfterViewInit]="true"
+ [fileTreeNodes]="fileTree">
+</texera-user-dataset-version-filetree>
+<br />
+<button
+ nz-button
+ nzType="primary"
+ [disabled]="!selectedPath"
+ (click)="onConfirmSelection()">
+ Select
+</button>
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupport.scala b/frontend/src/app/workspace/component/model-selection-modal/model-selection-modal.component.scss
similarity index 54%
copy from common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupport.scala
copy to frontend/src/app/workspace/component/model-selection-modal/model-selection-modal.component.scss
index 97548c344b..4b70d951d8 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupport.scala
+++ b/frontend/src/app/workspace/component/model-selection-modal/model-selection-modal.component.scss
@@ -1,4 +1,4 @@
-/*
+/**
* 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
@@ -17,21 +17,16 @@
* under the License.
*/
-package org.apache.texera.amber.operator.udf.python
-
-import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
-import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
-
-/** Shared serialized UI-parameter property and code-injection hook for Python UDF descriptors. */
-trait PythonUdfUiParameterSupport {
+.model-row {
+ display: flex;
+ justify-content: space-between;
+}
- @JsonProperty
- @JsonSchemaTitle("Parameters")
- @JsonPropertyDescription(
- "Parameters inferred from active self.UiParameter(...) calls in the Python script"
- )
- var uiParameters: List[UiUDFParameter] = List()
+.model-name {
+ overflow: hidden;
+ text-overflow: ellipsis;
+}
- protected final def injectUiParameters(code: String): String =
- PythonUdfUiParameterInjector.inject(code, uiParameters)
+.access-level {
+ color: grey;
}
diff --git a/frontend/src/app/workspace/component/model-selection-modal/model-selection-modal.component.spec.ts b/frontend/src/app/workspace/component/model-selection-modal/model-selection-modal.component.spec.ts
new file mode 100644
index 0000000000..82cbb3db9e
--- /dev/null
+++ b/frontend/src/app/workspace/component/model-selection-modal/model-selection-modal.component.spec.ts
@@ -0,0 +1,143 @@
+/**
+ * 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 { ComponentFixture, TestBed } from "@angular/core/testing";
+import { of } from "rxjs";
+import { NZ_MODAL_DATA, NzModalRef } from "ng-zorro-antd/modal";
+import type { Mock } from "vitest";
+import { vi as vitest } from "vitest";
+import { ModelSelectionModalComponent } from "./model-selection-modal.component";
+import { ModelService } from "../../../dashboard/service/user/model/model.service";
+import { DashboardModel } from "../../../dashboard/type/dashboard-model.interface";
+import { ModelVersion } from "../../../common/type/model";
+import { DatasetFileNode } from "../../../common/type/datasetVersionFileTree";
+
+const OWNER = "owner@x.com";
+
+const model: DashboardModel = {
+ isOwner: true,
+ ownerEmail: OWNER,
+ accessPrivilege: "WRITE",
+ size: 0,
+ model: {
+ mid: 7,
+ ownerUid: 1,
+ name: "resnet",
+ repositoryName: "model-7",
+ isPublic: false,
+ isDownloadable: true,
+ description: "",
+ creationTime: undefined,
+ coverImage: undefined,
+ framework: "pytorch",
+ format: "torchscript",
+ },
+};
+
+const version: ModelVersion = {
+ mvid: 70,
+ mid: 7,
+ creatorUid: 1,
+ name: "v1",
+ versionHash: undefined,
+ creationTime: undefined,
+ fileNodes: undefined,
+};
+
+const fileNode: DatasetFileNode = { name: "model.pt", type: "file", parentDir: `/${OWNER}/resnet/v1` };
+
+describe("ModelSelectionModalComponent", () => {
+ let component: ModelSelectionModalComponent;
+ let fixture: ComponentFixture<ModelSelectionModalComponent>;
+ let modalData: { selectedPath?: string | null };
+ let modalRef: { close: Mock };
+ let modelService: {
+ retrieveAccessibleModels: Mock;
+ retrieveModelVersionList: Mock;
+ retrieveModelVersionFileTree: Mock;
+ };
+
+ function build(): void {
+ fixture = TestBed.createComponent(ModelSelectionModalComponent);
+ component = fixture.componentInstance;
+ fixture.detectChanges();
+ }
+
+ beforeEach(async () => {
+ modalData = {};
+ modalRef = { close: vitest.fn() };
+ modelService = {
+ retrieveAccessibleModels: vitest.fn().mockReturnValue(of([model])),
+ retrieveModelVersionList: vitest.fn().mockReturnValue(of([version])),
+ retrieveModelVersionFileTree: vitest.fn().mockReturnValue(of({ fileNodes: [fileNode], size: 0 })),
+ };
+ await TestBed.configureTestingModule({
+ imports: [ModelSelectionModalComponent],
+ providers: [
+ { provide: NZ_MODAL_DATA, useValue: modalData },
+ { provide: NzModalRef, useValue: modalRef },
+ { provide: ModelService, useValue: modelService },
+ ],
+ }).compileComponents();
+ });
+
+ it("lists the models the caller can see, with nothing selected yet", () => {
+ build();
+ expect(component.models).toEqual([model]);
+ expect(component.selectedPath).toBeUndefined();
+ const button: HTMLButtonElement = fixture.nativeElement.querySelector("button[nz-button]");
+ expect(button.disabled).toBe(true);
+ });
+
+ it("reopens on the version it was given", () => {
+ modalData.selectedPath = `/model/${OWNER}/resnet/v1`;
+ build();
+
+ expect(component.selectedModel).toBe(model);
+ expect(component.selectedVersion).toBe(version);
+ expect(component.fileTree).toEqual([fileNode]);
+ expect(modelService.retrieveModelVersionFileTree).toHaveBeenCalledWith(7, 70);
+ });
+
+ it("closes with the chosen version's model path", () => {
+ build();
+ component.selectedModel = model;
+ component.onModelChange("v1");
+
+ expect(component.selectedPath).toBe(`/model/${OWNER}/resnet/v1`);
+ component.onConfirmSelection();
+ expect(modalRef.close).toHaveBeenCalledWith(`/model/${OWNER}/resnet/v1`);
+ });
+
+ it("clears the choice when the model is cleared", () => {
+ build();
+ component.selectedModel = undefined;
+ component.onModelChange();
+ expect(component.modelVersions).toEqual([]);
+ expect(component.selectedPath).toBeUndefined();
+ });
+
+ it("filters models by name or by #id", () => {
+ build();
+ const option = { nzValue: model } as any;
+ expect(component.modelFilterOption("res", option)).toBe(true);
+ expect(component.modelFilterOption("#7", option)).toBe(true);
+ expect(component.modelFilterOption("bert", option)).toBe(false);
+ });
+});
diff --git a/frontend/src/app/workspace/component/model-selection-modal/model-selection-modal.component.ts b/frontend/src/app/workspace/component/model-selection-modal/model-selection-modal.component.ts
new file mode 100644
index 0000000000..1c3b239ef3
--- /dev/null
+++ b/frontend/src/app/workspace/component/model-selection-modal/model-selection-modal.component.ts
@@ -0,0 +1,129 @@
+/**
+ * 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 { Component, inject, OnInit } from "@angular/core";
+import { NZ_MODAL_DATA, NzModalRef } from "ng-zorro-antd/modal";
+import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy";
+import { DatasetFileNode } from "../../../common/type/datasetVersionFileTree";
+import { ModelVersion } from "../../../common/type/model";
+import { ResourceType } from "../../../common/type/resource-type";
+import { DashboardModel } from "../../../dashboard/type/dashboard-model.interface";
+import { ModelService } from "../../../dashboard/service/user/model/model.service";
+import { NzRowDirective, NzColDirective } from "ng-zorro-antd/grid";
+import { NzSelectComponent, NzOptionComponent, NzSelectItemInterface } from "ng-zorro-antd/select";
+import { FormsModule } from "@angular/forms";
+import { NgFor } from "@angular/common";
+import { UserDatasetVersionFiletreeComponent } from "../../../dashboard/component/user/user-dataset/user-dataset-explorer/user-dataset-version-filetree/user-dataset-version-filetree.component";
+import { NzButtonComponent } from "ng-zorro-antd/button";
+import { NzWaveDirective } from "ng-zorro-antd/core/wave";
+import { ɵNzTransitionPatchDirective } from "ng-zorro-antd/core/transition-patch";
+import { datasetMatchesQuery } from "../dataset-selection-modal/dataset-search.util";
+
+/** Picks a model version; closes with its /model/ownerEmail/name/version path. */
+@UntilDestroy()
+@Component({
+ templateUrl: "model-selection-modal.component.html",
+ styleUrls: ["model-selection-modal.component.scss"],
+ imports: [
+ NzRowDirective,
+ NzSelectComponent,
+ NzColDirective,
+ FormsModule,
+ NgFor,
+ NzOptionComponent,
+ UserDatasetVersionFiletreeComponent,
+ NzButtonComponent,
+ NzWaveDirective,
+ ɵNzTransitionPatchDirective,
+ ],
+})
+export class ModelSelectionModalComponent implements OnInit {
+ private readonly data = inject(NZ_MODAL_DATA, { optional: true }) as { selectedPath?: string | null } | null;
+
+ models: ReadonlyArray<DashboardModel> = [];
+ modelVersions: ReadonlyArray<ModelVersion> = [];
+ fileTree: DatasetFileNode[] = [];
+ selectedModel?: DashboardModel;
+ selectedVersion?: ModelVersion;
+ selectedPath?: string;
+
+ constructor(
+ private modalRef: NzModalRef,
+ private modelService: ModelService
+ ) {}
+
+ // Matches the typed text against the model's name and its `#<id>`, as the dataset browser does.
+ modelFilterOption = (input: string, option: NzSelectItemInterface): boolean => {
+ const model = option.nzValue as DashboardModel | undefined;
+ return datasetMatchesQuery(model?.model?.name, model?.model?.mid, input ?? "");
+ };
+
+ ngOnInit(): void {
+ this.modelService
+ .retrieveAccessibleModels()
+ .pipe(untilDestroyed(this))
+ .subscribe(models => {
+ this.models = models;
+ const selectedPath = this.data?.selectedPath;
+ if (!selectedPath) return;
+ const [, ownerEmail, modelName, versionName] = selectedPath.split("/").filter(part => part.length > 0);
+ this.selectedModel = models.find(model => model.ownerEmail === ownerEmail && model.model.name === modelName);
+ this.onModelChange(versionName);
+ });
+ }
+
+ onModelChange(versionName?: string): void {
+ this.fileTree = [];
+ this.selectedVersion = undefined;
+ this.selectedPath = undefined;
+ const mid = this.selectedModel?.model.mid;
+ if (mid === undefined) {
+ this.modelVersions = [];
+ return;
+ }
+ this.modelService
+ .retrieveModelVersionList(mid)
+ .pipe(untilDestroyed(this))
+ .subscribe(versions => {
+ this.modelVersions = versions;
+ const preselected = versions.find(version => version.name === versionName);
+ if (preselected) {
+ this.selectedVersion = preselected;
+ this.onVersionChange();
+ }
+ });
+ }
+
+ onVersionChange(): void {
+ const model = this.selectedModel;
+ const version = this.selectedVersion;
+ this.fileTree = [];
+ this.selectedPath = undefined;
+ if (model?.model.mid === undefined || version?.mvid === undefined) return;
+ this.modelService
+ .retrieveModelVersionFileTree(model.model.mid, version.mvid)
+ .pipe(untilDestroyed(this))
+ .subscribe(data => (this.fileTree = data.fileNodes));
+ this.selectedPath = `/${ResourceType.Model}/${model.ownerEmail}/${model.model.name}/${version.name}`;
+ }
+
+ onConfirmSelection(): void {
+ this.modalRef.close(this.selectedPath);
+ }
+}
diff --git a/frontend/src/app/workspace/component/resource-value-selector/resource-value-selector.component.html b/frontend/src/app/workspace/component/resource-value-selector/resource-value-selector.component.html
new file mode 100644
index 0000000000..3bb9d9f3e5
--- /dev/null
+++ b/frontend/src/app/workspace/component/resource-value-selector/resource-value-selector.component.html
@@ -0,0 +1,32 @@
+<!--
+ 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.
+-->
+
+<button
+ nz-button
+ type="button"
+ class="resource-value-select"
+ [disabled]="formControl.disabled"
+ nz-tooltip
+ [nzTooltipTitle]="tooltip"
+ (click)="openPicker()">
+ <i
+ nz-icon
+ [nzType]="icon"></i>
+ <span class="resource-value-label">{{ label }}</span>
+</button>
diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupport.scala b/frontend/src/app/workspace/component/resource-value-selector/resource-value-selector.component.scss
similarity index 54%
copy from common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupport.scala
copy to frontend/src/app/workspace/component/resource-value-selector/resource-value-selector.component.scss
index 97548c344b..5d3f798c1c 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupport.scala
+++ b/frontend/src/app/workspace/component/resource-value-selector/resource-value-selector.component.scss
@@ -1,4 +1,4 @@
-/*
+/**
* 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
@@ -17,21 +17,16 @@
* under the License.
*/
-package org.apache.texera.amber.operator.udf.python
-
-import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription}
-import com.kjetland.jackson.jsonSchema.annotations.JsonSchemaTitle
-
-/** Shared serialized UI-parameter property and code-injection hook for Python UDF descriptors. */
-trait PythonUdfUiParameterSupport {
-
- @JsonProperty
- @JsonSchemaTitle("Parameters")
- @JsonPropertyDescription(
- "Parameters inferred from active self.UiParameter(...) calls in the Python script"
- )
- var uiParameters: List[UiUDFParameter] = List()
+.resource-value-select {
+ display: flex;
+ align-items: center;
+ gap: 6px;
+ width: 100%;
+ overflow: hidden;
+}
- protected final def injectUiParameters(code: String): String =
- PythonUdfUiParameterInjector.inject(code, uiParameters)
+.resource-value-label {
+ overflow: hidden;
+ text-overflow: ellipsis;
+ white-space: nowrap;
}
diff --git a/frontend/src/app/workspace/component/resource-value-selector/resource-value-selector.component.spec.ts b/frontend/src/app/workspace/component/resource-value-selector/resource-value-selector.component.spec.ts
new file mode 100644
index 0000000000..68e3e2ffdf
--- /dev/null
+++ b/frontend/src/app/workspace/component/resource-value-selector/resource-value-selector.component.spec.ts
@@ -0,0 +1,83 @@
+/**
+ * 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 { TestBed } from "@angular/core/testing";
+import { FormControl } from "@angular/forms";
+import { of } from "rxjs";
+import { NzModalService } from "ng-zorro-antd/modal";
+import type { Mock } from "vitest";
+import { vi as vitest } from "vitest";
+import { ResourceValueSelectorComponent } from "./resource-value-selector.component";
+import { ModelSelectionModalComponent } from "../model-selection-modal/model-selection-modal.component";
+import { DatasetSelectionModalComponent } from "../dataset-selection-modal/dataset-selection-modal.component";
+
+describe("ResourceValueSelectorComponent", () => {
+ let modalService: { create: Mock };
+
+ function selector(resource: string, value = "", disabled = false): ResourceValueSelectorComponent {
+ const component = TestBed.runInInjectionContext(
+ () => new ResourceValueSelectorComponent(modalService as unknown as NzModalService)
+ );
+ (component as any).field = {
+ props: { resource },
+ formControl: new FormControl({ value, disabled }),
+ };
+ return component;
+ }
+
+ beforeEach(() => {
+ modalService = { create: vitest.fn() };
+ });
+
+ it("labels a chosen version by name and version, and an empty one by what to pick", () => {
+ expect(selector("model", "/model/owner@x.com/resnet/v2").label).toBe("resnet · v2");
+ expect(selector("model").label).toBe("Select model");
+ expect(selector("dataset").label).toBe("Select dataset");
+ });
+
+ it("opens the model browser and stores the version it returns", () => {
+ const component = selector("model", "/model/owner@x.com/resnet/v1");
+ modalService.create.mockReturnValue({ afterClose: of("/model/owner@x.com/resnet/v2") });
+
+ component.openPicker();
+
+ const options = modalService.create.mock.calls[0][0];
+ expect(options.nzContent).toBe(ModelSelectionModalComponent);
+ expect(options.nzData.selectedPath).toBe("/model/owner@x.com/resnet/v1");
+ expect(component.formControl.value).toBe("/model/owner@x.com/resnet/v2");
+ expect(component.formControl.dirty).toBe(true);
+ });
+
+ it("opens the dataset browser on whole versions rather than files", () => {
+ const component = selector("dataset");
+ modalService.create.mockReturnValue({ afterClose: of(undefined) });
+
+ component.openPicker();
+
+ const options = modalService.create.mock.calls[0][0];
+ expect(options.nzContent).toBe(DatasetSelectionModalComponent);
+ expect(options.nzData).toEqual({ fileMode: false, selectedPath: null });
+ expect(component.formControl.value).toBe("");
+ });
+
+ it("does nothing while the field is disabled", () => {
+ selector("model", "", true).openPicker();
+ expect(modalService.create).not.toHaveBeenCalled();
+ });
+});
diff --git a/frontend/src/app/workspace/component/resource-value-selector/resource-value-selector.component.ts b/frontend/src/app/workspace/component/resource-value-selector/resource-value-selector.component.ts
new file mode 100644
index 0000000000..88fa3827f5
--- /dev/null
+++ b/frontend/src/app/workspace/component/resource-value-selector/resource-value-selector.component.ts
@@ -0,0 +1,118 @@
+/**
+ * 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 { Component, Type } from "@angular/core";
+import { FieldType, FieldTypeConfig } from "@ngx-formly/core";
+import { UntilDestroy, untilDestroyed } from "@ngneat/until-destroy";
+import { NzModalService } from "ng-zorro-antd/modal";
+import { NzButtonComponent } from "ng-zorro-antd/button";
+import { NzWaveDirective } from "ng-zorro-antd/core/wave";
+import { ɵNzTransitionPatchDirective } from "ng-zorro-antd/core/transition-patch";
+import { NzIconDirective } from "ng-zorro-antd/icon";
+import { NzTooltipDirective } from "ng-zorro-antd/tooltip";
+import { ModelSelectionModalComponent } from "../model-selection-modal/model-selection-modal.component";
+import { DatasetSelectionModalComponent } from "../dataset-selection-modal/dataset-selection-modal.component";
+import { DATASET_INPUT_TYPE, MODEL_INPUT_TYPE } from "../../service/code-editor/ui-udf-parameters-parser.service";
+import { MODEL_ICON } from "../../../common/icon/model-icon";
+
+type ResourceBrowser = Readonly<{
+ title: string;
+ icon: string;
+ emptyLabel: string;
+ component: Type<unknown>;
+ data?: Readonly<Record<string, unknown>>;
+}>;
+
+const BROWSERS: Readonly<Record<string, ResourceBrowser>> = {
+ [MODEL_INPUT_TYPE]: {
+ title: "Select a model version",
+ icon: MODEL_ICON,
+ emptyLabel: "Select model",
+ component: ModelSelectionModalComponent,
+ },
+ [DATASET_INPUT_TYPE]: {
+ title: "Select a dataset version",
+ icon: "database",
+ emptyLabel: "Select dataset",
+ component: DatasetSelectionModalComponent,
+ // A mount exposes a whole version, so the value is the version's path, not a file in it.
+ data: { fileMode: false },
+ },
+};
+
+/**
+ * Value editor for a UDF parameter declared with `value=Resource.MODEL` or `Resource.DATASET`.
+ * The value is still a string — the chosen version's path, picked from that resource's browser
+ * rather than typed.
+ */
+@UntilDestroy()
+@Component({
+ selector: "texera-resource-value-selector",
+ templateUrl: "./resource-value-selector.component.html",
+ styleUrls: ["./resource-value-selector.component.scss"],
+ imports: [NzButtonComponent, NzWaveDirective, ɵNzTransitionPatchDirective, NzIconDirective, NzTooltipDirective],
+})
+export class ResourceValueSelectorComponent extends FieldType<FieldTypeConfig> {
+ constructor(private modalService: NzModalService) {
+ super();
+ }
+
+ private get browser(): ResourceBrowser {
+ const resource = (this.props as { resource?: string } | undefined)?.resource ?? "";
+ return BROWSERS[resource] ?? BROWSERS[DATASET_INPUT_TYPE];
+ }
+
+ get icon(): string {
+ return this.browser.icon;
+ }
+
+ get selectedPath(): string {
+ return (this.formControl?.value as string) ?? "";
+ }
+
+ get tooltip(): string {
+ return this.selectedPath || this.browser.title;
+ }
+
+ /** Name and version only — the cell is narrow, and the full path is the tooltip. */
+ get label(): string {
+ const parts = this.selectedPath.split("/").filter(part => part.length > 0);
+ return parts.length >= 4 ? `${parts[2]} · ${parts[3]}` : this.selectedPath || this.browser.emptyLabel;
+ }
+
+ openPicker(): void {
+ if (this.formControl?.disabled) return;
+ const browser = this.browser;
+ const modal = this.modalService.create({
+ nzTitle: browser.title,
+ nzContent: browser.component,
+ nzFooter: null,
+ nzData: { ...(browser.data ?? {}), selectedPath: this.selectedPath || null },
+ // Explicit: the browsers size to their contents, so "fit-content" collapses the dialog.
+ nzWidth: 720,
+ nzBodyStyle: { overflow: "auto", minHeight: "200px", maxHeight: "70vh" },
+ });
+ modal.afterClose.pipe(untilDestroyed(this)).subscribe((selectedPath?: string) => {
+ if (!selectedPath) return;
+ this.formControl.setValue(selectedPath);
+ this.formControl.markAsDirty();
+ this.formControl.markAsTouched();
+ });
+ }
+}
diff --git a/frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.spec.ts b/frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.spec.ts
index 089c21948d..cc2b8beaa0 100644
--- a/frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.spec.ts
+++ b/frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.spec.ts
@@ -17,7 +17,7 @@
* under the License.
*/
-import { FormControl } from "@angular/forms";
+import { FormControl, UntypedFormArray } from "@angular/forms";
import { ComponentFixture, TestBed } from "@angular/core/testing";
import { By } from "@angular/platform-browser";
import { FormlyFieldConfig } from "@ngx-formly/core";
@@ -130,6 +130,91 @@ describe("UiUdfParametersComponent", () => {
});
});
+ it("should edit a row that names a resource with that resource's browser, and leave others alone", () => {
+ const columns = () =>
+ rowConfig([
+ { key: "value", formControl: new FormControl("") },
+ { key: "attributeName", formControl: new FormControl("SOURCE") },
+ { key: "attributeType", formControl: new FormControl("string") },
+ ]);
+ const resourceRow = columns();
+ const plainRow = columns();
+ const unknownRow = columns();
+
+ component.onPopulate({
+ model: [{ inputType: "model" }, {}, { inputType: "workflow" }],
+ fieldGroup: [resourceRow, plainRow, unknownRow],
+ } as FormlyFieldConfig);
+
+ const valueOf = (row: FormlyFieldConfig) => component.getColumnField(row, component.fieldColumns[0]);
+ expect(valueOf(resourceRow)?.type).toBe("resourcevalue");
+ expect(valueOf(resourceRow)?.props?.resource).toBe("model");
+ expect(valueOf(plainRow)?.type).toBeUndefined();
+ expect(valueOf(unknownRow)?.type).toBeUndefined();
+ });
+
+ it("should find the rows before Formly narrows the field's model to them", () => {
+ const rows = [{ inputType: "dataset", attribute: { attributeName: "DATA" } }];
+ const operatorProperties = { code: "", uiParameters: rows };
+ const field: FormlyFieldConfig = {
+ key: "uiParameters",
+ props: {},
+ formControl: new UntypedFormArray([]),
+ fieldArray: rowConfig([{ key: "value" }, { key: "attributeName" }, { key: "attributeType" }]),
+ fieldGroup: [],
+ };
+ // Formly hands this field the whole operator's properties and narrows `model` to the
+ // parameter array while the base class populates, so the first read sees the properties.
+ let narrowed = false;
+ Object.defineProperty(field, "model", {
+ get: () => {
+ const model = narrowed ? rows : operatorProperties;
+ narrowed = true;
+ return model;
+ },
+ configurable: true,
+ });
+
+ component.onPopulate(field);
+
+ expect(component.getColumnField(field.fieldGroup![0], component.fieldColumns[0])?.type).toBe("resourcevalue");
+ });
+
+ it("should rebuild a row whose resource changed, so its cell renders the editor it now needs", () => {
+ const columnKeys = [{ key: "value" }, { key: "attributeName" }, { key: "attributeType" }];
+ const rows: object[] = [{ attribute: { attributeName: "DATA" } }, { attribute: { attributeName: "count" } }];
+ const field: FormlyFieldConfig = {
+ model: rows,
+ fieldArray: rowConfig(columnKeys),
+ fieldGroup: [],
+ };
+ const valueOf = (index: number) => component.getColumnField(field.fieldGroup![index], component.fieldColumns[0]);
+
+ component.onPopulate(field);
+ const plainRow = field.fieldGroup![0];
+ expect(valueOf(0)?.type).toBeUndefined();
+
+ // The code now declares DATA with value=Resource.DATASET.
+ rows[0] = { inputType: "dataset", attribute: { attributeName: "DATA" } };
+ component.onPopulate(field);
+ expect(field.fieldGroup![0]).not.toBe(plainRow);
+ expect(valueOf(0)?.type).toBe("resourcevalue");
+ expect(valueOf(0)?.props?.resource).toBe("dataset");
+
+ // Populating again with nothing changed keeps the rows.
+ const resourceRow = field.fieldGroup![0];
+ component.onPopulate(field);
+ expect(field.fieldGroup![0]).toBe(resourceRow);
+
+ // And back to free text: the browser must not linger on the row.
+ rows[0] = { attribute: { attributeName: "DATA" } };
+ component.onPopulate(field);
+ expect(field.fieldGroup![0]).not.toBe(resourceRow);
+ expect(valueOf(0)?.type).toBeUndefined();
+ expect(valueOf(0)?.props?.["resource"]).toBeUndefined();
+ expect(field.fieldGroup).toHaveLength(2);
+ });
+
it("should apply disabled state to rows generated from the field array template", () => {
const field: FormlyFieldConfig = {
model: [{ value: "42", attribute: { attributeName: "threshold", attributeType: "double" } }],
diff --git a/frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.ts b/frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.ts
index d1aebb0cb9..15afa66ccd 100644
--- a/frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.ts
+++ b/frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.ts
@@ -30,10 +30,18 @@ import {
UiUdfParametersParseError,
} from "../../service/code-editor/ui-udf-parameters-parser.service";
import { UiUdfParametersSyncService } from "../../service/code-editor/ui-udf-parameters-sync.service";
+import { DATASET_INPUT_TYPE, MODEL_INPUT_TYPE } from "../../service/code-editor/ui-udf-parameters-parser.service";
import type { AttributeType } from "../../types/workflow-compiling.interface";
type UiUdfParameterColumn = Readonly<{ label: string; key: string; parentKey?: string; disabled: boolean }>;
+const VALUE_COLUMN: UiUdfParameterColumn = { label: "Value", key: "value", disabled: false };
+const RESOURCE_VALUE_EDITOR = "resourcevalue";
+const RESOURCE_INPUT_TYPES: ReadonlySet<string> = new Set([MODEL_INPUT_TYPE, DATASET_INPUT_TYPE]);
+
+/** The resource a row's value names, or "" for free text. */
+const resourceOf = (inputType?: string): string => (inputType && RESOURCE_INPUT_TYPES.has(inputType) ? inputType : "");
+
/** Renders inferred Python UDF UI parameters with editable values and locked name/type columns. */
@Component({
selector: "texera-ui-udf-parameters",
@@ -51,9 +59,11 @@ type UiUdfParameterColumn = Readonly<{ label: string; key: string; parentKey?: s
})
export class UiUdfParametersComponent extends FieldArrayType<FormlyFieldConfig> {
private readonly disabledStateConfigured = new WeakMap<FormlyFieldConfig, boolean>();
+ // The resource each row's value editor was configured for.
+ private readonly rowResources = new WeakMap<FormlyFieldConfig, string>();
readonly fieldColumns: UiUdfParameterColumn[] = [
- { label: "Value", key: "value", disabled: false },
+ VALUE_COLUMN,
{ label: "Name", key: "attributeName", parentKey: "attribute", disabled: true },
{ label: "Type", key: "attributeType", parentKey: "attribute", disabled: true },
];
@@ -87,8 +97,21 @@ export class UiUdfParametersComponent extends FieldArrayType<FormlyFieldConfig>
override onPopulate(field: FormlyFieldConfig): void {
this.configureRowTemplate(this.getFieldArrayTemplate(field));
+ this.dropRowsWhoseResourceChanged(field, this.parameterRows(field));
super.onPopulate(field);
- field.fieldGroup?.forEach(rowField => this.configureRowFields(rowField));
+ const rows = this.parameterRows(field);
+ field.fieldGroup?.forEach((rowField, index) => this.configureRowFields(rowField, rows[index]?.inputType));
+ }
+
+ /**
+ * The parameter rows. Before Formly narrows `model` to this field's own array it still holds
+ * the whole operator's properties, so the array is taken from this field's key in that case.
+ */
+ private parameterRows(field: FormlyFieldConfig): ReadonlyArray<{ inputType?: string } | undefined> {
+ const model: unknown = Array.isArray(field.model)
+ ? field.model
+ : (field.model as Record<string, unknown> | undefined)?.[String(field.key)];
+ return Array.isArray(model) ? model : [];
}
/** Finds the Formly field config that backs one visible column in a parameter row. */
@@ -104,8 +127,37 @@ export class UiUdfParametersComponent extends FieldArrayType<FormlyFieldConfig>
this.configureRowColumns(rowField, this.setDisabledMetadata.bind(this));
}
- private configureRowFields(rowField: FormlyFieldConfig | undefined): void {
+ private configureRowFields(rowField: FormlyFieldConfig | undefined, inputType?: string): void {
this.configureRowColumns(rowField, this.configureDisabledState.bind(this));
+ this.configureValueEditor(rowField, inputType);
+ }
+
+ /** A row whose value names a resource is edited with that resource's browser, not a text box. */
+ private configureValueEditor(rowField: FormlyFieldConfig | undefined, inputType?: string): void {
+ if (!rowField) return;
+ const resource = resourceOf(inputType);
+ this.rowResources.set(rowField, resource);
+ const valueField = this.getColumnField(rowField, VALUE_COLUMN);
+ if (!valueField || !resource) return;
+ valueField.type = RESOURCE_VALUE_EDITOR;
+ valueField.props = { ...(valueField.props ?? {}), resource };
+ }
+
+ /**
+ * Formly keeps a row's config when the parameters are rebuilt from the code, and re-renders a
+ * cell only when handed a new config. A row whose resource changed is therefore dropped, with
+ * every row after it, so Formly rebuilds them from the template with the editor they now need.
+ */
+ private dropRowsWhoseResourceChanged(
+ field: FormlyFieldConfig,
+ rows: ReadonlyArray<{ inputType?: string } | undefined>
+ ): void {
+ const rowFields = field.fieldGroup ?? [];
+ const firstChanged = rowFields.findIndex(
+ (rowField, index) =>
+ this.rowResources.has(rowField) && this.rowResources.get(rowField) !== resourceOf(rows[index]?.inputType)
+ );
+ if (firstChanged >= 0) rowFields.splice(firstChanged);
}
private configureRowColumns(
diff --git a/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-parser.service.spec.ts b/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-parser.service.spec.ts
index b9241ce1e2..ffe134e170 100644
--- a/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-parser.service.spec.ts
+++ b/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-parser.service.spec.ts
@@ -18,6 +18,8 @@
*/
import {
+ DATASET_INPUT_TYPE,
+ MODEL_INPUT_TYPE,
UiUdfParametersEditError,
UiUdfParametersParseError,
UiUdfParametersParserService,
@@ -612,3 +614,41 @@ ${openStatements}
function parameter(attributeName: string, attributeType: UiUdfParameter["attribute"]["attributeType"]): UiUdfParameter {
return { attribute: { attributeName, attributeType }, value: "" };
}
+
+describe("UiUdfParametersParserService resource parameters", () => {
+ const service = new UiUdfParametersParserService();
+ const inOpen = (...lines: string[]): string =>
+ `class ProcessTupleOperator(UDFOperatorV2):\n def open(self):\n${lines.map(line => ` ${line}\n`).join("")}`;
+
+ it("marks a parameter declared with value=Resource.X as naming that resource", () => {
+ expect(
+ service.parse(
+ inOpen(
+ 'self.UiParameter("MODEL", AttributeType.STRING, value=Resource.MODEL)',
+ 'self.UiParameter(name="DATA", type=AttributeType.STRING, value=Resource.DATASET)',
+ 'self.UiParameter("count", AttributeType.INT)'
+ )
+ )
+ ).toEqual([
+ { attribute: { attributeName: "MODEL", attributeType: "string" }, value: "", inputType: MODEL_INPUT_TYPE },
+ { attribute: { attributeName: "DATA", attributeType: "string" }, value: "", inputType: DATASET_INPUT_TYPE },
+ { attribute: { attributeName: "count", attributeType: "integer" }, value: "" },
+ ]);
+ });
+
+ it("ignores a declaration naming a resource it does not know, or not through Resource", () => {
+ expect(
+ service.parse(
+ inOpen(
+ 'self.UiParameter("A", AttributeType.STRING, value=Resource.WORKFLOW)',
+ 'self.UiParameter("B", AttributeType.STRING, value="model")',
+ 'self.UiParameter("C", AttributeType.STRING, value=Other.MODEL)'
+ )
+ )
+ ).toEqual([]);
+ });
+
+ it("ignores a resource parameter that is not a string, since it receives a directory path", () => {
+ expect(service.parse(inOpen('self.UiParameter("MODEL", AttributeType.INT, value=Resource.MODEL)'))).toEqual([]);
+ });
+});
diff --git a/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-parser.service.ts b/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-parser.service.ts
index 3db18b05a9..4d07fbd279 100644
--- a/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-parser.service.ts
+++ b/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-parser.service.ts
@@ -51,16 +51,27 @@ const ATTRIBUTE_TYPE_RECEIVER = "AttributeType";
const ARGUMENT_NAME = "name";
const ARGUMENT_TYPE = "type";
const ARGUMENT_ATTR_TYPE = "attr_type";
+const ARGUMENT_VALUE = "value";
const POSITIONAL_ARGUMENT_KEYS = [ARGUMENT_NAME, ARGUMENT_TYPE] as const;
+// A resource the value names, declared as `value=Resource.X`; keep in sync with pytexera's Resource.
+const RESOURCE_RECEIVER = "Resource";
+export const MODEL_INPUT_TYPE = "model";
+export const DATASET_INPUT_TYPE = "dataset";
+const INPUT_TYPES_BY_RESOURCE_TOKEN: Readonly<Record<string, string>> = {
+ MODEL: MODEL_INPUT_TYPE,
+ DATASET: DATASET_INPUT_TYPE,
+};
+
type ParserSyntaxNode = ReturnType<typeof parser.parse>["topNode"];
type ParsedArgument = Readonly<{ key?: string; value: ParserSyntaxNode }>;
type UiParameterArgument =
| Readonly<{ kind: typeof ARGUMENT_NAME; value: string }>
- | Readonly<{ kind: typeof ARGUMENT_TYPE; value: AttributeType }>;
+ | Readonly<{ kind: typeof ARGUMENT_TYPE; value: AttributeType }>
+ | Readonly<{ kind: typeof ARGUMENT_VALUE; value: string }>;
/** UI parameter row inferred from Python code, with backend-compatible attribute metadata and an editable value. */
-export type UiUdfParameter = Readonly<{ attribute: SchemaAttribute; value: string }>;
+export type UiUdfParameter = Readonly<{ attribute: SchemaAttribute; value: string; inputType?: string }>;
/** Raised when supported Python UDF code declares UI parameters that cannot be represented safely in the UI. */
export class UiUdfParametersParseError extends Error {}
@@ -265,16 +276,20 @@ function readCall(call: ParserSyntaxNode, code: string): UiUdfParameter | undefi
let attributeName: string | undefined;
let attributeType: AttributeType | undefined;
+ let inputType: string | undefined;
const uiParameterArguments = readUiParameterArguments(argumentList, code);
if (!uiParameterArguments) return undefined;
for (const argument of uiParameterArguments) {
if (argument.kind === ARGUMENT_NAME && !attributeName) attributeName = argument.value;
else if (argument.kind === ARGUMENT_TYPE && !attributeType) attributeType = argument.value;
+ else if (argument.kind === ARGUMENT_VALUE && !inputType) inputType = argument.value;
else return undefined;
}
- return attributeName && attributeType ? { attribute: { attributeName, attributeType }, value: "" } : undefined;
+ // What a resource parameter receives is a directory path, so it has to be a string.
+ if (!attributeName || !attributeType || (inputType && attributeType !== "string")) return undefined;
+ return { attribute: { attributeName, attributeType }, value: "", ...(inputType ? { inputType } : {}) };
}
function readUiParameterArguments(argumentList: ParserSyntaxNode, code: string): UiParameterArgument[] | undefined {
@@ -308,6 +323,10 @@ function readUiParameterArgument(
const attributeType = readType(value, code);
return attributeType ? { kind: ARGUMENT_TYPE, value: attributeType } : undefined;
}
+ if (key === ARGUMENT_VALUE) {
+ const inputType = readResource(value, code);
+ return inputType ? { kind: ARGUMENT_VALUE, value: inputType } : undefined;
+ }
return undefined;
}
@@ -351,6 +370,12 @@ function readType(value: ParserSyntaxNode, code: string): AttributeType | undefi
return token ? ATTRIBUTE_TYPES_BY_TOKEN[token] : undefined;
}
+function readResource(value: ParserSyntaxNode, code: string): string | undefined {
+ const parts = readMemberPath(value, code);
+ if (parts?.length !== 2 || parts[0] !== RESOURCE_RECEIVER) return undefined;
+ return INPUT_TYPES_BY_RESOURCE_TOKEN[parts[1].toUpperCase()];
+}
+
function isMemberPath(node: ParserSyntaxNode | null, code: string, expectedParts: string[]): boolean {
const parts = node ? readMemberPath(node, code) : undefined;
return parts?.length === expectedParts.length && parts.every((part, index) => part === expectedParts[index]);
diff --git a/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-sync.service.spec.ts b/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-sync.service.spec.ts
index b6777b5a75..3ce6e92ca3 100644
--- a/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-sync.service.spec.ts
+++ b/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-sync.service.spec.ts
@@ -94,6 +94,34 @@ describe("UiUdfParametersSyncService", () => {
expect(parametersChangedObserver).not.toHaveBeenCalled();
});
+ describe("resource rows", () => {
+ const row = (inputType: string | undefined, value = ""): UiUdfParameter => ({
+ attribute: { attributeName: "SOURCE", attributeType: "string" },
+ value,
+ ...(inputType ? { inputType } : {}),
+ });
+
+ it("should drop the value when the row starts naming a different kind of resource", () => {
+ operator.operatorProperties.uiParameters = [row("model", "/model/owner/iris/v1")];
+ parserServiceMock.parse.mockReturnValue([row("dataset")]);
+ const parametersChangedObserver = observeParameterChanges();
+
+ service.syncStructureFromCode(operatorId, code);
+
+ expect(parametersChangedObserver).toHaveBeenCalledWith({ operatorId, parameters: [row("dataset", "")] });
+ });
+
+ it("should keep the chosen version while the row still names the same resource", () => {
+ operator.operatorProperties.uiParameters = [row("model", "/model/owner/iris/v1")];
+ parserServiceMock.parse.mockReturnValue([row("model")]);
+ const parametersChangedObserver = observeParameterChanges();
+
+ service.syncStructureFromCode(operatorId, code);
+
+ expect(parametersChangedObserver).not.toHaveBeenCalled();
+ });
+ });
+
it("should not replay a previous parameter change to a late subscriber", () => {
parserServiceMock.parse.mockReturnValue([parameter("count", "integer")]);
diff --git a/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-sync.service.ts b/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-sync.service.ts
index 54aa5ec363..271bea653b 100644
--- a/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-sync.service.ts
+++ b/frontend/src/app/workspace/service/code-editor/ui-udf-parameters-sync.service.ts
@@ -112,13 +112,14 @@ export class UiUdfParametersSyncService {
private buildParsedShapeWithPreservedValues(code: string, existingParameters: UiUdfParameter[]): UiUdfParameter[] {
const parsedParameters = this.uiUdfParametersParserService.parse(code);
- const existingValues = new Map(
- existingParameters.map(parameter => [parameter.attribute.attributeName, parameter.value] as const)
- );
+ // A value carries over only while the row names the same kind of resource: a model path
+ // left in a row that now names a dataset would resolve to the wrong thing.
+ const rowKey = (parameter: UiUdfParameter) => `${parameter.attribute.attributeName}:${parameter.inputType ?? ""}`;
+ const existingValues = new Map(existingParameters.map(parameter => [rowKey(parameter), parameter.value] as const));
return parsedParameters.map(parameter => ({
...parameter,
- value: existingValues.get(parameter.attribute.attributeName) ?? "",
+ value: existingValues.get(rowKey(parameter)) ?? "",
}));
}