Copilot commented on code in PR #8327:
URL: https://github.com/apache/texera/pull/8327#discussion_r3936147015
##########
common/pybuilder/src/main/scala/org/apache/texera/amber/pybuilder/PythonTemplateBuilder.scala:
##########
@@ -209,6 +209,29 @@ object PythonTemplateBuilder {
def wrapWithPythonDecoderExpr(text: String): String =
s"self.decode_python_template('$text')"
+ /**
+ * Render `text` as a Python double-quoted string literal, quotes included.
+ *
+ * For generators that emit standalone Python source rather than an operator
+ * for the runtime: they cannot use the decode expression (it needs the
+ * operator's `decode_python_template`, and it is deliberately rejected
inside
+ * quotes), so they need the value as a *literal*. Writing `"$value"` by
hand
+ * instead lets any quote, backslash or newline in the value close the
literal
+ * early and change — or break — the emitted program.
+ *
+ * Escapes exactly what can end a double-quoted single-line literal.
+ */
+ def pyStringLiteral(text: String): String = {
+ val escaped = Option(text)
+ .getOrElse("")
+ .replace("\\", "\\\\")
+ .replace("\"", "\\\"")
+ .replace("\r", "\\r")
+ .replace("\n", "\\n")
+ .replace("\t", "\\t")
Review Comment:
A JSON string may contain `\u0000`, but this renderer leaves that character
verbatim. Python rejects generated source containing a null byte (`SyntaxError:
source code string cannot contain null bytes`), so a valid column name or
predicate value can make the export unusable. Escape NUL as a Python hex escape.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala:
##########
@@ -98,4 +104,31 @@ class ProjectionOpDesc extends MapOpDesc {
outputPorts = List(OutputPort())
)
}
+
+ override def generateStandaloneCode(): String = {
+ val units = Option(attributes).getOrElse(List.empty)
+ // JVM validates non-empty at runtime via Preconditions; emit passthrough
+ // as best-effort so the standalone script still runs.
+ if (units.isEmpty) return "out1df = in1df.copy()"
Review Comment:
This silently changes an invalid Projection into a successful pass-through.
The engine explicitly rejects an empty attribute list in schema propagation and
`ProjectionOpExec`, so the exported workflow can produce data where the Texera
workflow fails. Emit an explicit failure instead of copying the input.
This issue also appears in the following locations of the same file:
- line 114
- line 129
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala:
##########
@@ -60,4 +62,40 @@ class SpecializedFilterOpDesc extends FilterOpDesc {
supportReconfiguration = true
)
}
+
+ override def generateStandaloneCode(): String = {
+ if (predicates.isEmpty) return "out1df = in1df.copy()"
Review Comment:
An empty predicate list does not mean pass-through in the engine:
`SpecializedFilterOpExec` uses `desc.predicates.exists(...)`, which is false
for every tuple. This export therefore returns every row where Texera returns
none. Emit an empty frame with the same columns instead.
This issue also appears on line 80 of the same file.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/limit/LimitOpDesc.scala:
##########
@@ -80,4 +80,8 @@ class LimitOpDesc extends LogicalOp {
}
Success(newPhysicalOp, Some(stateTransferFunc))
}
+
+ override def generateStandaloneCode(): String = {
+ s"out1df = in1df.head($limit).reset_index(drop=True)"
Review Comment:
Negative limits diverge from the engine: `LimitOpExec` emits zero rows
because `count < limit` is immediately false, while pandas `head(-1)` emits
every row except the last. Clamp the generated argument at zero (or reject
negative limits consistently).
##########
workflow-compiling-service/src/main/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslator.scala:
##########
@@ -0,0 +1,198 @@
+/*
+ * 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.translator
+
+import com.typesafe.scalalogging.LazyLogging
+import org.apache.texera.amber.core.virtualidentity.OperatorIdentity
+import org.apache.texera.common.compiler.model.LogicalPlan
+import org.apache.texera.amber.operator.StandaloneCodeGenerator
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+import scala.jdk.CollectionConverters._
+
+class WorkflowToPythonTranslator extends LazyLogging {
+
+ // Output-port-level key. An operator with N output ports gets N entries
+ // (e.g. Split has port 0 and port 1, each with its own assigned dfN var).
+ private type PortKey = (String, Int) // (opId, portIdx)
+
+ def translate(logicalPlan: LogicalPlan): String = {
+ // Track downstream connections per (opId, fromPortIdx). A port is a leaf
+ // if it has no outgoing edges — operator-level "no outgoing links" is too
+ // coarse for multi-output ops (Split's port 0 may have downstream while
+ // port 1 doesn't, or vice versa).
+ val outgoingFromPort = mutable.Map[PortKey, Int]().withDefaultValue(0)
+ logicalPlan.links.foreach { link =>
+ outgoingFromPort((link.fromOpId.id, link.fromPortId.id)) += 1
+ }
+
+ val outputVar = mutable.Map[PortKey, String]()
+ var varCounter = 1
+ val script = ArrayBuffer[String]()
+
+ script += "import pandas as pd"
+ script += "import plotly.express as px"
+ script += "import plotly.graph_objects as go"
+ script += "import plotly.io"
Review Comment:
These unconditional imports make every exported workflow require Plotly even
though all generators introduced here use only pandas. A
Distinct/Filter/Limit/Projection/Union script fails at startup with
`ModuleNotFoundError` in an otherwise sufficient pandas environment. Collect
imports from the participating generators, as is done for helpers, or omit
Plotly until an operator needs it.
##########
common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/LogicalOp.scala:
##########
@@ -459,6 +459,16 @@ abstract class LogicalOp extends PortDescriptor with
Serializable {
def operatorInfo: OperatorInfo
+ /**
+ * Whether the row ORDER of this operator's output is part of its contract.
+ * Defaults to false: the engine runs operators across parallel workers, so
+ * for almost every operator the output row order is an
implementation-defined
+ * interleaving that consumers must not rely on. Only operators whose very
+ * purpose is to establish an order (the sort family) override this to true.
+ * Consumers that must not rely on a stable row order read this flag.
+ */
+ def orderSensitive: Boolean = false
Review Comment:
This documentation says the sort family overrides the flag and consumers
read it, but repository-wide usage currently finds neither an override nor a
consumer. The flag therefore has no effect and is unrelated to this PR's export
path; it appears to be parity-test scaffolding from a later issue in #8325.
Remove it from this PR or include the behavior that makes the contract true.
##########
workflow-compiling-service/src/main/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslator.scala:
##########
@@ -0,0 +1,198 @@
+/*
+ * 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.translator
+
+import com.typesafe.scalalogging.LazyLogging
+import org.apache.texera.amber.core.virtualidentity.OperatorIdentity
+import org.apache.texera.common.compiler.model.LogicalPlan
+import org.apache.texera.amber.operator.StandaloneCodeGenerator
+
+import scala.collection.mutable
+import scala.collection.mutable.ArrayBuffer
+import scala.jdk.CollectionConverters._
+
+class WorkflowToPythonTranslator extends LazyLogging {
+
+ // Output-port-level key. An operator with N output ports gets N entries
+ // (e.g. Split has port 0 and port 1, each with its own assigned dfN var).
+ private type PortKey = (String, Int) // (opId, portIdx)
+
+ def translate(logicalPlan: LogicalPlan): String = {
+ // Track downstream connections per (opId, fromPortIdx). A port is a leaf
+ // if it has no outgoing edges — operator-level "no outgoing links" is too
+ // coarse for multi-output ops (Split's port 0 may have downstream while
+ // port 1 doesn't, or vice versa).
+ val outgoingFromPort = mutable.Map[PortKey, Int]().withDefaultValue(0)
+ logicalPlan.links.foreach { link =>
+ outgoingFromPort((link.fromOpId.id, link.fromPortId.id)) += 1
+ }
+
+ val outputVar = mutable.Map[PortKey, String]()
+ var varCounter = 1
+ val script = ArrayBuffer[String]()
+
+ script += "import pandas as pd"
+ script += "import plotly.express as px"
+ script += "import plotly.graph_objects as go"
+ script += "import plotly.io"
+ script += ""
+
+ // getTopologicalOpIds() uses jgrapht internally — no need for a custom
topo sort
+ val topoOrder = logicalPlan.getTopologicalOpIds.asScala.toList
+
+ // Helper definitions the operator bodies below refer to. Collected across
the
+ // whole plan and deduplicated by text, so a workflow holding two operators
+ // that share one helper still emits it once. Order follows the topological
+ // order, which keeps the script stable for a given plan.
+ val helpers = topoOrder
+ .map(logicalPlan.getOperator)
+ .collect { case gen: StandaloneCodeGenerator => gen.standaloneHelpers() }
+ .flatten
+ .distinct
+ if (helpers.nonEmpty) {
+ helpers.foreach { helper => script += helper; script += "" }
+ }
+
+ for (opIdentity <- topoOrder) {
+ val opId = opIdentity.id
+ val op = logicalPlan.getOperator(opIdentity)
+ val displayName = op.operatorInfo.userFriendlyName
+
+ // Resolve upstream inputs in the consuming operator's input-port order
+ // (link.toPortId), NOT the order links happen to appear in the plan's
+ // link list. This makes in1df/in2df/... deterministic and correct for
+ // multi-input operators (joins, set ops) where port 0 vs port 1 carries
+ // semantics (e.g. build vs probe side). Ties on the same toPortId keep
+ // link order — relevant for variadic single-port operators like Union.
+ // Each upstream link is resolved via (fromOpId, fromPortId) so that a
+ // multi-output upstream (Split) hands each downstream the correct DF.
+ val inVars = logicalPlan
+ .getUpstreamLinks(opIdentity)
+ .sortBy(link => (link.toPortId.id, link.toPortId.internal))
+ .map(link => outputVar((link.fromOpId.id, link.fromPortId.id)))
+
+ // Allocate one dfN per declared output port. Existing single-output
+ // operators have outputPorts.size == 1, so they get exactly one var and
+ // their behavior is identical to the previous flat scheme.
+ val outVars = op.operatorInfo.outputPorts.map { port =>
+ val v = s"df$varCounter"
+ varCounter += 1
+ outputVar((opId, port.id.id)) = v
+ v
+ }
+
+ script += s"# [$displayName]"
+
+ // Jackson deserializes each operator into its concrete subclass via
@JsonSubTypes on LogicalOp,
+ // so the pattern match below will resolve to the correct descriptor
(e.g. BarChartOpDesc).
+ op match {
+ case gen: StandaloneCodeGenerator =>
+ // generateStandaloneCode() returns a code block using in{N}df /
out{N}df
+ // placeholders; substituteVars() replaces them with the assigned
vars.
+ script += substituteVars(gen.generateStandaloneCode(), inVars,
outVars, displayName)
+
+ case _ =>
+ logger.warn(
+ s"Operator '$displayName' does not implement
StandaloneCodeGenerator. Skipping."
+ )
+ script += s"# TODO: '$displayName' is not yet supported by the
translator."
+ outVars.zipWithIndex.foreach {
+ case (v, i) => script += s"# $v = <output port $i of $displayName>"
+ }
+ }
+
+ script += ""
+ }
+
+ // Leaf detection runs at the port level: a (opId, port) pair is a leaf
+ // if no link consumes it. For Split with one downstream port and one
+ // dangling port, only the dangling port is treated as a leaf to print.
+ val leafPorts = outputVar.keys.toList
+ .sortBy { case (_, portIdx) => portIdx }
+ .filter(key => outgoingFromPort(key) == 0)
+ val dataFrameLeafPorts = leafPorts.filter {
+ case (opId, _) =>
+ logicalPlan.getOperator(OperatorIdentity(opId)) match {
+ case gen: StandaloneCodeGenerator => gen.producesDataFrame()
+ case _ => false
+ }
+ }
+
+ if (dataFrameLeafPorts.nonEmpty) {
+ script += "# --- Output ---"
+ // Print in topological order of the producing operator so multi-port
+ // operators print contiguously and the order matches the script flow.
+ val topoIndex = topoOrder.map(_.id).zipWithIndex.toMap
+ dataFrameLeafPorts
+ .sortBy { case (opId, portIdx) => (topoIndex.getOrElse(opId,
Int.MaxValue), portIdx) }
+ .foreach {
+ case (opId, portIdx) =>
+ val varName = outputVar((opId, portIdx))
+ val displayName =
+
logicalPlan.getOperator(OperatorIdentity(opId)).operatorInfo.userFriendlyName
+ val portSuffix = if (outputVar.keys.count(_._1 == opId) > 1) s"
port $portIdx" else ""
+ script += s"""print("\\n[$displayName$portSuffix] $varName:")"""
+ script += s"print($varName.head())"
Review Comment:
This prints only the first five rows of each leaf DataFrame, although the
endpoint contract says the standalone script prints its results. Any workflow
producing more than five rows therefore has truncated observable output; print
the full leaf DataFrame instead.
##########
workflow-compiling-service/src/main/scala/org/apache/texera/service/resource/WorkflowToPythonResource.scala:
##########
@@ -0,0 +1,70 @@
+/*
+ * 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.service.resource
+
+import com.fasterxml.jackson.annotation.{JsonSubTypes, JsonTypeInfo}
+import com.typesafe.scalalogging.LazyLogging
+import jakarta.annotation.security.RolesAllowed
+import jakarta.ws.rs.core.MediaType
+import jakarta.ws.rs.{Consumes, POST, Path, Produces}
+import org.apache.texera.common.compiler.model.{LogicalPlan, LogicalPlanPojo}
+import org.apache.texera.amber.translator.WorkflowToPythonTranslator
+
+@JsonTypeInfo(
+ use = JsonTypeInfo.Id.NAME,
+ include = JsonTypeInfo.As.PROPERTY,
+ property = "type"
+)
+@JsonSubTypes(
+ Array(
+ new JsonSubTypes.Type(value = classOf[WorkflowToPythonSuccess], name =
"success"),
+ new JsonSubTypes.Type(value = classOf[WorkflowToPythonFailure], name =
"failure")
+ )
+)
+sealed trait WorkflowToPythonResponse
+
+case class WorkflowToPythonSuccess(pythonCode: String) extends
WorkflowToPythonResponse
+
+case class WorkflowToPythonFailure(errorMessage: String) extends
WorkflowToPythonResponse
+
+@Consumes(Array(MediaType.APPLICATION_JSON))
+@Produces(Array(MediaType.APPLICATION_JSON))
+@RolesAllowed(Array("REGULAR", "ADMIN"))
+@Path("/workflow-to-python")
+class WorkflowToPythonResource extends LazyLogging {
+
+ private val translator = new WorkflowToPythonTranslator()
+
+ @POST
+ @Path("")
+ def convertWorkflowToPython(
+ logicalPlanPojo: LogicalPlanPojo
+ ): WorkflowToPythonResponse = {
Review Comment:
The new HTTP contract has no resource-level coverage: unlike
`WorkflowCompilationResourceSpec`, nothing posts frontend-shaped JSON and
verifies the success/failure discriminator and generated payload, and
`WorkflowCompilingServiceRunSpec` still verifies only the pre-existing
resources. Add positive and malformed-plan endpoint tests plus
registration/access-control assertions so removal or serialization regressions
cannot pass CI.
--
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]
