Copilot commented on code in PR #4268:
URL: https://github.com/apache/texera/pull/4268#discussion_r3025288848
##########
amber/src/main/python/core/models/schema/attribute_type.py:
##########
@@ -78,6 +78,17 @@ class AttributeType(Enum):
}
+FROM_STRING_PARSER_MAPPING = {
+ AttributeType.STRING: str,
+ AttributeType.INT: int,
+ AttributeType.LONG: int,
+ AttributeType.DOUBLE: float,
+ AttributeType.BOOL: lambda v: str(v).strip().lower() in ("true", "1",
"yes"),
+ AttributeType.BINARY: lambda v: v if isinstance(v, bytes) else
str(v).encode(),
+ AttributeType.TIMESTAMP: lambda v: datetime.datetime.fromisoformat(v),
+ AttributeType.LARGE_BINARY: largebinary,
+}
Review Comment:
`AttributeType.TIMESTAMP` parsing uses `datetime.datetime.fromisoformat(v)`,
which raises `ValueError` for common ISO-8601 inputs with a trailing `Z` (e.g.,
`2024-01-01T00:00:00Z`, which is produced by many Java/Scala serializers). This
will break UiParameter timestamp values coming from the UI/backend. Update the
parser to handle `Z`/offset forms (e.g., normalize `Z` to `+00:00`) and add a
test case for the `Z` form.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterInjector.scala:
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.texera.amber.operator.udf.python
+
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder
+import
org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext
+
+import scala.util.matching.Regex
+
+object PythonUdfUiParameterInjector {
+
+ private val ReservedHookMethod = "_texera_injected_ui_parameters"
+
+ // Match user-facing UDF classes (the ones users write)
+ private val SupportedUserClassRegex: Regex =
+ """(?m)^([
\t]*)class\s+(ProcessTupleOperator|ProcessBatchOperator|ProcessTableOperator|GenerateOperator)\s*\([^)]*\)\s*:\s*(?:#.*)?$""".r
+
+ private def validate(uiParameters: List[UiUDFParameter]): Unit = {
+ uiParameters.foreach { parameter =>
+ if (parameter.attribute == null) {
+ throw new RuntimeException("UiParameter attribute is required.")
+ }
+ }
+
+ val grouped = uiParameters.groupBy(_.attribute.getName)
+ grouped.foreach {
+ case (key, values) =>
+ val typeSet = values.map(_.attribute.getType).toSet
+ if (typeSet.size > 1) {
+ throw new RuntimeException(
+ s"UiParameter key '$key' has multiple types:
${typeSet.map(_.name()).mkString(",")}."
+ )
+ }
+ }
+ }
+
+ private def buildInjectedParametersMap(
+ uiParameters: List[UiUDFParameter]
+ ): PythonTemplateBuilder = {
+ val entries = uiParameters.map { parameter =>
+ pyb"'${parameter.attribute.getName()}': ${parameter.value}"
+ }
+
+ entries.reduceOption((acc, entry) => acc + pyb", " +
entry).getOrElse(pyb"")
+ }
+
+ private def buildInjectedHookMethod(uiParameters: List[UiUDFParameter]):
String = {
+ val injectedParametersMap = buildInjectedParametersMap(uiParameters)
+
+ // unindented method; we indent it when inserting into the class body
+ (pyb"""|@overrides
+ |def """ + pyb"$ReservedHookMethod" + pyb"""(self) ->
typing.Dict[str, typing.Any]:
+ | return {""" +
+ injectedParametersMap +
+ pyb"""}
+ |""").encode
+ }
+
+ private def ensureTypingImport(encodedUserCode: String): String = {
+ val alreadyImported = encodedUserCode
+ .split("\n", -1)
+ .exists(_.trim == "import typing")
+
+ if (alreadyImported) {
+ encodedUserCode
+ } else {
+ "import typing\n" + encodedUserCode
+ }
Review Comment:
`ensureTypingImport` prepends `import typing` to the very top of the user
script. This can break valid Python that requires header ordering (notably
`from __future__ import ...` must appear before any other imports, and a
shebang/encoding cookie may need to remain first). Consider inserting `import
typing` after any shebang/encoding lines and any `__future__` imports instead
of always prefixing.
```suggestion
val lines = encodedUserCode.split("\n", -1)
// If 'import typing' is already present as a standalone line, leave the
code unchanged.
val alreadyImported = lines.exists(_.trim == "import typing")
if (alreadyImported) {
return encodedUserCode
}
// Determine where to insert 'import typing' while respecting Python
header ordering:
// - shebang line (#!...)
// - encoding cookie (e.g., # -*- coding: utf-8 -*- or # coding=utf-8)
// - 'from __future__ import ...' imports
var insertAt = 0
var i = 0
// Skip shebang line if present.
if (lines.nonEmpty && lines(0).startsWith("#!")) {
insertAt = 1
i = 1
}
// Regex for encoding declaration comments, case-insensitive on "coding".
val encodingRegex: Regex = """(?i)^#.*coding[:=]\s*[-\w.]+""".r
// Advance past encoding cookies and __future__ imports.
while (i < lines.length) {
val trimmed = lines(i).trim
trimmed match {
case encodingRegex() =>
insertAt = i + 1
i += 1
case t if t.startsWith("from __future__ import ") =>
insertAt = i + 1
i += 1
case _ =>
// Stop at the first line that is not an encoding cookie or
__future__ import.
i = lines.length
}
}
val newLines =
if (insertAt >= lines.length) {
// Append at the end (unlikely, but safe).
lines.toList :+ "import typing"
} else {
val (before, after) = lines.splitAt(insertAt)
before.toList ::: "import typing" :: after.toList
}
newLines.mkString("\n")
```
##########
frontend/src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.ts:
##########
@@ -193,6 +194,23 @@ 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.workflowActionService.setOperatorProperty(operatorId, newModel);
+ this.listeningToChange = true;
Review Comment:
In the `uiParametersChanged$` subscription, `listeningToChange` is set to
`false` around `setOperatorProperty`. Since operator property updates are
emitted synchronously via Yjs transactions, this prevents
`registerOperatorPropertyChangeHandler()` from updating `formData`, so the
property panel may not reflect the newly inferred `uiParameters`. Consider
either not disabling `listeningToChange` here, or explicitly updating
`this.formData`/rerendering after `setOperatorProperty`.
```suggestion
this.workflowActionService.setOperatorProperty(operatorId, newModel);
```
##########
frontend/src/app/workspace/service/code-editor/ui-udf-parameters-sync.service.ts:
##########
@@ -0,0 +1,130 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { Injectable } from "@angular/core";
+import { isEqual } from "lodash-es";
+import { ReplaySubject } from "rxjs";
+import { WorkflowActionService } from
"../workflow-graph/model/workflow-action.service";
+import { UiUdfParameter, UiUdfParametersParserService } from
"./ui-udf-parameters-parser.service";
+import { isDefined } from "../../../common/util/predicate";
+import {
+ DUAL_INPUT_PORTS_PYTHON_UDF_V2_OP_TYPE,
+ PYTHON_UDF_SOURCE_V2_OP_TYPE,
+ PYTHON_UDF_V2_OP_TYPE,
+} from "../workflow-graph/model/workflow-graph";
+import { YType } from "../../types/shared-editing.interface";
+import { YText } from "yjs/dist/src/types/YText";
+
+@Injectable({ providedIn: "root" })
+export class UiUdfParametersSyncService {
+ private readonly uiParametersChangedSubject = new ReplaySubject<{
operatorId: string; parameters: UiUdfParameter[] }>(
+ 1
+ );
+
+ readonly uiParametersChanged$ =
this.uiParametersChangedSubject.asObservable();
+
+ constructor(
+ private workflowActionService: WorkflowActionService,
+ private uiUdfParametersParserService: UiUdfParametersParserService
+ ) {}
+
+ /**
+ * Attach directly to YText and sync whenever it changes
+ */
+ attachToYCode(operatorId: string, yCode: YText): () => void {
+ const handler = () => {
+ const latestCode = yCode.toString();
+ this.syncStructureFromCode(operatorId, latestCode);
+ };
+
+ yCode.observe(handler);
+
+ handler();
+
+ // return cleanup function
+ return () => yCode.unobserve(handler);
+ }
+
+ syncStructureFromCode(operatorId: string, codeFromEditor?: string): void {
+ const operator =
this.workflowActionService.getTexeraGraph().getOperator(operatorId);
+
+ if (!operator || !this.isSupportedPythonUdfType(operator.operatorType)) {
+ return;
+ }
+
+ const code = codeFromEditor ?? this.getSharedCode(operatorId);
+ if (!isDefined(code)) {
+ return;
+ }
+
+ const existingParameters = operator.operatorProperties?.uiParameters ?? [];
+ const mergedUiParameters = this.buildParsedShapeWithPreservedValues(code,
existingParameters);
+
+ if (isEqual(existingParameters, mergedUiParameters)) {
+ return;
+ }
+
+ // Emit event so UI updates
+ this.uiParametersChangedSubject.next({
+ operatorId,
+ parameters: mergedUiParameters,
+ });
+
+ // optionally persist here if desired
+ // this.workflowActionService.setOperatorProperty(...)
+ }
+
+ private buildParsedShapeWithPreservedValues(code: string,
existingParameters: any[]): UiUdfParameter[] {
+ const parsedParameters = this.uiUdfParametersParserService.parse(code);
+
+ const existingValues = new Map<string, string>();
+ existingParameters.forEach((parameter: any) => {
+ const parameterName = parameter?.attribute?.attributeName ??
parameter?.attribute?.name;
+
+ if (isDefined(parameterName) && isDefined(parameter?.value)) {
+ existingValues.set(parameterName, parameter.value);
+ }
+ });
+
+ return parsedParameters.map(parameter => ({
+ ...parameter,
+ value: existingValues.get(parameter.attribute.attributeName) ?? "",
+ }));
+ }
Review Comment:
This new sync service drives core UX (auto-infer parameter shape from code
and preserve values), but it currently has no unit tests. Since this directory
already uses Angular/Jasmine tests (e.g.,
`ui-udf-parameters-parser.service.spec.ts`), consider adding tests for
`syncStructureFromCode` / `buildParsedShapeWithPreservedValues` to cover
preserving values, dropping removed params, and emitting only when the merged
model changes.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterInjector.scala:
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.texera.amber.operator.udf.python
+
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder
+import
org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext
+
+import scala.util.matching.Regex
+
+object PythonUdfUiParameterInjector {
+
+ private val ReservedHookMethod = "_texera_injected_ui_parameters"
+
+ // Match user-facing UDF classes (the ones users write)
+ private val SupportedUserClassRegex: Regex =
+ """(?m)^([
\t]*)class\s+(ProcessTupleOperator|ProcessBatchOperator|ProcessTableOperator|GenerateOperator)\s*\([^)]*\)\s*:\s*(?:#.*)?$""".r
+
+ private def validate(uiParameters: List[UiUDFParameter]): Unit = {
+ uiParameters.foreach { parameter =>
+ if (parameter.attribute == null) {
+ throw new RuntimeException("UiParameter attribute is required.")
+ }
+ }
+
+ val grouped = uiParameters.groupBy(_.attribute.getName)
+ grouped.foreach {
+ case (key, values) =>
+ val typeSet = values.map(_.attribute.getType).toSet
+ if (typeSet.size > 1) {
+ throw new RuntimeException(
+ s"UiParameter key '$key' has multiple types:
${typeSet.map(_.name()).mkString(",")}."
+ )
+ }
+ }
+ }
+
+ private def buildInjectedParametersMap(
+ uiParameters: List[UiUDFParameter]
+ ): PythonTemplateBuilder = {
+ val entries = uiParameters.map { parameter =>
+ pyb"'${parameter.attribute.getName()}': ${parameter.value}"
+ }
+
+ entries.reduceOption((acc, entry) => acc + pyb", " +
entry).getOrElse(pyb"")
+ }
+
+ private def buildInjectedHookMethod(uiParameters: List[UiUDFParameter]):
String = {
+ val injectedParametersMap = buildInjectedParametersMap(uiParameters)
+
+ // unindented method; we indent it when inserting into the class body
+ (pyb"""|@overrides
+ |def """ + pyb"$ReservedHookMethod" + pyb"""(self) ->
typing.Dict[str, typing.Any]:
+ | return {""" +
+ injectedParametersMap +
+ pyb"""}
+ |""").encode
Review Comment:
The injected hook method is decorated with `@overrides`, but the injector
only ensures `import typing` is present. If a user script does not import
`overrides` (e.g., uses `import pytexera` instead of `from pytexera import *`),
the injected code will fail at import time with `NameError: name 'overrides' is
not defined`. Consider removing the decorator from the injected method or
injecting an explicit `from overrides import overrides` / `import overrides`
consistently.
```suggestion
(pyb"""|def """ + pyb"$ReservedHookMethod" + pyb"""(self) ->
typing.Dict[str, typing.Any]:
| return {""" +
injectedParametersMap +
pyb"""}
|""").encode
```
##########
frontend/src/app/workspace/component/ui-udf-parameters/ui-udf-parameters.component.ts:
##########
@@ -0,0 +1,91 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+import { Component } from "@angular/core";
+import { FieldArrayType, FormlyFieldConfig } from "@ngx-formly/core";
+
+@Component({
+ selector: "texera-ui-udf-parameters",
+ templateUrl: "./ui-udf-parameters.component.html",
+ styleUrls: ["./ui-udf-parameters.component.scss"],
+})
+export class UiUdfParametersComponent extends FieldArrayType {
+ private getField(rowField: FormlyFieldConfig, key: string):
FormlyFieldConfig | undefined {
+ return rowField.fieldGroup?.find(f => f.key === key);
+ }
+
+ private getAttributeChild(rowField: FormlyFieldConfig, childKey: string):
FormlyFieldConfig | undefined {
+ const attributeGroup = this.getField(rowField, "attribute");
+ return attributeGroup?.fieldGroup?.find(f => f.key === childKey);
+ }
+
+ private setDisabled(field: FormlyFieldConfig | undefined, disabled:
boolean): FormlyFieldConfig | undefined {
+ if (!field) return undefined;
+
+ // 1) Modern Formly
+ field.props = { ...(field.props ?? {}), disabled };
+
+ // 2) Compatibility for templates/wrappers still using templateOptions
+ // (use `as any` so you don't get nagged by the @deprecated JSDoc)
+ (field as any).templateOptions = { ...((field as any).templateOptions ??
{}), disabled };
+
+ // 3) Enforce at the reactive form level
+ if (field.formControl) {
+ if (disabled) {
+ field.formControl.disable({ emitEvent: false });
+ } else {
+ field.formControl.enable({ emitEvent: false });
+ }
+ } else {
+ // If control isn't created yet, disable it at init time.
+ const prevOnInit = field.hooks?.onInit;
+ field.hooks = {
+ ...(field.hooks ?? {}),
+ onInit: f => {
+ prevOnInit?.(f);
+ if (disabled) {
+ f.formControl?.disable({ emitEvent: false });
+ } else {
+ f.formControl?.enable({ emitEvent: false });
+ }
+ },
+ };
Review Comment:
`setDisabled()` mutates `field.hooks.onInit` by wrapping it every time the
template calls `getNameField/getTypeField/getValueField`. Since these getters
run during change detection, this can repeatedly nest `onInit` wrappers (and
reapply disable/enable logic multiple times). Consider setting disabled state
once (e.g., in the component’s `ngOnInit`/`prePopulate`, or by configuring the
Formly field definitions up-front) rather than mutating fields on every render.
```suggestion
// 3) Enforce at the reactive form level when the control already exists.
if (field.formControl) {
if (disabled) {
field.formControl.disable({ emitEvent: false });
} else {
field.formControl.enable({ emitEvent: false });
}
```
##########
common/workflow-core/src/main/scala/org/apache/texera/amber/core/tuple/Attribute.java:
##########
@@ -21,6 +21,10 @@
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.texera.amber.pybuilder.EncodableStringAnnotation;
+import org.apache.texera.amber.pybuilder.PyStringTypes;
+import org.apache.texera.amber.pybuilder.PyStringTypes.EncodableStringFactory$;
Review Comment:
The new imports `PyStringTypes` and `PyStringTypes.EncodableStringFactory$`
are unused in this file. In Java, unused imports cause a compilation error, so
the build will fail unless these are removed (only `EncodableStringAnnotation`
is referenced below).
```suggestion
```
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/udf/python/PythonUdfUiParameterInjector.scala:
##########
@@ -0,0 +1,189 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.texera.amber.operator.udf.python
+
+import org.apache.texera.amber.pybuilder.PythonTemplateBuilder
+import
org.apache.texera.amber.pybuilder.PythonTemplateBuilder.PythonTemplateBuilderStringContext
+
+import scala.util.matching.Regex
+
+object PythonUdfUiParameterInjector {
+
+ private val ReservedHookMethod = "_texera_injected_ui_parameters"
+
+ // Match user-facing UDF classes (the ones users write)
+ private val SupportedUserClassRegex: Regex =
+ """(?m)^([
\t]*)class\s+(ProcessTupleOperator|ProcessBatchOperator|ProcessTableOperator|GenerateOperator)\s*\([^)]*\)\s*:\s*(?:#.*)?$""".r
+
+ private def validate(uiParameters: List[UiUDFParameter]): Unit = {
+ uiParameters.foreach { parameter =>
+ if (parameter.attribute == null) {
+ throw new RuntimeException("UiParameter attribute is required.")
+ }
+ }
+
+ val grouped = uiParameters.groupBy(_.attribute.getName)
+ grouped.foreach {
+ case (key, values) =>
+ val typeSet = values.map(_.attribute.getType).toSet
+ if (typeSet.size > 1) {
+ throw new RuntimeException(
+ s"UiParameter key '$key' has multiple types:
${typeSet.map(_.name()).mkString(",")}."
+ )
+ }
+ }
+ }
+
+ private def buildInjectedParametersMap(
+ uiParameters: List[UiUDFParameter]
+ ): PythonTemplateBuilder = {
+ val entries = uiParameters.map { parameter =>
+ pyb"'${parameter.attribute.getName()}': ${parameter.value}"
Review Comment:
`buildInjectedParametersMap` splices `parameter.attribute.getName()` into a
single-quoted Python string literal via Scala interpolation. This bypasses the
pybuilder encoding/escaping (and defeats the new `@EncodableStringAnnotation`
on `Attribute.getName()`), so attribute names containing quotes/backslashes can
generate invalid Python. Prefer rendering the key as a Python expression (e.g.,
via `pyb"${parameter.attribute.getName()}: ..."`) or otherwise ensure proper
escaping rather than manual `'${...}'` quoting.
```suggestion
pyb"${parameter.attribute.getName()}: ${parameter.value}"
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
