This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new a7f4386ba7 feat(udf): enable Python UDF UI parameters end to end
(#5912)
a7f4386ba7 is described below
commit a7f4386ba7ccb58155e44f79c8e10a25e60d2208
Author: carloea2 <[email protected]>
AuthorDate: Mon Aug 10 11:35:13 2026 -0700
feat(udf): enable Python UDF UI parameters end to end (#5912)
### What changes were proposed in this PR?
This PR wires the existing Python UDF UI parameter support into the
end-to-end execution path.
It changes:
| Area | Change |
| --- | --- |
| Python UDF descriptors | Adds the `uiParameters` property to Python
UDF, dual-input Python UDF, and Python UDF source operators. |
| Execution wiring | Applies `PythonUdfUiParameterInjector.inject(...)`
before creating Python physical operators. |
| Frontend wiring | Connects the existing UI-parameter sync service to
the code editor and property editor, and renders `uiParameters` with the
existing UI-parameter editor. |
| Operator templates | Adds commented `UiParameter` examples to Python
UDF templates. |
### Any related issues, documentation, discussions?
Part of the Python UDF UI parameter feature split from
`feat/ui-parameter`.
Related tracking issue / stack: #5044
Stack order:
1. Frontend UI parameter building blocks: #5043
2. Scala backend injection model: #5141
3. Python runtime support: #5603
4. End-to-end execution wiring: this PR
### How was this PR tested?
Manually +
Commands run:
```bash
sbt --no-server scalafmtAll scalafixAll
sbt --no-server scalafmtCheckAll "scalafixAll --check"
cd amber
ruff check src/main/python src/test/python
ruff format --check src/main/python src/test/python
cd ..
sbt --no-server "WorkflowExecutionService / Test / testOnly
org.apache.texera.amber.engine.architecture.pythonworker.PythonWorkflowWorkerStartupConfigSpec"
```
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Me
---
amber/src/main/python/pytexera/udf/udf_operator.py | 9 +++++
.../test/python/pytexera/udf/test_udf_operator.py | 27 ++++++++-------
.../python/DualInputPortsPythonUDFOpDescV2.scala | 11 ++++--
.../operator/udf/python/PythonUDFOpDescV2.scala | 11 ++++--
.../udf/python/PythonUdfUiParameterInjector.scala | 28 +++++++++++++--
.../udf/python/PythonUdfUiParameterSupport.scala | 37 ++++++++++++++++++++
.../python/source/PythonUDFSourceOpDescV2.scala | 17 +++++++--
.../DualInputPortsPythonUDFOpDescV2Spec.scala | 28 +++++++++++++++
.../udf/python/PythonUDFOpDescV2Spec.scala | 28 +++++++++++++++
.../python/PythonUdfUiParameterInjectorSpec.scala | 22 ++++++++++++
.../source/PythonUDFSourceOpDescV2Spec.scala | 29 ++++++++++++++++
.../python/1-out-python-udf.md | 5 +++
.../python/2-in-python-udf.md | 5 +++
.../user-defined-functions/python/_index.md | 20 +++++++++++
.../user-defined-functions/python/python-udf.md | 5 +++
frontend/LICENSE-binary | 4 +++
.../code-editor.component.spec.ts | 38 ++++++++++++++++++++
.../code-editor-dialog/code-editor.component.ts | 26 +++++++++++++-
.../operator-property-edit-frame.component.scss | 11 ++++++
.../operator-property-edit-frame.component.spec.ts | 40 ++++++++++++++++++++++
.../operator-property-edit-frame.component.ts | 27 ++++++++++++++-
.../ui-udf-parameters.component.html | 38 ++++++++++----------
.../ui-udf-parameters-sync.service.spec.ts | 21 ++++++++++++
.../code-editor/ui-udf-parameters-sync.service.ts | 8 ++---
24 files changed, 446 insertions(+), 49 deletions(-)
diff --git a/amber/src/main/python/pytexera/udf/udf_operator.py
b/amber/src/main/python/pytexera/udf/udf_operator.py
index 7c0fd239c7..4727c9399a 100644
--- a/amber/src/main/python/pytexera/udf/udf_operator.py
+++ b/amber/src/main/python/pytexera/udf/udf_operator.py
@@ -170,6 +170,15 @@ class _UiParameterSupport:
if value is None:
return None
+ if (
+ attr_type is not AttributeType.STRING
+ and isinstance(value, str)
+ and not value.strip()
+ ):
+ raise ValueError(
+ f"UiParameter value cannot be empty for type {attr_type.name}."
+ )
+
try:
return parser(value)
except Exception as e:
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 a04e52b3a1..1d93eba5e4 100644
--- a/amber/src/test/python/pytexera/udf/test_udf_operator.py
+++ b/amber/src/test/python/pytexera/udf/test_udf_operator.py
@@ -238,20 +238,23 @@ class TestUiParameterSupport:
assert _UiParameterSupport._parse(raw_value, attr_type) == expected
@pytest.mark.parametrize(
- ("raw_value", "attr_type", "expected"),
+ ("raw_value", "attr_type"),
[
- ("", AttributeType.INT, 0),
- (" ", AttributeType.LONG, 0),
- ("", AttributeType.DOUBLE, 0.0),
- (
- "",
- AttributeType.TIMESTAMP,
- datetime.datetime(1970, 1, 1, tzinfo=datetime.timezone.utc),
- ),
+ ("", AttributeType.INT),
+ (" ", AttributeType.LONG),
+ ("", AttributeType.DOUBLE),
+ ("\t", AttributeType.BOOL),
+ ("", AttributeType.TIMESTAMP),
],
)
- def test_parse_empty_values(self, raw_value, attr_type, expected):
- assert _UiParameterSupport._parse(raw_value, attr_type) == expected
+ def test_parse_empty_non_string_values_raises_value_error(
+ self, raw_value, attr_type
+ ):
+ with pytest.raises(ValueError, match="UiParameter value cannot be
empty"):
+ _UiParameterSupport._parse(raw_value, attr_type)
+
+ def test_parse_empty_string_value(self):
+ assert _UiParameterSupport._parse("", AttributeType.STRING) == ""
def test_java_attribute_type_aliases_parse_like_python_names(self):
assert AttributeType.INTEGER is AttributeType.INT
@@ -262,8 +265,6 @@ class TestUiParameterSupport:
@pytest.mark.parametrize(
("raw_value", "expected"),
[
- ("", False),
- (" ", False),
("True", True),
("true", True),
("1", True),
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 1cfa29c9f1..07fec7af92 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
@@ -29,11 +29,15 @@ import org.apache.texera.amber.core.workflow._
import org.apache.texera.amber.operator.LogicalOp
import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants,
OperatorInfo}
-class DualInputPortsPythonUDFOpDescV2 extends LogicalOp {
+class DualInputPortsPythonUDFOpDescV2 extends LogicalOp with
PythonUdfUiParameterSupport {
@JsonProperty(
required = true,
defaultValue =
"# Choose from the following templates:\n" +
+ "# \n" +
+ "# Define UiParameter inside open() of ProcessTupleOperator,
ProcessBatchOperator, or ProcessTableOperator.\n" +
+ "# Example: self.count = self.UiParameter(\"count\",
AttributeType.INT).value\n" +
+ "# See the Python UDF operator documentation for supported types and
behavior.\n" +
"# \n" +
"# from pytexera import *\n" +
"# \n" +
@@ -105,6 +109,7 @@ class DualInputPortsPythonUDFOpDescV2 extends LogicalOp {
)
trimmed
}
+ val codeWithParameters = injectUiParameters(code)
val physicalOp = if (workers > 1) {
PhysicalOp
@@ -112,7 +117,7 @@ class DualInputPortsPythonUDFOpDescV2 extends LogicalOp {
workflowId,
executionId,
operatorIdentifier,
- OpExecWithCode(code, "python")
+ OpExecWithCode(codeWithParameters, "python")
)
.withParallelizable(true)
.withSuggestedWorkerNum(workers)
@@ -122,7 +127,7 @@ class DualInputPortsPythonUDFOpDescV2 extends LogicalOp {
workflowId,
executionId,
operatorIdentifier,
- OpExecWithCode(code, "python")
+ OpExecWithCode(codeWithParameters, "python")
)
.withParallelizable(false)
}
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 6739041a53..01610979ec 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
@@ -31,11 +31,15 @@ import org.apache.texera.amber.operator.{LogicalOp,
PortDescription, StateTransf
import scala.util.{Success, Try}
-class PythonUDFOpDescV2 extends LogicalOp {
+class PythonUDFOpDescV2 extends LogicalOp with PythonUdfUiParameterSupport {
@JsonProperty(
required = true,
defaultValue =
"# Choose from the following templates:\n" +
+ "# \n" +
+ "# Define UiParameter inside open() of ProcessTupleOperator,
ProcessBatchOperator, or ProcessTableOperator.\n" +
+ "# Example: self.count = self.UiParameter(\"count\",
AttributeType.INT).value\n" +
+ "# See the Python UDF operator documentation for supported types and
behavior.\n" +
"# \n" +
"# from pytexera import *\n" +
"# \n" +
@@ -102,6 +106,7 @@ class PythonUDFOpDescV2 extends LogicalOp {
} else {
opInfo.inputPorts.map(_ => None)
}
+ val codeWithParameters = injectUiParameters(code)
val propagateSchema = (inputSchemas: Map[PortIdentity, Schema]) => {
val inputSchema = inputSchemas(operatorInfo.inputPorts.head.id)
@@ -130,7 +135,7 @@ class PythonUDFOpDescV2 extends LogicalOp {
workflowId,
executionId,
operatorIdentifier,
- OpExecWithCode(code, "python")
+ OpExecWithCode(codeWithParameters, "python")
)
.withParallelizable(true)
.withSuggestedWorkerNum(workers)
@@ -140,7 +145,7 @@ class PythonUDFOpDescV2 extends LogicalOp {
workflowId,
executionId,
operatorIdentifier,
- OpExecWithCode(code, "python")
+ OpExecWithCode(codeWithParameters, "python")
)
.withParallelizable(false)
}
diff --git
a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterInjector.scala
b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterInjector.scala
index 7b26761290..bd7d4334fb 100644
---
a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterInjector.scala
+++
b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterInjector.scala
@@ -38,14 +38,27 @@ object PythonUdfUiParameterInjector {
private val InjectedUiParametersHookMethodHeader =
s"def $InjectedUiParametersHookMethodName(self) -> Dict[str, Any]:"
private val UnsupportedUiParameterTypes = Set(AttributeType.BINARY,
AttributeType.LARGE_BINARY)
+ private val UiParameterTypesRequiringValues = Set(
+ AttributeType.INTEGER,
+ AttributeType.LONG,
+ AttributeType.DOUBLE,
+ AttributeType.BOOLEAN,
+ AttributeType.TIMESTAMP
+ )
// Keep supported user-facing UDF class names in sync with the frontend
parser.
private val SupportedPythonUdfClassHeaderRegex: Regex =
"""(?m)^([
\t]*)class\s+(ProcessTupleOperator|ProcessBatchOperator|ProcessTableOperator|GenerateOperator)\s*\([^)]*\)\s*:\s*(?:#.*)?$""".r
private def validate(uiParameters: List[UiUDFParameter]): Unit = {
- val attributes = uiParameters.map(parameterAttribute)
- attributes.foreach(validateSupportedType)
+ val parametersWithAttributes =
+ uiParameters.map(parameter => parameter -> parameterAttribute(parameter))
+ val attributes = parametersWithAttributes.map(_._2)
+ parametersWithAttributes.foreach {
+ case (parameter, attribute) =>
+ validateSupportedType(attribute)
+ validateRequiredValue(parameter, attribute)
+ }
attributes
.groupBy(_.getName)
@@ -71,6 +84,17 @@ object PythonUdfUiParameterInjector {
}
}
+ private def validateRequiredValue(parameter: UiUDFParameter, attribute:
Attribute): Unit = {
+ if (
+ UiParameterTypesRequiringValues.contains(attribute.getType) &&
+ Option(parameter.value).forall(_.trim.isEmpty)
+ ) {
+ throw new RuntimeException(
+ s"UiParameter '${attribute.getName}' requires a value for type
'${attribute.getType.name()}'."
+ )
+ }
+ }
+
private def buildInjectedParameterEntry(parameter: UiUDFParameter):
PythonTemplateBuilder = {
pyb"${parameter.attribute.getName}: ${parameter.value}"
}
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
new file mode 100644
index 0000000000..97548c344b
--- /dev/null
+++
b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterSupport.scala
@@ -0,0 +1,37 @@
+/*
+ * 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 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()
+
+ protected final def injectUiParameters(code: String): String =
+ PythonUdfUiParameterInjector.inject(code, uiParameters)
+}
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 840c37238a..6ddfa23e01 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
@@ -27,12 +27,17 @@ import
org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, Workflow
import org.apache.texera.amber.core.workflow.{OutputPort, PhysicalOp,
SchemaPropagationFunc}
import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants,
OperatorInfo}
import org.apache.texera.amber.operator.source.SourceOperatorDescriptor
+import org.apache.texera.amber.operator.udf.python.PythonUdfUiParameterSupport
-class PythonUDFSourceOpDescV2 extends SourceOperatorDescriptor {
+class PythonUDFSourceOpDescV2 extends SourceOperatorDescriptor with
PythonUdfUiParameterSupport {
@JsonProperty(
required = true,
- defaultValue = "# from pytexera import *\n" +
+ defaultValue = "# Define UiParameter inside GenerateOperator.open().\n" +
+ "# Example: self.count = self.UiParameter(\"count\",
AttributeType.INT).value\n" +
+ "# See the Python UDF operator documentation for supported types and
behavior.\n" +
+ "# \n" +
+ "# from pytexera import *\n" +
"# class GenerateOperator(UDFSourceOperator):\n" +
"# \n" +
"# @overrides\n" +
@@ -82,9 +87,15 @@ class PythonUDFSourceOpDescV2 extends
SourceOperatorDescriptor {
)
trimmed
}
+ val codeWithParameters = injectUiParameters(code)
val physicalOp = PhysicalOp
- .sourcePhysicalOp(workflowId, executionId, operatorIdentifier,
OpExecWithCode(code, "python"))
+ .sourcePhysicalOp(
+ workflowId,
+ executionId,
+ operatorIdentifier,
+ OpExecWithCode(codeWithParameters, "python")
+ )
.withInputPorts(operatorInfo.inputPorts)
.withOutputPorts(operatorInfo.outputPorts)
.withIsOneToManyOp(true)
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 6c246d5008..74caa27379 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
@@ -34,6 +34,13 @@ class DualInputPortsPythonUDFOpDescV2Spec extends
AnyFlatSpec with Matchers {
private val workflowId = WorkflowIdentity(1L)
private val executionId = ExecutionIdentity(1L)
+ private def uiParameter(name: String, value: String): UiUDFParameter = {
+ val parameter = new UiUDFParameter
+ parameter.attribute = new Attribute(name, AttributeType.INTEGER)
+ parameter.value = value
+ parameter
+ }
+
private def twoPortSchemas(dataSchema: Schema): Map[PortIdentity, Schema] =
Map(
PortIdentity() -> Schema().add(new Attribute("model",
AttributeType.STRING)),
@@ -87,6 +94,24 @@ class DualInputPortsPythonUDFOpDescV2Spec extends
AnyFlatSpec with Matchers {
physical.suggestedWorkerNum shouldBe Some(2)
}
+ it should "inject configured UI parameters into the generated Python code"
in {
+ val d = new DualInputPortsPythonUDFOpDescV2
+ d.code = """from pytexera import *
+ |
+ |class ProcessTupleOperator(UDFOperatorV2):
+ | def process_tuple(self, tuple_, port):
+ | yield tuple_
+ |""".stripMargin
+ d.uiParameters = List(uiParameter("count", "7"))
+
+ d.getPhysicalOp(workflowId, executionId).opExecInitInfo match {
+ case OpExecWithCode(code, "python") =>
+ code should include("def _texera_injected_ui_parameters")
+ code should include("self.decode_python_template")
+ case other => fail(s"expected Python OpExecWithCode, got $other")
+ }
+ }
+
it should "reject a blank virtual-environment name when the default env is
disabled" in {
val d = new DualInputPortsPythonUDFOpDescV2
d.defaultEnv = false
@@ -154,6 +179,7 @@ class DualInputPortsPythonUDFOpDescV2Spec extends
AnyFlatSpec with Matchers {
d.defaultEnv = false
d.envName = "myenv"
d.outputColumns = List(new Attribute("res", AttributeType.INTEGER))
+ d.uiParameters = List(uiParameter("count", "7"))
val restored = objectMapper.readValue(objectMapper.writeValueAsString(d),
classOf[LogicalOp])
restored shouldBe a[DualInputPortsPythonUDFOpDescV2]
val r = restored.asInstanceOf[DualInputPortsPythonUDFOpDescV2]
@@ -163,5 +189,7 @@ class DualInputPortsPythonUDFOpDescV2Spec extends
AnyFlatSpec with Matchers {
r.defaultEnv shouldBe false
r.envName shouldBe "myenv"
r.outputColumns shouldBe List(new Attribute("res", AttributeType.INTEGER))
+ r.uiParameters.map(_.attribute) shouldBe List(new Attribute("count",
AttributeType.INTEGER))
+ r.uiParameters.map(_.value) shouldBe List("7")
}
}
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 aea38193aa..a36dee6107 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,13 @@ class PythonUDFOpDescV2Spec extends AnyFlatSpec with
Matchers {
private val workflowId = WorkflowIdentity(1L)
private val executionId = ExecutionIdentity(1L)
+ private def uiParameter(name: String, value: String): UiUDFParameter = {
+ val parameter = new UiUDFParameter
+ parameter.attribute = new Attribute(name, AttributeType.INTEGER)
+ parameter.value = value
+ parameter
+ }
+
"PythonUDFOpDescV2.operatorInfo" should
"advertise the name, Python group, dynamic ports, and a default 1-in/1-out
shape" in {
val info = (new PythonUDFOpDescV2).operatorInfo
@@ -74,6 +81,24 @@ class PythonUDFOpDescV2Spec extends AnyFlatSpec with
Matchers {
intercept[IllegalArgumentException] { d.getPhysicalOp(workflowId,
executionId) }
}
+ it should "inject configured UI parameters into the generated Python code"
in {
+ val d = new PythonUDFOpDescV2
+ d.code = """from pytexera import *
+ |
+ |class ProcessTupleOperator(UDFOperatorV2):
+ | def process_tuple(self, tuple_, port):
+ | yield tuple_
+ |""".stripMargin
+ d.uiParameters = List(uiParameter("count", "7"))
+
+ d.getPhysicalOp(workflowId, executionId).opExecInitInfo match {
+ case OpExecWithCode(code, "python") =>
+ code should include("def _texera_injected_ui_parameters")
+ code should include("self.decode_python_template")
+ case other => fail(s"expected Python OpExecWithCode, got $other")
+ }
+ }
+
it should "reject a blank virtual-environment name when the default env is
disabled" in {
val d = new PythonUDFOpDescV2
d.defaultEnv = false
@@ -125,6 +150,7 @@ class PythonUDFOpDescV2Spec extends AnyFlatSpec with
Matchers {
d.defaultEnv = false
d.envName = "myenv"
d.outputColumns = List(new Attribute("res", AttributeType.INTEGER))
+ d.uiParameters = List(uiParameter("count", "7"))
val restored = objectMapper.readValue(objectMapper.writeValueAsString(d),
classOf[LogicalOp])
restored shouldBe a[PythonUDFOpDescV2]
val p = restored.asInstanceOf[PythonUDFOpDescV2]
@@ -134,6 +160,8 @@ class PythonUDFOpDescV2Spec extends AnyFlatSpec with
Matchers {
p.defaultEnv shouldBe false
p.envName shouldBe "myenv"
p.outputColumns shouldBe List(new Attribute("res", AttributeType.INTEGER))
+ p.uiParameters.map(_.attribute) shouldBe List(new Attribute("count",
AttributeType.INTEGER))
+ p.uiParameters.map(_.value) shouldBe List("7")
}
"PythonUDFOpDescV2.getPhysicalOp" should
diff --git
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterInjectorSpec.scala
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterInjectorSpec.scala
index a741dfbf03..bf387a7aa2 100644
---
a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterInjectorSpec.scala
+++
b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterInjectorSpec.scala
@@ -208,6 +208,28 @@ class PythonUdfUiParameterInjectorSpec extends AnyFlatSpec
with Matchers {
exception.getMessage should include("UiParameter name 'date' is declared
more than once")
}
+ Seq(
+ AttributeType.INTEGER,
+ AttributeType.LONG,
+ AttributeType.DOUBLE,
+ AttributeType.BOOLEAN,
+ AttributeType.TIMESTAMP
+ ).foreach { attributeType =>
+ it should s"throw when a ${attributeType.name()} UI parameter value is
blank" in {
+ val exception = the[RuntimeException] thrownBy {
+ inject(uiParameter("required", attributeType, " "))
+ }
+
+ exception.getMessage should include("UiParameter 'required' requires a
value")
+ }
+ }
+
+ it should "allow an empty string UI parameter value" in {
+ inject(uiParameter("optional_text", AttributeType.STRING, "")) should
include(
+ "def _texera_injected_ui_parameters"
+ )
+ }
+
Seq(AttributeType.BINARY, AttributeType.LARGE_BINARY).foreach {
unsupportedType =>
it should s"throw when a UI parameter uses ${unsupportedType.name()} type"
in {
val exception = the[RuntimeException] thrownBy {
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 c6320d775e..cda363511b 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
@@ -24,6 +24,7 @@ import org.apache.texera.amber.core.tuple.{Attribute,
AttributeType, Schema}
import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity,
WorkflowIdentity}
import org.apache.texera.amber.operator.LogicalOp
import org.apache.texera.amber.operator.metadata.OperatorGroupConstants
+import org.apache.texera.amber.operator.udf.python.UiUDFParameter
import org.apache.texera.amber.util.JSONUtils.objectMapper
import org.scalatest.flatspec.AnyFlatSpec
import org.scalatest.matchers.should.Matchers
@@ -33,6 +34,13 @@ class PythonUDFSourceOpDescV2Spec extends AnyFlatSpec with
Matchers {
private val workflowId = WorkflowIdentity(1L)
private val executionId = ExecutionIdentity(1L)
+ private def uiParameter(name: String, value: String): UiUDFParameter = {
+ val parameter = new UiUDFParameter
+ parameter.attribute = new Attribute(name, AttributeType.INTEGER)
+ parameter.value = value
+ parameter
+ }
+
"PythonUDFSourceOpDescV2.operatorInfo" should
"advertise the 1-out Python UDF source (no inputs, one output,
reconfigurable)" in {
val info = (new PythonUDFSourceOpDescV2).operatorInfo
@@ -70,6 +78,24 @@ class PythonUDFSourceOpDescV2Spec extends AnyFlatSpec with
Matchers {
intercept[IllegalArgumentException] { d.getPhysicalOp(workflowId,
executionId) }
}
+ it should "inject configured UI parameters into the generated Python code"
in {
+ val d = new PythonUDFSourceOpDescV2
+ d.code = """from pytexera import *
+ |
+ |class GenerateOperator(UDFSourceOperator):
+ | def produce(self):
+ | yield
+ |""".stripMargin
+ d.uiParameters = List(uiParameter("count", "7"))
+
+ d.getPhysicalOp(workflowId, executionId).opExecInitInfo match {
+ case OpExecWithCode(code, "python") =>
+ code should include("def _texera_injected_ui_parameters")
+ code should include("self.decode_python_template")
+ case other => fail(s"expected Python OpExecWithCode, got $other")
+ }
+ }
+
it should "reject a blank virtual-environment name when the default env is
disabled" in {
val d = new PythonUDFSourceOpDescV2
d.code = "yield"
@@ -104,6 +130,7 @@ class PythonUDFSourceOpDescV2Spec extends AnyFlatSpec with
Matchers {
d.defaultEnv = false
d.envName = "venv"
d.columns = List(new Attribute("a", AttributeType.STRING))
+ d.uiParameters = List(uiParameter("count", "7"))
val restored = objectMapper.readValue(objectMapper.writeValueAsString(d),
classOf[LogicalOp])
restored shouldBe a[PythonUDFSourceOpDescV2]
val p = restored.asInstanceOf[PythonUDFSourceOpDescV2]
@@ -112,5 +139,7 @@ class PythonUDFSourceOpDescV2Spec extends AnyFlatSpec with
Matchers {
p.defaultEnv shouldBe false
p.envName shouldBe "venv"
p.columns shouldBe List(new Attribute("a", AttributeType.STRING))
+ p.uiParameters.map(_.attribute) shouldBe List(new Attribute("count",
AttributeType.INTEGER))
+ p.uiParameters.map(_.value) shouldBe List("7")
}
}
diff --git
a/docs/reference/operators/user-defined-functions/python/1-out-python-udf.md
b/docs/reference/operators/user-defined-functions/python/1-out-python-udf.md
index 8c0729125b..69f68d25d8 100644
--- a/docs/reference/operators/user-defined-functions/python/1-out-python-udf.md
+++ b/docs/reference/operators/user-defined-functions/python/1-out-python-udf.md
@@ -36,12 +36,17 @@ tags: [user-defined-functions, python]
| Columns | | List<Attribute> | - | The columns of the source |
| ↳ Attribute Name | ✓ | String | - | |
| ↳ Attribute Type | ✓ | string, integer, long, double, boolean,<br>timestamp,
binary, large_binary | - | |
+| Parameters | | List<UiUDFParameter> | - | Values inferred from active
`self.UiParameter(...)` calls.<br>See [UI parameters](../#ui-parameters). |
#### Default Code Template
**Python script**
```python
+# Define UiParameter inside GenerateOperator.open().
+# Example: self.count = self.UiParameter("count", AttributeType.INT).value
+# See the Python UDF operator documentation for supported types and behavior.
+#
# from pytexera import *
# class GenerateOperator(UDFSourceOperator):
#
diff --git
a/docs/reference/operators/user-defined-functions/python/2-in-python-udf.md
b/docs/reference/operators/user-defined-functions/python/2-in-python-udf.md
index d305f5ec33..5c0c398a62 100644
--- a/docs/reference/operators/user-defined-functions/python/2-in-python-udf.md
+++ b/docs/reference/operators/user-defined-functions/python/2-in-python-udf.md
@@ -37,6 +37,7 @@ tags: [user-defined-functions, python]
| Extra output column(s) | | List<Attribute> | - | Name of the newly added
output columns that the<br>UDF will produce, if any |
| ↳ Attribute Name | ✓ | String | - | |
| ↳ Attribute Type | ✓ | string, integer, long, double, boolean,<br>timestamp,
binary, large_binary | - | |
+| Parameters | | List<UiUDFParameter> | - | Values inferred from active
`self.UiParameter(...)` calls.<br>See [UI parameters](../#ui-parameters). |
#### Default Code Template
@@ -44,6 +45,10 @@ tags: [user-defined-functions, python]
```python
# Choose from the following templates:
+#
+# Define UiParameter inside open() of ProcessTupleOperator,
ProcessBatchOperator, or ProcessTableOperator.
+# Example: self.count = self.UiParameter("count", AttributeType.INT).value
+# See the Python UDF operator documentation for supported types and behavior.
#
# from pytexera import *
#
diff --git a/docs/reference/operators/user-defined-functions/python/_index.md
b/docs/reference/operators/user-defined-functions/python/_index.md
index 2ad98d4d60..18ed8d59cd 100644
--- a/docs/reference/operators/user-defined-functions/python/_index.md
+++ b/docs/reference/operators/user-defined-functions/python/_index.md
@@ -38,3 +38,23 @@ tags: [user-defined-functions, python]
| [Python UDF](python-udf/) | User-defined function operator in Python script |
**Total**: 5 operators
+
+## UI parameters
+
+The Python UDF, 2-in Python UDF, and 1-out Python UDF operators can expose
values in the property panel. Declare each value with `self.UiParameter(...)`
inside the UDF class's `open()` method, then use its `.value` in later methods.
+
+```python
+from pytexera import *
+
+class ProcessTupleOperator(UDFOperatorV2):
+ @overrides
+ def open(self):
+ self.count = self.UiParameter("count", AttributeType.INT).value
+
+ @overrides
+ def process_tuple(self, tuple_: Tuple, port: int):
+ # self.count contains the value entered in the property panel.
+ yield tuple_
+```
+
+Active calls are inferred from the script; commented-out calls are ignored.
The supported classes are `ProcessTupleOperator`, `ProcessBatchOperator`,
`ProcessTableOperator`, and `GenerateOperator`. Supported types are `STRING`,
`INT`/`LONG`, `DOUBLE`, `BOOL`, and `TIMESTAMP`. Empty strings are valid for
`STRING`; all other types require a non-empty value before execution.
diff --git
a/docs/reference/operators/user-defined-functions/python/python-udf.md
b/docs/reference/operators/user-defined-functions/python/python-udf.md
index 93e951ee61..3f1a1e7f85 100644
--- a/docs/reference/operators/user-defined-functions/python/python-udf.md
+++ b/docs/reference/operators/user-defined-functions/python/python-udf.md
@@ -37,6 +37,7 @@ tags: [user-defined-functions, python]
| Extra output column(s) | | List<Attribute> | - | Name of the newly added
output columns that the<br>UDF will produce, if any |
| ↳ Attribute Name | ✓ | String | - | |
| ↳ Attribute Type | ✓ | string, integer, long, double, boolean,<br>timestamp,
binary, large_binary | - | |
+| Parameters | | List<UiUDFParameter> | - | Values inferred from active
`self.UiParameter(...)` calls.<br>See [UI parameters](../#ui-parameters). |
#### Default Code Template
@@ -44,6 +45,10 @@ tags: [user-defined-functions, python]
```python
# Choose from the following templates:
+#
+# Define UiParameter inside open() of ProcessTupleOperator,
ProcessBatchOperator, or ProcessTableOperator.
+# Example: self.count = self.UiParameter("count", AttributeType.INT).value
+# See the Python UDF operator documentation for supported types and behavior.
#
# from pytexera import *
#
diff --git a/frontend/LICENSE-binary b/frontend/LICENSE-binary
index 61e929c65c..a2c99bfa9d 100644
--- a/frontend/LICENSE-binary
+++ b/frontend/LICENSE-binary
@@ -291,6 +291,10 @@ Angular / npm packages:
- @codingame/[email protected]
- @codingame/[email protected]
- @ctrl/[email protected]
+ - @lezer/[email protected]
+ - @lezer/[email protected]
+ - @lezer/[email protected]
+ - @lezer/[email protected]
- @ngneat/[email protected]
- @ngx-formly/[email protected]
- @ngx-formly/[email protected]
diff --git
a/frontend/src/app/workspace/component/code-editor-dialog/code-editor.component.spec.ts
b/frontend/src/app/workspace/component/code-editor-dialog/code-editor.component.spec.ts
index c262036e87..6c9bd5af88 100644
---
a/frontend/src/app/workspace/component/code-editor-dialog/code-editor.component.spec.ts
+++
b/frontend/src/app/workspace/component/code-editor-dialog/code-editor.component.spec.ts
@@ -31,6 +31,8 @@ import { OperatorSchema } from
"../../types/operator-schema.interface";
import { of, Subject } from "rxjs";
import { AIAssistantService } from
"../../service/ai-assistant/ai-assistant.service";
import * as monaco from "monaco-editor";
+import { UiUdfParametersSyncService } from
"../../service/code-editor/ui-udf-parameters-sync.service";
+import { NotificationService } from
"../../../common/service/notification/notification.service";
// Operator types that the constructor's language-detection branch must map
// to a specific language. `RUDFSource` / `RUDF` -> `r`; the three V2 Python
@@ -91,12 +93,24 @@ const buildPredicate = (operatorID: string, operatorType:
string): OperatorPredi
describe("CodeEditorComponent", () => {
let workflowActionService: WorkflowActionService;
+ let uiParametersParseErrors: Subject<{ operatorId: string; message?: string
}>;
+ let notificationServiceMock: { error: ReturnType<typeof vi.fn> };
beforeEach(async () => {
+ uiParametersParseErrors = new Subject();
+ notificationServiceMock = { error: vi.fn() };
await TestBed.configureTestingModule({
providers: [
WorkflowActionService,
{ provide: OperatorMetadataService, useClass:
AugmentedStubMetadataService },
+ {
+ provide: UiUdfParametersSyncService,
+ useValue: {
+ attachToYCode: vi.fn(() => () => undefined),
+ uiParametersParseError$: uiParametersParseErrors.asObservable(),
+ },
+ },
+ { provide: NotificationService, useValue: notificationServiceMock },
...commonTestProviders,
],
imports: [CodeEditorComponent, HttpClientTestingModule],
@@ -119,6 +133,30 @@ describe("CodeEditorComponent", () => {
expect(fixture.componentInstance.currentOperatorId).toBe(mockJavaUDFPredicate.operatorID);
});
+ it("shows UI parameter parser errors for the edited operator", () => {
+ const predicate = buildPredicate("python-with-invalid-parameters",
"PythonUDFV2");
+ makeFixture(predicate);
+
+ uiParametersParseErrors.next({
+ operatorId: predicate.operatorID,
+ message: "UiParameter name 'count' is declared more than once.",
+ });
+
+ expect(notificationServiceMock.error).toHaveBeenCalledWith(
+ "Could not update UDF parameters: UiParameter name 'count' is declared
more than once."
+ );
+ });
+
+ it("ignores UI parameter parser events for other operators and successful
parses", () => {
+ const predicate = buildPredicate("python-with-valid-parameters",
"PythonUDFV2");
+ makeFixture(predicate);
+
+ uiParametersParseErrors.next({ operatorId: "another-operator", message:
"invalid" });
+ uiParametersParseErrors.next({ operatorId: predicate.operatorID });
+
+ expect(notificationServiceMock.error).not.toHaveBeenCalled();
+ });
+
// Language detection — the constructor maps `RUDFSource` / `RUDF` to `r`,
// the three V2-era Python operator types to `python`, and anything else
// to `java`. The exact branch lives in the constructor; the public
diff --git
a/frontend/src/app/workspace/component/code-editor-dialog/code-editor.component.ts
b/frontend/src/app/workspace/component/code-editor-dialog/code-editor.component.ts
index 5dd1a6791f..d68128312e 100644
---
a/frontend/src/app/workspace/component/code-editor-dialog/code-editor.component.ts
+++
b/frontend/src/app/workspace/component/code-editor-dialog/code-editor.component.ts
@@ -60,6 +60,8 @@ import { NzButtonComponent } from "ng-zorro-antd/button";
import { ɵNzTransitionPatchDirective } from
"ng-zorro-antd/core/transition-patch";
import { NzIconDirective } from "ng-zorro-antd/icon";
import { NgFor, NgComponentOutlet, NgIf } from "@angular/common";
+import { UiUdfParametersSyncService } from
"../../service/code-editor/ui-udf-parameters-sync.service";
+import { NotificationService } from
"../../../common/service/notification/notification.service";
type MonacoEditor = monaco.editor.IStandaloneCodeEditor;
@@ -110,6 +112,7 @@ export class CodeEditorComponent implements AfterViewInit,
SafeStyle, OnDestroy
private editorApp?: EditorApp;
private languageClientWrapper?: LanguageClientWrapper;
private monacoBinding?: MonacoBinding;
+ private detachYCodeListener?: () => void;
// Boolean to determine whether the suggestion UI should be shown
public showAnnotationSuggestion: boolean = false;
@@ -144,7 +147,9 @@ export class CodeEditorComponent implements AfterViewInit,
SafeStyle, OnDestroy
private workflowVersionService: WorkflowVersionService,
public coeditorPresenceService: CoeditorPresenceService,
private aiAssistantService: AIAssistantService,
- private config: GuiConfigService
+ private config: GuiConfigService,
+ private uiUdfParametersSyncService: UiUdfParametersSyncService,
+ private notificationService: NotificationService
) {
this.currentOperatorId =
this.workflowActionService.getJointGraphWrapper().getCurrentHighlightedOperatorIDs()[0];
const operatorType =
this.workflowActionService.getTexeraGraph().getOperator(this.currentOperatorId).operatorType;
@@ -171,6 +176,14 @@ export class CodeEditorComponent implements AfterViewInit,
SafeStyle, OnDestroy
const style = localStorage.getItem(this.currentOperatorId);
if (style) this.containerElement.nativeElement.style.cssText = style;
+ this.uiUdfParametersSyncService.uiParametersParseError$
+ .pipe(untilDestroyed(this))
+ .subscribe(({ operatorId, message }) => {
+ if (operatorId === this.currentOperatorId && message) {
+ this.notificationService.error(`Could not update UDF parameters:
${message}`);
+ }
+ });
+
// start editor
this.workflowVersionService
.getDisplayParticularVersionStream()
@@ -196,6 +209,10 @@ export class CodeEditorComponent implements AfterViewInit,
SafeStyle, OnDestroy
this.workflowVersionStreamSubject.next();
this.workflowVersionStreamSubject.complete();
+
+ if (this.detachYCodeListener) {
+ this.detachYCodeListener();
+ }
}
/**
@@ -403,6 +420,13 @@ export class CodeEditorComponent implements AfterViewInit,
SafeStyle, OnDestroy
}
this.setupAIAssistantActions(editor);
this.initCodeDebuggerComponent(editor);
+ if (this.detachYCodeListener) {
+ this.detachYCodeListener();
+ }
+
+ if (this.code) {
+ this.detachYCodeListener =
this.uiUdfParametersSyncService.attachToYCode(this.currentOperatorId,
this.code);
+ }
});
}
diff --git
a/frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.scss
b/frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.scss
index 740ac59204..92a8e9080d 100644
---
a/frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.scss
+++
b/frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.scss
@@ -212,3 +212,14 @@
padding-top: 8px;
border-top: 1px solid #f0f0f0;
}
+
+/* Style only the UDF Parameters field using the widget's stable class. */
+:host ::ng-deep nz-form-item:has(.ui-udf-parameters-field) {
+ border-top: 1px solid #d9d9d9;
+ padding-top: 12px;
+ margin-top: 8px;
+
+ label {
+ font-weight: 600;
+ }
+}
diff --git
a/frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.spec.ts
b/frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.spec.ts
index aba2811444..a127d0bc6d 100644
---
a/frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.spec.ts
+++
b/frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.spec.ts
@@ -60,6 +60,7 @@ import { commonTestProviders } from
"../../../../common/testing/test-utils";
import { DynamicSchemaService } from
"../../../service/dynamic-schema/dynamic-schema.service";
import { NotificationService } from
"../../../../common/service/notification/notification.service";
import { WorkflowGraph } from
"../../../service/workflow-graph/model/workflow-graph";
+import { UiUdfParametersSyncService } from
"../../../service/code-editor/ui-udf-parameters-sync.service";
const { marbles } = configure({ run: false });
@@ -208,6 +209,37 @@ describe("OperatorPropertyEditFrameComponent", () => {
expect(emitEventCounter).toEqual(1);
}));
+ it("keeps code-inferred UI parameters in the form model and subsequent form
edits", fakeAsync(() => {
+ const predicate = {
+ ...mockScanPredicate,
+ operatorProperties: { tableName: "before", uiParameters: [] },
+ };
+ workflowActionService.addOperator(predicate, mockPoint);
+ component.ngOnChanges({
+ currentOperatorId: new SimpleChange(undefined, predicate.operatorID,
true),
+ });
+ fixture.detectChanges();
+ tick(COLLAB_DEBOUNCE_TIME_MS);
+
+ const inferredParameters = [{ attribute: { attributeName: "count",
attributeType: "integer" }, value: "" }];
+ const syncService = TestBed.inject(UiUdfParametersSyncService);
+ (syncService as any).uiParametersChangedSubject.next({
+ operatorId: predicate.operatorID,
+ parameters: inferredParameters,
+ });
+
+ expect(component.formData.uiParameters).toEqual(inferredParameters);
+
+ component.onFormChanges({ ...component.formData, tableName: "after" });
+ tick(FORM_DEBOUNCE_TIME_MS + 10);
+
+
expect(workflowActionService.getTexeraGraph().getOperator(predicate.operatorID).operatorProperties).toEqual({
+ tableName: "after",
+ uiParameters: inferredParameters,
+ });
+ discardPeriodicTasks();
+ }));
+
it.skip(
"should debounce the user form input to avoid emitting event too
frequently",
marbles(m => {
@@ -1698,6 +1730,14 @@ describe("OperatorPropertyEditFrameComponent", () => {
expect(getField("datasetVersionPath")?.type).toBe("datasetversionselector");
});
+ it("maps uiParameters to the ui-udf-parameters field type", () => {
+ component.setFormlyFormBinding({
+ type: "object",
+ properties: { uiParameters: { type: "array" } },
+ });
+ expect(getField("uiParameters")?.type).toBe("ui-udf-parameters");
+ });
+
it("maps a field described as 'Input your code here' to the codearea field
type", () => {
component.setFormlyFormBinding({
type: "object",
diff --git
a/frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.ts
b/frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.ts
index f4b356c474..beedbabd90 100644
---
a/frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.ts
+++
b/frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.ts
@@ -74,6 +74,7 @@ import { WorkflowPveService } from
"../../../service/virtual-environment/virtual
import { ComputingUnitStatusService } from
"../../../../common/service/computing-unit/computing-unit-status/computing-unit-status.service";
import { of } from "rxjs";
import { map, switchMap, take } from "rxjs/operators";
+import { UiUdfParametersSyncService } from
"../../../service/code-editor/ui-udf-parameters-sync.service";
Quill.register("modules/cursors", QuillCursors);
@@ -460,7 +461,8 @@ export class OperatorPropertyEditFrameComponent implements
OnInit, OnChanges, On
private workflowStatusSerivce: WorkflowStatusService,
private config: GuiConfigService,
private workflowPveService: WorkflowPveService,
- private computingUnitStatusService: ComputingUnitStatusService
+ private computingUnitStatusService: ComputingUnitStatusService,
+ private uiUdfParametersSyncService: UiUdfParametersSyncService
) {}
private patchPythonUdfEnvironmentSchema(schema: CustomJSONSchema7,
environments: string[]): CustomJSONSchema7 {
@@ -516,6 +518,25 @@ export class OperatorPropertyEditFrameComponent implements
OnInit, OnChanges, On
this.currentOperatorStatus = update[this.currentOperatorId];
}
});
+
+ this.uiUdfParametersSyncService.uiParametersChanged$
+ .pipe(untilDestroyed(this))
+ .subscribe(({ operatorId, parameters }) => {
+ if (operatorId !== this.currentOperatorId) return;
+
+ const currentOperator =
this.workflowActionService.getTexeraGraph().getOperator(operatorId);
+
+ const newModel = {
+ ...cloneDeep(currentOperator.operatorProperties),
+ uiParameters: cloneDeep(parameters),
+ };
+
+ this.listeningToChange = false;
+ this.formData = cloneDeep(newModel);
+ this.workflowActionService.setOperatorProperty(operatorId, newModel);
+ this.listeningToChange = true;
+ this.changeDetectorRef.detectChanges();
+ });
}
private isHuggingFaceOperator(): boolean {
@@ -1038,6 +1059,10 @@ export class OperatorPropertyEditFrameComponent
implements OnInit, OnChanges, On
}
}
+ if (mappedField.key === "uiParameters") {
+ mappedField.type = "ui-udf-parameters";
+ }
+
if (mappedField.key === "datasetVersionPath") {
mappedField.type = "datasetversionselector";
}
diff --git
a/frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.html
b/frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.html
index c56391ab47..26ea63b827 100644
---
a/frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.html
+++
b/frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.html
@@ -16,28 +16,30 @@
specific language governing permissions and limitations
under the License.
-->
-<div
- class="ui-udf-parameter-list"
- *ngIf="model?.length">
- <div class="ui-udf-parameter-row header">
- <div
- class="field-cell"
- *ngFor="let column of fieldColumns">
- <span class="col-title">{{ column.label }}</span>
- </div>
- </div>
-
+<div class="ui-udf-parameters-field">
<div
- class="ui-udf-parameter-row"
- *ngFor="let parameter of (model || []); let i = index; trackBy:
trackByParameterName">
- <ng-container *ngIf="field.fieldGroup?.[i] as rowField">
+ class="ui-udf-parameter-list"
+ *ngIf="model?.length">
+ <div class="ui-udf-parameter-row header">
<div
class="field-cell"
*ngFor="let column of fieldColumns">
- <formly-field
- *ngIf="getColumnField(rowField, column) as columnField"
- [field]="columnField"></formly-field>
+ <span class="col-title">{{ column.label }}</span>
</div>
- </ng-container>
+ </div>
+
+ <div
+ class="ui-udf-parameter-row"
+ *ngFor="let parameter of (model || []); let i = index; trackBy:
trackByParameterName">
+ <ng-container *ngIf="field.fieldGroup?.[i] as rowField">
+ <div
+ class="field-cell"
+ *ngFor="let column of fieldColumns">
+ <formly-field
+ *ngIf="getColumnField(rowField, column) as columnField"
+ [field]="columnField"></formly-field>
+ </div>
+ </ng-container>
+ </div>
</div>
</div>
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 d755978ca6..3d3613fa5a 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
@@ -90,6 +90,15 @@ describe("UiUdfParametersSyncService", () => {
expect(parametersChangedObserver).not.toHaveBeenCalled();
});
+ it("should not replay a previous parameter change to a late subscriber", ()
=> {
+ parserServiceMock.parse.mockReturnValue([parameter("count", "integer")]);
+
+ service.syncStructureFromCode(operatorId, code);
+
+ const lateObserver = observeParameterChanges();
+ expect(lateObserver).not.toHaveBeenCalled();
+ });
+
it("should emit parser errors without replacing the current parameters", ()
=> {
operator.operatorProperties.uiParameters = [parameter("count", "integer",
"42")];
parserServiceMock.parse.mockImplementation(() => {
@@ -109,6 +118,18 @@ describe("UiUdfParametersSyncService", () => {
});
});
+ it("should not replay a previous parser error to a late subscriber", () => {
+ parserServiceMock.parse.mockImplementation(() => {
+ throw new UiUdfParametersParseError("invalid parameters");
+ });
+
+ service.syncStructureFromCode(operatorId, code);
+
+ const lateObserver = vitest.fn();
+ service.uiParametersParseError$.subscribe(lateObserver);
+ expect(lateObserver).not.toHaveBeenCalled();
+ });
+
it("should not parse code for non-Python UDF operators", () => {
operator.operatorType = "Projection";
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 519d54aaa9..bb3c832420 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
@@ -18,7 +18,7 @@
*/
import { Injectable } from "@angular/core";
import { isEqual } from "lodash-es";
-import { ReplaySubject, Subject } from "rxjs";
+import { Subject } from "rxjs";
import { debounceTime } from "rxjs/operators";
import { WorkflowActionService } from
"../workflow-graph/model/workflow-action.service";
import { UiUdfParametersParseError, UiUdfParametersParserService } from
"./ui-udf-parameters-parser.service";
@@ -38,10 +38,8 @@ const UI_PARAMETER_SYNC_DEBOUNCE_TIME_MS = 200;
/** Keeps Python UDF UI parameter structure in sync with the code editor and
workflow graph. */
@Injectable({ providedIn: "root" })
export class UiUdfParametersSyncService {
- private readonly uiParametersChangedSubject = new ReplaySubject<{
operatorId: string; parameters: UiUdfParameter[] }>(
- 1
- );
- private readonly uiParametersParseErrorSubject = new ReplaySubject<{
operatorId: string; message?: string }>(1);
+ private readonly uiParametersChangedSubject = new Subject<{ operatorId:
string; parameters: UiUdfParameter[] }>();
+ private readonly uiParametersParseErrorSubject = new Subject<{ operatorId:
string; message?: string }>();
/** Emits when parsed UI parameter structure changes; consumers should write
the parameters back to operatorProperties. */
readonly uiParametersChanged$ =
this.uiParametersChangedSubject.asObservable();
