From 4562b6306c8661454addb63a93bee8734317774f Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 2 Sep 2026 16:30:44 -0700 Subject: [PATCH 1/2] test(verify): run the script the export produces The other side of the comparison. `StandaloneRunner` writes the script the operator's generator emits, binds its inputs to the files the fixture wrote, runs it, and reads the frames it leaves behind. The script is kept where it ran, so an operator whose two answers differ can be opened as generated rather than described second-hand. `HarnessSpec` covers all three pieces on one operator whose answer is short enough to state in full. Co-Authored-By: Claude Opus 5 (1M context) --- .../resources/python/standalone_worker.py | 122 ++++++ .../amber/translator/verify/HarnessSpec.scala | 115 ++++++ .../translator/verify/StandaloneRunner.scala | 367 ++++++++++++++++++ 3 files changed, 604 insertions(+) create mode 100644 workflow-compiling-service/src/test/resources/python/standalone_worker.py create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/HarnessSpec.scala create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneRunner.scala diff --git a/workflow-compiling-service/src/test/resources/python/standalone_worker.py b/workflow-compiling-service/src/test/resources/python/standalone_worker.py new file mode 100644 index 00000000000..946fec23206 --- /dev/null +++ b/workflow-compiling-service/src/test/resources/python/standalone_worker.py @@ -0,0 +1,122 @@ +#!/usr/bin/env python3 +# +# 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. +""" +Persistent worker for the Path B (standalone) verify path. + +Motivation: forking a fresh interpreter per operator pays the pandas/plotly +import cost (~260-310 ms) on every spawn, while the operator's actual compute +on the tiny canonical fixtures is ~4 ms. Imports dominate ~96% of the per-spawn +cost. This worker imports those heavy libraries ONCE at startup, then executes +many operators' generated scripts over its lifetime — so the import cost is +paid once, not once per operator. + +It is a drop-in replacement for `python `: it runs the exact same +rendered script `StandaloneRunner` already produces (imports + prologue + body ++ epilogue). The script's own top-of-file `import pandas` becomes a ~0 ms +`sys.modules` cache hit. + +Protocol (line-delimited JSON, both directions): + + startup worker -> parent: {"ready": true} + request parent -> worker: {"scriptPath": "", "workDir": ""}\n + response worker -> parent: {"exit": 0, "stdout": "...", "stderr": "..."}\n + +`exit` is 0 on success or 1 if the script raised; on 1, `stderr` carries the +traceback — mirroring a nonzero subprocess exit so the Scala side's +StandaloneExecutionException path is unchanged. The worker keeps running after +a script error (only a hard interpreter crash ends it); parent closes stdin +(EOF) to shut it down. + +Isolation trade-off (accepted, per design discussion): all jobs share one +interpreter, so module-level state (e.g. pandas display options) can leak +between operators. Each job is exec'd in a FRESH namespace and chdir'd to its +own workDir to contain the common cases; this is weaker than the old +process-per-operator isolation. +""" +from __future__ import annotations + +import io +import json +import os +import sys +import traceback +from contextlib import redirect_stderr, redirect_stdout + +# --- Pay the heavy import cost ONCE, here, at startup. ---------------------- +# These mirror the imports StandaloneRunner injects at the top of every +# rendered script. Pre-importing them populates sys.modules, so each executed +# script's own `import pandas as pd` / `import plotly...` is a cache hit. +# numpy is intentionally NOT imported (see StandaloneRunner.renderScript: the +# production translator only provides pandas + plotly, so an operator needing +# numpy must import it itself — we must not mask that). +import pandas as pd # noqa: F401 +import plotly.express as px # noqa: F401 +import plotly.graph_objects as go # noqa: F401 +import plotly.io # noqa: F401 + + +def _run_one(script_path: str, work_dir: str) -> "dict[str, object]": + """Execute one rendered standalone script and capture its output. + + Runs in a fresh namespace with cwd = work_dir (generated code may use + relative paths, e.g. CSVScan's `pd.read_csv("sample.csv")`; absolute paths + written by the prologue/epilogue are unaffected). The script's stdout / + stderr are redirected into buffers so they never corrupt the protocol + channel on real stdout. + """ + out_buf, err_buf = io.StringIO(), io.StringIO() + # __name__ = "__main__" so scripts with a `if __name__ == "__main__"` guard + # still run their body (the translator does not emit one, but it is free + # insurance and matches `python script.py` semantics). + namespace = {"__name__": "__main__", "__file__": script_path} + try: + with open(script_path, "r", encoding="utf-8") as f: + source = f.read() + os.chdir(work_dir) + code = compile(source, script_path, "exec") + with redirect_stdout(out_buf), redirect_stderr(err_buf): + exec(code, namespace) # noqa: S102 (running generated verify code by design) + return {"exit": 0, "stdout": out_buf.getvalue(), "stderr": err_buf.getvalue()} + except BaseException: # noqa: BLE001 — a script error must NOT kill the worker + # Match a nonzero subprocess exit: traceback goes to stderr, exit = 1. + err = err_buf.getvalue() + traceback.format_exc() + return {"exit": 1, "stdout": out_buf.getvalue(), "stderr": err} + + +def main() -> None: + # Signal readiness only after the heavy imports above have completed, so the + # parent can warm a pool and attribute startup cost deterministically. + sys.stdout.write(json.dumps({"ready": True}) + "\n") + sys.stdout.flush() + + for line in sys.stdin: + line = line.strip() + if not line: + continue + try: + req = json.loads(line) + result = _run_one(req["scriptPath"], req["workDir"]) + except Exception: # malformed request — report, keep serving + result = {"exit": 1, "stdout": "", "stderr": traceback.format_exc()} + sys.stdout.write(json.dumps(result) + "\n") + sys.stdout.flush() + + +if __name__ == "__main__": + main() diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/HarnessSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/HarnessSpec.scala new file mode 100644 index 00000000000..685b6f056db --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/HarnessSpec.scala @@ -0,0 +1,115 @@ +/* + * 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.verify + +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.operator.distinct.DistinctOpDesc +import org.scalatest.Tag +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.nio.file.{Files, Path} + +/** The two ways of running one operator, and the file format they meet in. + * + * `Distinct` is the operator under test throughout, because what is being + * tested is the harness rather than the operator: it takes one input, needs no + * configuration, and its answer is short enough to state in full. + */ +class HarnessSpec extends AnyFlatSpec with Matchers { + + /** Only the standalone run needs an interpreter, so only it is held back from + * the job that provisions none. The other two are JVM-side and run there. + */ + private val NeedsPython = + Tag("org.apache.texera.amber.translator.verify.tags.IntegrationTest") + + private val schema = new Schema( + new Attribute("id", AttributeType.INTEGER), + new Attribute("name", AttributeType.STRING) + ) + + private def tuple(id: Int, name: String): Tuple = { + val b = Tuple.builder(schema) + b.add(schema.getAttribute("id"), Int.box(id)) + b.add(schema.getAttribute("name"), name) + b.build() + } + + /** Four rows, the last a repeat of the second. */ + private val rows = Seq(tuple(1, "a"), tuple(2, "b"), tuple(3, "c"), tuple(2, "b")) + + private def withInput(test: (Path, Path) => Unit): Unit = { + val dir = Files.createTempDirectory("harness-spec-") + val input = dir.resolve("input_port_0.jsonl") + TupleIO.writeTuples(input, rows.iterator, schema) + test(dir, input) + } + + "TupleIO" should "read back the rows and the schema it wrote" in { + withInput { (_, input) => + // The schema travels in a sidecar rather than in the JSONL, which carries + // values alone and so cannot say a column is INTEGER rather than a number. + TupleIO.readSchemaSidecar(input) shouldBe schema + val read = TupleIO.readTuples(input, schema).toSeq + read should have length 4 + read.map(_.getField[Integer]("id").intValue) shouldBe Seq(1, 2, 3, 2) + } + } + + "OpExecHarness" should "run an operator and write one file per output port" in { + withInput { (dir, input) => + val out = dir.resolve("actual") + val result = + OpExecHarness.execute(new DistinctOpDesc, Map(PortIdentity(0) -> input), out) + + result.outputs should have size 1 + val produced = result.outputs(PortIdentity(0)) + Files.exists(produced) shouldBe true + + val written = TupleIO.readTuples(produced, result.outputSchemas(PortIdentity(0))).toSeq + written.map(_.getField[Integer]("id").intValue) shouldBe Seq(1, 2, 3) + } + } + + "StandaloneRunner" should "run the generated script and reach the same answer" taggedAs NeedsPython in { + withInput { (dir, input) => + val work = dir.resolve("standalone") + Files.createDirectories(work) + val result = StandaloneRunner.run( + opDesc = new DistinctOpDesc, + inputs = Map(1 -> input), + outputPortCount = 1, + workDir = work + ) + + // The script is kept where it ran, so a failing operator can be opened as + // generated rather than described second-hand. + Files.exists(work.resolve("script.py")) shouldBe true + + val produced = result.outputs(1) + val lines = Files.readAllLines(produced) + lines should have size 3 + lines.get(0) should include("\"id\":1") + lines.get(2) should include("\"id\":3") + } + } +} diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneRunner.scala new file mode 100644 index 00000000000..359bad51e5a --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneRunner.scala @@ -0,0 +1,367 @@ +/* + * 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.verify + +import com.typesafe.scalalogging.LazyLogging +import org.apache.texera.amber.core.tuple.AttributeType +import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator} +import org.apache.texera.amber.util.python.PythonWorkerPool + +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Path} +import scala.collection.mutable.ArrayBuffer +import scala.sys.process._ + +/** + * Executes the Python code an OpDesc's [[StandaloneCodeGenerator]] emits and + * captures its DataFrame outputs as JSONL files (compatible with + * [[TupleIO]]'s sidecar-schema format on the comparison side). + * + * Wraps the operator's raw generated code with: + * + * ── prologue ────────────────────────────────────────────── + * in1df = pd.read_json("input_port_0.jsonl", lines=True) + * in2df = pd.read_json("input_port_1.jsonl", lines=True) + * ... + * inAlldf = [in1df, in2df] + * ── operator body (verbatim from generateStandaloneCode) ── + * out1df = in1df[in1df["age"] > 18] + * ── epilogue ───────────────────────────────────────────── + * out1df.to_json("output_port_0.jsonl", orient='records', lines=True) + * ... + * + * Port indexing matches the placeholder convention used by the translator: + * `inNdf`/`outNdf` is 1-based and corresponds to the operator's N-th external + * input/output port in declaration order. The harness key (a 1-based Int) is + * what the placeholder uses; the caller is responsible for ordering inputs + * the same way the operator's `generateStandaloneCode()` expects. + * + * The subprocess inherits the caller's environment so the Python interpreter + * picks up whatever pandas/plotly the test fixture installed. + */ +object StandaloneRunner extends LazyLogging { + + /** The value both paths seed numpy's global RNG with. Any fixed number does; + * what matters is that the two agree, so it is declared once here and + * referenced by name from py_op_driver's comment. + */ + private[verify] val VerifySeed: Int = 20260811 + + /** + * @param outputs paths to the per-port output JSONL files. Empty map iff + * the operator's `producesDataFrame()` returned false + * (visualizations, etc.) — caller handles those separately. + * @param stdout raw subprocess stdout (useful for failure diagnostics) + * @param stderr raw subprocess stderr + */ + final case class Result(outputs: Map[Int, Path], stdout: String, stderr: String) + + /** + * Generate, write, and execute the standalone Python script for `opDesc`. + * + * @param opDesc must mix in [[StandaloneCodeGenerator]]; otherwise we throw + * since there's nothing to test. + * @param inputs map from 1-based port index → JSONL fixture path. The + * script reads each into `inNdf`. + * @param outputPortCount how many `outNdf` variables the operator declares. + * Caller derives this from the OpDesc's output ports. + * @param workDir directory used for the generated `script.py` and output + * JSONL files. Created if missing. + * @param pythonExe path to the Python 3.12 interpreter. Defaults to + * the env var `UDF_PYTHON_PATH`, then `python3.12`, then + * `python3`. The same fallback chain used by the rest of + * the Texera test suite for Python-backed operators. + */ + def run( + opDesc: LogicalOp, + inputs: Map[Int, Path], + outputPortCount: Int, + workDir: Path, + pythonExe: String = resolvePython() + ): Result = { + val gen = opDesc match { + case g: StandaloneCodeGenerator => g + case other => + throw new IllegalArgumentException( + s"OpDesc ${other.getClass.getSimpleName} does not implement " + + s"StandaloneCodeGenerator; nothing to verify" + ) + } + + Files.createDirectories(workDir) + val scriptPath = workDir.resolve("script.py") + val outputPaths: Map[Int, Path] = + if (gen.producesDataFrame()) + (1 to outputPortCount).map(i => i -> workDir.resolve(s"output_port_${i - 1}.jsonl")).toMap + else Map.empty + + val source = + renderScript(gen.generateStandaloneCode(), inputs, outputPaths, gen.standaloneHelpers()) + Files.write(scriptPath, source.getBytes(StandardCharsets.UTF_8)) + + val (exit, stdout, stderr) = execute(scriptPath, workDir, pythonExe) + if (exit != 0) { + throw new StandaloneExecutionException(exit, scriptPath, source, stdout, stderr) + } + Result(outputPaths, stdout, stderr) + } + + private val WorkerResourcePath = "/python/standalone_worker.py" + + // Run the rendered script and return (exitCode, stdout, stderr). Prefers a + // pooled persistent worker (imports pandas/plotly once, ~18x faster per op — + // see PythonWorkerPool); a rare hard worker crash falls back to a one-shot + // subprocess so behavior is never worse than the original path. Both paths + // run with cwd = workDir and read results from files, so they are + // interchangeable — the executed script is byte-identical. + private def execute(scriptPath: Path, workDir: Path, pythonExe: String): (Int, String, String) = { + if (PythonWorkerPool.enabled) { + try { + val req = org.apache.texera.amber.util.JSONUtils.objectMapper.createObjectNode() + req.put("scriptPath", scriptPath.toString) + req.put("workDir", workDir.toString) + val o = PythonWorkerPool.run(WorkerResourcePath, Seq.empty, pythonExe, req) + return (o.exit, o.stdout, o.stderr) + } catch { + case e: PythonWorkerPool.WorkerDiedException => + logger.warn( + s"Standalone worker unavailable; falling back to one-shot subprocess " + + s"for $scriptPath: ${e.getMessage}" + ) + } + } + runSubprocess(scriptPath, workDir, pythonExe) + } + + // Original one-process-per-operator path. Retained as the fallback and as the + // behavior selected by TEXERA_TEST_PYTHON_WORKER=0. + private def runSubprocess( + scriptPath: Path, + workDir: Path, + pythonExe: String + ): (Int, String, String) = { + // Capture stdout/stderr separately. ProcessLogger's append is called from + // the subprocess's I/O thread, so we collect into ArrayBuffer (thread-safe + // append is fine for this serial use) and join at the end. + val outBuf = ArrayBuffer.empty[String] + val errBuf = ArrayBuffer.empty[String] + val logger = ProcessLogger(line => outBuf += line, line => errBuf += line) + // cwd = workDir so generated code using *relative* paths (e.g. CSVScan's + // basename-stripped `pd.read_csv("sample.csv")`) resolves against workDir. + // Absolute paths written by the prologue/epilogue are unaffected. + val exit = Process(Seq(pythonExe, scriptPath.toString), Some(workDir.toFile)).!(logger) + (exit, outBuf.mkString("\n"), errBuf.mkString("\n")) + } + + // Builds the full Python source: imports + prologue + verbatim operator body + // + epilogue. We intentionally do NOT substitute the inNdf/outNdf placeholders + // — the body keeps them so the var-bindings the prologue/epilogue introduce + // (also named inNdf/outNdf) reference the same names. + private def renderScript( + body: String, + inputs: Map[Int, Path], + outputs: Map[Int, Path], + helpers: Seq[String] + ): String = { + val sb = new StringBuilder + + sb.append("# Auto-generated by StandaloneRunner. Do not commit.\n") + sb.append("import json\n") + sb.append("import sys\n") + sb.append("import base64\n") + sb.append("import pickle\n") + // NOTE: numpy is intentionally NOT injected here. The production translator + // (WorkflowToPythonTranslator) only provides pandas + plotly to standalone + // scripts, so any operator whose standalone code needs numpy must import it + // itself. Injecting numpy here would mask that class of bug in verify tests. + sb.append("import pandas as pd\n") + sb.append("import plotly.express as px\n") + sb.append("import plotly.graph_objects as go\n") + sb.append("import plotly.io\n") + // Same seed as py_op_driver's run_config, for the reason given there. Bound + // under a private name and deleted so the note above still holds: a script + // that wants numpy has to import it, and this does not hand it one. + sb.append(s"import numpy as _texera_np; _texera_np.random.seed($VerifySeed); del _texera_np\n") + sb.append("\n") + + // Object columns holding non-primitive values (e.g. a trained sklearn model + // in a BINARY output column) can't go through to_json. Pickle+base64 them so + // the JSONL matches py_op_driver's BINARY write path exactly. Primitives + // (str/int/float/bool/None) pass through unchanged, so ordinary DataFrame + // outputs are unaffected. + sb.append("def _texera_encode_obj_cols(df):\n") + sb.append(" for _c in df.columns:\n") + sb.append(" if df[_c].dtype == object:\n") + sb.append( + " df[_c] = df[_c].map(lambda _v: base64.b64encode(pickle.dumps(_v)).decode('ascii') " + + "if not isinstance(_v, (str, int, float, bool, type(None))) else _v)\n" + ) + sb.append(" return df\n") + sb.append("\n") + + // TIMESTAMP columns are handed to the operator as datetime64 (see the + // prologue below) to match the schema-typed runtime path, but the runtime + // path serializes a TIMESTAMP back out with java.sql.Timestamp.toString — + // "yyyy-mm-dd hh:mm:ss.f", trailing zeros trimmed to at least one digit — + // whereas pandas' to_json would emit epoch millis. Convert datetime columns + // back to that exact form before writing so both paths' JSONL agree. + sb.append("def _texera_ts_str(_v):\n") + sb.append(" if pd.isna(_v):\n") + sb.append(" return None\n") + sb.append(" _s = _v.strftime('%Y-%m-%d %H:%M:%S.%f').rstrip('0')\n") + sb.append(" return _s + '0' if _s.endswith('.') else _s\n") + sb.append("\n") + sb.append("def _texera_encode_ts_cols(df):\n") + sb.append(" for _c in df.columns:\n") + sb.append(" if pd.api.types.is_datetime64_any_dtype(df[_c]):\n") + sb.append(" df[_c] = df[_c].map(_texera_ts_str)\n") + sb.append(" return df\n") + sb.append("\n") + + // Prologue: load each external input into in{N}df. Note: pd.read_json with + // lines=True correctly handles empty files (returns empty DataFrame). + // convert_dates=False: pd.read_json otherwise auto-coerces ISO-ish strings + // and columns named like dates ("date", "*_at", …) to datetime64, which the + // schema-typed runtime path (STRING) does not do — that divergence would + // make a plain date string column serialize as "...T00:00:00" on only one + // side. Operators that genuinely need datetimes convert explicitly, so both + // paths stay in sync. + // precise_float=True: pd.read_json's default (ujson) fast double parser is + // lossy in the last few ULPs, so a DOUBLE column would load slightly + // different values than the schema-typed runtime path (which parses doubles + // exactly). Operators that stringify raw cell values (e.g. Radar hover text) + // then diverge; precise_float=True keeps both paths bit-identical. + // The blanket convert_dates=False also leaves genuine TIMESTAMP columns as + // strings, which the runtime path delivers as datetime64 — a divergence for + // any operator that renders or computes on them. The fixture's schema + // sidecar says which columns those are, so cast exactly those back. + inputs.toSeq.sortBy(_._1).foreach { + case (n, path) => + sb.append( + s"in${n}df = pd.read_json(${py(path.toString)}, lines=True, convert_dates=False, precise_float=True)\n" + ) + timestampColumns(path).foreach { col => + sb.append(s"if ${py(col)} in in${n}df.columns:\n") + sb.append(s" in${n}df[${py(col)}] = pd.to_datetime(in${n}df[${py(col)}])\n") + } + doubleColumns(path).foreach { col => + sb.append(s"if ${py(col)} in in${n}df.columns:\n") + sb.append(s" in${n}df[${py(col)}] = in${n}df[${py(col)}].astype('float64')\n") + } + } + // The variadic placeholder, bound here for the same reason the numbered ones + // are: this script leaves the body's placeholders alone and defines names to + // match them, so an operator reading a variadic port finds its list here the + // way the translator would have written one out. + if (inputs.nonEmpty) { + sb.append( + inputs.keys.toSeq.sorted.map(n => s"in${n}df").mkString("inAlldf = [", ", ", "]\n") + ) + } + sb.append("\n") + + // Body verbatim — placeholders left in place. + // Emitted ahead of the body the way the translator does, so an operator that + // declares a helper is exercised here exactly as it runs in a real script. + helpers.foreach { helper => + sb.append(helper) + if (!helper.endsWith("\n")) sb.append('\n') + sb.append('\n') + } + + sb.append("# ── operator body ──\n") + sb.append(body) + if (!body.endsWith("\n")) sb.append('\n') + sb.append("\n") + + // Epilogue: dump each out{N}df to JSONL. When producesDataFrame() is false + // (visualization ops), `outputs` is empty and this block is a no-op — the + // caller is expected to verify viz outputs by other means. + outputs.toSeq.sortBy(_._1).foreach { + case (n, path) => + sb.append( + s"_texera_encode_obj_cols(_texera_encode_ts_cols(out${n}df))" + + s".to_json(${py(path.toString)}, orient='records', lines=True)\n" + ) + } + + sb.toString + } + + // TIMESTAMP-typed column names from a fixture's `.jsonl.schema.json` sidecar. + // A missing or unreadable sidecar means no casts — the prologue then behaves + // exactly as before. + private def timestampColumns(input: Path): Seq[String] = + columnsOfType(input, AttributeType.TIMESTAMP) + + // DOUBLE-typed column names. pd.read_json narrows a float column whose values + // are all integral to int64, while the runtime path keeps the schema's DOUBLE, + // so a column like 7.0 stringifies as "7" on one side and "7.0" on the other — + // invisible to numeric comparison, visible the moment an operator uses the + // column as a label (a trace name, a legend entry, hover text). + private def doubleColumns(input: Path): Seq[String] = + columnsOfType(input, AttributeType.DOUBLE) + + private def columnsOfType(input: Path, attributeType: AttributeType): Seq[String] = + scala.util + .Try(TupleIO.readSchemaSidecar(input)) + .toOption + .toSeq + .flatMap( + _.getAttributes.filter(_.getType == attributeType).map(_.getName) + ) + + // Python string literal, single-quoted with backslashes escaped. We + // deliberately don't use repr() in Scala (no such thing) — JSON.toString + // would also work but introduces double-quote escaping when the path has + // spaces. + private def py(s: String): String = + "'" + s.replace("\\", "\\\\").replace("'", "\\'") + "'" + + // Resolution chain mirrors the rest of the Texera test infra: env var first + // (set by CI / the shared-venv setup), then conventional names. + private def resolvePython(): String = { + val fromEnv = sys.env.get("UDF_PYTHON_PATH").filter(_.nonEmpty) + fromEnv.getOrElse { + // We don't try to probe `which` here — if neither env var nor a literal + // `python3.12` is on PATH, the subprocess invocation will fail and the + // error path below surfaces it. + "python3.12" + } + } +} + +final class StandaloneExecutionException( + val exitCode: Int, + val scriptPath: Path, + val source: String, + val stdout: String, + val stderr: String +) extends RuntimeException( + // The script path goes first in the message so a failing CI log makes it + // immediately obvious which file to open. stderr ends the message because + // the Python traceback (if any) is the most actionable signal. + s"""Standalone Python script exited with code $exitCode. + |Script: $scriptPath + |--- stdout --- + |$stdout + |--- stderr --- + |$stderr""".stripMargin + ) From 60fbc77992d04701842853a6f970347851708ce9 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 3 Sep 2026 15:45:59 -0700 Subject: [PATCH 2/2] test(verify): say it once, in the shape the code does not already give Three kinds of comment came out. A drawing of the string the code below assembles. A restatement of a branch the reader can see. And the word MVP, which dated the scope to a moment rather than stating it. What replaces them says the same thing shorter, or says what the code cannot: which cases the harness does not drive and why none of them has an operator asking for it. Co-Authored-By: Claude Opus 5 (1M context) --- .../translator/verify/StandaloneRunner.scala | 15 +++------------ 1 file changed, 3 insertions(+), 12 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneRunner.scala index 359bad51e5a..24a7c8fe405 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneRunner.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneRunner.scala @@ -34,18 +34,9 @@ import scala.sys.process._ * captures its DataFrame outputs as JSONL files (compatible with * [[TupleIO]]'s sidecar-schema format on the comparison side). * - * Wraps the operator's raw generated code with: - * - * ── prologue ────────────────────────────────────────────── - * in1df = pd.read_json("input_port_0.jsonl", lines=True) - * in2df = pd.read_json("input_port_1.jsonl", lines=True) - * ... - * inAlldf = [in1df, in2df] - * ── operator body (verbatim from generateStandaloneCode) ── - * out1df = in1df[in1df["age"] > 18] - * ── epilogue ───────────────────────────────────────────── - * out1df.to_json("output_port_0.jsonl", orient='records', lines=True) - * ... + * The operator's code is wrapped in a prologue that reads each input file into + * an `inNdf` and an epilogue that writes each `outNdf` back out, with the + * generated body verbatim between them. * * Port indexing matches the placeholder convention used by the translator: * `inNdf`/`outNdf` is 1-based and corresponds to the operator's N-th external