From f9f6877d200ace115d03a01ea7ed803f73ff7b3a Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 2 Sep 2026 16:30:43 -0700 Subject: [PATCH 01/17] test(verify): run the source operators, and report what is covered A source reads a file rather than an upstream port, so it is configured from a file the runner writes rather than from a table, and the variants follow the configuration: a CSV with a header and without, an encoding per charset. Two reports come with it. One prints which operators run and which are withheld and why, so the state of the set is a command rather than a claim. The other parses every generated script and fails on one that a hostile column name would break, which is a whole class of defect a comparison cannot see. Co-Authored-By: Claude Opus 5 (1M context) --- .../verify/ConfigCoverageSpec.scala | 122 +++++ .../verify/SourceCategoryRunner.scala | 471 ++++++++++++++++++ .../verify/StandaloneEscapingCheck.scala | 156 ++++++ 3 files changed, 749 insertions(+) create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ConfigCoverageSpec.scala create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ConfigCoverageSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ConfigCoverageSpec.scala new file mode 100644 index 00000000000..a2a55e4235a --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ConfigCoverageSpec.scala @@ -0,0 +1,122 @@ +/* + * 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.operator.LogicalOp +import org.apache.texera.amber.operator.PythonOperatorDescriptor +import org.apache.texera.amber.operator.source.SourceOperatorDescriptor +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +/** + * Reports the harness verification tier per discovered [[StandaloneCodeGenerator]] + * operator: RUNNABLE (auto) / RUNNABLE (curated) / FLAG (reason). The `info` + * table printed by this spec is the coverage artifact shown at the handoff demo. + * + * Hard-asserts the must-run set: the operators that must appear as RUNNABLE for + * the demo to be credible. Flagged operators are always reported with a reason — + * never silently passed, and neither is a single withheld KIND of run + * (see [[reportWithheldRuns]]). + */ +class ConfigCoverageSpec extends AnyFlatSpec with Matchers { + + // The ops the demo must show as runnable; extend as triage flips flags. + private val mustRun = Set( + "IntersectOpDesc", + "DifferenceOpDesc", + "SymmetricDifferenceOpDesc", + "HashJoinOpDesc", + "SpecializedFilterOpDesc", + "SortOpDesc", + "LimitOpDesc" + ) + + "the harness" should "classify every discovered operator into a tier and report coverage" in { + val operators = OperatorBehaviorSpec.discoverStandaloneOperators() + + val rows = operators.map { opClass => + val name = opClass.getSimpleName + val kind = + if (classOf[SourceOperatorDescriptor].isAssignableFrom(opClass)) "source" + else if (classOf[PythonOperatorDescriptor].isAssignableFrom(opClass)) "python-udf" + else "jvm" + val tier = + if (kind == "source") { + if (SourceCategoryRunner.canRun(opClass)) + s"RUNNABLE (${SourceCategoryRunner.tier(opClass)})" + else s"FLAG (${SourceCategoryRunner.flagReason(opClass)})" + } else + TransformVerificationRunner.disposition(opClass) match { + case TransformVerificationRunner.Runnable(t) => s"RUNNABLE ($t)" + case TransformVerificationRunner.Flagged(reason) => s"FLAG ($reason)" + } + (name, kind, tier) + } + + val runnable = rows.count(_._3.startsWith("RUNNABLE")) + info(s"Coverage: $runnable/${rows.size} operators runnable, ${rows.size - runnable} flagged") + Seq("jvm", "python-udf", "source").foreach { k => + val of = rows.filter(_._2 == k) + info(s" $k: ${of.count(_._3.startsWith("RUNNABLE"))}/${of.size} runnable") + } + rows.sortBy { case (n, k, t) => (!t.startsWith("RUNNABLE"), k, n) }.foreach { + case (name, kind, tier) => info(f" $tier%-50s [$kind%-10s] $name") + } + + reportWithheldRuns(operators) + + val failedTargets = rows.collect { + case (name, _, tier) if mustRun.contains(name) && !tier.startsWith("RUNNABLE") => + s"$name → $tier" + } + withClue(s"must-run operators not runnable: $failedTargets") { + failedTargets shouldBe empty + } + } + + /** RUNNABLE is per operator, but a runnable operator can still be missing one + * KIND of run. Report those too, or the table reads as fuller than it is. + * + * Split by whether anyone should be waiting: a pending fix is a line someone + * deletes when an issue closes, by design is an answer. Both name the operator, + * so a family entry expands to the estimators it actually covers. + */ + private def reportWithheldRuns(operators: Seq[Class[_ <: LogicalOp]]): Unit = { + import TransformVerificationRunner._ + + val withheld = for { + opClass <- operators + (kind, reason) <- withheldRunsFor(opClass) + } yield (opClass.getSimpleName, kind, reason) + + val pending = withheld.collect { case (n, k, PendingFix(issue)) => (n, k, issue) } + val byDesign = withheld.collect { case (n, k, ByDesign(why)) => (n, k, why) } + info( + s"Runs withheld: ${pending.size} pending a fix, " + + s"${byDesign.size} not applicable by design" + ) + pending.sorted.foreach { + case (name, kind, issue) => info(f" PENDING $kind%-20s $name%-45s $issue") + } + byDesign.sorted.foreach { + case (name, kind, why) => info(f" BY-DESIGN $kind%-20s $name%-45s $why") + } + } +} diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala new file mode 100644 index 00000000000..01c13bf777a --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala @@ -0,0 +1,471 @@ +/* + * 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.{Schema, Tuple} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.operator.LogicalOp +import org.apache.texera.amber.operator.source.fetcher.URLFetcherOpDesc +import org.apache.texera.amber.operator.source.scan.ScanSourceOpDesc +import org.apache.texera.amber.operator.source.scan.file.{FileScanOpDesc, FileScanSourceOpDesc} +import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc +import com.fasterxml.jackson.databind.node.ObjectNode +import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.apache.arrow.memory.RootAllocator +import org.apache.arrow.vector.ipc.ArrowFileWriter +import org.apache.arrow.vector.VectorSchemaRoot +import org.apache.texera.amber.util.ArrowUtils + +import java.nio.channels.FileChannel +import java.nio.charset.{Charset, StandardCharsets} +import java.nio.file.{Files, Path, StandardOpenOption} +import scala.collection.mutable +import scala.util.{Try, Using} + +/** + * Per-category test runner for source operators (operators with no input + * ports — they read from an external resource and emit tuples). + * + * Dispatch is auto-first: + * - Auto tier: a scan source declares the file format it reads via + * [[ScanSourceOpDesc.fileTypeName]]. If that tag is in [[encoderByFileType]], + * the operator is fixtured with zero per-operator code — the shared + * [[CanonicalSourceFixture]] is encoded into that format and `fileName` + * points at it. A newly added file-scan source in a known format + * (CSV/JSONL/Arrow/…) is verified the moment it is registered in + * [[LogicalOp]]'s `@JsonSubTypes`, no edit here. + * - Curated tier: sources that can't take the shared table (text-family + * single-`line` output, or inline-config data) keep a hand-written + * [[SourceHandler]] in [[curatedHandlersByClass]]. + * - Otherwise the test is flagged (a [[knownIssues]] reason, an unsupported + * declared format, or no match) — never silently skipped. + * + * The runner itself is operator-agnostic: it builds an OpDesc, drives + * [[OpExecHarness]] (Path A) and [[StandaloneRunner]] (Path B), compares via + * [[Comparator]]. Sources have no input ports so `inputs = Map.empty` for both. + */ +object SourceCategoryRunner { + + /** + * The curated tier: sources that keep a hand-written handler because they + * can't go through the shared-fixture + encoder (auto) path — their output + * isn't the shared 3-column table (text-family, single `line` column) or + * their data is inline config rather than a file. Mirrors the transform + * side's [[CuratedHandlers]] (hand-written vs auto-generated fixture). + */ + private val curatedHandlersByClass: Map[Class[_ <: LogicalOp], SourceHandler] = + Seq[SourceHandler](TextInputHandler, FileScanSourceHandler) + .map(h => h.opDescClass -> h) + .toMap + + /** + * The auto tier. A scan source declares the file format it reads via + * [[ScanSourceOpDesc.fileTypeName]] ("CSV", "JSONL", "Arrow", …). Map that + * tag to the [[CanonicalSourceFixture]] encoder that writes a file in that + * format. Any source whose `fileTypeName` is a key here runs with zero + * per-operator code, so a newly added file-scan source in a known format is + * verified the moment it is registered in `@JsonSubTypes` — no handler, no + * edit here. (ParallelCSV also declares "CSV" and would be covered for free, + * but it is currently commented out of `@JsonSubTypes`, so the suite doesn't + * enumerate it.) + */ + private val encoderByFileType: Map[String, (Path, Charset) => Path] = Map( + "CSV" -> CanonicalSourceFixture.writeCsv, + "CSVOld" -> CanonicalSourceFixture.writeCsv, + "JSONL" -> CanonicalSourceFixture.writeJsonl, + // Arrow is binary and its descriptor declares fileEncoding ignored, so the + // charset a variant asks for has nothing to apply to. + "Arrow" -> ((dir, _) => CanonicalSourceFixture.writeArrow(dir)) + ) + + /** + * Sources this runner cannot verify, with the honest reason. Mirrors + * `TransformVerificationRunner.knownIssues`: the reason surfaces in the + * ignored test's name and the coverage table. + */ + private val knownIssues: Map[Class[_ <: LogicalOp], String] = Map( + classOf[FileScanOpDesc] -> + ("input-driven source: filenames arrive on an input port at runtime, but this runner " + + "feeds sources no inputs — Path B's generated code references an undefined in1df"), + classOf[URLFetcherOpDesc] -> + ("live-network source: the operator fetches a real URL over the network, so its " + + "output is non-deterministic and depends on external connectivity — it cannot be " + + "verified against a fixed fixture in isolation") + ) + + /** The format tag a source declares, or `None` if it isn't an instantiable + * ScanSourceOpDesc (non-scan sources, or ones that fail to construct). + */ + private def declaredFileType(opDescClass: Class[_ <: LogicalOp]): Option[String] = + Try(opDescClass.getDeclaredConstructor().newInstance()).toOption.collect { + case scan: ScanSourceOpDesc => scan.fileTypeName + }.flatten + + def canRun(opDescClass: Class[_ <: LogicalOp]): Boolean = + curatedHandlersByClass.contains(opDescClass) || + declaredFileType(opDescClass).exists(encoderByFileType.contains) + + /** + * Tier label for a runnable source, mirroring the transform side's + * auto/curated distinction: `"curated source"` when a hand-written + * [[SourceHandler]] serves it, else `"auto source"` (a declared-format scan + * source fixtured by an [[encoderByFileType]] encoder with zero per-op code). + */ + def tier(opDescClass: Class[_ <: LogicalOp]): String = + if (curatedHandlersByClass.contains(opDescClass)) "curated source" else "auto source" + + /** Why a non-runnable source is flagged: a specific known issue, an + * unsupported declared format, or no handler/format match at all. + */ + def flagReason(opDescClass: Class[_ <: LogicalOp]): String = + knownIssues.getOrElse( + opDescClass, + declaredFileType(opDescClass) match { + case Some(fileType) => + s"unsupported source format '$fileType' — no encoder registered in SourceCategoryRunner" + case None => "no source handler registered yet" + } + ) + + private def newScanSource(opDescClass: Class[_ <: LogicalOp]): ScanSourceOpDesc = + opDescClass.getDeclaredConstructor().newInstance() match { + case s: ScanSourceOpDesc => s + case other => + throw new IllegalArgumentException( + s"${opDescClass.getSimpleName} has no curated handler and is not a " + + s"ScanSourceOpDesc (${other.getClass.getName})" + ) + } + + /** + * Every configuration of one source worth running, as (label, op, its own + * directory). + * + * Each variant gets a directory of its own holding its OWN copy of the fixture, + * because the generated script reads the file by bare name (`pd.read_csv( + * "sample.csv")`) out of the directory it runs in. Two variants wanting two + * different `sample.csv` files cannot share one. + */ + private def variantsFor( + opDescClass: Class[_ <: LogicalOp], + testRoot: Path + ): Seq[(String, LogicalOp, Path)] = { + // Punctuation collapses to '_', so two labels differing only in punctuation + // would name the same directory and share one fixture and one output dir. Fail + // loudly instead of letting a variant quietly run someone else's file. + val taken = mutable.Set.empty[String] + def dirFor(label: String): Path = { + val name = label.replaceAll("[^A-Za-z0-9]+", "_") + require(taken.add(name), s"two variants of $opDescClass both map to the directory '$name'") + Files.createDirectories(testRoot.resolve(name)) + } + + curatedHandlersByClass.get(opDescClass) match { + case Some(handler) => + val baseDir = dirFor("default") + val base = handler.makeOpDesc(baseDir) + // Every variant calls the handler AGAIN rather than reusing `base`: the handler + // writes its fixture into the directory it is given, and a second op carrying + // the first one's `fileName` would read a file outside the directory it runs in. + // No enum sweep: both curated sources are the text family, whose `attributeType` + // says how to PARSE the fixture (`alice` is not an integer) and whose + // `fileEncoding` describes its BYTES — flipping either without rewriting the + // fixture compares nothing but how the two paths fail. The auto branch below + // rewrites its fixture per variant and does sweep them. + ConfigGenerator + .fullVariantEditsOf(base, Map.empty, handler.rowCount, sweepEnums = false) + .fold( + reason => + throw new IllegalStateException( + s"cannot vary ${opDescClass.getSimpleName}: $reason" + ), + identity + ) + .map { variant => + if (variant.at.isEmpty) ("default", base, baseDir) + else { + val dir = dirFor(variant.label) + val op = ConfigGenerator + .applyVariant(handler.makeOpDesc(dir), variant) + .fold( + reason => + throw new IllegalStateException( + s"cannot build ${opDescClass.getSimpleName} variant '${variant.label}': $reason" + ), + identity + ) + (variant.label, op, dir) + } + } + case None => + val fileType = declaredFileType(opDescClass).getOrElse("") + val encoder = encoderByFileType.getOrElse( + fileType, + throw new IllegalArgumentException( + s"No encoder for ${opDescClass.getSimpleName} (fileTypeName='$fileType')" + ) + ) + val base = { + val dir = dirFor("default") + val op = newScanSource(opDescClass) + op.fileName = Some(encoder(dir, op.fileEncoding.getCharset).toUri.toString) + ("default", op: LogicalOp, dir) + } + base +: generatedVariants(opDescClass, encoder, dirFor) + } + } + + /** + * The variants the shared [[ConfigGenerator]] derives from the operator's own + * fields — the base config with every knob filled, plus one per enum branch + * (`hasHeader`, JSONL's `flatten`). Nothing to register per operator: a knob + * added to a source is swept the day it is added. + * + * `fileEncoding` is swept like any other enum, and the fixture FOLLOWS it: each + * variant's file is written in the charset that variant declares. Encoding is a + * statement about the bytes, so a UTF_16 config over a file left in UTF-8 would + * only compare how each path fails. + * + * Variants that serialize identically are dropped — an operator that ignores + * `fileEncoding` (Arrow declares `@JsonIgnoreProperties`) would otherwise run the + * same config three times. + */ + private def generatedVariants( + opDescClass: Class[_ <: LogicalOp], + encoder: (Path, Charset) => Path, + dirFor: String => Path + ): Seq[(String, LogicalOp, Path)] = { + val seen = mutable.Set.empty[String] + ConfigGenerator + .generateVariants(opDescClass, Map.empty, CanonicalSourceFixture.rows.size) + .fold( + reason => + throw new IllegalStateException( + s"cannot auto-configure ${opDescClass.getSimpleName}: $reason" + ), + identity + ) + .flatMap { + case (label, op) => + val scan = op.asInstanceOf[ScanSourceOpDesc] + val shape = objectMapper.valueToTree[ObjectNode](scan) + shape.remove("fileName") // every variant reads its own copy of the file + if (!seen.add(shape.toString)) None + else { + // "default" is already the bare newInstance config above; this one is + // the generator's, which additionally fills limit and offset. + val name = if (label == "default") "auto-base" else label + val dir = dirFor(name) + scan.fileName = Some(encoder(dir, scan.fileEncoding.getCharset).toUri.toString) + Some((name, scan: LogicalOp, dir)) + } + } + } + + /** Runs the parity test for the operator, once per variant. Throws on mismatch. */ + def run(opDescClass: Class[_ <: LogicalOp]): Unit = { + val testRoot = Files.createTempDirectory(s"op-behavior-${opDescClass.getSimpleName}-") + variantsFor(opDescClass, testRoot).foreach { + case (label, opDesc, workDir) => + try runVariant(opDesc, workDir) + catch { + case e: Throwable => + throw new AssertionError(s"[variant: $label] ${e.getMessage}", e) + } + } + } + + /** Drive one configured source through both paths inside `workDir`, which holds + * that variant's fixture, and assert the two tables match. + */ + private def runVariant(opDesc: LogicalOp, workDir: Path): Unit = { + val actualDir = workDir.resolve("actual") + Files.createDirectories(actualDir) + + val pathA = OpExecHarness.execute(opDesc, inputs = Map.empty, outputDir = actualDir) + val pathB = StandaloneRunner.run( + opDesc = opDesc, + inputs = Map.empty, + outputPortCount = 1, + workDir = workDir + ) + + val actual = pathA.outputs(PortIdentity(0)) + val expected = pathB.outputs(1) + Comparator.assertEqual(actual, expected) + } +} + +/** + * A hand-written recipe for one source that can't use the auto tier + * (fileTypeName + [[CanonicalSourceFixture]] encoder): which OpDesc class it + * handles and how to fixture a working instance. Used for the text-family + * sources ([[TextInputHandler]], [[FileScanSourceHandler]]). + */ +trait SourceHandler { + + /** The concrete OpDesc class this handler tests. */ + def opDescClass: Class[_ <: LogicalOp] + + /** + * Generate the fixture file inside `testRoot` and return a configured + * OpDesc instance whose `fileName` (or analogous URI field) points at it. + */ + def makeOpDesc(testRoot: Path): LogicalOp + + /** How many rows the fixture holds. Only the handler knows — it writes its own, + * rather than the shared [[CanonicalSourceFixture]]. A row-window knob the + * variants fill (`limit`, `offset`) is sized against this, so that the value they + * take keeps some rows and drops some instead of landing past the end. + */ + def rowCount: Int +} + +/** + * The rows every structured-file source reads: [[CanonicalFixture]]'s, whole. + * + * A source has no input port, so the fixture is delivered not as an input JSONL + * but as a file the operator opens itself. Each `writeXxx` encodes these rows + * into one on-disk format (CSV / JSONL / Arrow); a source handler picks the + * encoder its operator understands and points `fileName` at the result. So CSV, + * CSVOld, JSONL and Arrow all verify that the operator reconstructs one shared + * table, instead of each asserting against its own ad-hoc sample. + * + * It reads the canonical table rather than a narrow one of its own. A source + * fixture picked for the types that survive a round trip would be choosing not + * to ask the question this suite exists to ask: these files carry no types, both + * readers infer, and where they infer differently is exactly what should show. A + * date column does part them, and [[StandaloneRunner.sourceCasts]] is where that + * is settled — on Path B's reading, not by leaving the column out. + */ +object CanonicalSourceFixture { + + val schema: Schema = CanonicalFixture.schema + + val rows: Vector[Tuple] = CanonicalFixture.allRows + + /** Write the rows as a header-first, comma-delimited CSV encoded in `charset`. + * + * The charset is a parameter because it describes the BYTES, not the config: a + * variant declaring `fileEncoding = UTF_16` over a file left in UTF-8 would + * compare nothing but how each path fails. + */ + def writeCsv(dir: Path, charset: Charset): Path = { + val path = dir.resolve("sample.csv") + val header = schema.getAttributes.map(a => csvField(a.getName)).mkString(",") + val body = rows.map { t => + schema.getAttributes + .map(a => csvField(Option(t.getField[AnyRef](a.getName)).map(_.toString).orNull)) + .mkString(",") + } + Files.write(path, ((header +: body).mkString("\n") + "\n").getBytes(charset)) + path + } + + /** One CSV field, quoted per RFC 4180. + * + * The table carries commas inside values — a bracketed edge pair, a + * comma-delimited list, an ordinary English sentence — and writing those raw + * shifts every column after them. What the two paths then disagree about is a + * broken file rather than anything either of them does. + */ + private def csvField(value: String): String = + if (value == null) "" + else if (value.exists(c => c == ',' || c == '"' || c == '\n' || c == '\r')) + "\"" + value.replace("\"", "\"\"") + "\"" + else value + + /** Write the rows as JSON Lines (one object per line, keys in schema order). + * Reuses [[TupleIO.writeTuples]] — the same writer the transform fixtures + * use; it also drops a `.schema.json` sidecar the source ignores. + * + * That writer is shared and always writes UTF-8, so a variant asking for another + * charset gets the bytes transcoded afterwards rather than a second writer. + */ + def writeJsonl(dir: Path, charset: Charset): Path = { + val path = dir.resolve("sample.jsonl") + TupleIO.writeTuples(path, rows.iterator, schema) + if (charset != StandardCharsets.UTF_8) { + val text = new String(Files.readAllBytes(path), StandardCharsets.UTF_8) + Files.write(path, text.getBytes(charset)) + } + path + } + + /** Write the rows as an uncompressed Arrow IPC ("file" format) stream — the + * format both `ArrowFileReader` (Path A) and `pd.read_feather` (Path B) + * read. + */ + def writeArrow(dir: Path): Path = { + val path = dir.resolve("sample.arrow") + // Texera's own Schema-to-Arrow mapping and tuple writer, so the file carries + // exactly the types `ArrowUtils.toTexeraSchema` reads back on the other side. + // Hand-listing the fields is what let the table outgrow them unnoticed: the + // columns past the list were simply not written, and both paths went on + // agreeing about the few that were. + val arrowSchema = ArrowUtils.fromTexeraSchema(schema) + Using.Manager { use => + val allocator = use(new RootAllocator()) + val root = use(VectorSchemaRoot.create(arrowSchema, allocator)) + root.allocateNew() + rows.zipWithIndex.foreach { case (t, i) => ArrowUtils.setTexeraTuple(t, i, root) } + root.setRowCount(rows.size) + val channel = use( + FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.WRITE) + ) + val writer = use(new ArrowFileWriter(root, null, channel)) + writer.start() + writer.writeBatch() + writer.end() + }.get + path + } +} + +/** Handler for `TextInputSourceOpDesc`. The text lives in the config — no fixture file. */ +object TextInputHandler extends SourceHandler { + + override val opDescClass: Class[_ <: LogicalOp] = classOf[TextInputSourceOpDesc] + + override val rowCount: Int = 3 + + override def makeOpDesc(testRoot: Path): LogicalOp = { + val desc = new TextInputSourceOpDesc() + desc.textInput = "alice\nbob\ncarol" + desc // defaults: attributeType STRING (one row per line), attributeName "line" + } +} + +/** Handler for `FileScanSourceOpDesc`. Plain text file read in default line mode. */ +object FileScanSourceHandler extends SourceHandler { + + override val opDescClass: Class[_ <: LogicalOp] = classOf[FileScanSourceOpDesc] + + override val rowCount: Int = 3 + + override def makeOpDesc(testRoot: Path): LogicalOp = { + val txtPath = testRoot.resolve("sample.txt") + Files.write(txtPath, "alice\nbob\ncarol\n".getBytes(StandardCharsets.UTF_8)) + + val desc = new FileScanSourceOpDesc() + desc.fileName = Some(txtPath.toUri.toString) + desc // defaults: attributeType STRING (one row per line), attributeName "line" + } +} diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala new file mode 100644 index 00000000000..5280b4cf1e7 --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala @@ -0,0 +1,156 @@ +/* + * 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.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.{ArrayNode, ObjectNode} +import org.apache.texera.amber.operator.source.SourceOperatorDescriptor +import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator} +import org.apache.texera.amber.util.JSONUtils.objectMapper + +import java.nio.charset.StandardCharsets.UTF_8 +import java.nio.file.{Files, Path} +import scala.jdk.CollectionConverters._ +import scala.sys.process.{Process, ProcessLogger} +import scala.util.Try + +/** + * Configures every operator against hostile column names and parses what + * `generateStandaloneCode` produces. A value spliced in without + * `pyStringLiteral` is silent until someone names a column `a"b`, and then the + * exported script will not compile. + * + * The check has to be behavioural. A generator that builds its quoted literal + * inside a helper shows no quotes in its template, which is how RadarPlot and + * Aggregate survived a source-level sweep that reported zero remaining sites. + */ +object StandaloneEscapingCheck { + + private val schemas = CanonicalFixture.schemasByPort + private val columns = CanonicalFixture.schema.getAttributes.map(_.getName).toSet + + /** Every problem found, empty when the suite is clean. An operator whose config + * cannot be built lands here too: dropping it would check less while still + * passing. + */ + def run(): Seq[String] = { + // Sources have no input schema and so no column knobs. Their free-text knobs + // already get a hostile variant in the normal verify run. + val operators = OperatorBehaviorSpec + .discoverStandaloneOperators() + .filterNot(classOf[SourceOperatorDescriptor].isAssignableFrom) + val dir = Files.createTempDirectory("standalone-escaping-") + dir.toFile.deleteOnExit() + + val (unconfigurable, code) = operators + .map(op => op.getSimpleName -> codeFor(op, dir)) + .partitionMap { + case (name, Left(why)) => Left(s"$name: $why") + case (name, Right(variants)) => + Right(variants.map { case (label, src) => s"$name/$label" -> src }) + } + unconfigurable ++ parse(code.flatten, dir) + } + + /** Holds the four characters that end a Python literal, and carries the column + * it replaces so that two knobs never collide on one name. + */ + private def hostile(column: String): String = "a\"b'c\\d\ne_" + column + + private def hostilize(node: JsonNode): Unit = + node match { + case obj: ObjectNode => + obj.fields().asScala.toSeq.foreach { e => + if (isColumn(e.getValue)) obj.put(e.getKey, hostile(e.getValue.asText)) + else hostilize(e.getValue) + } + case arr: ArrayNode => + (0 until arr.size).foreach { i => + if (isColumn(arr.get(i))) + arr.set(i, objectMapper.getNodeFactory.textNode(hostile(arr.get(i).asText))) + else hostilize(arr.get(i)) + } + case _ => () + } + + private def isColumn(n: JsonNode): Boolean = n.isTextual && columns.contains(n.asText) + + /** Split the way the runner splits: a curated operator's config is hand-written + * because the generator cannot derive one. Auto operators contribute every + * variant, since a knob only `optionals` fills is a site the base never reaches. + */ + private def codeFor( + opClass: Class[_ <: LogicalOp], + dir: Path + ): Either[String, Seq[(String, String)]] = { + val configs = CuratedHandlers.byClass.get(opClass) match { + // The handler's config, not its enum sweep: sweeping moves an enum value, + // never a column name, and would need schemas only the runner holds. + case Some(h) => + Try(Seq("curated" -> h.fixture(dir)._1)).toEither.left.map(e => s"curated: ${e.getMessage}") + case None => ConfigGenerator.generateVariants(opClass, schemas) + } + configs.flatMap { variants => + val results = variants.map { + case (label, op) => + val node = objectMapper.valueToTree[ObjectNode](op) + hostilize(node) + Try( + objectMapper + .treeToValue(node, opClass) + .asInstanceOf[StandaloneCodeGenerator] + .generateStandaloneCode() + ).toEither.left.map(e => s"$label: $e").map(label -> _) + } + results.collectFirst { case Left(why) => why }.toLeft(results.collect { case Right(r) => r }) + } + } + + /** One Python process for all of them; the cost is startup, not parsing. The + * snippets stay in `dir` so a reported operator can be opened as generated. + */ + private def parse(snippets: Seq[(String, String)], dir: Path): Seq[String] = { + val payload = objectMapper.createObjectNode() + snippets.foreach { case (name, src) => payload.put(name, src) } + val input = write(dir, "snippets.json", payload.toString) + val script = write( + dir, + "parse_all.py", + """import ast, json, sys + |for name, code in json.load(open(sys.argv[1])).items(): + | try: + | ast.parse(code) + | except SyntaxError as e: + | print(f"{name}: line {e.lineno}: {e.msg}") + |""".stripMargin + ) + val out = Seq.newBuilder[String] + val err = Seq.newBuilder[String] + val python = sys.env.get("UDF_PYTHON_PATH").filter(_.nonEmpty).getOrElse("python3.12") + val exit = Process(Seq(python, script.toString, input.toString)) + .!(ProcessLogger(out += _, err += _)) + // Without this a parser that never ran reads as "nothing failed". + require(exit == 0, s"parse_all.py exited $exit: ${err.result().mkString("\n")}") + out.result() + } + + private def write(dir: Path, name: String, content: String): Path = + Files.write(dir.resolve(name), content.getBytes(UTF_8)) +} From 59f79841f5ffdbde60f1db09510cf55ac4b31228 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 2 Sep 2026 17:17:36 -0700 Subject: [PATCH 02/17] test(verify): move the per-operator spec here, where its callees already are It routes a source to the source runner and asks the escaping check for its findings, and the coverage report in turn reads the operators it discovered. Split across two changes those references point both ways, and neither compiles until both land whatever order they land in. Co-Authored-By: Claude Opus 5 (1M context) --- .../verify/OperatorBehaviorSpec.scala | 151 ++++++++++++++++++ 1 file changed, 151 insertions(+) create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala new file mode 100644 index 00000000000..f6d8b6759be --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala @@ -0,0 +1,151 @@ +/* + * 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.fasterxml.jackson.annotation.JsonSubTypes +import org.apache.texera.amber.operator.{LogicalOp, StandaloneCodeGenerator} +import org.apache.texera.amber.operator.source.SourceOperatorDescriptor +import org.apache.texera.amber.translator.verify.tags.IntegrationTest +import org.scalatest.ParallelTestExecution +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +/** + * Auto-discovered behavioral-parity tests: for every operator registered + * with [[LogicalOp]]'s `@JsonSubTypes` that implements + * [[StandaloneCodeGenerator]], emit a test that runs both Path A (Texera + * exec) and Path B (translator-generated Python via [[StandaloneRunner]]) + * and asserts their outputs are equivalent. + * + * Dispatch is auto-first: [[TransformVerificationRunner]] classifies each + * non-source transform as `Runnable("auto")` (auto-configured fixture), + * `Runnable("curated")` (hand-written fixture from [[CuratedHandlers]]), + * or `Flagged(reason)` (shown as ignored with the reason in the test name). + * Sources route to [[SourceCategoryRunner]] unchanged. + * + * No edits to this spec are needed when a new operator is added — reflection + * discovers it automatically via `@JsonSubTypes`. The tier label appears in + * the test name so the report shows which path exercised each operator. + * + * Requires Python 3 with pandas on the [[Comparator]] / [[StandaloneRunner]] + * resolution chain (`UDF_PYTHON_PATH` env var, then `python3.12`). + */ +// Tagged @IntegrationTest: this is the only verify spec that forks a real +// Python process end-to-end, so CI routes it to the Python-provisioned +// integration job (see workflow-compiling-service/build.sbt WCS_TEST_FILTER). +@IntegrationTest +class OperatorBehaviorSpec extends AnyFlatSpec with Matchers with ParallelTestExecution { + + // Build the test list at class construction. Each branch below registers + // one test (`in` for runnable, `ignore` for skipped) so the test report + // shows every translator-eligible operator and why it did or didn't run. + OperatorBehaviorSpec.discoverStandaloneOperators().foreach { opClass => + val name = opClass.getSimpleName + + if (!OperatorBehaviorSpec.isSelected(name)) { + // Narrowed out by VERIFY_ONLY / VERIFY_SKIP, which only a local run sets. + // Still registered, as an `ignore`, so the report lists every operator + // rather than reading as though the narrowed-out ones do not exist. + name should "NARROWED OUT — outside this run's VERIFY_ONLY / VERIFY_SKIP" ignore {} + } else if (classOf[SourceOperatorDescriptor].isAssignableFrom(opClass)) { + // Sources keep their handler-per-source design: each needs a real file + // in its specific format, which a generic fixture can't supply. + if (SourceCategoryRunner.canRun(opClass)) { + name should "produce equivalent output in Texera and standalone Python (source)" in { + SourceCategoryRunner.run(opClass) + } + } else { + name should s"FLAGGED — ${SourceCategoryRunner.flagReason(opClass)}" ignore {} + } + } else { + TransformVerificationRunner.disposition(opClass) match { + case TransformVerificationRunner.Runnable(tier) => + name should s"produce equivalent output in Texera and standalone Python ($tier)" in { + TransformVerificationRunner.run(opClass) + } + case TransformVerificationRunner.Flagged(reason) => + name should s"FLAGGED — $reason" ignore { + // Reason is in the test name so the report carries it; the + // coverage table in ConfigCoverageSpec aggregates these. + } + } + } + } + + // Not one test per operator like the rest of this spec: it is one assertion + // over all of them, and it deliberately ignores the selection knobs above so a + // VERIFY_ONLY run still cannot hide a broken splice site. + "Generated standalone code" should "stay parseable when the column names are hostile" in { + StandaloneEscapingCheck.run() shouldBe empty + } +} + +object OperatorBehaviorSpec { + + // Narrowing knobs for a local run, both unset by default, so the default run + // is every operator: VERIFY_ONLY names the only ones to run, VERIFY_SKIP the + // ones to leave out. Case-sensitive substrings against the operator's simple + // name, comma-separated. Neither is set in CI, which therefore runs the lot. + // + // There is deliberately no third list withholding operators by default. What + // stays withheld is narrower than an operator and lives where it can say why: + // a single variant in [[TransformVerificationRunner.variantsNotRun]], or an + // operator that cannot be run at all in its `knownIssues`, each against an + // issue or a reason. A name here would withdraw an operator's every variant + // and record nothing about what is wrong with it. + private def patterns(envVar: String): Seq[String] = + sys.env.getOrElse(envVar, "").split(",").iterator.map(_.trim).filter(_.nonEmpty).toSeq + + private lazy val onlyPatterns: Seq[String] = patterns("VERIFY_ONLY") + private lazy val skipPatterns: Seq[String] = patterns("VERIFY_SKIP") + + /** True if `name` should run: in VERIFY_ONLY when that is set, and not in + * VERIFY_SKIP. True for everything when neither is set. + */ + def isSelected(name: String): Boolean = { + val included = onlyPatterns.isEmpty || onlyPatterns.exists(name.contains) + val excluded = skipPatterns.exists(name.contains) + included && !excluded + } + + /** + * Enumerates every concrete subclass of [[LogicalOp]] declared in its + * `@JsonSubTypes` annotation, filters to those implementing + * [[StandaloneCodeGenerator]], and returns them sorted by simple name + * (stable test report order). + * + * Uses the same registry Jackson uses to deserialize operators — no + * separate discovery mechanism needed. Adding an operator to + * `LogicalOp.@JsonSubTypes` makes it visible here automatically. + */ + def discoverStandaloneOperators(): Seq[Class[_ <: LogicalOp]] = { + val annotation = classOf[LogicalOp].getAnnotation(classOf[JsonSubTypes]) + if (annotation == null) Seq.empty + else + annotation + .value() + .toSeq + .map(_.value()) + .filter(classOf[StandaloneCodeGenerator].isAssignableFrom) + .map(_.asInstanceOf[Class[_ <: LogicalOp]]) + .distinct + .sortBy(_.getSimpleName) + } +} From c4bc40eabc487b1a1cf3c9221afbb4969f312fa3 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 3 Sep 2026 15:07:22 -0700 Subject: [PATCH 03/17] test(verify): keep the guard, drop the table it printed around it The spec printed a tier per operator, a per-kind tally, and every withheld run with its reason. That is a progress report: it read the same dispositions the run itself reads and asserted nothing about them, so it could only ever say what the run had already decided. What it did assert stays: a small set of operators that must not become unrunnable. They take the shared table as it is and nothing about them is hard to run, so one turning up flagged means something upstream of the disposition broke. Co-Authored-By: Claude Opus 5 (1M context) --- .../verify/ConfigCoverageSpec.scala | 100 ++++-------------- 1 file changed, 23 insertions(+), 77 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ConfigCoverageSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ConfigCoverageSpec.scala index a2a55e4235a..9165b44d599 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ConfigCoverageSpec.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/ConfigCoverageSpec.scala @@ -20,24 +20,23 @@ package org.apache.texera.amber.translator.verify import org.apache.texera.amber.operator.LogicalOp -import org.apache.texera.amber.operator.PythonOperatorDescriptor import org.apache.texera.amber.operator.source.SourceOperatorDescriptor import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers /** - * Reports the harness verification tier per discovered [[StandaloneCodeGenerator]] - * operator: RUNNABLE (auto) / RUNNABLE (curated) / FLAG (reason). The `info` - * table printed by this spec is the coverage artifact shown at the handoff demo. + * Holds the floor under what the harness can run. * - * Hard-asserts the must-run set: the operators that must appear as RUNNABLE for - * the demo to be credible. Flagged operators are always reported with a reason — - * never silently passed, and neither is a single withheld KIND of run - * (see [[reportWithheldRuns]]). + * Every operator carrying standalone code is classified runnable or flagged, and + * a flag is how an operator stops being checked at all. For most that is a state + * someone is working through. For the ones below it would be a regression: they + * are the plainest operators the harness has, they take the shared table as it + * is, and nothing about them is hard to run. An operator arriving here means + * something upstream of the disposition broke rather than that this operator + * became difficult. */ class ConfigCoverageSpec extends AnyFlatSpec with Matchers { - // The ops the demo must show as runnable; extend as triage flips flags. private val mustRun = Set( "IntersectOpDesc", "DifferenceOpDesc", @@ -48,75 +47,22 @@ class ConfigCoverageSpec extends AnyFlatSpec with Matchers { "LimitOpDesc" ) - "the harness" should "classify every discovered operator into a tier and report coverage" in { - val operators = OperatorBehaviorSpec.discoverStandaloneOperators() - - val rows = operators.map { opClass => - val name = opClass.getSimpleName - val kind = - if (classOf[SourceOperatorDescriptor].isAssignableFrom(opClass)) "source" - else if (classOf[PythonOperatorDescriptor].isAssignableFrom(opClass)) "python-udf" - else "jvm" - val tier = - if (kind == "source") { - if (SourceCategoryRunner.canRun(opClass)) - s"RUNNABLE (${SourceCategoryRunner.tier(opClass)})" - else s"FLAG (${SourceCategoryRunner.flagReason(opClass)})" - } else - TransformVerificationRunner.disposition(opClass) match { - case TransformVerificationRunner.Runnable(t) => s"RUNNABLE ($t)" - case TransformVerificationRunner.Flagged(reason) => s"FLAG ($reason)" - } - (name, kind, tier) - } - - val runnable = rows.count(_._3.startsWith("RUNNABLE")) - info(s"Coverage: $runnable/${rows.size} operators runnable, ${rows.size - runnable} flagged") - Seq("jvm", "python-udf", "source").foreach { k => - val of = rows.filter(_._2 == k) - info(s" $k: ${of.count(_._3.startsWith("RUNNABLE"))}/${of.size} runnable") - } - rows.sortBy { case (n, k, t) => (!t.startsWith("RUNNABLE"), k, n) }.foreach { - case (name, kind, tier) => info(f" $tier%-50s [$kind%-10s] $name") - } - - reportWithheldRuns(operators) - - val failedTargets = rows.collect { - case (name, _, tier) if mustRun.contains(name) && !tier.startsWith("RUNNABLE") => - s"$name → $tier" - } - withClue(s"must-run operators not runnable: $failedTargets") { - failedTargets shouldBe empty + "the harness" should "keep runnable every operator it must be able to run" in { + val flagged = OperatorBehaviorSpec + .discoverStandaloneOperators() + .filter(opClass => mustRun.contains(opClass.getSimpleName)) + .filterNot(runnable) + .map(_.getSimpleName) + withClue(s"must-run operators no longer runnable: $flagged") { + flagged shouldBe empty } } - /** RUNNABLE is per operator, but a runnable operator can still be missing one - * KIND of run. Report those too, or the table reads as fuller than it is. - * - * Split by whether anyone should be waiting: a pending fix is a line someone - * deletes when an issue closes, by design is an answer. Both name the operator, - * so a family entry expands to the estimators it actually covers. - */ - private def reportWithheldRuns(operators: Seq[Class[_ <: LogicalOp]]): Unit = { - import TransformVerificationRunner._ - - val withheld = for { - opClass <- operators - (kind, reason) <- withheldRunsFor(opClass) - } yield (opClass.getSimpleName, kind, reason) - - val pending = withheld.collect { case (n, k, PendingFix(issue)) => (n, k, issue) } - val byDesign = withheld.collect { case (n, k, ByDesign(why)) => (n, k, why) } - info( - s"Runs withheld: ${pending.size} pending a fix, " + - s"${byDesign.size} not applicable by design" - ) - pending.sorted.foreach { - case (name, kind, issue) => info(f" PENDING $kind%-20s $name%-45s $issue") - } - byDesign.sorted.foreach { - case (name, kind, why) => info(f" BY-DESIGN $kind%-20s $name%-45s $why") - } - } + private def runnable(opClass: Class[_ <: LogicalOp]): Boolean = + if (classOf[SourceOperatorDescriptor].isAssignableFrom(opClass)) + SourceCategoryRunner.canRun(opClass) + else + TransformVerificationRunner + .disposition(opClass) + .isInstanceOf[TransformVerificationRunner.Runnable] } From 2f66569ecdc7e8d068d00f9da7835a46bb31ffb2 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 3 Sep 2026 15:46:23 -0700 Subject: [PATCH 04/17] test(verify): say it once, in the shape the code does not already give Two comments here restated what the code beside them shows: the three tiers a source dispatches through, which `canRun` and `flagReason` already spell out, and this spec's own repetition of the runner's dispatch, which belongs to the runner. What replaces them says what each file is for, once. Co-Authored-By: Claude Opus 5 (1M context) --- .../verify/OperatorBehaviorSpec.scala | 10 ++++----- .../verify/SourceCategoryRunner.scala | 21 +++++++------------ 2 files changed, 13 insertions(+), 18 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala index f6d8b6759be..2f984a3ba3b 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala @@ -34,11 +34,11 @@ import org.scalatest.matchers.should.Matchers * exec) and Path B (translator-generated Python via [[StandaloneRunner]]) * and asserts their outputs are equivalent. * - * Dispatch is auto-first: [[TransformVerificationRunner]] classifies each - * non-source transform as `Runnable("auto")` (auto-configured fixture), - * `Runnable("curated")` (hand-written fixture from [[CuratedHandlers]]), - * or `Flagged(reason)` (shown as ignored with the reason in the test name). - * Sources route to [[SourceCategoryRunner]] unchanged. + * [[TransformVerificationRunner]] decides how each non-source transform is + * configured and whether it can run at all; sources go to + * [[SourceCategoryRunner]]. An operator it cannot run is registered as an + * ignored test carrying the reason, so the report lists every operator rather + * than reading as though the unrunnable ones do not exist. * * No edits to this spec are needed when a new operator is added — reflection * discovers it automatically via `@JsonSubTypes`. The tier label appears in diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala index 01c13bf777a..ab0561c0164 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala @@ -43,19 +43,14 @@ import scala.util.{Try, Using} * Per-category test runner for source operators (operators with no input * ports — they read from an external resource and emit tuples). * - * Dispatch is auto-first: - * - Auto tier: a scan source declares the file format it reads via - * [[ScanSourceOpDesc.fileTypeName]]. If that tag is in [[encoderByFileType]], - * the operator is fixtured with zero per-operator code — the shared - * [[CanonicalSourceFixture]] is encoded into that format and `fileName` - * points at it. A newly added file-scan source in a known format - * (CSV/JSONL/Arrow/…) is verified the moment it is registered in - * [[LogicalOp]]'s `@JsonSubTypes`, no edit here. - * - Curated tier: sources that can't take the shared table (text-family - * single-`line` output, or inline-config data) keep a hand-written - * [[SourceHandler]] in [[curatedHandlersByClass]]. - * - Otherwise the test is flagged (a [[knownIssues]] reason, an unsupported - * declared format, or no match) — never silently skipped. + * Dispatch is auto-first. A scan source declares the format it reads via + * [[ScanSourceOpDesc.fileTypeName]], and where [[encoderByFileType]] knows that + * format the shared [[CanonicalSourceFixture]] is encoded into it with no + * per-operator code at all: a newly registered file-scan source in a known + * format is verified the moment it appears in [[LogicalOp]]'s `@JsonSubTypes`. + * A source that cannot take the shared table — the text family emits a single + * `line` column, and some carry their data inline — keeps a hand-written + * [[SourceHandler]] instead. Anything else is flagged, never silently skipped. * * The runner itself is operator-agnostic: it builds an OpDesc, drives * [[OpExecHarness]] (Path A) and [[StandaloneRunner]] (Path B), compares via From d2e44e4ac57c60cbf76c331552617e97da044314 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 4 Sep 2026 13:22:26 -0700 Subject: [PATCH 05/17] test(verify): report an operator that writes to the frame it was handed The translator names a variable per output PORT, not per reader, so a workflow that draws two charts from one upstream hands both the same name. An operator that drops rows into it, or assigns a column through it, changes what the other branch reads. Ten operators did, and every one of them passed: a fixture with a single branch has nothing to notice it with. `StandaloneInputCheck` reads the generated code instead of waiting for a run to disagree, and reports all three spellings: the assignment, the `inplace=True`, and the rebinding that looks local but renames the shared variable. Co-Authored-By: Claude Opus 5 (1M context) --- .../verify/OperatorBehaviorSpec.scala | 7 ++ .../verify/StandaloneEscapingCheck.scala | 2 +- .../verify/StandaloneInputCheck.scala | 78 +++++++++++++++++++ 3 files changed, 86 insertions(+), 1 deletion(-) create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala index 2f984a3ba3b..72abc6be0f3 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala @@ -95,6 +95,13 @@ class OperatorBehaviorSpec extends AnyFlatSpec with Matchers with ParallelTestEx "Generated standalone code" should "stay parseable when the column names are hostile" in { StandaloneEscapingCheck.run() shouldBe empty } + + // Also one assertion over all of them: a workflow that draws two charts from + // one upstream hands both the same variable, and only a plan with a branch + // ever notices an operator writing to it. + it should "leave the frame it was handed alone" in { + StandaloneInputCheck.run() shouldBe empty + } } object OperatorBehaviorSpec { diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala index 5280b4cf1e7..46564ada47a 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala @@ -96,7 +96,7 @@ object StandaloneEscapingCheck { * because the generator cannot derive one. Auto operators contribute every * variant, since a knob only `optionals` fills is a site the base never reaches. */ - private def codeFor( + private[verify] def codeFor( opClass: Class[_ <: LogicalOp], dir: Path ): Either[String, Seq[(String, String)]] = { diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala new file mode 100644 index 00000000000..33063e97bf8 --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala @@ -0,0 +1,78 @@ +/* + * 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.operator.source.SourceOperatorDescriptor + +import java.nio.file.Files + +/** + * Reads every operator's generated code and reports one that writes to the + * frame it was handed. + * + * The translator names a variable per output PORT, not per reader, so two + * operators drawn from one upstream are handed the same name. An operator that + * drops rows into that name, or assigns a column through it, changes what the + * other branch goes on to read: a bar chart that drops its own nulls would + * leave a pie chart beside it drawing a table it never received. + * + * Both spellings of the mistake matter, and one of them does not look like a + * mistake. `inplace=True` mutates the frame outright. `in1df = in1df.dropna()` + * reads like a local rebinding, but the operator bodies are concatenated at + * module scope, so the name it rebinds is the shared one. + * + * A single-branch workflow never notices either, which is why this reads the + * code rather than waiting for a run to disagree. + */ +object StandaloneInputCheck { + + private val Assignment = """(?m)^\s*in\d+df\s*\[[^\]]*\]\s*=""".r + private val InPlace = """in\d+df[^\n]*inplace\s*=\s*True""".r + private val Rebind = """(?m)^\s*in\d+df\s*=""".r + + /** Every operator that writes to its input, empty when the suite is clean. */ + def run(): Seq[String] = { + val operators = OperatorBehaviorSpec + .discoverStandaloneOperators() + .filterNot(classOf[SourceOperatorDescriptor].isAssignableFrom) + val dir = Files.createTempDirectory("standalone-input-") + dir.toFile.deleteOnExit() + + operators.flatMap { op => + StandaloneEscapingCheck.codeFor(op, dir) match { + // An operator whose config cannot be built is reported by the escaping + // check, which builds the same ones; repeating it here would say the + // same thing twice. + case Left(_) => Seq.empty + case Right(variants) => + variants.flatMap { + case (label, code) => + val how = Seq( + Assignment.findFirstIn(code).map(_ => "assigns a column through it"), + InPlace.findFirstIn(code).map(_ => "mutates it with inplace=True"), + Rebind.findFirstIn(code).map(_ => "rebinds the name it was given") + ).flatten + if (how.isEmpty) None + else Some(s"${op.getSimpleName}/$label: ${how.mkString(", ")}") + } + } + } + } +} From f3355ba2f31e941a31e33f311d12324f77cb3ecd Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 4 Sep 2026 14:08:15 -0700 Subject: [PATCH 06/17] test(verify): read the accessor spellings of the same assignment `in1df.loc[:, "c"] = v` writes to the input exactly as `in1df["c"] = v` does, and the pattern read only the plain subscript, so an operator would have passed for choosing the accessor. What it still cannot see is a mutation through another name: bind the frame to something else first and the writes below are invisible to anything reading the code. The scaladoc says so rather than leaving the check looking airtight. Co-Authored-By: Claude Opus 5 (1M context) --- .../translator/verify/StandaloneInputCheck.scala | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala index 33063e97bf8..a4de776f07a 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala @@ -39,11 +39,21 @@ import java.nio.file.Files * module scope, so the name it rebinds is the shared one. * * A single-branch workflow never notices either, which is why this reads the - * code rather than waiting for a run to disagree. + * code rather than waiting for a run to disagree. Every fixture the runner + * builds has one reader, so a comparison of the two paths agrees while the + * frame is being altered underneath a branch that the fixture does not have. + * + * What reading the code cannot see is a mutation through another name: bind + * the frame to something else first and the writes below are invisible here. + * No operator does that today, and an operator that starts to would be saying + * something a reader has to work out anyway. */ object StandaloneInputCheck { - private val Assignment = """(?m)^\s*in\d+df\s*\[[^\]]*\]\s*=""".r + // `.loc` and `.iloc` are here because they are the other way to write the + // assignment, and a pattern that reads only the plain subscript would call an + // operator clean for choosing the accessor. + private val Assignment = """(?m)^\s*in\d+df(\.i?loc|\.i?at)?\s*\[[^\]]*\]\s*=""".r private val InPlace = """in\d+df[^\n]*inplace\s*=\s*True""".r private val Rebind = """(?m)^\s*in\d+df\s*=""".r From f66d6e5fef45d67257906bb51a78044f2ebdc13f Mon Sep 17 00:00:00 2001 From: kary zheng Date: Tue, 8 Sep 2026 23:50:39 -0700 Subject: [PATCH 07/17] test(verify): say what each of the two code-reading checks reaches A hostile-column run now exists, and the two look alike enough that a reader could take one for a duplicate of the other. What this end cannot do is run the script; what the run cannot do is make every knob hostile at once, since a table holds one hostile column per type and the first knob takes it. The rest is length. StandaloneInputCheck kept a paragraph on what it cannot see, which rested on no operator doing that today, and the dispatcher stated the verdict-in-the-test-name rule in four places where one will do. Co-Authored-By: Claude Opus 5 (1M context) --- .../verify/OperatorBehaviorSpec.scala | 51 +++++++------------ .../verify/SourceCategoryRunner.scala | 5 +- .../verify/StandaloneEscapingCheck.scala | 6 +++ .../verify/StandaloneInputCheck.scala | 28 +++------- 4 files changed, 32 insertions(+), 58 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala index 72abc6be0f3..635868d0302 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala @@ -28,21 +28,15 @@ import org.scalatest.flatspec.AnyFlatSpec import org.scalatest.matchers.should.Matchers /** - * Auto-discovered behavioral-parity tests: for every operator registered - * with [[LogicalOp]]'s `@JsonSubTypes` that implements - * [[StandaloneCodeGenerator]], emit a test that runs both Path A (Texera - * exec) and Path B (translator-generated Python via [[StandaloneRunner]]) - * and asserts their outputs are equivalent. + * A parity test per operator: every one registered in [[LogicalOp]]'s + * `@JsonSubTypes` that implements [[StandaloneCodeGenerator]] runs through the + * Texera executor and through its generated Python, and the two outputs are + * compared. Adding an operator needs no edit here. * - * [[TransformVerificationRunner]] decides how each non-source transform is - * configured and whether it can run at all; sources go to - * [[SourceCategoryRunner]]. An operator it cannot run is registered as an - * ignored test carrying the reason, so the report lists every operator rather - * than reading as though the unrunnable ones do not exist. - * - * No edits to this spec are needed when a new operator is added — reflection - * discovers it automatically via `@JsonSubTypes`. The tier label appears in - * the test name so the report shows which path exercised each operator. + * [[TransformVerificationRunner]] configures each non-source transform and + * decides whether it can run; sources go to [[SourceCategoryRunner]]. Every + * test's NAME carries its verdict, the tier that ran it or the reason it could + * not, so the report lists every operator and says what it did not check. * * Requires Python 3 with pandas on the [[Comparator]] / [[StandaloneRunner]] * resolution chain (`UDF_PYTHON_PATH` env var, then `python3.12`). @@ -53,16 +47,12 @@ import org.scalatest.matchers.should.Matchers @IntegrationTest class OperatorBehaviorSpec extends AnyFlatSpec with Matchers with ParallelTestExecution { - // Build the test list at class construction. Each branch below registers - // one test (`in` for runnable, `ignore` for skipped) so the test report - // shows every translator-eligible operator and why it did or didn't run. + // The test list is built at class construction, one test per operator. OperatorBehaviorSpec.discoverStandaloneOperators().foreach { opClass => val name = opClass.getSimpleName if (!OperatorBehaviorSpec.isSelected(name)) { - // Narrowed out by VERIFY_ONLY / VERIFY_SKIP, which only a local run sets. - // Still registered, as an `ignore`, so the report lists every operator - // rather than reading as though the narrowed-out ones do not exist. + // Only a local run sets those, and CI therefore runs the lot. name should "NARROWED OUT — outside this run's VERIFY_ONLY / VERIFY_SKIP" ignore {} } else if (classOf[SourceOperatorDescriptor].isAssignableFrom(opClass)) { // Sources keep their handler-per-source design: each needs a real file @@ -81,10 +71,8 @@ class OperatorBehaviorSpec extends AnyFlatSpec with Matchers with ParallelTestEx TransformVerificationRunner.run(opClass) } case TransformVerificationRunner.Flagged(reason) => - name should s"FLAGGED — $reason" ignore { - // Reason is in the test name so the report carries it; the - // coverage table in ConfigCoverageSpec aggregates these. - } + // ConfigCoverageSpec aggregates these into its table. + name should s"FLAGGED — $reason" ignore {} } } } @@ -106,17 +94,12 @@ class OperatorBehaviorSpec extends AnyFlatSpec with Matchers with ParallelTestEx object OperatorBehaviorSpec { - // Narrowing knobs for a local run, both unset by default, so the default run - // is every operator: VERIFY_ONLY names the only ones to run, VERIFY_SKIP the - // ones to leave out. Case-sensitive substrings against the operator's simple - // name, comma-separated. Neither is set in CI, which therefore runs the lot. + // Case-sensitive substrings of the operator's simple name, comma-separated. // - // There is deliberately no third list withholding operators by default. What - // stays withheld is narrower than an operator and lives where it can say why: - // a single variant in [[TransformVerificationRunner.variantsNotRun]], or an - // operator that cannot be run at all in its `knownIssues`, each against an - // issue or a reason. A name here would withdraw an operator's every variant - // and record nothing about what is wrong with it. + // There is deliberately no third list withholding operators by default: a + // name here would withdraw an operator's every variant and record nothing + // about what is wrong with it. Withholding lives where it can say why, in + // [[TransformVerificationRunner]]'s `variantsNotRun` or its `knownIssues`. private def patterns(envVar: String): Seq[String] = sys.env.getOrElse(envVar, "").split(",").iterator.map(_.trim).filter(_.nonEmpty).toSeq diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala index ab0561c0164..4e9d53e3168 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala @@ -52,9 +52,8 @@ import scala.util.{Try, Using} * `line` column, and some carry their data inline — keeps a hand-written * [[SourceHandler]] instead. Anything else is flagged, never silently skipped. * - * The runner itself is operator-agnostic: it builds an OpDesc, drives - * [[OpExecHarness]] (Path A) and [[StandaloneRunner]] (Path B), compares via - * [[Comparator]]. Sources have no input ports so `inputs = Map.empty` for both. + * The runner itself is operator-agnostic, and drives the same two paths every + * other operator takes, with no inputs on either. */ object SourceCategoryRunner { diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala index 46564ada47a..d4b43c315cf 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneEscapingCheck.scala @@ -40,6 +40,12 @@ import scala.util.Try * The check has to be behavioural. A generator that builds its quoted literal * inside a helper shows no quotes in its template, which is how RadarPlot and * Aggregate survived a source-level sweep that reported zero remaining sites. + * + * The `hostileColumns` run covers the same ground by running the script, which + * this end cannot. What it cannot do is make every knob hostile at once: a + * fixture holds one hostile column per type, so whichever knob takes it leaves + * a chart's `color` and `pattern` on ordinary names. Renaming in place carries + * the old column in the new name, so no number of knobs can collide. */ object StandaloneEscapingCheck { diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala index a4de776f07a..2c233e1b2b0 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/StandaloneInputCheck.scala @@ -24,29 +24,15 @@ import org.apache.texera.amber.operator.source.SourceOperatorDescriptor import java.nio.file.Files /** - * Reads every operator's generated code and reports one that writes to the - * frame it was handed. + * Reports an operator whose generated code writes to the frame it was handed. * - * The translator names a variable per output PORT, not per reader, so two - * operators drawn from one upstream are handed the same name. An operator that - * drops rows into that name, or assigns a column through it, changes what the - * other branch goes on to read: a bar chart that drops its own nulls would - * leave a pie chart beside it drawing a table it never received. + * The translator names a variable per output PORT, so two operators drawn from + * one upstream share it, and a write changes what the other reads. Rebinding + * counts: the bodies are concatenated at module scope, so + * `in1df = in1df.dropna()` rebinds the shared name. * - * Both spellings of the mistake matter, and one of them does not look like a - * mistake. `inplace=True` mutates the frame outright. `in1df = in1df.dropna()` - * reads like a local rebinding, but the operator bodies are concatenated at - * module scope, so the name it rebinds is the shared one. - * - * A single-branch workflow never notices either, which is why this reads the - * code rather than waiting for a run to disagree. Every fixture the runner - * builds has one reader, so a comparison of the two paths agrees while the - * frame is being altered underneath a branch that the fixture does not have. - * - * What reading the code cannot see is a mutation through another name: bind - * the frame to something else first and the writes below are invisible here. - * No operator does that today, and an operator that starts to would be saying - * something a reader has to work out anyway. + * Read rather than run, because every fixture has a single reader, so both + * paths agree while the frame is altered under a branch that is not there. */ object StandaloneInputCheck { From 1ee3dd291e8feabba3978b4afb4b8bd101cc1e48 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 10 Sep 2026 15:56:03 -0700 Subject: [PATCH 08/17] build(verify): bound the pool the operator sweep runs in OperatorBehaviorSpec forks a Python subprocess per operator, and at core-count concurrency the contention caused rare flakes. A fixed 4 is deterministic across machines and still runs about three times faster than serial; it also matches PythonWorkerPool's own default cap, so the two bounds agree rather than multiply. Co-Authored-By: Claude Opus 5 (1M context) --- workflow-compiling-service/build.sbt | 20 +++++++++++++++++++- 1 file changed, 19 insertions(+), 1 deletion(-) diff --git a/workflow-compiling-service/build.sbt b/workflow-compiling-service/build.sbt index 2af92efc4d4..e09263a62f1 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 the split adds. +Test / testOptions += Tests.Argument(TestFrameworks.ScalaTest, "-P4") + ///////////////////////////////////////////////////////////////////////////// // Compiler Options ///////////////////////////////////////////////////////////////////////////// From f8d0015c48073ab1f05e717b894e9189dd76e088 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 10 Sep 2026 16:12:22 -0700 Subject: [PATCH 09/17] Revert "build(verify): bound the pool the operator sweep runs in" This reverts commit 1ee3dd291e8feabba3978b4afb4b8bd101cc1e48. --- workflow-compiling-service/build.sbt | 20 +------------------- 1 file changed, 1 insertion(+), 19 deletions(-) diff --git a/workflow-compiling-service/build.sbt b/workflow-compiling-service/build.sbt index e09263a62f1..2af92efc4d4 100644 --- a/workflow-compiling-service/build.sbt +++ b/workflow-compiling-service/build.sbt @@ -41,27 +41,9 @@ 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. This caps how many -// test *suites* run concurrently; ParallelTestExecution still parallelizes the -// tests *within* a suite (e.g. OperatorBehaviorSpec) via ScalaTest's own pool. +// Restrict parallel execution of tests to avoid conflicts 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 the split adds. -Test / testOptions += Tests.Argument(TestFrameworks.ScalaTest, "-P4") - ///////////////////////////////////////////////////////////////////////////// // Compiler Options ///////////////////////////////////////////////////////////////////////////// From cb6125c26d33f1bfd279530e0a149135ae77976c Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 10 Sep 2026 16:45:47 -0700 Subject: [PATCH 10/17] test(verify): cut the doc to what the code cannot say It restated its own signature and listed a codec table that lives in TupleIO. The reasons a reader cannot derive stay. Co-Authored-By: Claude Opus 5 (1M context) --- .../amber/translator/verify/SourceCategoryRunner.scala | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala index 4e9d53e3168..9892867ae85 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala @@ -48,12 +48,10 @@ import scala.util.{Try, Using} * format the shared [[CanonicalSourceFixture]] is encoded into it with no * per-operator code at all: a newly registered file-scan source in a known * format is verified the moment it appears in [[LogicalOp]]'s `@JsonSubTypes`. - * A source that cannot take the shared table — the text family emits a single - * `line` column, and some carry their data inline — keeps a hand-written - * [[SourceHandler]] instead. Anything else is flagged, never silently skipped. - * - * The runner itself is operator-agnostic, and drives the same two paths every - * other operator takes, with no inputs on either. + * A source that cannot take the shared table keeps a hand-written + * [[SourceHandler]] instead: the text family emits a single `line` column, and + * some carry their data inline. Anything else is flagged, never silently + * skipped. */ object SourceCategoryRunner { From cfcddaf7d7706b292ffe6fabca1716527e194845 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 10 Sep 2026 23:29:35 -0700 Subject: [PATCH 11/17] build: open the packages Kryo reaches once both halves run together Four more --add-opens. Kryo recurses into a timestamp's internals when it serializes one, and reaches them only in a JVM that has already forked a Python worker, which is what running this module's tests without WCS_TEST_FILTER does. The two CI jobs each take half, so CI never gets there; a local full-module run does, and reports an InaccessibleObjectException that looks like a defect in the diff. They arrive with the spec that makes the combination possible. Co-Authored-By: Claude Opus 5 (1M context) --- .jvmopts | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/.jvmopts b/.jvmopts index 3f4c694a764..7aaa6d2866a 100644 --- a/.jvmopts +++ b/.jvmopts @@ -26,3 +26,12 @@ # Required by Apache Pekko: https://pekko.apache.org/docs/pekko/snapshot/release-notes/releases-2.0.html --add-opens=java.base/jdk.internal.misc=ALL-UNNAMED + +# Kryo again, reached only once a suite that serializes a timestamp runs in the +# same JVM as one that has already forked a Python worker, which is what running +# this module's tests without WCS_TEST_FILTER does. The two CI jobs each take +# half, so CI never reaches it and a local full-module run does. +--add-opens=java.base/sun.util.calendar=ALL-UNNAMED +--add-opens=java.base/java.time=ALL-UNNAMED +--add-opens=java.base/java.time.zone=ALL-UNNAMED +--add-opens=java.base/java.util.concurrent=ALL-UNNAMED From de34e86d120ced3aae68b16f9b758bbf49e61cf1 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 11 Sep 2026 00:36:22 -0700 Subject: [PATCH 12/17] test(verify): encode the canonical table as Parquet A scan source is fixtured by the format it declares, so registering an encoder for "Parquet" is the whole of what the new source needs from this runner: no handler, and no mention of the operator anywhere here. The writer is the inverse of ParquetSchemaMapping, which is what the operator reads the file back with. Every column is written optional, since Parquet has no null value and only a field that repeats zero times can carry a hole. A timestamp is written as the wall clock counted from the epoch, the reading both paths agree on. Co-Authored-By: Claude Opus 5 (1M context) --- .../verify/SourceCategoryRunner.scala | 87 ++++++++++++++++++- 1 file changed, 85 insertions(+), 2 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala index 9892867ae85..aec74bbe498 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala @@ -19,7 +19,14 @@ package org.apache.texera.amber.translator.verify -import org.apache.texera.amber.core.tuple.{Schema, Tuple} +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.fs.{Path => HadoopPath} +import org.apache.parquet.example.data.Group +import org.apache.parquet.example.data.simple.SimpleGroupFactory +import org.apache.parquet.hadoop.example.{ExampleParquetWriter, GroupWriteSupport} +import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName +import org.apache.parquet.schema.{LogicalTypeAnnotation, MessageType, Types} +import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple} import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.operator.source.fetcher.URLFetcherOpDesc @@ -36,6 +43,7 @@ import org.apache.texera.amber.util.ArrowUtils import java.nio.channels.FileChannel import java.nio.charset.{Charset, StandardCharsets} import java.nio.file.{Files, Path, StandardOpenOption} +import java.time.ZoneOffset import scala.collection.mutable import scala.util.{Try, Using} @@ -84,7 +92,8 @@ object SourceCategoryRunner { "JSONL" -> CanonicalSourceFixture.writeJsonl, // Arrow is binary and its descriptor declares fileEncoding ignored, so the // charset a variant asks for has nothing to apply to. - "Arrow" -> ((dir, _) => CanonicalSourceFixture.writeArrow(dir)) + "Arrow" -> ((dir, _) => CanonicalSourceFixture.writeArrow(dir)), + "Parquet" -> ((dir, _) => CanonicalSourceFixture.writeParquet(dir)) ) /** @@ -429,6 +438,80 @@ object CanonicalSourceFixture { }.get path } + + /** Write the rows as a Parquet file carrying the canonical table's own types. + * + * The mapping below is the inverse of `ParquetSchemaMapping`, which is what + * the operator reads the file back with. Writing every column as optional is + * what lets a hole be a hole: Parquet has no null value, only a field that + * repeats zero times, so a required column could not hold one. + */ + def writeParquet(dir: Path): Path = { + val path = dir.resolve("sample.parquet") + val messageType = parquetSchemaOf(schema) + val conf = new Configuration() + GroupWriteSupport.setSchema(messageType, conf) + val factory = new SimpleGroupFactory(messageType) + Using( + ExampleParquetWriter + .builder(new HadoopPath(path.toString)) + .withConf(conf) + .withType(messageType) + .build() + ) { writer => rows.foreach(t => writer.write(groupOf(factory, t))) }.get + path + } + + /** The canonical schema as Parquet states types. */ + private def parquetSchemaOf(schema: Schema): MessageType = { + val fields = schema.getAttributes.map { attribute => + val name = attribute.getName + attribute.getType match { + case AttributeType.STRING => + Types + .optional(PrimitiveTypeName.BINARY) + .as(LogicalTypeAnnotation.stringType()) + .named(name) + case AttributeType.INTEGER => Types.optional(PrimitiveTypeName.INT32).named(name) + case AttributeType.LONG => Types.optional(PrimitiveTypeName.INT64).named(name) + case AttributeType.DOUBLE => Types.optional(PrimitiveTypeName.DOUBLE).named(name) + case AttributeType.BOOLEAN => Types.optional(PrimitiveTypeName.BOOLEAN).named(name) + case AttributeType.TIMESTAMP => + Types + .optional(PrimitiveTypeName.INT64) + .as(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS)) + .named(name) + case other => + throw new UnsupportedOperationException(s"no Parquet type for $other in column $name") + } + } + Types.buildMessage().addFields(fields.toSeq: _*).named("canonical") + } + + /** One row, with an absent field wherever the tuple holds a null. */ + private def groupOf(factory: SimpleGroupFactory, tuple: Tuple): Group = { + val group = factory.newGroup() + schema.getAttributes.foreach { attribute => + val name = attribute.getName + Option(tuple.getField[AnyRef](name)).foreach { + case v: java.lang.String => group.append(name, v) + case v: java.lang.Integer => group.append(name, v.intValue()) + case v: java.lang.Long => group.append(name, v.longValue()) + case v: java.lang.Double => group.append(name, v.doubleValue()) + case v: java.lang.Boolean => group.append(name, v.booleanValue()) + // The wall clock counted from the epoch with no zone in the picture, + // which is what `ArrowUtils` means by a Texera TIMESTAMP and what the + // operator reads back. `getTime` would bake in the writer's own zone. + case v: java.sql.Timestamp => + group.append(name, v.toLocalDateTime.toInstant(ZoneOffset.UTC).toEpochMilli) + case other => + throw new UnsupportedOperationException( + s"no Parquet value for ${other.getClass.getSimpleName} in column $name" + ) + } + } + group + } } /** Handler for `TextInputSourceOpDesc`. The text lives in the config — no fixture file. */ From 556bbca160355f108674c6610e39dc0d838544f3 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 18 Sep 2026 17:05:03 -0700 Subject: [PATCH 13/17] test(verify): hand the script the schemas the read produced The epilogue writes a declared-integral column as an integer, the way the engine's writer does, and a source's declared columns are what the Texera path already read back here. Co-Authored-By: Claude Opus 5 (1M context) --- .../amber/translator/verify/SourceCategoryRunner.scala | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala index aec74bbe498..142d6391ccd 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala @@ -305,7 +305,10 @@ object SourceCategoryRunner { opDesc = opDesc, inputs = Map.empty, outputPortCount = 1, - workDir = workDir + workDir = workDir, + // What the read declared its columns to be, so an integral one is written + // the way the engine's writer wrote it. + outputSchemas = pathA.outputSchemas ) val actual = pathA.outputs(PortIdentity(0)) From 92e6713e8987a6207e347508b06d2057483711d8 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 25 Sep 2026 14:25:07 -0700 Subject: [PATCH 14/17] test(verify): read each source through the cases a reader gets wrong The source runner handed every reader the same full table on its default config, so a case a reader had been fixed for was asked only by a spec with a file of its own. It now asks them of every source: - Holes: the table is read once more with one empty cell per column. The JSONL writer leaves out the key for a hole in every other column and writes null in the rest, since the executor reads the two differently. - Values: two columns the transforms never see, a nullable long past 2^53 and the text NA; CSV headers left blank and one spelled "Unnamed: 1"; JSONL records nesting an array of objects, an array of text and a month-first date; Arrow columns in single precision, 8 and 16 bits, three unsigned widths and a zoned timestamp, and the schema note pandas leaves for an index. - Windows: a limit of 0, a negative offset, a negative limit, both, an offset of 1 with the largest limit, the largest of both, and for JSONL a first line that is no JSON, read past. - Text sources: a handler can list scenarios the sweep cannot reach. File Scan reads an integer past a skipped line, boolean lines, a UTF-16 file, and a zip holding the entry macOS adds, line by line and whole, with and without each entry's name. Text Input reads the first two. Values alone cannot tell a width or a zone apart, and an output with no row reads back with no column on either side, so each run also checks the columns and dtypes the script left against what the read declared. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../verify/SourceCategoryRunner.scala | 403 ++++++++++++++++-- 1 file changed, 375 insertions(+), 28 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala index 142d6391ccd..a307de685c2 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala @@ -26,25 +26,38 @@ import org.apache.parquet.example.data.simple.SimpleGroupFactory import org.apache.parquet.hadoop.example.{ExampleParquetWriter, GroupWriteSupport} import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName import org.apache.parquet.schema.{LogicalTypeAnnotation, MessageType, Types} -import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple} +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.LogicalOp import org.apache.texera.amber.operator.source.fetcher.URLFetcherOpDesc -import org.apache.texera.amber.operator.source.scan.ScanSourceOpDesc +import org.apache.texera.amber.operator.source.scan.{FileAttributeType, ScanSourceOpDesc} import org.apache.texera.amber.operator.source.scan.file.{FileScanOpDesc, FileScanSourceOpDesc} import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc import com.fasterxml.jackson.databind.node.ObjectNode import org.apache.texera.amber.util.JSONUtils.objectMapper import org.apache.arrow.memory.RootAllocator import org.apache.arrow.vector.ipc.ArrowFileWriter -import org.apache.arrow.vector.VectorSchemaRoot +import org.apache.arrow.vector.types.pojo.{Schema => ArrowSchema} +import org.apache.arrow.vector.{ + FieldVector, + Float4Vector, + SmallIntVector, + TimeStampMilliTZVector, + TinyIntVector, + UInt1Vector, + UInt2Vector, + UInt4Vector, + VectorSchemaRoot +} import org.apache.texera.amber.util.ArrowUtils import java.nio.channels.FileChannel import java.nio.charset.{Charset, StandardCharsets} import java.nio.file.{Files, Path, StandardOpenOption} import java.time.ZoneOffset +import java.util.zip.{ZipEntry, ZipOutputStream} import scala.collection.mutable +import scala.jdk.CollectionConverters._ import scala.util.{Try, Using} /** @@ -86,14 +99,14 @@ object SourceCategoryRunner { * but it is currently commented out of `@JsonSubTypes`, so the suite doesn't * enumerate it.) */ - private val encoderByFileType: Map[String, (Path, Charset) => Path] = Map( + private val encoderByFileType: Map[String, (Path, Charset, Seq[Tuple]) => Path] = Map( "CSV" -> CanonicalSourceFixture.writeCsv, "CSVOld" -> CanonicalSourceFixture.writeCsv, "JSONL" -> CanonicalSourceFixture.writeJsonl, // Arrow is binary and its descriptor declares fileEncoding ignored, so the // charset a variant asks for has nothing to apply to. - "Arrow" -> ((dir, _) => CanonicalSourceFixture.writeArrow(dir)), - "Parquet" -> ((dir, _) => CanonicalSourceFixture.writeParquet(dir)) + "Arrow" -> ((dir, _, rows) => CanonicalSourceFixture.writeArrow(dir, rows)), + "Parquet" -> ((dir, _, rows) => CanonicalSourceFixture.writeParquet(dir, rows)) ) /** @@ -214,7 +227,11 @@ object SourceCategoryRunner { ) (variant.label, op, dir) } - } + } ++ handler.scenarios.map { + case (label, write) => + val dir = dirFor(label) + (label, write(dir), dir) + } case None => val fileType = declaredFileType(opDescClass).getOrElse("") val encoder = encoderByFileType.getOrElse( @@ -223,16 +240,66 @@ object SourceCategoryRunner { s"No encoder for ${opDescClass.getSimpleName} (fileTypeName='$fileType')" ) ) - val base = { - val dir = dirFor("default") + def bare( + label: String, + rows: Seq[Tuple], + window: (Option[Int], Option[Int]) = (None, None) + ): (String, LogicalOp, Path) = { + val dir = dirFor(label) val op = newScanSource(opDescClass) - op.fileName = Some(encoder(dir, op.fileEncoding.getCharset).toUri.toString) - ("default", op: LogicalOp, dir) + op.fileName = Some(encoder(dir, op.fileEncoding.getCharset, rows).toUri.toString) + op.offset = window._1 + op.limit = window._2 + (label, op, dir) } - base +: generatedVariants(opDescClass, encoder, dirFor) + // The nulls variant takes the bare config rather than crossing with the + // others: how a reader treats an empty cell is a property of the reader, + // and multiplying it across every knob buys runtime, not signal. + // + // The windows are the ones the generator never fills, since it fills a + // knob with a value the property editor accepts: a limit of no rows, which + // still has columns to declare, a negative bound, which only a plan posted + // to the API can carry and which pandas reads from the end, and bounds + // whose sum is past what an Int holds. + val rows = CanonicalSourceFixture.rows + Seq( + bare("default", rows), + bare("nulls", CanonicalSourceFixture.holedRows), + bare("window: limit 0", rows, (None, Some(0))), + bare("window: negative offset", rows, (Some(-1), None)), + bare("window: negative limit", rows, (None, Some(-1))), + bare("window: negative", rows, (Some(-1), Some(-1))), + bare("window: long end", rows, (Some(1), Some(Int.MaxValue))), + bare("window: largest", rows, (Some(Int.MaxValue), Some(Int.MaxValue))) + ) ++ skippedBadLine(opDescClass, fileType, encoder, dirFor) ++ + generatedVariants(opDescClass, encoder, dirFor) } } + /** A JSONL file whose first line is no JSON at all, read past it. The executor + * drops and takes the raw lines before it parses any, so a line outside the + * window is never read, however malformed. The other formats have no line a + * reader refuses to parse. + */ + private def skippedBadLine( + opDescClass: Class[_ <: LogicalOp], + fileType: String, + encoder: (Path, Charset, Seq[Tuple]) => Path, + dirFor: String => Path + ): Seq[(String, LogicalOp, Path)] = + if (fileType != "JSONL") Seq.empty + else { + val label = "window: skips a bad line" + val dir = dirFor(label) + val op = newScanSource(opDescClass) + val path = encoder(dir, op.fileEncoding.getCharset, CanonicalSourceFixture.rows) + val body = new String(Files.readAllBytes(path), op.fileEncoding.getCharset) + Files.write(path, ("not json at all\n" + body).getBytes(op.fileEncoding.getCharset)) + op.fileName = Some(path.toUri.toString) + op.offset = Some(1) + Seq((label, op, dir)) + } + /** * The variants the shared [[ConfigGenerator]] derives from the operator's own * fields — the base config with every knob filled, plus one per enum branch @@ -250,7 +317,7 @@ object SourceCategoryRunner { */ private def generatedVariants( opDescClass: Class[_ <: LogicalOp], - encoder: (Path, Charset) => Path, + encoder: (Path, Charset, Seq[Tuple]) => Path, dirFor: String => Path ): Seq[(String, LogicalOp, Path)] = { val seen = mutable.Set.empty[String] @@ -274,7 +341,9 @@ object SourceCategoryRunner { // the generator's, which additionally fills limit and offset. val name = if (label == "default") "auto-base" else label val dir = dirFor(name) - scan.fileName = Some(encoder(dir, scan.fileEncoding.getCharset).toUri.toString) + scan.fileName = Some( + encoder(dir, scan.fileEncoding.getCharset, CanonicalSourceFixture.rows).toUri.toString + ) Some((name, scan: LogicalOp, dir)) } } @@ -314,6 +383,50 @@ object SourceCategoryRunner { val actual = pathA.outputs(PortIdentity(0)) val expected = pathB.outputs(1) Comparator.assertEqual(actual, expected) + assertDtypesFit(pathA.outputSchemas(PortIdentity(0)), expected) + } + + /** Whether a pandas dtype is one a Texera column of this type can be left in. + * + * The values alone cannot tell a width or a zone apart: a float32 writes the + * same text as a float64, and the zone is gone once the timestamp is text. Both + * still reach the next operator, where a float32 sums to another number and a + * zoned column refuses a cast to a plain datetime. So what the script left + * each column in is compared with what the read declared, apart from the + * values. Only the widths and zones Texera has no column for are refused. + */ + private def fits(attributeType: AttributeType, dtype: String): Boolean = + attributeType match { + case AttributeType.INTEGER => Set("int32", "Int32", "int64", "Int64").contains(dtype) + case AttributeType.LONG => Set("int64", "Int64").contains(dtype) + case AttributeType.DOUBLE => Set("float64", "Float64").contains(dtype) + case AttributeType.BOOLEAN => Set("bool", "boolean").contains(dtype) + case AttributeType.STRING => dtype == "object" || dtype == "str" || dtype.startsWith("string") + case AttributeType.TIMESTAMP => dtype.startsWith("datetime64[") && !dtype.contains(",") + case _ => true + } + + private def assertDtypesFit(declared: Schema, output: Path): Unit = { + val dtypes = + objectMapper.readTree(Files.readAllBytes(Path.of(output.toString + ".dtypes.json"))) + // An output with no rows reads back with no columns on either side, so the + // values cannot say which columns the frame had. What the read declared and + // what the script left are both still on record. + val left = dtypes.fieldNames.asScala.toList + if (left != declared.getAttributeNames) + throw new AssertionError( + s"the script left the columns $left where the read declared ${declared.getAttributeNames}" + ) + val misfits = declared.getAttributes.flatMap { attribute => + Option(dtypes.get(attribute.getName)) + .map(_.asText) + .filterNot(fits(attribute.getType, _)) + .map(dtype => s"${attribute.getName}: declared ${attribute.getType}, left as $dtype") + } + if (misfits.nonEmpty) + throw new AssertionError( + s"the script left columns in a dtype the read did not declare:\n ${misfits.mkString("\n ")}" + ) } } @@ -340,6 +453,13 @@ trait SourceHandler { * take keeps some rows and drops some instead of landing past the end. */ def rowCount: Int + + /** Configurations the sweep cannot reach, each with the file it has to read: + * a type the default text cannot be parsed as, or bytes in another charset. + * Each one writes into the directory it is handed and returns the configured + * op. Default: none. + */ + def scenarios: Seq[(String, Path => LogicalOp)] = Seq.empty } /** @@ -361,19 +481,59 @@ trait SourceHandler { */ object CanonicalSourceFixture { - val schema: Schema = CanonicalFixture.schema + /** Columns only a reader has a question about, appended to the canonical ones. + * They stay out of [[CanonicalFixture]], which every transform reads. + * + * `a"b\c_big` holds odd integers past 2^53, none of which a float can hold, so + * a reader that widens the column through one rounds every value. `a"b\c_region` + * holds the text `NA`, which pandas reads as missing unless told not to. + */ + private val sourceColumns: Seq[(Attribute, Int => AnyRef)] = Seq( + new Attribute("a\"b\\c_big", AttributeType.LONG) -> (i => Long.box(9007199254740993L + 2 * i)), + new Attribute("a\"b\\c_region", AttributeType.STRING) -> (i => Seq("NA", "EU", "APAC")(i % 3)) + ) + + val schema: Schema = + new Schema(CanonicalFixture.schema.getAttributes ++ sourceColumns.map(_._1): _*) + + val rows: Vector[Tuple] = CanonicalFixture.allRows.zipWithIndex.map { + case (canonical, i) => + val b = Tuple.builder(schema) + CanonicalFixture.schema.getAttributes.foreach { a => + b.add(a, canonical.getField[AnyRef](a.getName)) + } + sourceColumns.foreach { case (a, value) => b.add(a, value(i)) } + b.build() + } - val rows: Vector[Tuple] = CanonicalFixture.allRows + /** [[rows]] with one empty cell per column, for the `nulls` variant. Nothing is + * kept filled: a source reads every column the same way and pairs no rows, so + * no column's value is load-bearing. + */ + val holedRows: Seq[Tuple] = SharedFixture.emptyOneCellPerColumn(rows, schema, Set.empty) /** Write the rows as a header-first, comma-delimited CSV encoded in `charset`. * * The charset is a parameter because it describes the BYTES, not the config: a * variant declaring `fileEncoding = UTF_16` over a file left in UTF-8 would * compare nothing but how each path fails. + * + * Two headers are left blank, a timestamp's and a long's, which the reader + * names column-N and pandas names "Unnamed: N" until it is told otherwise, so + * a type asked for by the schema's name finds no column. The header at + * position 1 is written as the text "Unnamed: 1", a name a user can really + * choose and pandas' placeholder for the same place. */ - def writeCsv(dir: Path, charset: Charset): Path = { + def writeCsv(dir: Path, charset: Charset, rows: Seq[Tuple]): Path = { val path = dir.resolve("sample.csv") - val header = schema.getAttributes.map(a => csvField(a.getName)).mkString(",") + val blank = Set("start_ts", "a\"b\\c_big") + val header = schema.getAttributes.zipWithIndex + .map { + case (_, 1) => "Unnamed: 1" + case (a, _) if blank.contains(a.getName) => "" + case (a, _) => csvField(a.getName) + } + .mkString(",") val body = rows.map { t => schema.getAttributes .map(a => csvField(Option(t.getField[AnyRef](a.getName)).map(_.toString).orNull)) @@ -402,22 +562,56 @@ object CanonicalSourceFixture { * * That writer is shared and always writes UTF-8, so a variant asking for another * charset gets the bytes transcoded afterwards rather than a second writer. + * + * A hole is written two ways, because the executor reads them differently: a + * key left out is a null, and a key holding null is the text "null". A column + * at an even position keeps the key, one at an odd position leaves it out. + * + * Each record also nests, the one thing JSON holds that a table does not: an + * array of objects, an array of text whose length differs between records, and + * an object holding a date written month first. Flattening names every element + * by its position, and the date is only a date once it is parsed. */ - def writeJsonl(dir: Path, charset: Charset): Path = { + def writeJsonl(dir: Path, charset: Charset, rows: Seq[Tuple]): Path = { val path = dir.resolve("sample.jsonl") TupleIO.writeTuples(path, rows.iterator, schema) - if (charset != StandardCharsets.UTF_8) { - val text = new String(Files.readAllBytes(path), StandardCharsets.UTF_8) - Files.write(path, text.getBytes(charset)) + val leftOut = schema.getAttributes.zipWithIndex.collect { + case (attribute, i) if i % 2 == 1 => attribute.getName } + val lines = new String(Files.readAllBytes(path), StandardCharsets.UTF_8) + .split("\n") + .filter(_.nonEmpty) + .zipWithIndex + .map { + case (line, i) => + val node = objectMapper.readTree(line).asInstanceOf[ObjectNode] + leftOut.filter(name => node.has(name) && node.get(name).isNull).foreach(node.remove) + addNested(node, i) + objectMapper.writeValueAsString(node) + } + Files.write(path, lines.map(_ + "\n").mkString.getBytes(charset)) path } + /** The nested values of record `i`. The first item's `id` is a long past 2^53, + * missing from record 1, so the flattened `items1.id` is a nullable long. + */ + private def addNested(node: ObjectNode, i: Int): Unit = { + val items = node.putArray("a\"b\\c_items") + val first = items.addObject() + if (i != 1) first.put("id", 9007199254740993L + 2 * i) + items.addObject().put("id", i) + val tags = node.putArray("a\"b\\c_tags") + tags.add("x") + if (i % 2 == 0) tags.add("y") + node.putObject("a\"b\\c_meta").put("when", f"${i % 12 + 1}%02d/15/${2024 + i % 2} 10:00:00") + } + /** Write the rows as an uncompressed Arrow IPC ("file" format) stream — the * format both `ArrowFileReader` (Path A) and `pd.read_feather` (Path B) * read. */ - def writeArrow(dir: Path): Path = { + def writeArrow(dir: Path, rows: Seq[Tuple]): Path = { val path = dir.resolve("sample.arrow") // Texera's own Schema-to-Arrow mapping and tuple writer, so the file carries // exactly the types `ArrowUtils.toTexeraSchema` reads back on the other side. @@ -427,10 +621,14 @@ object CanonicalSourceFixture { val arrowSchema = ArrowUtils.fromTexeraSchema(schema) Using.Manager { use => val allocator = use(new RootAllocator()) - val root = use(VectorSchemaRoot.create(arrowSchema, allocator)) - root.allocateNew() - rows.zipWithIndex.foreach { case (t, i) => ArrowUtils.setTexeraTuple(t, i, root) } - root.setRowCount(rows.size) + val texera = use(VectorSchemaRoot.create(arrowSchema, allocator)) + texera.allocateNew() + rows.zipWithIndex.foreach { case (t, i) => ArrowUtils.setTexeraTuple(t, i, texera) } + val extras = arrowOnlyColumns(allocator, rows.size).map(use(_)) + val vectors = texera.getFieldVectors.asScala.toSeq ++ extras + val withNote = + new ArrowSchema(vectors.map(_.getField).asJava, Map("pandas" -> PandasIndexNote).asJava) + val root = use(new VectorSchemaRoot(withNote, vectors.asJava, rows.size)) val channel = use( FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.WRITE) ) @@ -442,6 +640,56 @@ object CanonicalSourceFixture { path } + /** The note pandas leaves in the schema of a file it wrote from a frame keyed + * by `id`. pandas reads it back and makes `id` the index rather than a column; + * the executor reads the columns the file states and knows of no index. + */ + private val PandasIndexNote: String = + """{"index_columns": ["id"], "column_indexes": [], "columns": [{"name": "id", """ + + """"field_name": "id", "pandas_type": "int32", "numpy_type": "int32", "metadata": null}], """ + + """"pandas_version": "2.2.3"}""" + + /** Columns in the widths and zones an Arrow file states and Texera's own + * mapping never writes, since it writes every float as a double and every + * integer as a signed 32 or 64 bits. Real files carry them: pyarrow writes a + * numpy float32, int16 or uint32 array as one. + * + * Row 0 holds the value each width cannot be read out of by accident: 2^24 in + * single precision, a negative in eight bits, and the largest value each + * unsigned width counts to, which its storage holds as -1. Row 1 is null in + * every one. The timestamp names a zone other than UTC, so the wall clock read + * is the file's. + * + * The values reach the executor's reading. The width and zone the script + * leaves each column in write out as the same values, and are what + * [[SourceCategoryRunner.assertDtypesFit]] checks. + */ + private def arrowOnlyColumns(allocator: RootAllocator, rowCount: Int): Seq[FieldVector] = { + val f32 = new Float4Vector("a\"b\\c_f32", allocator) + val i16 = new SmallIntVector("a\"b\\c_i16", allocator) + val i8 = new TinyIntVector("a\"b\\c_i8", allocator) + val u8 = new UInt1Vector("a\"b\\c_u8", allocator) + val u16 = new UInt2Vector("a\"b\\c_u16", allocator) + val u32 = new UInt4Vector("a\"b\\c_u32", allocator) + val zoned = new TimeStampMilliTZVector("a\"b\\c_zoned", allocator, "Asia/Tokyo") + val all = Seq[FieldVector](f32, i16, i8, u8, u16, u32, zoned) + all.foreach(_.allocateNew()) + (0 until rowCount).foreach { i => + if (i == 1) all.foreach(_.setNull(i)) + else { + f32.setSafe(i, if (i == 0) 16777216.0f else i.toFloat) + i16.setSafe(i, (i * 7).toShort) + i8.setSafe(i, (if (i == 0) -3 else i).toByte) + u8.setSafe(i, if (i == 0) 0xff else i) + u16.setSafe(i, if (i == 0) 0xffff else i) + u32.setSafe(i, if (i == 0) -1 else i) + zoned.setSafe(i, i * 3600000L) + } + } + all.foreach(_.setValueCount(rowCount)) + all + } + /** Write the rows as a Parquet file carrying the canonical table's own types. * * The mapping below is the inverse of `ParquetSchemaMapping`, which is what @@ -449,7 +697,7 @@ object CanonicalSourceFixture { * what lets a hole be a hole: Parquet has no null value, only a field that * repeats zero times, so a required column could not hold one. */ - def writeParquet(dir: Path): Path = { + def writeParquet(dir: Path, rows: Seq[Tuple]): Path = { val path = dir.resolve("sample.parquet") val messageType = parquetSchemaOf(schema) val conf = new Configuration() @@ -529,6 +777,29 @@ object TextInputHandler extends SourceHandler { desc.textInput = "alice\nbob\ncarol" desc // defaults: attributeType STRING (one row per line), attributeName "line" } + + private def typed( + text: String, + attributeType: FileAttributeType, + offset: Option[Int] + ): LogicalOp = { + val desc = new TextInputSourceOpDesc() + desc.textInput = text + desc.attributeType = attributeType + desc.fileScanOffset = offset + desc + } + + /** A line the window skips is never parsed, so it may be one the type refuses. + * A boolean line is read as the engine reads one, where 1 is true. + */ + override def scenarios: Seq[(String, Path => LogicalOp)] = + Seq( + "integers past a line skipped" -> (_ => + typed("not a number\n1\n2\n3", FileAttributeType.INTEGER, Some(1)) + ), + "booleans" -> (_ => typed("true\nFALSE\n1\n0", FileAttributeType.BOOLEAN, None)) + ) } /** Handler for `FileScanSourceOpDesc`. Plain text file read in default line mode. */ @@ -546,4 +817,80 @@ object FileScanSourceHandler extends SourceHandler { desc.fileName = Some(txtPath.toUri.toString) desc // defaults: attributeType STRING (one row per line), attributeName "line" } + + /** `extract`, `outputFileName` and `encoding` are vals, so they are + * deserialized in rather than set. + */ + private def described( + file: Path, + attributeType: FileAttributeType, + fields: String* + ): LogicalOp = { + val desc = objectMapper.readValue( + (""""operatorType":"FileScan"""" +: fields).mkString("{", ",", "}"), + classOf[FileScanSourceOpDesc] + ) + desc.fileName = Some(file.toUri.toString) + desc.attributeType = attributeType + desc + } + + private def text(dir: Path, content: String, charset: Charset = StandardCharsets.UTF_8): Path = + Files.write(dir.resolve("sample.txt"), content.getBytes(charset)) + + /** Two entries and the one macOS adds beside them, which the engine skips. */ + private def archive(dir: Path): Path = { + val path = dir.resolve("sample.zip") + Using(new ZipOutputStream(Files.newOutputStream(path))) { out => + Seq("a.txt" -> "line1\nline2\n", "b.txt" -> "line3\n", "__MACOSX/a.txt" -> "junk\n").foreach { + case (name, content) => + out.putNextEntry(new ZipEntry(name)) + out.write(content.getBytes(StandardCharsets.UTF_8)) + out.closeEntry() + } + }.get + path + } + + /** A line the window skips is never parsed, so it may be one the type refuses. + * A boolean line is read as the engine reads one, where 1 is true. The charset + * the Encoding field names is the one the bytes are in. With Extract on, the + * engine reads the entries inside the archive and names each one by its own + * name, line by line or whole. + */ + override def scenarios: Seq[(String, Path => LogicalOp)] = + Seq( + "integers past a line skipped" -> (dir => + described( + text(dir, "not a number\n1\n2\n3\n"), + FileAttributeType.INTEGER, + """"fileScanOffset":1""" + ) + ), + "booleans" -> (dir => described(text(dir, "true\nFALSE\n1\n0\n"), FileAttributeType.BOOLEAN)), + "UTF-16" -> (dir => + described( + text(dir, "première\ndeuxième\n", StandardCharsets.UTF_16), + FileAttributeType.STRING, + """"encoding":"UTF_16"""" + ) + ), + "archive" -> (dir => described(archive(dir), FileAttributeType.STRING, """"extract":true""")), + "archive, lines named" -> (dir => + described( + archive(dir), + FileAttributeType.STRING, + """"extract":true""", + """"outputFileName":true""" + ) + ), + "archive, entries named" -> (dir => + described( + archive(dir), + FileAttributeType.SINGLE_STRING, + """"extract":true""", + """"outputFileName":true""" + ) + ) + ) } From a0254b5f7b4fbad655fa199d6d9a78ba04c293b6 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Sat, 26 Sep 2026 20:09:49 -0700 Subject: [PATCH 15/17] test(verify): read source timestamps past the millisecond, with an offset, and past 2262 Co-Authored-By: Claude Opus 5.5 (1M context) --- .../verify/OperatorBehaviorSpec.scala | 10 ++-- .../verify/SourceCategoryRunner.scala | 46 +++++++++++++++---- 2 files changed, 41 insertions(+), 15 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala index 635868d0302..92d46d24e1d 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/OperatorBehaviorSpec.scala @@ -41,9 +41,9 @@ import org.scalatest.matchers.should.Matchers * Requires Python 3 with pandas on the [[Comparator]] / [[StandaloneRunner]] * resolution chain (`UDF_PYTHON_PATH` env var, then `python3.12`). */ -// Tagged @IntegrationTest: this is the only verify spec that forks a real -// Python process end-to-end, so CI routes it to the Python-provisioned -// integration job (see workflow-compiling-service/build.sbt WCS_TEST_FILTER). +// Tagged @IntegrationTest: it forks Python, so CI routes it to the +// Python-provisioned integration job (see workflow-compiling-service/build.sbt +// WCS_TEST_FILTER). @IntegrationTest class OperatorBehaviorSpec extends AnyFlatSpec with Matchers with ParallelTestExecution { @@ -55,8 +55,8 @@ class OperatorBehaviorSpec extends AnyFlatSpec with Matchers with ParallelTestEx // Only a local run sets those, and CI therefore runs the lot. name should "NARROWED OUT — outside this run's VERIFY_ONLY / VERIFY_SKIP" ignore {} } else if (classOf[SourceOperatorDescriptor].isAssignableFrom(opClass)) { - // Sources keep their handler-per-source design: each needs a real file - // in its specific format, which a generic fixture can't supply. + // A source has no input port, so its own runner writes the shared table + // into the file format each one reads. if (SourceCategoryRunner.canRun(opClass)) { name should "produce equivalent output in Texera and standalone Python (source)" in { SourceCategoryRunner.run(opClass) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala index a307de685c2..a8edad720c9 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala @@ -79,7 +79,7 @@ object SourceCategoryRunner { /** * The curated tier: sources that keep a hand-written handler because they * can't go through the shared-fixture + encoder (auto) path — their output - * isn't the shared 3-column table (text-family, single `line` column) or + * isn't the shared table (text-family, single `line` column) or * their data is inline config rather than a file. Mirrors the transform * side's [[CuratedHandlers]] (hand-written vs auto-generated fixture). */ @@ -455,7 +455,8 @@ trait SourceHandler { def rowCount: Int /** Configurations the sweep cannot reach, each with the file it has to read: - * a type the default text cannot be parsed as, or bytes in another charset. + * a type the default text cannot be parsed as, bytes in another charset, or + * an archive to extract. * Each one writes into the directory it is handed and returns the configured * op. Default: none. */ @@ -467,17 +468,17 @@ trait SourceHandler { * * A source has no input port, so the fixture is delivered not as an input JSONL * but as a file the operator opens itself. Each `writeXxx` encodes these rows - * into one on-disk format (CSV / JSONL / Arrow); a source handler picks the - * encoder its operator understands and points `fileName` at the result. So CSV, - * CSVOld, JSONL and Arrow all verify that the operator reconstructs one shared - * table, instead of each asserting against its own ad-hoc sample. + * into one on-disk format (CSV / JSONL / Arrow / Parquet); [[encoderByFileType]] + * picks the one the operator declares and points `fileName` at the result. So + * CSV, CSVOld, JSONL, Arrow and Parquet all verify that the operator reconstructs + * one shared table, instead of each asserting against its own ad-hoc sample. * * It reads the canonical table rather than a narrow one of its own. A source * fixture picked for the types that survive a round trip would be choosing not * to ask the question this suite exists to ask: these files carry no types, both * readers infer, and where they infer differently is exactly what should show. A - * date column does part them, and [[StandaloneRunner.sourceCasts]] is where that - * is settled — on Path B's reading, not by leaving the column out. + * date column does part them, and the exported reader settles that by parsing it + * as the engine does, not by leaving the column out. */ object CanonicalSourceFixture { @@ -487,10 +488,24 @@ object CanonicalSourceFixture { * `a"b\c_big` holds odd integers past 2^53, none of which a float can hold, so * a reader that widens the column through one rounds every value. `a"b\c_region` * holds the text `NA`, which pandas reads as missing unless told not to. + * `a"b\c_nanos` holds digits past the millisecond, which a reader keeps or cuts + * as the engine's own does. It stays out of the canonical table, where a + * Python operator is handed a datetime, which counts no finer than microseconds. */ private val sourceColumns: Seq[(Attribute, Int => AnyRef)] = Seq( new Attribute("a\"b\\c_big", AttributeType.LONG) -> (i => Long.box(9007199254740993L + 2 * i)), - new Attribute("a\"b\\c_region", AttributeType.STRING) -> (i => Seq("NA", "EU", "APAC")(i % 3)) + new Attribute("a\"b\\c_region", AttributeType.STRING) -> (i => Seq("NA", "EU", "APAC")(i % 3)), + new Attribute("a\"b\\c_nanos", AttributeType.TIMESTAMP) -> (i => + java.sql.Timestamp.valueOf( + Seq( + "2024-01-01 00:00:00.123456789", + "2024-06-30 23:59:59.999999999", + "2024-03-05 14:09:07.5" + )( + i % 3 + ) + ) + ) ) val schema: Schema = @@ -790,6 +805,13 @@ object TextInputHandler extends SourceHandler { desc } + /** Timestamp lines the engine reads its own way: digits past the millisecond, + * which it cuts, an offset, which it moves to the local zone, and a year + * past pandas' nanosecond range. + */ + val TimestampLines: String = + "2024-01-01 00:00:00.123456789\n2024-01-01T00:00:00+05:30\n2500-01-01 00:00:00" + /** A line the window skips is never parsed, so it may be one the type refuses. * A boolean line is read as the engine reads one, where 1 is true. */ @@ -798,7 +820,8 @@ object TextInputHandler extends SourceHandler { "integers past a line skipped" -> (_ => typed("not a number\n1\n2\n3", FileAttributeType.INTEGER, Some(1)) ), - "booleans" -> (_ => typed("true\nFALSE\n1\n0", FileAttributeType.BOOLEAN, None)) + "booleans" -> (_ => typed("true\nFALSE\n1\n0", FileAttributeType.BOOLEAN, None)), + "timestamps" -> (_ => typed(TimestampLines, FileAttributeType.TIMESTAMP, None)) ) } @@ -868,6 +891,9 @@ object FileScanSourceHandler extends SourceHandler { ) ), "booleans" -> (dir => described(text(dir, "true\nFALSE\n1\n0\n"), FileAttributeType.BOOLEAN)), + "timestamps" -> (dir => + described(text(dir, TextInputHandler.TimestampLines + "\n"), FileAttributeType.TIMESTAMP) + ), "UTF-16" -> (dir => described( text(dir, "première\ndeuxième\n", StandardCharsets.UTF_16), From 2c5cd2412bb623dd379500a5e60546a959db2bba Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 7 Oct 2026 23:33:50 -0700 Subject: [PATCH 16/17] test(verify): write each CSV variant's fixture with its own delimiter The encoders now take the variant's op rather than its charset, so a CSV fixture is separated by the delimiter the variant reads with, as it is already encoded in the charset the variant declares. A semicolon config over a comma file reads each line as one field on both paths, and their agreeing on that tests no semicolon. A value holding the delimiter is quoted, whichever it is. Co-Authored-By: Claude Opus 5.5 --- .../verify/SourceCategoryRunner.scala | 73 +++++++++++++------ 1 file changed, 49 insertions(+), 24 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala index a8edad720c9..1afb8788613 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala @@ -31,6 +31,8 @@ import org.apache.texera.amber.core.workflow.PortIdentity import org.apache.texera.amber.operator.LogicalOp import org.apache.texera.amber.operator.source.fetcher.URLFetcherOpDesc import org.apache.texera.amber.operator.source.scan.{FileAttributeType, ScanSourceOpDesc} +import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc +import org.apache.texera.amber.operator.source.scan.csvOld.CSVOldScanSourceOpDesc import org.apache.texera.amber.operator.source.scan.file.{FileScanOpDesc, FileScanSourceOpDesc} import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc import com.fasterxml.jackson.databind.node.ObjectNode @@ -98,17 +100,37 @@ object SourceCategoryRunner { * edit here. (ParallelCSV also declares "CSV" and would be covered for free, * but it is currently commented out of `@JsonSubTypes`, so the suite doesn't * enumerate it.) + * + * An encoder is handed the op it writes for, because a source's config states + * how the bytes look as well as how to read them: a delimiter, like an + * encoding, has to be in the file before a reader can be asked to honour it. */ - private val encoderByFileType: Map[String, (Path, Charset, Seq[Tuple]) => Path] = Map( - "CSV" -> CanonicalSourceFixture.writeCsv, - "CSVOld" -> CanonicalSourceFixture.writeCsv, - "JSONL" -> CanonicalSourceFixture.writeJsonl, + private val encoderByFileType: Map[String, (Path, ScanSourceOpDesc, Seq[Tuple]) => Path] = Map( + "CSV" -> ((dir, op, rows) => + CanonicalSourceFixture.writeCsv(dir, op.fileEncoding.getCharset, delimiterOf(op), rows) + ), + "CSVOld" -> ((dir, op, rows) => + CanonicalSourceFixture.writeCsv(dir, op.fileEncoding.getCharset, delimiterOf(op), rows) + ), + "JSONL" -> ((dir, op, rows) => + CanonicalSourceFixture.writeJsonl(dir, op.fileEncoding.getCharset, rows) + ), // Arrow is binary and its descriptor declares fileEncoding ignored, so the // charset a variant asks for has nothing to apply to. "Arrow" -> ((dir, _, rows) => CanonicalSourceFixture.writeArrow(dir, rows)), "Parquet" -> ((dir, _, rows) => CanonicalSourceFixture.writeParquet(dir, rows)) ) + /** The delimiter a CSV variant reads with, resolved as the readers resolve it: + * the first character, and a comma when the field is empty or unset. + */ + private def delimiterOf(op: ScanSourceOpDesc): Char = + (op match { + case csv: CSVScanSourceOpDesc => csv.customDelimiter + case csvOld: CSVOldScanSourceOpDesc => csvOld.customDelimiter + case _ => None + }).filter(_.nonEmpty).map(_.charAt(0)).getOrElse(',') + /** * Sources this runner cannot verify, with the honest reason. Mirrors * `TransformVerificationRunner.knownIssues`: the reason surfaces in the @@ -247,7 +269,7 @@ object SourceCategoryRunner { ): (String, LogicalOp, Path) = { val dir = dirFor(label) val op = newScanSource(opDescClass) - op.fileName = Some(encoder(dir, op.fileEncoding.getCharset, rows).toUri.toString) + op.fileName = Some(encoder(dir, op, rows).toUri.toString) op.offset = window._1 op.limit = window._2 (label, op, dir) @@ -284,7 +306,7 @@ object SourceCategoryRunner { private def skippedBadLine( opDescClass: Class[_ <: LogicalOp], fileType: String, - encoder: (Path, Charset, Seq[Tuple]) => Path, + encoder: (Path, ScanSourceOpDesc, Seq[Tuple]) => Path, dirFor: String => Path ): Seq[(String, LogicalOp, Path)] = if (fileType != "JSONL") Seq.empty @@ -292,7 +314,7 @@ object SourceCategoryRunner { val label = "window: skips a bad line" val dir = dirFor(label) val op = newScanSource(opDescClass) - val path = encoder(dir, op.fileEncoding.getCharset, CanonicalSourceFixture.rows) + val path = encoder(dir, op, CanonicalSourceFixture.rows) val body = new String(Files.readAllBytes(path), op.fileEncoding.getCharset) Files.write(path, ("not json at all\n" + body).getBytes(op.fileEncoding.getCharset)) op.fileName = Some(path.toUri.toString) @@ -309,7 +331,9 @@ object SourceCategoryRunner { * `fileEncoding` is swept like any other enum, and the fixture FOLLOWS it: each * variant's file is written in the charset that variant declares. Encoding is a * statement about the bytes, so a UTF_16 config over a file left in UTF-8 would - * only compare how each path fails. + * only compare how each path fails. A CSV delimiter is swept the same way and + * followed the same way: a semicolon config over a comma file reads each line as + * one field on both paths, and agreeing on that tests no semicolon at all. * * Variants that serialize identically are dropped — an operator that ignores * `fileEncoding` (Arrow declares `@JsonIgnoreProperties`) would otherwise run the @@ -317,7 +341,7 @@ object SourceCategoryRunner { */ private def generatedVariants( opDescClass: Class[_ <: LogicalOp], - encoder: (Path, Charset, Seq[Tuple]) => Path, + encoder: (Path, ScanSourceOpDesc, Seq[Tuple]) => Path, dirFor: String => Path ): Seq[(String, LogicalOp, Path)] = { val seen = mutable.Set.empty[String] @@ -341,9 +365,7 @@ object SourceCategoryRunner { // the generator's, which additionally fills limit and offset. val name = if (label == "default") "auto-base" else label val dir = dirFor(name) - scan.fileName = Some( - encoder(dir, scan.fileEncoding.getCharset, CanonicalSourceFixture.rows).toUri.toString - ) + scan.fileName = Some(encoder(dir, scan, CanonicalSourceFixture.rows).toUri.toString) Some((name, scan: LogicalOp, dir)) } } @@ -527,11 +549,12 @@ object CanonicalSourceFixture { */ val holedRows: Seq[Tuple] = SharedFixture.emptyOneCellPerColumn(rows, schema, Set.empty) - /** Write the rows as a header-first, comma-delimited CSV encoded in `charset`. + /** Write the rows as a header-first CSV separated by `delimiter` and encoded in + * `charset`. * - * The charset is a parameter because it describes the BYTES, not the config: a - * variant declaring `fileEncoding = UTF_16` over a file left in UTF-8 would - * compare nothing but how each path fails. + * Both are parameters because they describe the BYTES, not the config: a variant + * declaring `fileEncoding = UTF_16` over a file left in UTF-8, or a tab over a + * file separated by commas, would compare nothing but how each path misreads it. * * Two headers are left blank, a timestamp's and a long's, which the reader * names column-N and pandas names "Unnamed: N" until it is told otherwise, so @@ -539,20 +562,20 @@ object CanonicalSourceFixture { * position 1 is written as the text "Unnamed: 1", a name a user can really * choose and pandas' placeholder for the same place. */ - def writeCsv(dir: Path, charset: Charset, rows: Seq[Tuple]): Path = { + def writeCsv(dir: Path, charset: Charset, delimiter: Char, rows: Seq[Tuple]): Path = { val path = dir.resolve("sample.csv") val blank = Set("start_ts", "a\"b\\c_big") val header = schema.getAttributes.zipWithIndex .map { case (_, 1) => "Unnamed: 1" case (a, _) if blank.contains(a.getName) => "" - case (a, _) => csvField(a.getName) + case (a, _) => csvField(a.getName, delimiter) } - .mkString(",") + .mkString(delimiter.toString) val body = rows.map { t => schema.getAttributes - .map(a => csvField(Option(t.getField[AnyRef](a.getName)).map(_.toString).orNull)) - .mkString(",") + .map(a => csvField(Option(t.getField[AnyRef](a.getName)).map(_.toString).orNull, delimiter)) + .mkString(delimiter.toString) } Files.write(path, ((header +: body).mkString("\n") + "\n").getBytes(charset)) path @@ -563,11 +586,13 @@ object CanonicalSourceFixture { * The table carries commas inside values — a bracketed edge pair, a * comma-delimited list, an ordinary English sentence — and writing those raw * shifts every column after them. What the two paths then disagree about is a - * broken file rather than anything either of them does. + * broken file rather than anything either of them does. The same holds for + * whichever character separates the fields: under a space, the sentence needs + * its quotes as much as the list does under a comma. */ - private def csvField(value: String): String = + private def csvField(value: String, delimiter: Char): String = if (value == null) "" - else if (value.exists(c => c == ',' || c == '"' || c == '\n' || c == '\r')) + else if (value.exists(c => c == delimiter || c == '"' || c == '\n' || c == '\r')) "\"" + value.replace("\"", "\"\"") + "\"" else value From acb0cc9adca997619564f1db833a28041b149e30 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 7 Oct 2026 23:37:14 -0700 Subject: [PATCH 17/17] test(verify): quote the CSV fixture's Unnamed header under any delimiter The header at position 1 is written as the literal Unnamed: 1, which skipped the quoting every other header gets. Under a space it split into two columns, and every row then came up one field short of the header. Co-Authored-By: Claude Opus 5.5 --- .../texera/amber/translator/verify/SourceCategoryRunner.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala index 1afb8788613..168f608216b 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SourceCategoryRunner.scala @@ -567,7 +567,7 @@ object CanonicalSourceFixture { val blank = Set("start_ts", "a\"b\\c_big") val header = schema.getAttributes.zipWithIndex .map { - case (_, 1) => "Unnamed: 1" + case (_, 1) => csvField("Unnamed: 1", delimiter) case (a, _) if blank.contains(a.getName) => "" case (a, _) => csvField(a.getName, delimiter) }