carloea2 commented on code in PR #8358: URL: https://github.com/apache/texera/pull/8358#discussion_r4253847542
########## workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/HarnessSpec.scala: ########## @@ -0,0 +1,368 @@ +/* + * 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.virtualidentity.{ExecutionIdentity, WorkflowIdentity} +import org.apache.texera.amber.core.workflow.{InputPort, OutputPort, PhysicalOp, PortIdentity} +import org.apache.texera.amber.operator.distinct.DistinctOpDesc +import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo} +import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator} +import org.scalatest.Tag +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Path} +import java.util.Base64 +import scala.sys.process._ + +/** 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) + } + + /** A source, in the one respect this spec is about: it reads no input port and + * names the file it reads by placeholder, leaving the naming to whoever + * assembles the script. + */ + private class StubSource(path: Path) extends LogicalOp with StandaloneCodeGenerator { + override def getPhysicalOp( + workflowId: WorkflowIdentity, + executionId: ExecutionIdentity + ): PhysicalOp = + throw new UnsupportedOperationException("the harness never builds a physical op") + + override def operatorInfo: OperatorInfo = + OperatorInfo( + "StubSource", + "Stands in for a source that names its file by placeholder", + OperatorGroupConstants.INPUT_GROUP, + inputPorts = List.empty, + outputPorts = List(OutputPort()) + ) + + override def standaloneSourcePath(): Option[String] = Some(path.toUri.toString) + + override def generateStandaloneCode(): String = + s"out1df = pd.read_json(${StandaloneCodeGenerator.SourceFilePlaceholder}, lines=True)" + } + + /** Reports the Python type of each cell of its `blob` column. */ + private class BlobTypeOp extends LogicalOp with StandaloneCodeGenerator { + override def getPhysicalOp( + workflowId: WorkflowIdentity, + executionId: ExecutionIdentity + ): PhysicalOp = + throw new UnsupportedOperationException("the harness never builds a physical op") + + override def operatorInfo: OperatorInfo = + OperatorInfo( + "BlobType", + "Reports the type of each binary cell", + OperatorGroupConstants.UTILITY_GROUP, + inputPorts = List(InputPort()), + outputPorts = List(OutputPort()) + ) + + override def generateStandaloneCode(): String = + """out1df = pd.DataFrame({"kind": [type(_v).__name__ for _v in in1df["blob"]]})""" + } + + /** Calls predict on each cell of its `blob` column, as a model's consumer does. */ + private class ModelPredictOp extends BlobTypeOp { + override def generateStandaloneCode(): String = + """out1df = pd.DataFrame({"kind": [str(_v.predict([[1.0]])[0]) for _v in in1df["blob"]]})""" + } + + // A fitted DecisionTreeClassifier behind the marker the cast writes, pickled + // by the interpreter the script runs so the two share a sklearn. + private def fittedTreeCell(): Array[Byte] = { + val script = + """import base64, pickle, sys + |from sklearn.tree import DecisionTreeClassifier + |model = DecisionTreeClassifier(random_state=0).fit([[0.0], [1.0]], [0, 1]) + |sys.stdout.write(base64.b64encode(b"pickle " + pickle.dumps(model)).decode("ascii")) + |""".stripMargin + val encoded = Process(Seq(PyOpExecHarness.resolvePython(), "-c", script)).!! Review Comment: This copy of the model fixture has the same Windows quoting failure as #8357: Python receives the truncated bytes literal and raises SyntaxError before the runner starts. It is the only failure among the nine existing harness cases. Please run the helper from a temporary Python file here too. ########## workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneRunner.scala: ########## @@ -0,0 +1,638 @@ +/* + * 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, Schema} +import org.apache.texera.amber.core.workflow.PortIdentity +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). + * + * 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. `N` is 1-based and counts the + * operator's external ports in declaration order, the translator's own + * convention, so the caller has to hand inputs over in that order. + */ +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`. + * 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(), + exactIntegers: Boolean = false, + outputSchemas: Map[PortIdentity, Schema] = Map.empty + ): 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 inputSchemas = schemasOf(inputs) + val source = + renderScript( + // The sidecar says what each column was DECLARED as, which the JSONL + // cannot carry. + gen.generateStandaloneCode(inputSchemas), + inputs, + outputPaths, + gen.standaloneHelpers(), + gen.standaloneImports(), + exactIntegers, + integralOutputColumns(outputSchemas, outputPaths.keys.toSeq), + gen.standaloneSourceName() + ) + 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], + imports: Seq[String], + exactIntegers: Boolean, + integralOutputs: Map[Int, Seq[String]], + sourceFileName: Option[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: nothing beyond pandas is injected here. The production translator + // (WorkflowToPythonTranslator) emits pandas for every script and then only + // what the operators in the plan ask for, so an operator whose standalone + // code needs numpy, or plotly, must say so. Injecting either here would mask + // that class of bug: the script would run in verify and fail on export. + sb.append("import pandas as pd\n") + imports.foreach(line => sb.append(line).append("\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") + // A numpy scalar is unwrapped rather than pickled, and pd.NA is written as + // null. An operator that rebuilds rows out of an integer column hands back + // neither a Python scalar nor None, so the branch below would put a base64 + // pickle into a column the schema declares a number. + sb.append(" def _texera_plain(_v):\n") + sb.append(" if _v is pd.NA:\n") + sb.append(" return None\n") + sb.append(" if hasattr(_v, 'item') and getattr(_v, 'ndim', None) == 0:\n") + sb.append(" return _v.item()\n") + sb.append(" return _v\n") + sb.append(" for _c in df.columns:\n") + sb.append(" if df[_c].dtype == object:\n") + sb.append(" df[_c] = df[_c].map(_texera_plain)\n") + // Bytes are written as the base64 the engine's writer writes, not as a + // base64 pickle of them: the prologue hands a BINARY column over as bytes, + // the way the engine hands one to an operator, and the branch below would + // otherwise answer a column the run wrote as `aGk=` with a pickle of it. + sb.append( + " df[_c] = df[_c].map(lambda _v: base64.b64encode(_v).decode('ascii') " + + "if isinstance(_v, (bytes, bytearray)) else _v)\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") + + // A model column arrives as the cast's pickle marker followed by the pickle, + // and the worker's ArrowTableTupleProvider unpickles such a cell before the + // operator sees it. Same test and slice, so the script is handed the model + // the run was handed and any other bytes stay bytes. + sb.append("def _texera_read_binary(_v):\n") + sb.append(" if pd.isna(_v):\n") + sb.append(" return None\n") + sb.append(" _b = base64.b64decode(_v)\n") + sb.append(" return pickle.loads(_b[10:]) if _b[:6] == b'pickle' else _b\n") + sb.append("\n") + + // Both paths have to be handed the same numbers. `read_json` parses a column + // holding a null through float64, so a LONG of 9007199254740993 arrives as + // 9007199254740992 while the engine still has the tuple. Python's json reads + // it exactly. Only for a JVM Path A: a Python operator's own table goes + // through pandas too, so there the float is what BOTH sides see. + if (exactIntegers) { + sb.append("def _texera_exact_ints(_path, _columns):\n") + sb.append(" _values = {_c: [] for _c in _columns}\n") + sb.append(" with open(_path, 'r', encoding='utf-8') as _f:\n") + sb.append(" for _line in _f:\n") + sb.append(" _line = _line.strip()\n") + sb.append(" if not _line:\n") + sb.append(" continue\n") + sb.append(" _row = json.loads(_line)\n") + sb.append(" for _c in _columns:\n") + sb.append(" _v = _row.get(_c)\n") + sb.append(" _values[_c].append(pd.NA if _v is None else int(_v))\n") + sb.append(" return {_c: pd.array(_v, dtype='Int64') for _c, _v in _values.items()}\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. + // + // The fraction is built from the nanoseconds and not from %f, which stops at + // six digits, while Timestamp.toString writes all nine. + sb.append("def _texera_ts_str(_v):\n") + sb.append(" if pd.isna(_v):\n") + sb.append(" return None\n") + sb.append(" _f = f'{_v.microsecond * 1000 + _v.nanosecond:09d}'.rstrip('0') or '0'\n") + sb.append(" return _v.strftime('%Y-%m-%d %H:%M:%S.') + _f\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") + + // A TIMESTAMP is a java.sql.Timestamp on the engine side, which holds the + // year 2500 as readily as 2024. pandas' default nanoseconds reach only 1677 + // to 2262, so a column outside that window is read at microseconds instead + // of failing before the operator runs. A column inside it keeps the + // nanoseconds, which Timestamp counts too. + // + // Cell by cell through dateutil and not astype: astype takes its unit from + // the text, so a fraction of nine digits is parsed as nanoseconds again and + // fails the same way. A datetime holds microseconds, so the rest is dropped. + sb.append("def _texera_read_ts(_s):\n") + sb.append(" try:\n") + sb.append(" return pd.to_datetime(_s)\n") + sb.append(" except pd.errors.OutOfBoundsDatetime:\n") + sb.append(" from dateutil.parser import parse as _parse_date\n") + sb.append(" _read = _s.map(lambda _v: None if pd.isna(_v) else _parse_date(_v))\n") + sb.append(" return _read.astype('datetime64[us]')\n") Review Comment: This still narrows timestamps inside pandas' range when another row is outside it. I ran Distinct on 2024-01-01 00:00:00.123456001, 2024-01-01 00:00:00.123456789 and 2500-01-01 00:00:00.0. The native harness keeps three rows; the standalone harness rounds both 2024 values to .123456 and keeps two. Without the 2500 row, both paths keep the two precise values. Please preserve those input values and add a mixed-range regression so the harness does not change an operator's answer before comparing it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
