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 73f4a3846dd..008c07ee4ed 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 acbed3031fc..1ec9ad749e7 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/StandaloneCodeGenerator.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneCodeGenerator.scala
new file mode 100644
index 00000000000..6b4d4ab7feb
--- /dev/null
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/StandaloneCodeGenerator.scala
@@ -0,0 +1,72 @@
+/*
+ * 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
+
+import java.net.URLDecoder
+import java.nio.charset.StandardCharsets
+
+trait StandaloneCodeGenerator {
+
+  def generateStandaloneCode(): String
+
+  /**
+    * The file's own name, for a script that reads it from its own directory
+    * rather than through Texera's resolved URI.
+    *
+    * Taken from the last path segment instead of by parsing the whole string as a
+    * URI: the resolver percent-encodes the file-relative segments but leaves the
+    * repository and version names as the user typed them, so a dataset version
+    * called `v3 - with long text` makes `new URI` throw on the space and no code
+    * is generated at all.
+    */
+  protected def sourceBasename(rawPath: String): String = {
+    val segment = rawPath.split("/").lastOption.getOrElse("")
+    // Percent-decoding only, matching what `URI.getPath` used to return here: form
+    // decoding would also turn a literal `+` in a file name into a space.
+    URLDecoder.decode(segment.replace("+", "%2B"), StandardCharsets.UTF_8)
+  }
+
+  def producesDataFrame(): Boolean = true
+
+  /**
+    * Definitions this operator's standalone code depends on, emitted once near
+    * the top of the script rather than inline.
+    *
+    * The translator concatenates operator bodies into a single module, so an
+    * operator needing a helper class has nowhere to put it that another operator
+    * would not duplicate. Helpers returned here are collected across the whole
+    * plan and deduplicated by their text, so two sampling operators in one
+    * workflow yield one copy of the generator they share.
+    */
+  def standaloneHelpers(): Seq[String] = Seq.empty
+
+  /**
+    * Modules this operator's standalone code needs, written as the import
+    * statements themselves, collected across the plan and emitted once at the
+    * top of the script.
+    *
+    * pandas is not named here: the translator emits it for every script, since
+    * an operator body reads and writes frames whatever else it does. What an
+    * operator states here is what it needs beyond that, so a script built from
+    * operators that only reshape a table does not require a plotting library
+    * to start.
+    */
+  def standaloneImports(): Seq[String] = Seq.empty
+}
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 9e75e648bb4..17b646bfa10 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 9e86773df7c..e5d47ee5232 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 6e1b7f37af3..ec68ff7c5cc 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/metadata/annotations/SampleColumn.java b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/annotations/SampleColumn.java
new file mode 100644
index 00000000000..cb35c9aab91
--- /dev/null
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/metadata/annotations/SampleColumn.java
@@ -0,0 +1,45 @@
+/*
+ * 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.metadata.annotations;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * Test-only metadata for transform verification: names the column in the shared
+ * verification fixture the operator runs on that should fill this
+ * {@code @AutofillAttributeName} field when the operator is auto-configured.
+ *
+ * <p>It lets a field declare a <em>semantic</em> sample that the column's
+ * {@code AttributeType} alone cannot express — e.g. a valid three-letter ISO
+ * country code, or a genuine OHLC price column — so the parity test exercises
+ * the operator on realistic input instead of a degenerate first-column pick
+ * (which can hide translation bugs and produce vacuous passes).
+ *
+ * <p>This has <em>no effect on production</em>: it is not a Jackson / JSON-schema
+ * annotation and is read only by the test-side ConfigGenerator.
+ */
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.FIELD})
+public @interface SampleColumn {
+    String value();
+}
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 fb9258410cb..2004fb594c6 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 82e292c8f38..460aa7990b1 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 2aba788acfe..15a9cf5d64e 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 84c7ec93779..39e1ec59988 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 f01e8f63f08..0ead5cd1ccf 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 66e6f556afb..848a2b4a647 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 a9c58bbcda4..16c7e027bc5 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/build.sbt b/workflow-compiling-service/build.sbt
index 2af92efc4d4..1c440d14b06 100644
--- a/workflow-compiling-service/build.sbt
+++ b/workflow-compiling-service/build.sbt
@@ -41,9 +41,27 @@ ThisBuild / semanticdbVersion := scalafixSemanticdb.revision
 // Manage dependency conflicts by always using the latest revision
 ThisBuild / conflictManager := ConflictManager.latestRevision
 
-// Restrict parallel execution of tests to avoid conflicts
+// Restrict parallel execution of tests to avoid conflicts. This caps how many
+// test *suites* run concurrently; ParallelTestExecution still parallelizes the
+// tests *within* a suite (e.g. OperatorBehaviorSpec) via ScalaTest's own pool.
 Global / concurrentRestrictions += Tags.limit(Tags.Test, 1)
 
+// -P4 bounds ScalaTest's ParallelTestExecution pool, and only this module wants
+// it: OperatorBehaviorSpec forks a Python subprocess per operator, and at
+// core-count concurrency (e.g. 12) resource contention caused rare flakes. A
+// fixed 4 stays deterministic across machines (incl. CI runners) while still
+// running ~3x faster than serial, and it matches PythonWorkerPool's own default
+// worker cap so the two bounds agree rather than multiply. Unconditional, so a
+// local run reproduces the concurrency CI runs at instead of a faster one that
+// flakes differently; WCS_TEST_FILTER selects which tests run, which is a
+// separate question from how many run at once. The fast-unit job is unaffected
+// either way, since OperatorBehaviorSpec is the only spec here that
+// parallelizes and that job excludes it. It lives here rather than in the
+// shared helper so that helper stays identical for every module. sbt
+// concatenates the ScalaTest arguments of every testOptions entry, so this
+// lands in the same argument list as the -n above.
+Test / testOptions += Tests.Argument(TestFrameworks.ScalaTest, "-P4")
+
 /////////////////////////////////////////////////////////////////////////////
 // Compiler Options
 /////////////////////////////////////////////////////////////////////////////
diff --git a/workflow-compiling-service/src/main/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslator.scala b/workflow-compiling-service/src/main/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslator.scala
new file mode 100644
index 00000000000..d0bdf2a131f
--- /dev/null
+++ b/workflow-compiling-service/src/main/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslator.scala
@@ -0,0 +1,208 @@
+/*
+ * 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]()
+
+    // getTopologicalOpIds() uses jgrapht internally — no need for a custom topo sort
+    val topoOrder = logicalPlan.getTopologicalOpIds.asScala.toList
+
+    // pandas is the one module every generator uses: an operator body reads and
+    // writes frames whatever else it does. Everything beyond that is asked of the
+    // operators in the plan, so a script that draws nothing does not require a
+    // plotting library to start.
+    script += "import pandas as pd"
+    topoOrder
+      .map(logicalPlan.getOperator)
+      .collect { case gen: StandaloneCodeGenerator => gen.standaloneImports() }
+      .flatten
+      .distinct
+      .foreach(script += _)
+    script += ""
+
+    // 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:")"""
+            // The frame itself rather than head(): pandas already elides the
+            // middle of a long one, and it states the row and column count,
+            // which head() hides.
+            script += s"print($varName)"
+            script += ""
+        }
+    }
+
+    script.mkString("\n")
+  }
+
+  // Replaces in{N}df / out{N}df placeholders with concrete variable names.
+  // Substitutes in reverse index order to prevent partial matches (e.g. in1df
+  // inside in10df). After substitution, scans for any leftover placeholders
+  // and logs a warning — that signals a mismatch between an operator's
+  // declared port count and what its generateStandaloneCode actually emits.
+  private def substituteVars(
+      code: String,
+      inVars: List[String],
+      outVars: List[String],
+      displayName: String
+  ): String = {
+    var result = code
+
+    // A variadic port takes as many upstream links as the user draws, and an
+    // operator reading one cannot name them: `in1df`/`in2df` state a count, and
+    // whichever count it states is wrong for every other workflow. This one
+    // placeholder becomes the whole list, so the operator writes the same line
+    // whether it is fed one table or five.
+    result = result.replaceAll("""\binAlldf\b""", inVars.mkString("[", ", ", "]"))
+    inVars.zipWithIndex.reverse.foreach {
+      case (v, idx) => result = result.replaceAll(s"\\bin${idx + 1}df\\b", v)
+    }
+    outVars.zipWithIndex.reverse.foreach {
+      case (v, idx) => result = result.replaceAll(s"\\bout${idx + 1}df\\b", v)
+    }
+
+    val leftoverIn = """\bin\d+df\b""".r.findAllIn(result).toSet
+    val leftoverOut = """\bout\d+df\b""".r.findAllIn(result).toSet
+    if (leftoverIn.nonEmpty || leftoverOut.nonEmpty) {
+      logger.warn(
+        s"Operator '$displayName' emitted placeholders that don't match its port " +
+          s"count: leftover inputs=$leftoverIn, leftover outputs=$leftoverOut. " +
+          s"Generated script will reference unbound variables."
+      )
+    }
+
+    result
+  }
+}
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 a69ef545246..46649647a0a 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 00000000000..18b75ef338f
--- /dev/null
+++ b/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 = {
+    try {
+      val logicalPlan = LogicalPlan(logicalPlanPojo)
+      val pythonCode = translator.translate(logicalPlan)
+      WorkflowToPythonSuccess(pythonCode)
+    } catch {
+      case e: Exception =>
+        logger.error("Failed to translate workflow to Python", e)
+        WorkflowToPythonFailure(e.getMessage)
+    }
+  }
+}
diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslatorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslatorSpec.scala
new file mode 100644
index 00000000000..b04f967ab7d
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/WorkflowToPythonTranslatorSpec.scala
@@ -0,0 +1,136 @@
+/*
+ * 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 org.apache.texera.amber.core.workflow.PortIdentity
+import org.apache.texera.amber.operator.LogicalOp
+import org.apache.texera.amber.operator.distinct.DistinctOpDesc
+import org.apache.texera.amber.operator.union.UnionOpDesc
+import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlan}
+import org.scalatest.flatspec.AnyFlatSpec
+import org.scalatest.matchers.should.Matchers
+
+/** The placeholder substitution, which is where an operator's generated code
+  * meets the variables the script actually binds. A variadic port is the case
+  * the numbered placeholders cannot state, so it is the case worth pinning.
+  */
+class WorkflowToPythonTranslatorSpec extends AnyFlatSpec with Matchers {
+
+  private def upstream(id: String): LogicalOp = {
+    val op = new DistinctOpDesc
+    op.setOperatorId(id)
+    op
+  }
+
+  /** `n` upstreams, all drawn into the union's single port, which is what a
+    * variadic port looks like in a plan.
+    */
+  private def unionOf(n: Int): String = {
+    val union = new UnionOpDesc
+    union.setOperatorId("union")
+    val ups = (1 to n).map(i => upstream(s"up$i"))
+    val links = ups.map { up =>
+      LogicalLink(
+        up.operatorIdentifier,
+        PortIdentity(0),
+        union.operatorIdentifier,
+        PortIdentity(0)
+      )
+    }
+    new WorkflowToPythonTranslator().translate(
+      LogicalPlan(ups.toList :+ union, links.toList)
+    )
+  }
+
+  "WorkflowToPythonTranslator" should "hand a variadic port every upstream it was drawn" in {
+    unionOf(3) should include("pd.concat([df1, df2, df3], ignore_index=True)")
+  }
+
+  it should "hand a variadic port a one-element list when only one link is drawn" in {
+    // The case the old fixed `[in1df, in2df]` got wrong in the other direction:
+    // it named a second frame the script never bound.
+    unionOf(1) should include("pd.concat([df1], ignore_index=True)")
+  }
+
+  it should "leave no placeholder behind for a variadic port" in {
+    unionOf(2) should not include "inAlldf"
+  }
+
+  // head() shows five rows and does not say how many there were, so a script whose
+  // leaf holds more reads as if that were the whole answer.
+  it should "print the leaf frame rather than its first rows" in {
+    val script = unionOf(2)
+    script should include("print(df3)")
+    script should not include ".head())"
+  }
+
+  // A script that only reshapes a table should run wherever pandas is installed,
+  // so an import no operator in the plan asked for must not be in the header.
+  it should "import pandas alone for a plan that asks for nothing else" in {
+    val script = unionOf(2)
+    script should include("import pandas as pd")
+    script should not include "import plotly"
+  }
+
+  // Two operators naming the same module yield one import, the way two operators
+  // sharing one helper yield one copy of it.
+  it should "emit an operator's declared import once per plan" in {
+    val ops = List("a", "b").map { id =>
+      val op = new DistinctOpDesc {
+        override def standaloneImports(): Seq[String] = Seq("import numpy as np")
+      }
+      op.setOperatorId(id)
+      op
+    }
+    val script = new WorkflowToPythonTranslator().translate(LogicalPlan(ops, List.empty))
+    script.linesIterator.count(_ == "import numpy as np") shouldBe 1
+  }
+
+  it should "still resolve a numbered placeholder against its own upstream" in {
+    // The variadic form is an addition, not a replacement: a chain of ordinary
+    // single-input operators has to keep reading `in1df` as its predecessor.
+    val first = upstream("first")
+    val second = upstream("second")
+    val script = new WorkflowToPythonTranslator().translate(
+      LogicalPlan(
+        List(first, second),
+        List(
+          LogicalLink(
+            first.operatorIdentifier,
+            PortIdentity(0),
+            second.operatorIdentifier,
+            PortIdentity(0)
+          )
+        )
+      )
+    )
+    script should include("df2 = df1.drop_duplicates(ignore_index=True)")
+  }
+
+  /** The translator's own contract when it meets an operator it cannot render:
+    * a comment rather than a silently wrong line.
+    */
+  it should "leave a TODO for an operator with no standalone code generator" in {
+    val op = new org.apache.texera.amber.operator.udf.python.PythonUDFOpDescV2
+    op.setOperatorId("udf")
+    val script = new WorkflowToPythonTranslator().translate(LogicalPlan(List(op), List.empty))
+    script should include("# TODO:")
+  }
+}
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 88b04116677..840685ba160 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 00000000000..a8068ae9c74
--- /dev/null
+++ b/workflow-compiling-service/src/test/scala/org/apache/texera/service/resource/WorkflowToPythonResourceSpec.scala
@@ -0,0 +1,162 @@
+/*
+ * 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.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)
+  }
+
+  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
+  }
+}