This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5912-8aec22903261ddc69d43fe60def80e9b08547aa1 in repository https://gitbox.apache.org/repos/asf/texera.git
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();
