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-8502-4d7fd493a8f47b28cc2155bd79c946699cec89a1 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 75c85aa7e771d9b3d3c79396b08b196928238858 Author: Kary Zheng <[email protected]> AuthorDate: Fri Sep 11 06:47:02 2026 +0000 feat(workflow-operator): export the first five operators, and serve the script (#8502) ### What changes were proposed in this PR? Five operators implement `StandaloneCodeGenerator`, which is what turns the translator from a mechanism into something that produces a script that runs: Distinct, Filter, Limit, Projection and Union. Four of them read a single input; Union reads a variadic port, so it names the whole list of upstreams rather than a fixed count, which a fixed count gets wrong in both directions. The endpoint the editor calls comes with them, because what it is worth testing on is a script that runs rather than one made of placeholders. It takes a plan, compiles it first so a scan source can read the schema off the file it points at, hands the translator the output schemas that gives, and returns the script. A failed compile is logged and translation goes on without them, so a workflow whose file is not chosen yet still exports. `pyStringLiteral` comes with them too. A generator has to write a column name into the source it emits, and writing the quotes by hand lets any quote, backslash or newline in the name close the literal early and change, or break, the emitted program. #8327 is the trait and the translator, and nothing that uses them. ### Any related issues, documentation, discussions? Part of #8325, 2 of 27; that issue lists the set in order. Closes #8501, the task this change is the whole of. ### How was this PR tested? Each operator asserts the block it emits in its own spec. `PythonTemplateBuilderApiSpec` covers what `pyStringLiteral` escapes, including the NUL that Python refuses to compile anywhere in a source file. `WorkflowToPythonResourceSpec` drives the endpoint over a plan built from these operators and reads the script back. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Claude Opus 5) 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-authored-by: Claude Opus 5 (1M context) <[email protected]> --- .../amber/pybuilder/PythonTemplateBuilder.scala | 26 +++ .../pybuilder/PythonTemplateBuilderApiSpec.scala | 29 +++ .../amber/operator/distinct/DistinctOpDesc.scala | 9 +- .../operator/filter/SpecializedFilterOpDesc.scala | 43 ++++- .../texera/amber/operator/limit/LimitOpDesc.scala | 11 +- .../operator/projection/ProjectionOpDesc.scala | 36 +++- .../texera/amber/operator/union/UnionOpDesc.scala | 11 +- .../operator/distinct/DistinctOpDescSpec.scala | 8 + .../filter/SpecializedFilterOpDescSpec.scala | 18 ++ .../amber/operator/limit/LimitOpDescSpec.scala | 17 ++ .../operator/projection/ProjectionOpDescSpec.scala | 22 +++ .../amber/operator/union/UnionOpDescSpec.scala | 13 ++ .../texera/service/WorkflowCompilingService.scala | 9 +- .../resource/WorkflowToPythonResource.scala | 99 ++++++++++ .../service/WorkflowCompilingServiceRunSpec.scala | 15 +- .../resource/WorkflowToPythonResourceSpec.scala | 200 +++++++++++++++++++++ 16 files changed, 554 insertions(+), 12 deletions(-) diff --git a/common/pybuilder/src/main/scala/org/apache/texera/amber/pybuilder/PythonTemplateBuilder.scala b/common/pybuilder/src/main/scala/org/apache/texera/amber/pybuilder/PythonTemplateBuilder.scala index 73f4a3846d..008c07ee4e 100644 --- a/common/pybuilder/src/main/scala/org/apache/texera/amber/pybuilder/PythonTemplateBuilder.scala +++ b/common/pybuilder/src/main/scala/org/apache/texera/amber/pybuilder/PythonTemplateBuilder.scala @@ -209,6 +209,32 @@ 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, plus NUL: + * Python refuses to compile source that holds one anywhere, so a column name + * carrying it would break the whole script rather than only this literal. + */ + def pyStringLiteral(text: String): String = { + val escaped = Option(text) + .getOrElse("") + .replace("\\", "\\\\") + .replace("\"", "\\\"") + .replace("\r", "\\r") + .replace("\n", "\\n") + .replace("\t", "\\t") + .replace(0.toChar.toString, "\\x00") + "\"" + escaped + "\"" + } + sealed trait RenderMode extends Product with Serializable object RenderMode { case object Plain extends RenderMode diff --git a/common/pybuilder/src/test/scala/org/apache/texera/amber/pybuilder/PythonTemplateBuilderApiSpec.scala b/common/pybuilder/src/test/scala/org/apache/texera/amber/pybuilder/PythonTemplateBuilderApiSpec.scala index acbed3031f..1ec9ad749e 100644 --- a/common/pybuilder/src/test/scala/org/apache/texera/amber/pybuilder/PythonTemplateBuilderApiSpec.scala +++ b/common/pybuilder/src/test/scala/org/apache/texera/amber/pybuilder/PythonTemplateBuilderApiSpec.scala @@ -244,4 +244,33 @@ class PythonTemplateBuilderApiSpec extends AnyFunSuite { test("hasUnclosedQuote: three opening single quotes count as unclosed") { assert(PythonLexerUtils.hasUnclosedQuote("'''abc")) } + + // -------- pyStringLiteral -------- + + // Every character that can end a double-quoted single-line literal, since one that + // slips through does not fail here but changes the emitted program. + test("pyStringLiteral: quotes the value and escapes what would close the literal") { + assert(PythonTemplateBuilder.pyStringLiteral("plain") == "\"plain\"") + assert(PythonTemplateBuilder.pyStringLiteral("say \"hi\"") == "\"say \\\"hi\\\"\"") + assert(PythonTemplateBuilder.pyStringLiteral("a\\b") == "\"a\\\\b\"") + assert(PythonTemplateBuilder.pyStringLiteral("one\ntwo") == "\"one\\ntwo\"") + assert(PythonTemplateBuilder.pyStringLiteral("a\tb") == "\"a\\tb\"") + assert(PythonTemplateBuilder.pyStringLiteral("a\rb") == "\"a\\rb\"") + } + + // A column name arrives from JSON and can be absent; an empty literal is a value the + // emitted program can carry, where `null` would reach it as the four letters. + test("pyStringLiteral: renders a null as the empty literal") { + assert(PythonTemplateBuilder.pyStringLiteral(null) == "\"\"") + } + + // NUL is the one character that a literal cannot carry verbatim: Python refuses to + // compile a source file holding one, so it takes the whole script down rather than + // this value alone. + test("pyStringLiteral: escapes a NUL rather than emitting it") { + val literal = PythonTemplateBuilder.pyStringLiteral("a" + 0.toChar + "b") + assert(literal == "\"a\\x00b\"") + assert(!literal.contains(0.toChar)) + } + } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/distinct/DistinctOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/distinct/DistinctOpDesc.scala index 9e75e648bb..17b646bfa1 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/distinct/DistinctOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/distinct/DistinctOpDesc.scala @@ -22,10 +22,10 @@ package org.apache.texera.amber.operator.distinct import org.apache.texera.amber.core.executor.OpExecWithClassName import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{HashPartition, InputPort, OutputPort, PhysicalOp} -import org.apache.texera.amber.operator.LogicalOp +import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} -class DistinctOpDesc extends LogicalOp { +class DistinctOpDesc extends LogicalOp with StandaloneCodeGenerator { override def getPhysicalOp( workflowId: WorkflowIdentity, @@ -54,4 +54,9 @@ class DistinctOpDesc extends LogicalOp { outputPorts = List(OutputPort(blocking = true)) ) + override def generateStandaloneCode(): String = { + // JVM op uses LinkedHashSet to preserve first-occurrence order; + // pandas drop_duplicates does the same by default. + "out1df = in1df.drop_duplicates(ignore_index=True)" + } } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala index 9e86773df7..e5d47ee523 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDesc.scala @@ -23,10 +23,12 @@ import com.fasterxml.jackson.annotation.{JsonProperty, JsonPropertyDescription} import org.apache.texera.amber.core.executor.OpExecWithClassName import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp} +import org.apache.texera.amber.operator.StandaloneCodeGenerator import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral import org.apache.texera.amber.util.JSONUtils.objectMapper -class SpecializedFilterOpDesc extends FilterOpDesc { +class SpecializedFilterOpDesc extends FilterOpDesc with StandaloneCodeGenerator { @JsonProperty(value = "predicates", required = true) @JsonPropertyDescription("multiple predicates in OR") @@ -60,4 +62,43 @@ class SpecializedFilterOpDesc extends FilterOpDesc { supportReconfiguration = true ) } + + override def generateStandaloneCode(): String = { + // No predicate keeps no row: the executor's filter is `predicates.exists`, + // which answers false on an empty list. Passing the frame through would be + // the opposite answer. + if (predicates.isEmpty) return "out1df = in1df.iloc[0:0].copy()" + val conditions = predicates.map { p => + val colLit = pyStringLiteral(p.attribute) + p.condition match { + case ComparisonType.IS_NULL => s"""(in1df[$colLit].isna())""" + case ComparisonType.IS_NOT_NULL => s"""(in1df[$colLit].notna())""" + case other => + val op = other.getName // returns "=", ">=", "<", etc. (see ComparisonType.java) + val pyOp = if (op == "=") "==" else op + // notna mirrors FilterPredicate, which answers false for every condition + // but IS_NULL / IS_NOT_NULL once the field is null. Only `!=` needs it — + // pandas answers True there, where every other operator answers False — + // but guarding all of them keeps the one rule visible in one place. + s"""(in1df[$colLit].notna() & (in1df[$colLit] $pyOp ${coerceValue(p.value)}))""" + } + } + s"out1df = in1df[${conditions.mkString(" | ")}].reset_index(drop=True)" + } + + // Try numeric coercion so generated code compares column values against the right type. + // Strings that don't parse fall through to a quoted string literal. + private def coerceValue(raw: String): String = { + try { + raw.toInt.toString + } catch { + case _: NumberFormatException => + try { + raw.toDouble.toString + } catch { + case _: NumberFormatException => + pyStringLiteral(raw) + } + } + } } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/limit/LimitOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/limit/LimitOpDesc.scala index 6e1b7f37af..ec68ff7c5c 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/limit/LimitOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/limit/LimitOpDesc.scala @@ -25,12 +25,12 @@ import org.apache.texera.amber.core.executor.OpExecWithClassName import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp} import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} -import org.apache.texera.amber.operator.{LogicalOp, StateTransferFunc} +import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator, StateTransferFunc} import org.apache.texera.amber.util.JSONUtils.objectMapper import scala.util.{Success, Try} -class LimitOpDesc extends LogicalOp { +class LimitOpDesc extends LogicalOp with StandaloneCodeGenerator { @JsonProperty(required = true) @JsonSchemaTitle("Limit") @@ -80,4 +80,11 @@ class LimitOpDesc extends LogicalOp { } Success(newPhysicalOp, Some(stateTransferFunc)) } + + override def generateStandaloneCode(): String = { + // Clamped, because the two sides read a negative limit differently: the + // executor's `count < limit` is false from the first tuple and emits + // nothing, while pandas' head(-n) drops only the last n rows. + s"out1df = in1df.head(${math.max(0, limit)}).reset_index(drop=True)" + } } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala index fb9258410c..2004fb594c 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/projection/ProjectionOpDesc.scala @@ -26,17 +26,23 @@ import org.apache.texera.amber.core.tuple.Schema import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.PhysicalOp.oneToOnePhysicalOp import org.apache.texera.amber.core.workflow._ +import org.apache.texera.amber.operator.StandaloneCodeGenerator import org.apache.texera.amber.operator.map.MapOpDesc import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} +import org.apache.texera.amber.pybuilder.PythonTemplateBuilder.pyStringLiteral import org.apache.texera.amber.util.JSONUtils.objectMapper -class ProjectionOpDesc extends MapOpDesc { +class ProjectionOpDesc extends MapOpDesc with StandaloneCodeGenerator { @JsonProperty(required = true, defaultValue = "false") @JsonSchemaTitle("Drop Option") @JsonPropertyDescription("check to drop the selected attributes") var isDrop: Boolean = false + // Named explicitly, without `required`: the form already asks for these and must go + // on accepting an empty list, but a field carrying no annotation is invisible to + // anything reading the operator's config by reflection. + @JsonProperty var attributes: List[AttributeUnit] = List() override def getPhysicalOp( @@ -98,4 +104,32 @@ class ProjectionOpDesc extends MapOpDesc { outputPorts = List(OutputPort()) ) } + + override def generateStandaloneCode(): String = { + val units = Option(attributes).getOrElse(List.empty) + // The engine refuses an empty selection, so the script says so too. Passing + // the frame through would hand back data where a run would have stopped. + if (units.isEmpty) + return """raise ValueError("Please select at least one attribute to project.")""" + + if (isDrop) { + // Drop mode ignores aliases (matches ProjectionOpExec). + val cols = units.map(u => pyStringLiteral(u.getOriginalAttribute)).mkString("[", ", ", "]") + s"out1df = in1df.drop(columns=$cols)" + } else { + val originals = + units.map(u => pyStringLiteral(u.getOriginalAttribute)).mkString("[", ", ", "]") + // AttributeUnit.getAlias returns originalAttribute when alias is blank, + // so an explicit rename is only needed when they differ. + val renames = units + .filter(u => u.getAlias != u.getOriginalAttribute) + .map(u => s"""${pyStringLiteral(u.getOriginalAttribute)}: ${pyStringLiteral(u.getAlias)}""") + if (renames.isEmpty) { + s"out1df = in1df[$originals].copy()" + } else { + val renameMap = renames.mkString("{", ", ", "}") + s"out1df = in1df[$originals].rename(columns=$renameMap)" + } + } + } } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/union/UnionOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/union/UnionOpDesc.scala index 82e292c8f3..460aa7990b 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/union/UnionOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/union/UnionOpDesc.scala @@ -22,10 +22,10 @@ package org.apache.texera.amber.operator.union import org.apache.texera.amber.core.executor.OpExecWithClassName import org.apache.texera.amber.core.virtualidentity.{ExecutionIdentity, WorkflowIdentity} import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp} -import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} +import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator} -class UnionOpDesc extends LogicalOp { +class UnionOpDesc extends LogicalOp with StandaloneCodeGenerator { override def getPhysicalOp( workflowId: WorkflowIdentity, @@ -50,4 +50,11 @@ class UnionOpDesc extends LogicalOp { inputPorts = List(InputPort()), outputPorts = List(OutputPort()) ) + + // UNION ALL: UnionOpExec passes tuples through without dedup. The port is + // variadic, so the code names the whole list of upstreams rather than a fixed + // two — naming two dropped a third and left the second unbound when only one + // was drawn. + override def generateStandaloneCode(): String = + "out1df = pd.concat(inAlldf, ignore_index=True)" } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/distinct/DistinctOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/distinct/DistinctOpDescSpec.scala index 2aba788acf..15a9cf5d64 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/distinct/DistinctOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/distinct/DistinctOpDescSpec.scala @@ -106,4 +106,12 @@ class DistinctOpDescSpec extends AnyFlatSpec with Matchers { val b = new DistinctOpDesc a.operatorIdentifier should not equal b.operatorIdentifier } + + // The JVM operator keeps the first occurrence of a duplicate, which is what + // drop_duplicates does by default, so the emitted line says nothing about order. + "DistinctOpDesc.generateStandaloneCode" should "drop duplicates in place" in { + (new DistinctOpDesc).generateStandaloneCode() shouldBe + "out1df = in1df.drop_duplicates(ignore_index=True)" + } + } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala index 84c7ec9377..39e1ec5998 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/filter/SpecializedFilterOpDescSpec.scala @@ -70,4 +70,22 @@ class SpecializedFilterOpDescSpec extends AnyFlatSpec with Matchers { restored shouldBe a[SpecializedFilterOpDesc] restored.asInstanceOf[SpecializedFilterOpDesc].predicates shouldBe empty } + + // A null answers false for every condition but IS_NULL / IS_NOT_NULL, which pandas + // does not do on its own for `!=`, so the emitted condition carries the guard. + "SpecializedFilterOpDesc.generateStandaloneCode" should "emit one condition per predicate" in { + val d = new SpecializedFilterOpDesc + d.predicates = List(new FilterPredicate("age", ComparisonType.GREATER_THAN, "18")) + val code = d.generateStandaloneCode() + code should include("in1df[\"age\"]") + code should include("out1df") + } + + // The executor filters on `predicates.exists`, which answers false on an empty + // list, so no predicate keeps no row. The columns survive; the rows do not. + it should "keep no row when there is no predicate" in { + (new SpecializedFilterOpDesc).generateStandaloneCode() shouldBe + "out1df = in1df.iloc[0:0].copy()" + } + } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/limit/LimitOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/limit/LimitOpDescSpec.scala index f01e8f63f0..0ead5cd1cc 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/limit/LimitOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/limit/LimitOpDescSpec.scala @@ -94,4 +94,21 @@ class LimitOpDescSpec extends AnyFlatSpec with Matchers { transfer(oldExec, newExec) newExec.count shouldBe 3 } + + // The index is reset because the operator hands its downstream a fresh table + // rather than a view of the one it read. + "LimitOpDesc.generateStandaloneCode" should "take the first N rows" in { + val d = new LimitOpDesc + d.limit = 3 + d.generateStandaloneCode() shouldBe "out1df = in1df.head(3).reset_index(drop=True)" + } + + // head(-1) would drop the last row and keep the rest, where the executor's + // `count < limit` is false from the first tuple and keeps nothing. + it should "keep nothing when the limit is negative" in { + val d = new LimitOpDesc + d.limit = -1 + d.generateStandaloneCode() shouldBe "out1df = in1df.head(0).reset_index(drop=True)" + } + } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpDescSpec.scala index 66e6f556af..848a2b4a64 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/projection/ProjectionOpDescSpec.scala @@ -216,4 +216,26 @@ class ProjectionOpDescSpec extends AnyFlatSpec with BeforeAndAfter { assert(out == SinglePartition()) } + // Drop mode names the columns to remove; keep mode names the ones to hold on to, + // in the order the user put them in. + "ProjectionOpDesc.generateStandaloneCode" should "select or drop the named columns" in { + val keep = new ProjectionOpDesc + keep.attributes = List(new AttributeUnit("a", ""), new AttributeUnit("b", "")) + assert(keep.generateStandaloneCode().contains("""in1df[["a", "b"]]""")) + + val drop = new ProjectionOpDesc + drop.attributes = List(new AttributeUnit("a", "")) + drop.isDrop = true + assert(drop.generateStandaloneCode() == """out1df = in1df.drop(columns=["a"])""") + } + + // Schema propagation and the executor both refuse an empty selection, so the + // script stops where a run would have. Passing the frame through would answer + // with data a run never produces. + it should "stop on an empty selection rather than pass the frame through" in { + val code = (new ProjectionOpDesc).generateStandaloneCode() + assert(code.startsWith("raise ValueError(")) + assert(code.contains("Please select at least one attribute to project.")) + } + } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/union/UnionOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/union/UnionOpDescSpec.scala index a9c58bbcda..16c7e027bc 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/union/UnionOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/union/UnionOpDescSpec.scala @@ -87,6 +87,19 @@ class UnionOpDescSpec extends AnyFlatSpec with Matchers { physical.partitionRequirement shouldBe empty } + // --------------------------------------------------------------------------- + // generateStandaloneCode + // --------------------------------------------------------------------------- + + // UNION ALL: UnionOpExec passes tuples through without dedup, so the + // generated concat must not drop duplicates either. It names the whole list + // of upstreams rather than a fixed two, because the port is variadic and any + // count the code stated would be wrong for some workflow. + "UnionOpDesc.generateStandaloneCode" should "concatenate every input without dedup" in { + (new UnionOpDesc).generateStandaloneCode() shouldBe + "out1df = pd.concat(inAlldf, ignore_index=True)" + } + // --------------------------------------------------------------------------- // Independent instances // --------------------------------------------------------------------------- diff --git a/workflow-compiling-service/src/main/scala/org/apache/texera/service/WorkflowCompilingService.scala b/workflow-compiling-service/src/main/scala/org/apache/texera/service/WorkflowCompilingService.scala index a69ef54524..46649647a0 100644 --- a/workflow-compiling-service/src/main/scala/org/apache/texera/service/WorkflowCompilingService.scala +++ b/workflow-compiling-service/src/main/scala/org/apache/texera/service/WorkflowCompilingService.scala @@ -27,7 +27,11 @@ import org.apache.texera.common.config.StorageConfig import org.apache.texera.amber.util.ObjectMapperUtils import org.apache.texera.auth.{AuthFeatures, RoleAnnotationEnforcer} import org.apache.texera.dao.SqlServer -import org.apache.texera.service.resource.{HealthCheckResource, WorkflowCompilationResource} +import org.apache.texera.service.resource.{ + HealthCheckResource, + WorkflowCompilationResource, + WorkflowToPythonResource +} import org.eclipse.jetty.servlet.FilterHolder import java.nio.file.Path @@ -67,6 +71,9 @@ class WorkflowCompilingService extends Application[WorkflowCompilingServiceConfi // register the compilation endpoint environment.jersey.register(classOf[WorkflowCompilationResource]) + // register the workflow-to-python endpoint + environment.jersey.register(classOf[WorkflowToPythonResource]) + RoleAnnotationEnforcer.enforce( environment.jersey.getResourceConfig, "WorkflowCompilingService" diff --git a/workflow-compiling-service/src/main/scala/org/apache/texera/service/resource/WorkflowToPythonResource.scala b/workflow-compiling-service/src/main/scala/org/apache/texera/service/resource/WorkflowToPythonResource.scala new file mode 100644 index 0000000000..38381e51ae --- /dev/null +++ b/workflow-compiling-service/src/main/scala/org/apache/texera/service/resource/WorkflowToPythonResource.scala @@ -0,0 +1,99 @@ +/* + * 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.amber.core.tuple.Schema +import org.apache.texera.amber.core.virtualidentity.{OperatorIdentity, WorkflowIdentity} +import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext} +import org.apache.texera.common.compiler.model.{LogicalPlan, LogicalPlanPojo} +import org.apache.texera.common.compiler.{CompilationErrorHandling, WorkflowCompiler} +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 = { + try { + val logicalPlan = LogicalPlan(logicalPlanPojo) + // Two things come from compiling first. A scan source generates its reader from the + // schema it reads off the file, and that schema can only be read from a resolved URI, + // which compilation does before it expands the plan; without it the name stayed as + // typed and the CSV reader quietly dropped the timestamp columns it would have + // parsed. And a downstream operator that renders a column as text needs the type the + // column was DECLARED as, which the file it reads cannot carry. + val pythonCode = translator.translate(logicalPlan, outputSchemasOf(logicalPlanPojo)) + WorkflowToPythonSuccess(pythonCode) + } catch { + case e: Exception => + logger.error("Failed to translate workflow to Python", e) + WorkflowToPythonFailure(e.getMessage) + } + } + + /** + * What each operator's output ports carry, from the same Lenient compile the + * editing path runs. A failure is logged and the translation goes on without + * them, so a workflow whose file is not chosen yet still exports, as it did + * before, only without what the schema adds. + */ + private def outputSchemasOf( + logicalPlanPojo: LogicalPlanPojo + ): Map[OperatorIdentity, Map[PortIdentity, Option[Schema]]] = + try { + new WorkflowCompiler(new WorkflowContext(workflowId = WorkflowIdentity(0))) + .compile(logicalPlanPojo, CompilationErrorHandling.Lenient) + .operatorIdToOutputSchemas + } catch { + case e: Exception => + logger.warn("Could not resolve output schemas; translating without them", e) + Map.empty + } +} diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/service/WorkflowCompilingServiceRunSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/service/WorkflowCompilingServiceRunSpec.scala index 88b0411667..840685ba16 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/service/WorkflowCompilingServiceRunSpec.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/service/WorkflowCompilingServiceRunSpec.scala @@ -32,7 +32,11 @@ import jakarta.servlet.{DispatcherType, Filter, FilterChain} import jakarta.servlet.http.{HttpServletRequest, HttpServletResponse} import org.apache.texera.auth.{RoleAnnotationEnforcer, UnauthorizedExceptionMapper} import org.apache.texera.service.WorkflowCompilingServiceRunSpec.SpecPayload -import org.apache.texera.service.resource.{HealthCheckResource, WorkflowCompilationResource} +import org.apache.texera.service.resource.{ + HealthCheckResource, + WorkflowCompilationResource, + WorkflowToPythonResource +} import org.eclipse.jetty.servlet.{FilterHolder, ServletHandler} import org.glassfish.jersey.server.filter.RolesAllowedDynamicFeature import org.mockito.ArgumentCaptor @@ -75,10 +79,11 @@ class WorkflowCompilingServiceRunSpec extends AnyFlatSpec with Matchers { verify(jersey).setUrlPattern("/api/*") } - it should "register the health check and compilation endpoints" in { + it should "register the health check, compilation and export endpoints" in { val (jersey, _) = ranService verify(jersey).register(classOf[HealthCheckResource]) verify(jersey).register(classOf[WorkflowCompilationResource]) + verify(jersey).register(classOf[WorkflowToPythonResource]) } it should "install the auth stack" in { @@ -184,7 +189,11 @@ class WorkflowCompilingServiceRunSpec extends AnyFlatSpec with Matchers { // Every endpoint this service registers declares @RolesAllowed/@PermitAll/@DenyAll. "WorkflowCompilingService's registered resources" should "all declare access control" in { RoleAnnotationEnforcer.findUnannotatedEndpoints( - Seq(classOf[WorkflowCompilationResource], classOf[HealthCheckResource]) + Seq( + classOf[WorkflowCompilationResource], + classOf[HealthCheckResource], + classOf[WorkflowToPythonResource] + ) ) shouldBe empty } diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowToPythonResourceSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowToPythonResourceSpec.scala new file mode 100644 index 0000000000..e9bc1c96fe --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowToPythonResourceSpec.scala @@ -0,0 +1,200 @@ +/* + * 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.databind.node.ObjectNode +import io.dropwizard.testing.junit5.ResourceExtension +import jakarta.ws.rs.client.Entity +import jakarta.ws.rs.core.{MediaType, Response} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.operator.distinct.DistinctOpDesc +import org.apache.texera.amber.operator.limit.LimitOpDesc +import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc +import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlanPojo} +import org.assertj.core.api.Assertions.assertThat +import org.scalatest.BeforeAndAfterAll +import org.scalatest.flatspec.AnyFlatSpec + +/** + * Resource-layer tests for `/workflow-to-python`. Owns only what the REST + * envelope adds on top of the translation itself: HTTP status, the + * `@JsonTypeInfo` discriminator the frontend routes on, and the JSON shape + * the resource expects on the wire. + * + * What the translator does with a plan is asserted in + * `WorkflowToPythonTranslatorSpec`, and what an operator emits in its own + * spec, so a regression lands where it belongs. + */ +class WorkflowToPythonResourceSpec extends AnyFlatSpec with BeforeAndAfterAll { + + private val resources: ResourceExtension = ResourceExtension + .builder() + .addResource(new WorkflowToPythonResource()) + .setMapper(objectMapper) + .build() + + override protected def beforeAll(): Unit = resources.before() + override protected def afterAll(): Unit = resources.after() + + private def distinctOp(id: String): DistinctOpDesc = { + val op = new DistinctOpDesc() + op.setOperatorId(id) + op + } + + private def limitOp(id: String, rows: Int): LimitOpDesc = { + val op = new LimitOpDesc() + op.setOperatorId(id) + op.limit = rows + op + } + + // The frontend serializes LogicalLink with `fromOpId` / `toOpId` as flat + // strings, but the Scala case class stores them as nested `OperatorIdentity` + // records. This helper mirrors the wire shape so the test exercises the + // resource's actual JSON contract instead of a Scala-only round trip. + private def encodePojoAsFrontendJson(pojo: LogicalPlanPojo): String = { + val jsonNode = objectMapper.valueToTree[ObjectNode](pojo) + val linksArray = jsonNode.withArray("links") + linksArray.forEach { linkNode => + val fromOpIdNode = linkNode.get("fromOpId") + linkNode.asInstanceOf[ObjectNode].put("fromOpId", fromOpIdNode.get("id").asText()) + val toOpIdNode = linkNode.get("toOpId") + linkNode.asInstanceOf[ObjectNode].put("toOpId", toOpIdNode.get("id").asText()) + } + objectMapper.writeValueAsString(jsonNode) + } + + private def postExport(pojo: LogicalPlanPojo): Response = + resources + .target("/workflow-to-python") + .request(MediaType.APPLICATION_JSON) + .post(Entity.json(encodePojoAsFrontendJson(pojo))) + + private def chainOf(from: DistinctOpDesc, to: LimitOpDesc): LogicalPlanPojo = + LogicalPlanPojo( + operators = List(from, to), + links = List( + LogicalLink( + from.operatorIdentifier, + PortIdentity(0), + to.operatorIdentifier, + PortIdentity(0) + ) + ), + opsToViewResult = List.empty, + opsToReuseResult = List.empty + ) + + "POST /workflow-to-python" should "return HTTP 200 for a well-formed plan" in { + val response = postExport(chainOf(distinctOp("distinct"), limitOp("limit", 5))) + assertThat(response.getStatus).isEqualTo(200) + } + + // The panel sends a scan source's file name as the user gave it, and a source can only read + // its schema from a resolved URI, so the export resolves the names the way compilation does. + // A name that will not resolve, which is what an unfinished workflow has, must not cost the + // user the export: the failure is collected and the script is still built. + it should "still export a workflow whose scan source names a file that cannot be resolved" in { + val scan = new CSVScanSourceOpDesc() + scan.setOperatorId("scan") + scan.fileName = Some("/nonexistent/never-written.csv") + scan.customDelimiter = Some(",") + scan.hasHeader = true + + val limit = limitOp("limit", 5) + val response = postExport( + LogicalPlanPojo( + operators = List(scan, limit), + links = List( + LogicalLink( + scan.operatorIdentifier, + PortIdentity(0), + limit.operatorIdentifier, + PortIdentity(0) + ) + ), + opsToViewResult = List.empty, + opsToReuseResult = List.empty + ) + ) + + assertThat(response.getStatus).isEqualTo(200) + val parsed = + objectMapper.readValue( + response.readEntity(classOf[String]), + classOf[WorkflowToPythonResponse] + ) + assert(parsed.isInstanceOf[WorkflowToPythonSuccess], s"export failed: $parsed") + } + + it should "tag the body with type=success and carry the script the plan translates to" in { + // The @JsonTypeInfo on WorkflowToPythonResponse writes a `type` field. Both + // polymorphic deserialization and a raw-JSON `type == "success"` check need + // to hold, so the Angular client can branch without depending on Scala class + // names. + val response = postExport(chainOf(distinctOp("distinct"), limitOp("limit", 5))) + val body = response.readEntity(classOf[String]) + + val node = objectMapper.readTree(body) + assert( + node.has("type") && node.get("type").asText() == "success", + s"expected type:success discriminator, got $body" + ) + + val parsed = objectMapper.readValue(body, classOf[WorkflowToPythonResponse]) + assert(parsed.isInstanceOf[WorkflowToPythonSuccess]) + val code = parsed.asInstanceOf[WorkflowToPythonSuccess].pythonCode + // Both operators of the chain, and the import every script carries: enough + // to show the payload is the translated plan rather than an empty string. + assert(code.contains("import pandas as pd")) + assert(code.contains("drop_duplicates")) + assert(code.contains("head(5)")) + } + + it should "return a failure body rather than HTTP 500 when the plan cannot be read" in { + // A link naming an operator the plan does not carry: the DAG refuses the + // edge, and the resource has to answer with a reason rather than a stack + // trace the frontend cannot render. + val distinct = distinctOp("distinct") + val absent = limitOp("absent", 5) + val response = postExport( + LogicalPlanPojo( + operators = List(distinct), + links = List( + LogicalLink( + distinct.operatorIdentifier, + PortIdentity(0), + absent.operatorIdentifier, + PortIdentity(0) + ) + ), + opsToViewResult = List.empty, + opsToReuseResult = List.empty + ) + ) + + assertThat(response.getStatus).isEqualTo(200) + val node = objectMapper.readTree(response.readEntity(classOf[String])) + assertThat(node.get("type").asText()).isEqualTo("failure") + assertThat(node.has("errorMessage")).isTrue + } +}
