From 94cfb7734016ab1256b29292aa18469b81ab7f18 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 2 Sep 2026 16:30:43 -0700 Subject: [PATCH 01/15] test(verify): hand-write the configurations a schema cannot describe Some operators need an input a schema alone cannot produce. A join needs two tables that share a key. A dictionary matcher needs a dictionary whose entries appear in the column it reads. Deriving those from annotations would mean teaching the generator what each operator means. These are written by hand instead, one entry per operator, and the entry says why the generator could not do it. Co-Authored-By: Claude Opus 5 (1M context) --- .../translator/verify/CuratedHandlers.scala | 666 ++++++++++++++++++ 1 file changed, 666 insertions(+) create mode 100644 workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala new file mode 100644 index 00000000000..52ecf4140c8 --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -0,0 +1,666 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.translator.verify + +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.operator.LogicalOp +import org.apache.texera.amber.operator.distinct.DistinctOpDesc +import org.apache.texera.amber.operator.aggregate.{ + AggregateOpDesc, + AggregationFunction, + AggregationOperation +} +import org.apache.texera.amber.operator.filter.{ + ComparisonType, + FilterPredicate, + SpecializedFilterOpDesc +} +import org.apache.texera.amber.operator.hashJoin.{HashJoinOpDesc, JoinType} +import org.apache.texera.amber.operator.keywordSearch.KeywordSearchOpDesc +import org.apache.texera.amber.operator.projection.{AttributeUnit, ProjectionOpDesc} +import org.apache.texera.amber.operator.regex.RegexOpDesc +import org.apache.texera.amber.operator.typecasting.{TypeCastingOpDesc, TypeCastingUnit} +import org.apache.texera.amber.operator.visualization.ImageViz.ImageVisualizerOpDesc + +import org.apache.texera.amber.operator.visualization.dumbbellPlot.{ + DumbbellDotConfig, + DumbbellPlotOpDesc +} +import org.apache.texera.amber.operator.sklearn.training.SklearnTrainingOpDesc +import org.apache.texera.amber.operator.sklearn.SklearnClassifierOpDesc +import org.apache.texera.amber.operator.sklearn.SklearnLinearRegressionOpDesc +import org.apache.texera.amber.operator.machineLearning.sklearnAdvanced.base.SklearnMLOperatorDescriptor +import org.apache.texera.amber.operator.ifStatement.IfOpDesc +import java.nio.file.{Files, Path} +import java.util + +/** + * A curated handler ships a configured OpDesc and the input fixtures it + * needs, written once into `testRoot`. Register it in [[CuratedHandlers.all]] + * to override the auto-config tier for that operator. + */ +trait TransformHandler { + def opDescClass: Class[_ <: LogicalOp] + def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) + + /** Extra independent scenarios beyond [[fixture]], each a self-contained + * (label, configured op, its own inputs). The runner runs each as a PINNED + * config (no enum sweep), in its own work subdir. Default: none. + * + * Used where one operator needs structurally different inputs per config + * branch that a single swept fixture can't cover — e.g. the sklearn + * `countVectorizer=true` text path, whose feature column must be text and so + * is incompatible with the numeric default fixture (`X = table.drop(target)` + * would feed a string column to a numeric estimator). Each scenario must + * write its input files somewhere unique (e.g. a `testRoot` subdir) so it + * does not clobber the primary fixture's files. + */ + def extraScenarios(testRoot: Path): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = + Seq.empty + + /** Opts this fixture into the `nulls` case, naming the columns it must never + * empty because their VALUE is what the fixture was built to arrange rather + * than data under test: a join key that has to pair, a grouping key that has + * to group. Emptying one of those changes what the test asks instead of asking + * what the operator does with a null. + * + * The default is `Some(Set.empty)`: most curated tables arrange nothing that a + * hole would disturb, so taking part is the normal case and a fixture that + * cannot afford a hole says so. `None` sits the case out entirely, for a table + * whose every column is load-bearing. + */ + def nullsKeepFilled: Option[Set[String]] = Some(Set.empty) +} + +/** + * The curated override tier of the config/fixture resolution chain: an + * operator listed here is verified with its hand-written fixture instead of + * the auto-generated one. This is also the seam where Xuan's curated + * operator-field-values JSON plugs in later, as a second curated source. + */ +object CuratedHandlers { + + /** Concrete `LogicalOp` classes discovered from the `@JsonSubTypes` registry + * on [[LogicalOp]] — the same source [[ConfigGenerator]] enumerates. The + * sklearn handler families below are auto-derived from this list, so a newly + * registered sklearn estimator is picked up with zero per-operator + * boilerplate here. + */ + private val registeredOps: Seq[Class[_ <: LogicalOp]] = + Option(classOf[LogicalOp].getAnnotation(classOf[com.fasterxml.jackson.annotation.JsonSubTypes])) + .map(_.value().toSeq.map(_.value().asInstanceOf[Class[_ <: LogicalOp]])) + .getOrElse(Seq.empty) + + private def isConcrete(cls: Class[_]): Boolean = + !java.lang.reflect.Modifier.isAbstract(cls.getModifiers) + + /** The concrete leaf ops under one sklearn base, excluding the base itself. + * + * No hard-coded baseline: a new sklearn operator is picked up automatically + * the moment it is registered in LogicalOp's @JsonSubTypes — zero per-op code + * here. The test suite (ConfigCoverageSpec / TransformVerificationRunnerSpec) + * is the safety net: a mis-discovered or misbehaving op fails its own parity + * check rather than being frozen by an assertion. + */ + private def sklearnFamily(base: Class[_]): Seq[Class[_ <: LogicalOp]] = + registeredOps.filter(c => base.isAssignableFrom(c) && c != base && isConcrete(c)) + + private def trainingOps = sklearnFamily(classOf[SklearnTrainingOpDesc]) + private def classifierOps = sklearnFamily(classOf[SklearnClassifierOpDesc]) + private def advancedOps = sklearnFamily(classOf[SklearnMLOperatorDescriptor[_]]) + + /** Every sklearn op, whichever tier serves it. `X = table.drop(target)` feeds + * each remaining column to `fit`, so these take canonical's petal-and-label + * projection rather than the whole table, whose string columns end the fit. + * + * Linear Regression is named on its own because it descends from + * `PythonOperatorDescriptor` directly rather than from one of the three + * bases, so no family picks it up. + */ + val sklearnNumericClasses: Set[Class[_ <: LogicalOp]] = + (trainingOps ++ classifierOps ++ advancedOps).toSet + classOf[SklearnLinearRegressionOpDesc] + + val all: Seq[TransformHandler] = Seq( + AggregateTransformHandler, + SpecializedFilterTransformHandler, + DistinctTransformHandler, + ProjectionTransformHandler, + HashJoinTransformHandler, + TypeCastingTransformHandler, + KeywordSearchTransformHandler, + DumbbellPlotVisualizationHandler, + ImageVisualizerVisualizationHandler, + IfTransformHandler, + RegexTransformHandler + ) + + val byClass: Map[Class[_ <: LogicalOp], TransformHandler] = + all.map(h => h.opDescClass -> h).toMap + + /** Generic fixture writer: builds a JSONL file with the given typed columns + * and rows, boxing each value per its declared [[AttributeType]]. Lets a + * curated handler declare bespoke per-operator input data in one call + * instead of hand-rolling a Schema + Tuple.builder loop. + */ + def writeFixture( + path: Path, + columns: Seq[(String, AttributeType)], + rows: Seq[Seq[Any]] + ): Path = { + val schema = new Schema(columns.map { case (n, t) => new Attribute(n, t) }: _*) + val tuples = rows.map { row => + val builder = Tuple.builder(schema) + columns.zip(row).foreach { + case ((name, attrType), value) => + val boxed: AnyRef = (attrType, value) match { + case (_, null) => null + case (AttributeType.INTEGER, x: Int) => Int.box(x) + case (AttributeType.INTEGER, x: Long) => Int.box(x.toInt) + case (AttributeType.INTEGER, x: Double) => Int.box(x.toInt) + case (AttributeType.LONG, x: Long) => Long.box(x) + case (AttributeType.LONG, x: Int) => Long.box(x.toLong) + case (AttributeType.DOUBLE, x: Double) => Double.box(x) + case (AttributeType.DOUBLE, x: Int) => Double.box(x.toDouble) + case (AttributeType.DOUBLE, x: Long) => Double.box(x.toDouble) + case (AttributeType.BOOLEAN, x: Boolean) => Boolean.box(x) + case (AttributeType.STRING, x) => x.toString + case (_, x) => x.toString + } + builder.add(schema.getAttribute(name), boxed) + } + builder.build() + } + TupleIO.writeTuples(path, tuples.iterator, schema) + path + } + +} + +/** + * Handler for `SpecializedFilterOpDesc`. Curated CONFIG over the shared + * canonical fixture: the auto tier fills a free-form predicate `value` with + * the canonical "1", which pins the shape of the comparison but not its + * corners. `id > 8 OR name == "eve"` exercises numeric comparison, string + * equality (the JSON predicate `value` is always a string) and OR-combination + * in one run, and keeps 5 of port 0's 10 rows — a proper subset either way. + * + * Both JVM `SpecializedFilterOpExec` and pandas boolean indexing preserve + * input row order, so positional comparator equality holds. + */ +object SpecializedFilterTransformHandler extends TransformHandler { + + override val opDescClass: Class[_ <: LogicalOp] = classOf[SpecializedFilterOpDesc] + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val desc = new SpecializedFilterOpDesc() + desc.predicates = List( + new FilterPredicate("id", ComparisonType.GREATER_THAN, "8"), + new FilterPredicate("name", ComparisonType.EQUAL_TO, "eve") + ) + + (desc, CanonicalFixture.writeInputs(testRoot, 1)) + } +} + +/** Handler for `DistinctOpDesc`. The canonical auto-fixture is all-distinct + * (uniq_name is globally unique by invariant), so it never exercises dedup. + * This 5-row table repeats two rows so both paths must actually drop + * duplicates; survivors keep first-occurrence order (JVM LinkedHashSet == + * pandas drop_duplicates keep="first"), so the positional comparator holds. + */ +/** + * Curated handler for [[ProjectionOpDesc]]. Its `attributes` list is not declared + * `required`, so the auto tier starts it empty the way the UI does — and + * `getPhysicalOp` refuses an empty list. Pinning one row is all this needs; the + * runner derives the rest of the variants from it. + */ +object ProjectionTransformHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[ProjectionOpDesc] + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val columns = Seq( + ("id", AttributeType.INTEGER), + ("name", AttributeType.STRING), + ("score", AttributeType.DOUBLE) + ) + val rows = Seq( + Seq[Any](1, "a", 1.5), + Seq[Any](2, "b", 2.5), + Seq[Any](3, "c", 3.5) + ) + val inputPath = + CuratedHandlers.writeFixture(testRoot.resolve("input_port_0.jsonl"), columns, rows) + val op = new ProjectionOpDesc() + // A blank alias is the untouched state of the row the `+` button adds, and it is + // the branch where the operator keeps the original name. + op.attributes = List(new AttributeUnit("id", "")) + (op, Map(PortIdentity(0) -> inputPath)) + } +} + +object DistinctTransformHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[DistinctOpDesc] + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val columns = Seq( + ("id", AttributeType.INTEGER), + ("name", AttributeType.STRING) + ) + val rows = Seq( + Seq[Any](1, "a"), + Seq[Any](2, "b"), + Seq[Any](1, "a"), // duplicate of row 0 + Seq[Any](3, "c"), + Seq[Any](2, "b") // duplicate of row 1 + ) + val inputPath = + CuratedHandlers.writeFixture(testRoot.resolve("input_port_0.jsonl"), columns, rows) + (new DistinctOpDesc(), Map(PortIdentity(0) -> inputPath)) + } +} + +/** + * Curated handler for [[RegexOpDesc]]. The auto tier only ever feeds it the + * trivial pattern `"1"` against the first column, which never exercises real + * regex semantics. This handler pins genuine patterns so the JVM↔Python engine + * parity is actually tested: + * + * - Primary fixture: `[a-z]+` over a mixed-case `text` column. The runner + * enum-sweeps the Boolean `caseInsensitive`, so BOTH branches run against + * the same data. The two branches select DIFFERENT row sets (case-sensitive + * keeps only rows with a lowercase letter; case-insensitive also keeps the + * all-caps rows), proving the flag actually flows through to both paths. + * - `extraScenarios`: `\d+` (a backslash class — verifies the escape survives + * `toPyDoubleQuotedLiteral` into Python's engine) and `\.` (an escaped + * metachar — an escaping bug would turn it into "match any char" and change + * the result, so this pins literal-vs-metachar handling). + * + * All fixture data is ASCII, where Java `\d` / `[a-z]` / CASE_INSENSITIVE and + * Python's `re` agree exactly; each pattern yields a proper subset (never + * all/none) so the comparison is meaningful. + */ +object RegexTransformHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[RegexOpDesc] + + private def regexOp(attribute: String, regex: String, caseInsensitive: Boolean): RegexOpDesc = { + val op = new RegexOpDesc() + op.attribute = attribute + op.regex = regex + op.caseInsensitive = caseInsensitive + op + } + + // Rows chosen so `[a-z]+` differs by case flag: "ABC"/"XY9" have no lowercase + // (dropped when case-sensitive) but are all-letter (kept when insensitive). + private val textColumn = Seq(("text", AttributeType.STRING)) + private val caseRows: Seq[Seq[Any]] = + Seq(Seq[Any]("abc"), Seq[Any]("ABC"), Seq[Any]("123"), Seq[Any]("a1B"), Seq[Any]("XY9")) + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val inputPath = + CuratedHandlers.writeFixture(testRoot.resolve("input_port_0.jsonl"), textColumn, caseRows) + (regexOp("text", "[a-z]+", caseInsensitive = false), Map(PortIdentity(0) -> inputPath)) + } + + override def extraScenarios( + testRoot: Path + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = { + // `\d+`: rows where a digit is present form a proper subset. + val digitDir = testRoot.resolve("digits") + Files.createDirectories(digitDir) + val digitRows: Seq[Seq[Any]] = + Seq(Seq[Any]("abc"), Seq[Any]("a1B"), Seq[Any]("XY9"), Seq[Any]("123"), Seq[Any]("ab")) + val digitInput = + CuratedHandlers.writeFixture(digitDir.resolve("input_port_0.jsonl"), textColumn, digitRows) + + // `\.`: only rows with a literal dot match. If the backslash were lost, the + // pattern would become bare `.` (match any char) and select every row. + val dotDir = testRoot.resolve("dot") + Files.createDirectories(dotDir) + val dotRows: Seq[Seq[Any]] = + Seq(Seq[Any]("a.b"), Seq[Any]("abc"), Seq[Any]("x.y.z"), Seq[Any]("no")) + val dotInput = + CuratedHandlers.writeFixture(dotDir.resolve("input_port_0.jsonl"), textColumn, dotRows) + + Seq( + ( + "regex=\\d+", + regexOp("text", "\\d+", caseInsensitive = false), + Map(PortIdentity(0) -> digitInput) + ), + ( + "regex=\\.", + regexOp("text", "\\.", caseInsensitive = false), + Map(PortIdentity(0) -> dotInput) + ) + ) + } +} + +/** HashJoin INNER on `id`. Build (port 0) and probe (port 1) intentionally + * arrive in different id orders so any probe-major / left-major mismatch + * between the JVM emit and `pd.merge` shows up. HashJoin inherits the + * unordered `LogicalOp.orderSensitive` default, so rows compare as a set. + */ +object HashJoinTransformHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[HashJoinOpDesc[_]] + + /** `id` is what the two sides pair on: empty it and the rows stop matching, so + * the run would be asking about an inner join that finds nothing rather than + * about a null. The payload columns carry no arrangement and take the holes. + */ + override def nullsKeepFilled: Option[Set[String]] = Some(Set("id")) + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val buildSchema = new Schema( + new Attribute("id", AttributeType.INTEGER), + new Attribute("name", AttributeType.STRING) + ) + val probeSchema = new Schema( + new Attribute("id", AttributeType.INTEGER), + new Attribute("score", AttributeType.INTEGER) + ) + + def buildTup(id: Int, name: String): Tuple = { + val b = Tuple.builder(buildSchema) + b.add(buildSchema.getAttribute("id"), Int.box(id)) + b.add(buildSchema.getAttribute("name"), name) + b.build() + } + def probeTup(id: Int, score: Int): Tuple = { + val b = Tuple.builder(probeSchema) + b.add(probeSchema.getAttribute("id"), Int.box(id)) + b.add(probeSchema.getAttribute("score"), Int.box(score)) + b.build() + } + + val buildRows = Seq( + buildTup(3, "carol"), + buildTup(1, "alice"), + buildTup(5, "eve"), + buildTup(2, "bob"), + buildTup(4, "dave") + ) + val probeRows = Seq( + probeTup(1, 95), + probeTup(2, 80), + probeTup(3, 88), + probeTup(4, 72), + probeTup(5, 91) + ) + val buildPath = testRoot.resolve("input_port_0.jsonl") + val probePath = testRoot.resolve("input_port_1.jsonl") + TupleIO.writeTuples(buildPath, buildRows.iterator, buildSchema) + TupleIO.writeTuples(probePath, probeRows.iterator, probeSchema) + + val desc = new HashJoinOpDesc[Integer]() + desc.buildAttributeName = "id" + desc.probeAttributeName = "id" + desc.joinType = JoinType.INNER + + (desc, Map(PortIdentity(0) -> buildPath, PortIdentity(1) -> probePath)) + } +} + +/** + * Handler for `TypeCastingOpDesc`. The auto tier points `attribute` at the + * canonical fixture's first column (`id`, INTEGER) and then sweeps `resultType` + * across ALL `AttributeType` values — but `TypeCastingUnit`'s attributeTypeRules + * only permit certain source types per target (e.g. `timestamp` accepts only + * string/long), and the native `TypeCastingOpExec` throws on an illegal cast + * (INTEGER → Timestamp). So the auto variant `resultType=timestamp` crashes + * Path A before any comparison. + * + * This fixture gives each cast a type-compatible source column and a value that + * round-trips identically on both paths (JVM `AttributeTypeUtils` vs the + * generated pandas), covering the value-comparable branches of + * `generateStandaloneCode`'s `resultType` match: STRING, INTEGER, LONG, DOUBLE, + * BOOLEAN. The op has an `enumSweep` row in + * [[TransformVerificationRunner.variantsNotRun]], suppressing the blind + * one-enum-at-a-time sweep that would re-pair each fixed column with every target + * type; the units below already exercise each branch. Map op: both paths keep + * input row order, so strict positional equality holds. + * + * TIMESTAMP is intentionally omitted: the two runtimes serialize a Timestamp + * differently to JSONL (native emits an ISO string `"2024-01-01 09:00:00.0"`, + * pandas emits epoch millis `1704099600000`), so the dataframe comparator flags + * a representation mismatch even though the instant is identical — a harness-wide + * timestamp-serialization gap, not a TypeCasting translation defect. + */ +object TypeCastingTransformHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[TypeCastingOpDesc] + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + // One dedicated source column per target so the casts don't chain. + val columns = Seq( + ("str_to_int", AttributeType.STRING), // numeric string → INTEGER + ("int_to_dbl", AttributeType.INTEGER), // integer → DOUBLE + ("int_to_str", AttributeType.INTEGER), // integer → STRING + ("int_to_lng", AttributeType.INTEGER), // integer → LONG + ("int_to_bool", AttributeType.INTEGER) // 1/0 → BOOLEAN + ) + val rows = Seq( + Seq[Any]("10", 1, 6, 11, 1), + Seq[Any]("20", 2, 7, 12, 0), + Seq[Any]("30", 3, 8, 13, 1), + Seq[Any]("40", 4, 9, 14, 0), + Seq[Any]("50", 5, 10, 15, 1) + ) + val inputPath = + CuratedHandlers.writeFixture(testRoot.resolve("input_port_0.jsonl"), columns, rows) + + def unit(attr: String, t: AttributeType): TypeCastingUnit = { + val u = new TypeCastingUnit() + u.attribute = attr + u.resultType = t + u + } + val desc = new TypeCastingOpDesc() + desc.typeCastingUnits = List( + unit("str_to_int", AttributeType.INTEGER), + unit("int_to_dbl", AttributeType.DOUBLE), + unit("int_to_str", AttributeType.STRING), + unit("int_to_lng", AttributeType.LONG), + unit("int_to_bool", AttributeType.BOOLEAN) + ) + + (desc, Map(PortIdentity(0) -> inputPath)) + } +} + +/** + * Handler for `KeywordSearchOpDesc`. The auto tier points `attribute` at the + * canonical fixture's first column (`id`) and fills `keyword` with the canonical + * "1", so the search runs against numeric ids and never touches a real text + * column. This fixture searches a genuine free-text column with a two-term + * query, exercising the standalone regex's meaningful branches — multi-term OR, + * whole-word boundaries — that both the JVM Lucene path and the pandas path + * agree on. Query "love day" keeps rows 1 and 2 (contain the whole words + * love/day); row 3 has neither; row 4's "lovely"/"today" are different tokens, + * so the shared word-boundary rule drops it. 4 rows → 2 kept. + * + * The rows are intentionally punctuation-free. The `isCaseSensitive` enum is + * swept (true and false), and the case-sensitive path uses `CaseSensitiveAnalyzer` + * (a `WhitespaceTokenizer` that leaves punctuation attached, e.g. "perfect."), + * which diverges from the standalone regex's `\b`-boundary matching on any + * punctuated word — and the standalone does NOT honor case at all. Clean + * whitespace-delimited words keep both tokenizers (and both case modes) in + * agreement; this is why the canonical fixture's punctuated `short_text` column + * cannot be reused here. Lucene phrase/boolean/wildcard syntax is likewise + * avoided — the regex approximation cannot reproduce it. + */ +object KeywordSearchTransformHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[KeywordSearchOpDesc] + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val columns = Seq(("txt", AttributeType.STRING)) + val rows = Seq( + Seq[Any]("i love this product"), + Seq[Any]("what a great day"), + Seq[Any]("terrible experience"), + Seq[Any]("lovely weather today") + ) + val inputPath = + CuratedHandlers.writeFixture(testRoot.resolve("input_port_0.jsonl"), columns, rows) + + val desc = new KeywordSearchOpDesc() + desc.attribute = "txt" + desc.keyword = "love day" + desc.isCaseSensitive = false + + (desc, Map(PortIdentity(0) -> inputPath)) + } +} + +/** DumbbellPlot: curated CONFIG over the shared canonical fixture. A dumbbell is + * one line per entity between the entity's value in two categories, so the two + * category values have to be values the entity actually has — which the auto tier + * cannot know: it fills both with the canonical string, leaving start == end and + * every line a point. + * + * `node_src` = n3 / n1 is the pair that works on this fixture: `bob` holds both + * (score 1.2 → 2.8) and so does `1` (1.7 → 0.5), giving two real dumbbells, while + * eve, dave and grace hold one each and stay single points — both branches drawn + * at once. `comparedColumnName` is a STRING column on purpose: plotly's trace + * `name` rejects a numpy number, so a numeric column raises there instead of + * plotting (reported upstream, not worked around here). + */ +object DumbbellPlotVisualizationHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[DumbbellPlotOpDesc] + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val dots = new util.ArrayList[DumbbellDotConfig]() + val dot = new DumbbellDotConfig() + dot.dotValue = "open" + dots.add(dot) + + val desc = new DumbbellPlotOpDesc() + desc.categoryColumnName = "node_src" + desc.dumbbellStartValue = "n3" + desc.dumbbellEndValue = "n1" + desc.measurementColumnName = "score" + desc.comparedColumnName = "name" + desc.dots = dots + + (desc, CanonicalFixture.writeInputs(testRoot, 1)) + } +} + +/** ImageVisualizer fixture. Uses deterministic binary payloads; the operator + * base64-encodes them into img tags. + */ +object ImageVisualizerVisualizationHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[ImageVisualizerOpDesc] + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val schema = new Schema(new Attribute("image_bytes", AttributeType.BINARY)) + + def tup(bytes: Array[Byte]): Tuple = { + val builder = Tuple.builder(schema) + builder.add(schema.getAttribute("image_bytes"), bytes) + builder.build() + } + + val rows = Seq( + tup(Array[Byte](1, 2, 3, 4)), + tup(Array[Byte](10, 20, 30, 40)) + ) + val inputPath = testRoot.resolve("input_port_0.jsonl") + TupleIO.writeTuples(inputPath, rows.iterator, schema) + + val desc = new ImageVisualizerOpDesc() + desc.binaryContent = "image_bytes" + + (desc, Map(PortIdentity(0) -> inputPath)) + } +} + +/** If operator: routes the data port (port 1) to the True (port 1) or False + * (port 0) output. We feed an EMPTY Condition port (port 0) so IfOpExec + * forwards no condition rows; with no State message it keeps its default + * active output (True), matching the standalone's default-True branch — so + * the True output gets all data rows and the False output is empty on both + * paths. + */ +/** Aggregate fixture exercising every aggregation function in one op, including + * COUNT(*) (empty attribute). Auto-config can't build this: upstream #5896 made + * `AggregationOperation.attribute` optional (required only for non-count via a + * conditional JSON-schema rule), so ConfigGenerator skips the optional autofill + * field and leaves it null — invalid for a non-count function, which NPEs in + * AggregateOpExec. This pins valid (function, column) pairs. Enum-sweep-exempt + * (see [[TransformVerificationRunner.variantsNotRun]]): the sweep flips each + * element's function in isolation and would re-pair, e.g., concat with a numeric + * column; the fixture already covers each function with a type-compatible column. + * Aggregate inherits the unordered `orderSensitive` default, so + * the comparator lex-sorts rows before comparing. + */ +object AggregateTransformHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[AggregateOpDesc] + + private def agg( + fn: AggregationFunction, + attr: String, + result: String + ): AggregationOperation = { + val a = new AggregationOperation() + a.aggFunction = fn + a.attribute = attr + a.resultAttribute = result + a + } + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val desc = new AggregateOpDesc() + desc.groupByKeys = List("name") + desc.aggregations = List( + agg(AggregationFunction.SUM, "score", "sum_score"), + agg(AggregationFunction.COUNT, "", "count_all"), // empty attribute => COUNT(*) + agg(AggregationFunction.COUNT, "score", "count_score"), + agg(AggregationFunction.AVERAGE, "score", "avg_score"), + agg(AggregationFunction.MIN, "score", "min_score"), + agg(AggregationFunction.MAX, "score", "max_score"), + agg(AggregationFunction.CONCAT, "iso_country", "cat_country") + ) + (desc, CanonicalFixture.writeInputs(testRoot, 1)) + } +} + +object IfTransformHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[IfOpDesc] + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val cols = Seq("id" -> AttributeType.INTEGER, "name" -> AttributeType.STRING) + val condition = + CuratedHandlers.writeFixture( + testRoot.resolve("input_port_0.jsonl"), + cols, + Seq.empty[Seq[Any]] + ) + val data = CuratedHandlers.writeFixture( + testRoot.resolve("input_port_1.jsonl"), + cols, + Seq(Seq(1, "a"), Seq(2, "b"), Seq(3, "c")) + ) + val desc = new IfOpDesc() + desc.conditionName = "cond" + (desc, Map(PortIdentity(0) -> condition, PortIdentity(1) -> data)) + } +} From 817bace3b2e217718d74bd52f6c592d768ac2dd4 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 2 Sep 2026 16:56:19 -0700 Subject: [PATCH 02/15] docs: state why the aggregate is curated, without the issue number The behaviour is what a reader needs; the change that introduced it is what a blame is for. Co-Authored-By: Claude Opus 5 (1M context) --- .../texera/amber/translator/verify/CuratedHandlers.scala | 9 ++++----- 1 file changed, 4 insertions(+), 5 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index 52ecf4140c8..5dd5bb0e37c 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -601,11 +601,10 @@ object ImageVisualizerVisualizationHandler extends TransformHandler { * paths. */ /** Aggregate fixture exercising every aggregation function in one op, including - * COUNT(*) (empty attribute). Auto-config can't build this: upstream #5896 made - * `AggregationOperation.attribute` optional (required only for non-count via a - * conditional JSON-schema rule), so ConfigGenerator skips the optional autofill - * field and leaves it null — invalid for a non-count function, which NPEs in - * AggregateOpExec. This pins valid (function, column) pairs. Enum-sweep-exempt + * COUNT(*) (empty attribute). Auto-config cannot build it: `attribute` is + * optional, required only for the functions other than count, so the generator + * leaves it unset and any other function then reaches the executor with no + * column to read. This pins valid (function, column) pairs. Enum-sweep-exempt * (see [[TransformVerificationRunner.variantsNotRun]]): the sweep flips each * element's function in isolation and would re-pair, e.g., concat with a numeric * column; the fixture already covers each function with a type-compatible column. From e1cf224888862ce7523235b10f1c2a2c50367b09 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 3 Sep 2026 15:36:00 -0700 Subject: [PATCH 03/15] test(verify): let Projection read the shared table like the rest The handler wrote a three-column table of its own: id, name, score. Those are the first three columns of the shared one, so the table was never the reason this operator is curated. Its config is: `attributes` is not declared required, the auto tier starts it empty as the UI does, and the operator refuses an empty list. Pinning that one row is all it needs. Dropping the table puts the operator on the same thirty-four columns every other one reads, where a projection has something to leave out. Co-Authored-By: Claude Opus 5 (1M context) --- .../translator/verify/CuratedHandlers.scala | 23 +++++-------------- 1 file changed, 6 insertions(+), 17 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index 5dd5bb0e37c..cf37b4852b5 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -227,32 +227,21 @@ object SpecializedFilterTransformHandler extends TransformHandler { * pandas drop_duplicates keep="first"), so the positional comparator holds. */ /** - * Curated handler for [[ProjectionOpDesc]]. Its `attributes` list is not declared - * `required`, so the auto tier starts it empty the way the UI does — and - * `getPhysicalOp` refuses an empty list. Pinning one row is all this needs; the - * runner derives the rest of the variants from it. + * Curated CONFIG for [[ProjectionOpDesc]] over the shared table. Its `attributes` + * list is not declared `required`, so the auto tier starts it empty the way the UI + * does — and `getPhysicalOp` refuses an empty list. Pinning one row is all this + * needs; the runner derives the rest of the variants from it, and the table stays + * the one every other operator reads. */ object ProjectionTransformHandler extends TransformHandler { override val opDescClass: Class[_ <: LogicalOp] = classOf[ProjectionOpDesc] override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { - val columns = Seq( - ("id", AttributeType.INTEGER), - ("name", AttributeType.STRING), - ("score", AttributeType.DOUBLE) - ) - val rows = Seq( - Seq[Any](1, "a", 1.5), - Seq[Any](2, "b", 2.5), - Seq[Any](3, "c", 3.5) - ) - val inputPath = - CuratedHandlers.writeFixture(testRoot.resolve("input_port_0.jsonl"), columns, rows) val op = new ProjectionOpDesc() // A blank alias is the untouched state of the row the `+` button adds, and it is // the branch where the operator keeps the original name. op.attributes = List(new AttributeUnit("id", "")) - (op, Map(PortIdentity(0) -> inputPath)) + (op, CanonicalFixture.writeInputs(testRoot, 1)) } } From bd18bdf60a511cf3cdf4162b48cabbe4d9015e05 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 4 Sep 2026 13:09:50 -0700 Subject: [PATCH 04/15] test(verify): give the type cast a string to read as a boolean Every source the fixture held was a number, and a number is the one source where Python and the engine cannot disagree: `bool(x)` is false exactly when x is zero, which is what the engine's `x != 0` says. The disagreement lives in the string column the fixture had none of, where Python answers true for "false" and for "0". The values are the ones both sides accept, since text the engine refuses would end the run before there is anything to compare. That half is pinned against `AttributeTypeUtils` in the operator's own spec. Co-Authored-By: Claude Opus 5 (1M context) --- .../translator/verify/CuratedHandlers.scala | 21 ++++++++++++------- 1 file changed, 14 insertions(+), 7 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index cf37b4852b5..f8a39d4ac46 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -444,14 +444,20 @@ object TypeCastingTransformHandler extends TransformHandler { ("int_to_dbl", AttributeType.INTEGER), // integer → DOUBLE ("int_to_str", AttributeType.INTEGER), // integer → STRING ("int_to_lng", AttributeType.INTEGER), // integer → LONG - ("int_to_bool", AttributeType.INTEGER) // 1/0 → BOOLEAN + ("int_to_bool", AttributeType.INTEGER), // 1/0 → BOOLEAN + // A string is the only source a boolean reads differently on the two + // sides: Python answers true for every non-empty one, the engine reads + // the word and then the number. Text the engine refuses cannot go here, + // since Path A would end before there is anything to compare; that half + // is pinned in TypeCastingOpDescSpec. + ("str_to_bool", AttributeType.STRING) // "true"/"0" → BOOLEAN ) val rows = Seq( - Seq[Any]("10", 1, 6, 11, 1), - Seq[Any]("20", 2, 7, 12, 0), - Seq[Any]("30", 3, 8, 13, 1), - Seq[Any]("40", 4, 9, 14, 0), - Seq[Any]("50", 5, 10, 15, 1) + Seq[Any]("10", 1, 6, 11, 1, "true"), + Seq[Any]("20", 2, 7, 12, 0, "false"), + Seq[Any]("30", 3, 8, 13, 1, "0"), + Seq[Any]("40", 4, 9, 14, 0, "1"), + Seq[Any]("50", 5, 10, 15, 1, "TRUE") ) val inputPath = CuratedHandlers.writeFixture(testRoot.resolve("input_port_0.jsonl"), columns, rows) @@ -468,7 +474,8 @@ object TypeCastingTransformHandler extends TransformHandler { unit("int_to_dbl", AttributeType.DOUBLE), unit("int_to_str", AttributeType.STRING), unit("int_to_lng", AttributeType.LONG), - unit("int_to_bool", AttributeType.BOOLEAN) + unit("int_to_bool", AttributeType.BOOLEAN), + unit("str_to_bool", AttributeType.BOOLEAN) ) (desc, Map(PortIdentity(0) -> inputPath)) From 082a6c33a9a4cb6698aa7ca41d09ced6e33e8621 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Tue, 8 Sep 2026 23:50:21 -0700 Subject: [PATCH 05/15] test(verify): give the curated tables hostile column names too These four write their own tables, so the shared table's renaming does not reach them. Each renames the column its own configuration points at, which is what puts the escaping question to the code it emits. Three of the headers here also claimed their rows compare by position. None of those operators declares itself order-sensitive, so the comparator lex-sorts both sides first, and the claim is gone. Co-Authored-By: Claude Opus 5 (1M context) --- .../translator/verify/CuratedHandlers.scala | 127 +++++++----------- 1 file changed, 49 insertions(+), 78 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index f8a39d4ac46..944e7023969 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -93,8 +93,7 @@ trait TransformHandler { /** * The curated override tier of the config/fixture resolution chain: an * operator listed here is verified with its hand-written fixture instead of - * the auto-generated one. This is also the seam where Xuan's curated - * operator-field-values JSON plugs in later, as a second curated source. + * the auto-generated one. */ object CuratedHandlers { @@ -201,9 +200,6 @@ object CuratedHandlers { * corners. `id > 8 OR name == "eve"` exercises numeric comparison, string * equality (the JSON predicate `value` is always a string) and OR-combination * in one run, and keeps 5 of port 0's 10 rows — a proper subset either way. - * - * Both JVM `SpecializedFilterOpExec` and pandas boolean indexing preserve - * input row order, so positional comparator equality holds. */ object SpecializedFilterTransformHandler extends TransformHandler { @@ -221,10 +217,10 @@ object SpecializedFilterTransformHandler extends TransformHandler { } /** Handler for `DistinctOpDesc`. The canonical auto-fixture is all-distinct - * (uniq_name is globally unique by invariant), so it never exercises dedup. + * (the name column is unique by invariant), so it never exercises dedup. * This 5-row table repeats two rows so both paths must actually drop - * duplicates; survivors keep first-occurrence order (JVM LinkedHashSet == - * pandas drop_duplicates keep="first"), so the positional comparator holds. + * duplicates, and both keep the first occurrence: JVM LinkedHashSet and pandas + * `drop_duplicates(keep="first")` agree on which of a pair survives. */ /** * Curated CONFIG for [[ProjectionOpDesc]] over the shared table. Its `attributes` @@ -268,23 +264,17 @@ object DistinctTransformHandler extends TransformHandler { /** * Curated handler for [[RegexOpDesc]]. The auto tier only ever feeds it the - * trivial pattern `"1"` against the first column, which never exercises real - * regex semantics. This handler pins genuine patterns so the JVM↔Python engine - * parity is actually tested: + * pattern `"1"`, which asks nothing of either regex engine. Three real ones do: * - * - Primary fixture: `[a-z]+` over a mixed-case `text` column. The runner - * enum-sweeps the Boolean `caseInsensitive`, so BOTH branches run against - * the same data. The two branches select DIFFERENT row sets (case-sensitive - * keeps only rows with a lowercase letter; case-insensitive also keeps the - * all-caps rows), proving the flag actually flows through to both paths. - * - `extraScenarios`: `\d+` (a backslash class — verifies the escape survives - * `toPyDoubleQuotedLiteral` into Python's engine) and `\.` (an escaped - * metachar — an escaping bug would turn it into "match any char" and change - * the result, so this pins literal-vs-metachar handling). + * - `[a-z]+` over a mixed-case column, with `caseInsensitive` swept. The two + * branches keep DIFFERENT rows, so the flag is shown to reach both paths. + * - `\d+`, a backslash class, which checks the escape survives + * `toPyDoubleQuotedLiteral` into Python's engine. + * - `\.`, an escaped metachar: an escaping bug turns it into "match any char" + * and changes the answer. * - * All fixture data is ASCII, where Java `\d` / `[a-z]` / CASE_INSENSITIVE and - * Python's `re` agree exactly; each pattern yields a proper subset (never - * all/none) so the comparison is meaningful. + * The data is ASCII, where Java and Python's `re` agree exactly, and each + * pattern keeps a proper subset rather than all or none of the rows. */ object RegexTransformHandler extends TransformHandler { override val opDescClass: Class[_ <: LogicalOp] = classOf[RegexOpDesc] @@ -299,14 +289,14 @@ object RegexTransformHandler extends TransformHandler { // Rows chosen so `[a-z]+` differs by case flag: "ABC"/"XY9" have no lowercase // (dropped when case-sensitive) but are all-letter (kept when insensitive). - private val textColumn = Seq(("text", AttributeType.STRING)) + private val textColumn = Seq(("a\"b\\c_text", AttributeType.STRING)) private val caseRows: Seq[Seq[Any]] = Seq(Seq[Any]("abc"), Seq[Any]("ABC"), Seq[Any]("123"), Seq[Any]("a1B"), Seq[Any]("XY9")) override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { val inputPath = CuratedHandlers.writeFixture(testRoot.resolve("input_port_0.jsonl"), textColumn, caseRows) - (regexOp("text", "[a-z]+", caseInsensitive = false), Map(PortIdentity(0) -> inputPath)) + (regexOp("a\"b\\c_text", "[a-z]+", caseInsensitive = false), Map(PortIdentity(0) -> inputPath)) } override def extraScenarios( @@ -332,12 +322,12 @@ object RegexTransformHandler extends TransformHandler { Seq( ( "regex=\\d+", - regexOp("text", "\\d+", caseInsensitive = false), + regexOp("a\"b\\c_text", "\\d+", caseInsensitive = false), Map(PortIdentity(0) -> digitInput) ), ( "regex=\\.", - regexOp("text", "\\.", caseInsensitive = false), + regexOp("a\"b\\c_text", "\\.", caseInsensitive = false), Map(PortIdentity(0) -> dotInput) ) ) @@ -356,27 +346,27 @@ object HashJoinTransformHandler extends TransformHandler { * the run would be asking about an inner join that finds nothing rather than * about a null. The payload columns carry no arrangement and take the holes. */ - override def nullsKeepFilled: Option[Set[String]] = Some(Set("id")) + override def nullsKeepFilled: Option[Set[String]] = Some(Set("a\"b\\c_id")) override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { val buildSchema = new Schema( - new Attribute("id", AttributeType.INTEGER), + new Attribute("a\"b\\c_id", AttributeType.INTEGER), new Attribute("name", AttributeType.STRING) ) val probeSchema = new Schema( - new Attribute("id", AttributeType.INTEGER), + new Attribute("a\"b\\c_id", AttributeType.INTEGER), new Attribute("score", AttributeType.INTEGER) ) def buildTup(id: Int, name: String): Tuple = { val b = Tuple.builder(buildSchema) - b.add(buildSchema.getAttribute("id"), Int.box(id)) + b.add(buildSchema.getAttribute("a\"b\\c_id"), Int.box(id)) b.add(buildSchema.getAttribute("name"), name) b.build() } def probeTup(id: Int, score: Int): Tuple = { val b = Tuple.builder(probeSchema) - b.add(probeSchema.getAttribute("id"), Int.box(id)) + b.add(probeSchema.getAttribute("a\"b\\c_id"), Int.box(id)) b.add(probeSchema.getAttribute("score"), Int.box(score)) b.build() } @@ -401,8 +391,8 @@ object HashJoinTransformHandler extends TransformHandler { TupleIO.writeTuples(probePath, probeRows.iterator, probeSchema) val desc = new HashJoinOpDesc[Integer]() - desc.buildAttributeName = "id" - desc.probeAttributeName = "id" + desc.buildAttributeName = "a\"b\\c_id" + desc.probeAttributeName = "a\"b\\c_id" desc.joinType = JoinType.INNER (desc, Map(PortIdentity(0) -> buildPath, PortIdentity(1) -> probePath)) @@ -410,29 +400,17 @@ object HashJoinTransformHandler extends TransformHandler { } /** - * Handler for `TypeCastingOpDesc`. The auto tier points `attribute` at the - * canonical fixture's first column (`id`, INTEGER) and then sweeps `resultType` - * across ALL `AttributeType` values — but `TypeCastingUnit`'s attributeTypeRules - * only permit certain source types per target (e.g. `timestamp` accepts only - * string/long), and the native `TypeCastingOpExec` throws on an illegal cast - * (INTEGER → Timestamp). So the auto variant `resultType=timestamp` crashes - * Path A before any comparison. - * - * This fixture gives each cast a type-compatible source column and a value that - * round-trips identically on both paths (JVM `AttributeTypeUtils` vs the - * generated pandas), covering the value-comparable branches of - * `generateStandaloneCode`'s `resultType` match: STRING, INTEGER, LONG, DOUBLE, - * BOOLEAN. The op has an `enumSweep` row in - * [[TransformVerificationRunner.variantsNotRun]], suppressing the blind - * one-enum-at-a-time sweep that would re-pair each fixed column with every target - * type; the units below already exercise each branch. Map op: both paths keep - * input row order, so strict positional equality holds. + * Handler for `TypeCastingOpDesc`. A blind sweep of `resultType` crashes Path A, + * for the reason its `enumSweep` row in + * [[TransformVerificationRunner.variantsNotRun]] gives. Each unit below instead + * pairs a target type with a source column that type accepts, holding a value + * that round-trips identically through JVM `AttributeTypeUtils` and through the + * generated pandas. * - * TIMESTAMP is intentionally omitted: the two runtimes serialize a Timestamp - * differently to JSONL (native emits an ISO string `"2024-01-01 09:00:00.0"`, - * pandas emits epoch millis `1704099600000`), so the dataframe comparator flags - * a representation mismatch even though the instant is identical — a harness-wide - * timestamp-serialization gap, not a TypeCasting translation defect. + * TIMESTAMP is left out: the two runtimes serialize one differently to JSONL, + * native as an ISO string and pandas as epoch millis, so the comparator flags a + * representation mismatch for an identical instant. That is a harness-wide gap + * rather than a TypeCasting defect. */ object TypeCastingTransformHandler extends TransformHandler { override val opDescClass: Class[_ <: LogicalOp] = classOf[TypeCastingOpDesc] @@ -493,21 +471,18 @@ object TypeCastingTransformHandler extends TransformHandler { * love/day); row 3 has neither; row 4's "lovely"/"today" are different tokens, * so the shared word-boundary rule drops it. 4 rows → 2 kept. * - * The rows are intentionally punctuation-free. The `isCaseSensitive` enum is - * swept (true and false), and the case-sensitive path uses `CaseSensitiveAnalyzer` - * (a `WhitespaceTokenizer` that leaves punctuation attached, e.g. "perfect."), - * which diverges from the standalone regex's `\b`-boundary matching on any - * punctuated word — and the standalone does NOT honor case at all. Clean - * whitespace-delimited words keep both tokenizers (and both case modes) in - * agreement; this is why the canonical fixture's punctuated `short_text` column - * cannot be reused here. Lucene phrase/boolean/wildcard syntax is likewise - * avoided — the regex approximation cannot reproduce it. + * The rows are intentionally punctuation-free. Sweeping `isCaseSensitive` puts + * the JVM's `CaseSensitiveAnalyzer` in play, a `WhitespaceTokenizer` that leaves + * punctuation attached to the word ("perfect."), while the standalone regex + * matches on `\b` boundaries and honours no case at all. Clean words keep the + * two in agreement, which is why the canonical table's punctuated `short_text` + * cannot be reused here, and why no Lucene phrase or wildcard syntax appears. */ object KeywordSearchTransformHandler extends TransformHandler { override val opDescClass: Class[_ <: LogicalOp] = classOf[KeywordSearchOpDesc] override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { - val columns = Seq(("txt", AttributeType.STRING)) + val columns = Seq(("a\"b\\c_txt", AttributeType.STRING)) val rows = Seq( Seq[Any]("i love this product"), Seq[Any]("what a great day"), @@ -518,7 +493,7 @@ object KeywordSearchTransformHandler extends TransformHandler { CuratedHandlers.writeFixture(testRoot.resolve("input_port_0.jsonl"), columns, rows) val desc = new KeywordSearchOpDesc() - desc.attribute = "txt" + desc.attribute = "a\"b\\c_txt" desc.keyword = "love day" desc.isCaseSensitive = false @@ -567,11 +542,11 @@ object ImageVisualizerVisualizationHandler extends TransformHandler { override val opDescClass: Class[_ <: LogicalOp] = classOf[ImageVisualizerOpDesc] override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { - val schema = new Schema(new Attribute("image_bytes", AttributeType.BINARY)) + val schema = new Schema(new Attribute("a\"b\\c_image_bytes", AttributeType.BINARY)) def tup(bytes: Array[Byte]): Tuple = { val builder = Tuple.builder(schema) - builder.add(schema.getAttribute("image_bytes"), bytes) + builder.add(schema.getAttribute("a\"b\\c_image_bytes"), bytes) builder.build() } @@ -583,7 +558,7 @@ object ImageVisualizerVisualizationHandler extends TransformHandler { TupleIO.writeTuples(inputPath, rows.iterator, schema) val desc = new ImageVisualizerOpDesc() - desc.binaryContent = "image_bytes" + desc.binaryContent = "a\"b\\c_image_bytes" (desc, Map(PortIdentity(0) -> inputPath)) } @@ -597,15 +572,11 @@ object ImageVisualizerVisualizationHandler extends TransformHandler { * paths. */ /** Aggregate fixture exercising every aggregation function in one op, including - * COUNT(*) (empty attribute). Auto-config cannot build it: `attribute` is + * COUNT(*) with its empty attribute. Auto-config cannot build it: `attribute` is * optional, required only for the functions other than count, so the generator - * leaves it unset and any other function then reaches the executor with no - * column to read. This pins valid (function, column) pairs. Enum-sweep-exempt - * (see [[TransformVerificationRunner.variantsNotRun]]): the sweep flips each - * element's function in isolation and would re-pair, e.g., concat with a numeric - * column; the fixture already covers each function with a type-compatible column. - * Aggregate inherits the unordered `orderSensitive` default, so - * the comparator lex-sorts rows before comparing. + * leaves it unset and any other function reaches the executor with no column to + * read. This pins a valid (function, column) pair for each. The matching + * `enumSweep` withholding is in [[TransformVerificationRunner.variantsNotRun]]. */ object AggregateTransformHandler extends TransformHandler { override val opDescClass: Class[_ <: LogicalOp] = classOf[AggregateOpDesc] From 22ae04f81ec6e110f1c1c2df17312d25cbeded8e Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 10 Sep 2026 13:14:49 -0700 Subject: [PATCH 06/15] test(verify): ask the curated tables what a rendered column looks like Regex, Keyword Search and Type Casting each gained the case their table could not put: an integer column and a boolean one, whose text a hole changes; a column named by two casting units, which the executor collapses to the last; a 64-bit value past the Int bounds; a number read as a timestamp; and text NumberFormat reads where Python's int refuses. Aggregate gained a leading empty string, which is the only value that tells the two spellings of CONCAT apart: a leading null reads the same either way, and the canonical table holds no empty strings. Keyword Search also gained a case-sensitive run over a column carrying one word in three casings. Its base table is lower-case throughout, so sweeping the flag there decided nothing. Co-Authored-By: Claude Opus 5 (1M context) --- .../translator/verify/CuratedHandlers.scala | 157 +++++++++++++++++- 1 file changed, 150 insertions(+), 7 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index 944e7023969..0925d2ab079 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -319,6 +319,38 @@ object RegexTransformHandler extends TransformHandler { val dotInput = CuratedHandlers.writeFixture(dotDir.resolve("input_port_0.jsonl"), textColumn, dotRows) + // A hole widens an integer column to float, so `^6$` meets "6.0" and selects + // nothing unless the rendering reads the declared type. + val numberDir = testRoot.resolve("numbers") + Files.createDirectories(numberDir) + // `a"b\c_big` only rides along. It is past the float64 exact-integer window, + // and its column has a hole, so a reader that parses it through a float + // hands the operator 9007199254740992 where the engine still has ...993. + val numberColumn = + Seq(("a\"b\\c_n", AttributeType.INTEGER), ("a\"b\\c_big", AttributeType.LONG)) + val numberRows: Seq[Seq[Any]] = Seq( + Seq[Any](6, 9007199254740993L), + Seq[Any](null, null), + Seq[Any](7, 1L), + Seq[Any](60, 2L), + Seq[Any](16, 3L) + ) + val numberInput = CuratedHandlers.writeFixture( + numberDir.resolve("input_port_0.jsonl"), + numberColumn, + numberRows + ) + + // A boolean column. The engine matches "true", where a hole leaves pandas + // holding 1.0. Case-sensitive on purpose, since case is half the difference. + val boolDir = testRoot.resolve("booleans") + Files.createDirectories(boolDir) + val boolColumn = Seq(("a\"b\\c_b", AttributeType.BOOLEAN)) + val boolRows: Seq[Seq[Any]] = + Seq(Seq[Any](true), Seq[Any](false), Seq[Any](null), Seq[Any](true)) + val boolInput = + CuratedHandlers.writeFixture(boolDir.resolve("input_port_0.jsonl"), boolColumn, boolRows) + Seq( ( "regex=\\d+", @@ -329,6 +361,16 @@ object RegexTransformHandler extends TransformHandler { "regex=\\.", regexOp("a\"b\\c_text", "\\.", caseInsensitive = false), Map(PortIdentity(0) -> dotInput) + ), + ( + "regex=^6$ on an integer column", + regexOp("a\"b\\c_n", "^6$", caseInsensitive = false), + Map(PortIdentity(0) -> numberInput) + ), + ( + "regex=^true$ on a boolean column", + regexOp("a\"b\\c_b", "^true$", caseInsensitive = false), + Map(PortIdentity(0) -> boolInput) ) ) } @@ -428,14 +470,40 @@ object TypeCastingTransformHandler extends TransformHandler { // the word and then the number. Text the engine refuses cannot go here, // since Path A would end before there is anything to compare; that half // is pinned in TypeCastingOpDescSpec. - ("str_to_bool", AttributeType.STRING) // "true"/"0" → BOOLEAN + ("str_to_bool", AttributeType.STRING), // "true"/"0" → BOOLEAN + // The other direction, which the five above cannot ask: the engine writes + // "true" where Python's str() writes "True". + ("bool_to_str", AttributeType.BOOLEAN), // true/false → STRING + // Named by two units at once. `tupleCasting` takes a Map, so the second + // replaces the first and casts the original value; running both would put + // the column through a datetime on the way. + ("twice_cast", AttributeType.STRING), // → TIMESTAMP, then → STRING + // Long.toInt keeps the low 32 bits rather than raising or saturating, so + // these straddle the Int bounds on purpose. + ("lng_to_int", AttributeType.LONG), // beyond Int range → INTEGER + // `new Timestamp(long)` reads the number as milliseconds; pd.to_datetime + // defaults to nanoseconds and puts every one of these in 1970. + ("ms_to_ts", AttributeType.LONG) // epoch millis → TIMESTAMP ) + // `str_to_int` carries the text NumberFormat reads and Python's int refuses: + // a decimal point, a grouping comma, trailing letters. val rows = Seq( - Seq[Any]("10", 1, 6, 11, 1, "true"), - Seq[Any]("20", 2, 7, 12, 0, "false"), - Seq[Any]("30", 3, 8, 13, 1, "0"), - Seq[Any]("40", 4, 9, 14, 0, "1"), - Seq[Any]("50", 5, 10, 15, 1, "TRUE") + Seq[Any]("10", 1, 6, 11, 1, "true", true, "2024-01-07 00:00:00", 2147483648L, 1700000000000L), + Seq[Any]("6.7", 2, 7, 12, 0, "false", false, "2024-03-15 08:30:00", 2147483647L, 0L), + Seq[Any]( + "1,234", + 3, + 8, + 13, + 1, + "0", + true, + "2024-06-01 12:00:00", + -2147483649L, + 1000000000000L + ), + Seq[Any]("12abc", 4, 9, 14, 0, "1", false, "2024-09-20 23:59:59", 0L, 1893456000000L), + Seq[Any]("-6.7", 5, 10, 15, 1, "TRUE", true, "2024-12-31 01:02:03", -1L, 946684800000L) ) val inputPath = CuratedHandlers.writeFixture(testRoot.resolve("input_port_0.jsonl"), columns, rows) @@ -453,7 +521,12 @@ object TypeCastingTransformHandler extends TransformHandler { unit("int_to_str", AttributeType.STRING), unit("int_to_lng", AttributeType.LONG), unit("int_to_bool", AttributeType.BOOLEAN), - unit("str_to_bool", AttributeType.BOOLEAN) + unit("str_to_bool", AttributeType.BOOLEAN), + unit("bool_to_str", AttributeType.STRING), + unit("twice_cast", AttributeType.TIMESTAMP), + unit("twice_cast", AttributeType.STRING), + unit("lng_to_int", AttributeType.INTEGER), + unit("ms_to_ts", AttributeType.TIMESTAMP) ) (desc, Map(PortIdentity(0) -> inputPath)) @@ -499,6 +572,49 @@ object KeywordSearchTransformHandler extends TransformHandler { (desc, Map(PortIdentity(0) -> inputPath)) } + + override def extraScenarios( + testRoot: Path + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = { + // A hole widens an integer column to float and every value grows a ".0" the + // term "0" then matches on a word boundary, so the rendering decides between + // one row and all of them. A digit term cannot: `\b6\b` finds "6.0" too. + val numberDir = testRoot.resolve("numbers") + Files.createDirectories(numberDir) + val numberColumn = Seq(("a\"b\\c_n", AttributeType.INTEGER)) + val numberRows: Seq[Seq[Any]] = + Seq(Seq[Any](6), Seq[Any](null), Seq[Any](70), Seq[Any](0)) + val numberInput = CuratedHandlers.writeFixture( + numberDir.resolve("input_port_0.jsonl"), + numberColumn, + numberRows + ) + + val desc = new KeywordSearchOpDesc() + desc.attribute = "a\"b\\c_n" + desc.keyword = "0" + desc.isCaseSensitive = false + + // The base fixture is all lower-case, so sweeping the flag there decides + // nothing: the two analyzers agree on every row it holds. + val casedDir = testRoot.resolve("cased") + Files.createDirectories(casedDir) + val casedColumn = Seq(("a\"b\\c_txt", AttributeType.STRING)) + val casedRows: Seq[Seq[Any]] = + Seq(Seq[Any]("i love this"), Seq[Any]("I Love this"), Seq[Any]("LOVE it")) + val casedInput = + CuratedHandlers.writeFixture(casedDir.resolve("input_port_0.jsonl"), casedColumn, casedRows) + + val cased = new KeywordSearchOpDesc() + cased.attribute = "a\"b\\c_txt" + cased.keyword = "Love" + cased.isCaseSensitive = true + + Seq( + ("keyword=0 on an integer column", desc, Map(PortIdentity(0) -> numberInput)), + ("keyword=Love, case sensitive", cased, Map(PortIdentity(0) -> casedInput)) + ) + } } /** DumbbellPlot: curated CONFIG over the shared canonical fixture. A dumbbell is @@ -607,6 +723,33 @@ object AggregateTransformHandler extends TransformHandler { ) (desc, CanonicalFixture.writeInputs(testRoot, 1)) } + + /** CONCAT over a column that starts a group with an empty string. + * + * The accumulator earns its separator only once it holds something, so a + * leading empty value contributes neither text nor comma. A null cannot ask + * this: it reads as the empty string either way. Canonical has none. + */ + override def extraScenarios( + testRoot: Path + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = { + val dir = testRoot.resolve("leading-empty") + Files.createDirectories(dir) + val columns = Seq(("a\"b\\c_grp", AttributeType.STRING), ("a\"b\\c_txt", AttributeType.STRING)) + val rows: Seq[Seq[Any]] = Seq( + Seq[Any]("g", ""), + Seq[Any]("g", "a"), + Seq[Any]("g", ""), + Seq[Any]("h", "x") + ) + val input = CuratedHandlers.writeFixture(dir.resolve("input_port_0.jsonl"), columns, rows) + + val desc = new AggregateOpDesc() + desc.groupByKeys = List("a\"b\\c_grp") + desc.aggregations = List(agg(AggregationFunction.CONCAT, "a\"b\\c_txt", "joined")) + + Seq(("concat over a leading empty string", desc, Map(PortIdentity(0) -> input))) + } } object IfTransformHandler extends TransformHandler { From a6c0f4329c4dc97e8b6778acbe92a8f6791ba4af Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 10 Sep 2026 16:45:42 -0700 Subject: [PATCH 07/15] 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) --- .../texera/amber/translator/verify/CuratedHandlers.scala | 7 ++----- 1 file changed, 2 insertions(+), 5 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index 0925d2ab079..ce6916ce469 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -538,11 +538,8 @@ object TypeCastingTransformHandler extends TransformHandler { * canonical fixture's first column (`id`) and fills `keyword` with the canonical * "1", so the search runs against numeric ids and never touches a real text * column. This fixture searches a genuine free-text column with a two-term - * query, exercising the standalone regex's meaningful branches — multi-term OR, - * whole-word boundaries — that both the JVM Lucene path and the pandas path - * agree on. Query "love day" keeps rows 1 and 2 (contain the whole words - * love/day); row 3 has neither; row 4's "lovely"/"today" are different tokens, - * so the shared word-boundary rule drops it. 4 rows → 2 kept. + * query, so the branches both the JVM Lucene path and the pandas path have to + * agree on, multi-term OR and whole-word boundaries, actually run. * * The rows are intentionally punctuation-free. Sweeping `isCaseSensitive` puts * the JVM's `CaseSensitiveAnalyzer` in play, a `WhitespaceTokenizer` that leaves From b1f39ec9464137e16e56ccb63e072bf22da35b61 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 18 Sep 2026 11:35:42 -0700 Subject: [PATCH 08/15] docs(verify): say why the If fixture reaches only the True route The comment sat above AggregateTransformHandler rather than the handler it describes, and it read as a choice rather than a limit. If's route is decided by a State message on the Condition port, and the harness has no State channel, so a fixture can only reach the engine's default, True. Say that, and say where the False route is covered instead. Co-Authored-By: Claude Opus 5 (1M context) --- .../translator/verify/CuratedHandlers.scala | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index ce6916ce469..5f2778e9c4d 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -677,13 +677,6 @@ object ImageVisualizerVisualizationHandler extends TransformHandler { } } -/** If operator: routes the data port (port 1) to the True (port 1) or False - * (port 0) output. We feed an EMPTY Condition port (port 0) so IfOpExec - * forwards no condition rows; with no State message it keeps its default - * active output (True), matching the standalone's default-True branch — so - * the True output gets all data rows and the False output is empty on both - * paths. - */ /** Aggregate fixture exercising every aggregation function in one op, including * COUNT(*) with its empty attribute. Auto-config cannot build it: `attribute` is * optional, required only for the functions other than count, so the generator @@ -749,6 +742,18 @@ object AggregateTransformHandler extends TransformHandler { } } +/** If routes the data port (port 1) to the True (port 1) or False (port 0) + * output, and which one is active is decided by a State message on the + * Condition port. The harness writes rows per port and has no State channel, + * so only the engine's default route is reachable here, and that default is + * True: the Condition port is fed empty, the True output takes every data row, + * and the False output is empty on both paths. + * + * What this fixture cannot reach is covered where it can be. The engine's + * False route is in IfOpExecSpec, and the exported block's own False branch in + * IfOpDescSpec, which sets the global the block reads and asserts the rows + * leave by the False output alone. + */ object IfTransformHandler extends TransformHandler { override val opDescClass: Class[_ <: LogicalOp] = classOf[IfOpDesc] From 5bd68fe688e41444409f77799d0945b7ff96e1c8 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 18 Sep 2026 16:41:28 -0700 Subject: [PATCH 09/15] test(verify): search a punctuated word with the case-sensitive analyzer The note explaining the punctuation-free rows described an analyzer that tokenized on whitespace, which stopped being true when CaseSensitiveAnalyzer moved to the StandardTokenizer: a term next to punctuation is a term to the engine and to a word-boundary regex alike. Say what the two do part on instead, a word punctuation belongs inside, and let the case-sensitive scenario carry a row that ends its term in a period. Co-Authored-By: Claude Opus 5 (1M context) --- .../translator/verify/CuratedHandlers.scala | 24 ++++++++++++------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index 5f2778e9c4d..6175ac94374 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -541,12 +541,13 @@ object TypeCastingTransformHandler extends TransformHandler { * query, so the branches both the JVM Lucene path and the pandas path have to * agree on, multi-term OR and whole-word boundaries, actually run. * - * The rows are intentionally punctuation-free. Sweeping `isCaseSensitive` puts - * the JVM's `CaseSensitiveAnalyzer` in play, a `WhitespaceTokenizer` that leaves - * punctuation attached to the word ("perfect."), while the standalone regex - * matches on `\b` boundaries and honours no case at all. Clean words keep the - * two in agreement, which is why the canonical table's punctuated `short_text` - * cannot be reused here, and why no Lucene phrase or wildcard syntax appears. + * Both analyzers tokenize on Unicode word boundaries, so a term next to + * punctuation is a term to either path, and the case-sensitive scenario below + * carries such a row. What the two do part on is a word punctuation belongs + * inside: the tokenizer keeps "don't" whole while `\bdon\b` finds the "don" in + * it, so no row holds one. No Lucene phrase or wildcard syntax appears either, + * since the script matches the terms a query is written with rather than + * running the query. */ object KeywordSearchTransformHandler extends TransformHandler { override val opDescClass: Class[_ <: LogicalOp] = classOf[KeywordSearchOpDesc] @@ -593,12 +594,19 @@ object KeywordSearchTransformHandler extends TransformHandler { desc.isCaseSensitive = false // The base fixture is all lower-case, so sweeping the flag there decides - // nothing: the two analyzers agree on every row it holds. + // nothing: the two analyzers agree on every row it holds. The last row ends + // the term in a period, which the case-sensitive analyzer tokenizes away and + // `\b` reads as a boundary, so both paths keep it. val casedDir = testRoot.resolve("cased") Files.createDirectories(casedDir) val casedColumn = Seq(("a\"b\\c_txt", AttributeType.STRING)) val casedRows: Seq[Seq[Any]] = - Seq(Seq[Any]("i love this"), Seq[Any]("I Love this"), Seq[Any]("LOVE it")) + Seq( + Seq[Any]("i love this"), + Seq[Any]("I Love this"), + Seq[Any]("LOVE it"), + Seq[Any]("everything was absolutely Love.") + ) val casedInput = CuratedHandlers.writeFixture(casedDir.resolve("input_port_0.jsonl"), casedColumn, casedRows) From adab605e03aa99c7bee492a17fa3520123dd919d Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 24 Sep 2026 20:44:14 -0700 Subject: [PATCH 10/15] test(verify): hand the model-port operators a fitted model Sklearn Prediction and Sklearn Testing read a fitted model on port 0, so no auto fixture can run them. The handlers fit a DecisionTree in Python on the table they write to port 1 and write it to port 0 the way Tuple.cast_to_schema writes a model into a BINARY field. The last row is held out of the fit and lands on the wrong side, so no score is perfect and the averaging behind it is actually compared. Testing also runs a regressor on a numeric target, which covers the isRegression branch. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../translator/verify/CuratedHandlers.scala | 133 +++++++++++++++++- 1 file changed, 132 insertions(+), 1 deletion(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index 6175ac94374..409470a3d53 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -47,6 +47,8 @@ import org.apache.texera.amber.operator.visualization.dumbbellPlot.{ import org.apache.texera.amber.operator.sklearn.training.SklearnTrainingOpDesc import org.apache.texera.amber.operator.sklearn.SklearnClassifierOpDesc import org.apache.texera.amber.operator.sklearn.SklearnLinearRegressionOpDesc +import org.apache.texera.amber.operator.sklearn.SklearnPredictionOpDesc +import org.apache.texera.amber.operator.sklearn.testing.SklearnTestingOpDesc import org.apache.texera.amber.operator.machineLearning.sklearnAdvanced.base.SklearnMLOperatorDescriptor import org.apache.texera.amber.operator.ifStatement.IfOpDesc import java.nio.file.{Files, Path} @@ -148,7 +150,9 @@ object CuratedHandlers { DumbbellPlotVisualizationHandler, ImageVisualizerVisualizationHandler, IfTransformHandler, - RegexTransformHandler + RegexTransformHandler, + SklearnPredictionTransformHandler, + SklearnTestingTransformHandler ) val byClass: Map[Class[_ <: LogicalOp], TransformHandler] = @@ -783,3 +787,130 @@ object IfTransformHandler extends TransformHandler { (desc, Map(PortIdentity(0) -> condition, PortIdentity(1) -> data)) } } + +/** A model port holding one fitted estimator, and the table it is scored on. + * + * The model is fitted in Python, since no JVM value is an sklearn model, and + * written the way Tuple.cast_to_schema writes one into a BINARY field: the + * pickle marker, then the pickle. Both paths unpickle a cell carrying the + * marker before the operator sees it, as the worker does. + * + * The label comes first because the operators' optional label knobs are filled + * with port 1's first column, and a feature in that place would drop a column + * the model was fitted on. + */ +private object FittedModelFixture { + val label = "a\"b\\c_species" + val model = "a\"b\\c_model" + val columns: Seq[(String, AttributeType)] = Seq( + label -> AttributeType.STRING, + "petal_length" -> AttributeType.DOUBLE, + "petal_width" -> AttributeType.DOUBLE + ) + val rows: Seq[Seq[Any]] = Seq( + Seq("setosa", 1.4, 0.2), + Seq("setosa", 1.3, 0.2), + Seq("versicolor", 4.7, 1.4), + Seq("versicolor", 4.5, 1.5), + Seq("virginica", 6.0, 2.5), + Seq("virginica", 5.1, 1.9), + // Held out of the fit, and on virginica's side of both features, so the + // model gets it wrong and no score is perfect. A perfect score reads the + // same whatever averaging computed it. + Seq("versicolor", 5.0, 1.8) + ) + + /** Writes the table to port 1 and a model fitted on all but its last row to + * port 0. + * + * @param estimator a DecisionTree class from `sklearn.tree` + * @param target the column the model predicts; every other numeric column + * is a feature + */ + def write(dir: Path, estimator: String, target: String): Map[PortIdentity, Path] = { + val data = CuratedHandlers.writeFixture(dir.resolve("input_port_1.jsonl"), columns, rows) + val fit = + s"""import base64, pickle, sys + |import pandas as pd + |from sklearn.tree import $estimator + |table = pd.read_json(sys.argv[1], lines=True).iloc[:-1] + |X = table.drop(sys.argv[2], axis=1).select_dtypes("number") + |fitted = $estimator(random_state=0).fit(X, table[sys.argv[2]]) + |print(base64.b64encode(b"pickle " + pickle.dumps(fitted)).decode("ascii")) + |""".stripMargin + val python = sys.env.get("UDF_PYTHON_PATH").filter(_.nonEmpty).getOrElse("python3.12") + val out = new java.io.ByteArrayOutputStream() + val exit = + scala.sys.process.Process(Seq(python, "-c", fit, data.toString, target)).#>(out).! + require(exit == 0, s"fitting the fixture's $estimator exited with $exit") + val cell = java.util.Base64.getDecoder.decode(out.toString("US-ASCII").trim) + + val modelSchema = new Schema(new Attribute(model, AttributeType.BINARY)) + val modelPath = dir.resolve("input_port_0.jsonl") + TupleIO.writeTuples( + modelPath, + Iterator(Tuple.builder(modelSchema).add(modelSchema.getAttribute(model), cell).build()), + modelSchema + ) + Map(PortIdentity(0) -> modelPath, PortIdentity(1) -> data) + } +} + +/** Prediction needs a fitted model on its model port, which no auto fixture can + * hold. The model cell stays filled under the nulls case: without it there is + * no model to predict with, which asks nothing about a null in the data. + */ +object SklearnPredictionTransformHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[SklearnPredictionOpDesc] + + override def nullsKeepFilled: Option[Set[String]] = Some(Set(FittedModelFixture.model)) + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val desc = new SklearnPredictionOpDesc() + desc.model = FittedModelFixture.model + desc.resultAttribute = "prediction" + ( + desc, + FittedModelFixture.write(testRoot, "DecisionTreeClassifier", FittedModelFixture.label) + ) + } +} + +/** Testing scores a fitted model on its model port. `isRegression` has to match + * the model the port carries, so the classifier fixture is not swept to it + * (see [[TransformVerificationRunner.variantsNotRun]]) and a regressor fitted on + * a numeric target covers the other branch. + */ +object SklearnTestingTransformHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[SklearnTestingOpDesc] + + override def nullsKeepFilled: Option[Set[String]] = Some(Set(FittedModelFixture.model)) + + private def testing(target: String, isRegression: Boolean): SklearnTestingOpDesc = { + val desc = new SklearnTestingOpDesc() + desc.model = FittedModelFixture.model + desc.target = target + desc.isRegression = isRegression + desc + } + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = + ( + testing(FittedModelFixture.label, isRegression = false), + FittedModelFixture.write(testRoot, "DecisionTreeClassifier", FittedModelFixture.label) + ) + + override def extraScenarios( + testRoot: Path + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = { + val dir = testRoot.resolve("regression") + Files.createDirectories(dir) + Seq( + ( + "regression", + testing("petal_width", isRegression = true), + FittedModelFixture.write(dir, "DecisionTreeRegressor", "petal_width") + ) + ) + } +} From 59592c694dc442ab1936a4c5b68257c0eb5e8860 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 24 Sep 2026 20:59:05 -0700 Subject: [PATCH 11/15] test(verify): fit the fixture's model with the interpreter the harness runs The fixture read UDF_PYTHON_PATH itself. PyOpExecHarness.resolvePython is the one place that says which interpreter the harness runs, and the model has to be pickled by the same one that unpickles it. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../texera/amber/translator/verify/CuratedHandlers.scala | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index 409470a3d53..4e01bcca126 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -838,10 +838,11 @@ private object FittedModelFixture { |fitted = $estimator(random_state=0).fit(X, table[sys.argv[2]]) |print(base64.b64encode(b"pickle " + pickle.dumps(fitted)).decode("ascii")) |""".stripMargin - val python = sys.env.get("UDF_PYTHON_PATH").filter(_.nonEmpty).getOrElse("python3.12") val out = new java.io.ByteArrayOutputStream() - val exit = - scala.sys.process.Process(Seq(python, "-c", fit, data.toString, target)).#>(out).! + val exit = scala.sys.process + .Process(Seq(PyOpExecHarness.resolvePython(), "-c", fit, data.toString, target)) + .#>(out) + .! require(exit == 0, s"fitting the fixture's $estimator exited with $exit") val cell = java.util.Base64.getDecoder.decode(out.toString("US-ASCII").trim) From 12c224c18257bb34a86a971b64e87677e60a6a7e Mon Sep 17 00:00:00 2001 From: kary zheng Date: Fri, 25 Sep 2026 23:59:18 -0700 Subject: [PATCH 12/15] test(verify): join and aggregate the cases the operator specs used to pin HashJoin gains scenarios for names that collide across and within the two sides and for outer joins whose unmatched rows leave holes in integer, long and differently typed columns. Aggregate gains the engine's own arithmetic: an INTEGER sum that wraps, a timestamp sum and average, and CONCAT over booleans. The fixture writer now takes a TIMESTAMP value. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../translator/verify/CuratedHandlers.scala | 158 ++++++++++++++++-- 1 file changed, 145 insertions(+), 13 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index 4e01bcca126..d9844f8c181 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -174,18 +174,19 @@ object CuratedHandlers { columns.zip(row).foreach { case ((name, attrType), value) => val boxed: AnyRef = (attrType, value) match { - case (_, null) => null - case (AttributeType.INTEGER, x: Int) => Int.box(x) - case (AttributeType.INTEGER, x: Long) => Int.box(x.toInt) - case (AttributeType.INTEGER, x: Double) => Int.box(x.toInt) - case (AttributeType.LONG, x: Long) => Long.box(x) - case (AttributeType.LONG, x: Int) => Long.box(x.toLong) - case (AttributeType.DOUBLE, x: Double) => Double.box(x) - case (AttributeType.DOUBLE, x: Int) => Double.box(x.toDouble) - case (AttributeType.DOUBLE, x: Long) => Double.box(x.toDouble) - case (AttributeType.BOOLEAN, x: Boolean) => Boolean.box(x) - case (AttributeType.STRING, x) => x.toString - case (_, x) => x.toString + case (_, null) => null + case (AttributeType.INTEGER, x: Int) => Int.box(x) + case (AttributeType.INTEGER, x: Long) => Int.box(x.toInt) + case (AttributeType.INTEGER, x: Double) => Int.box(x.toInt) + case (AttributeType.LONG, x: Long) => Long.box(x) + case (AttributeType.LONG, x: Int) => Long.box(x.toLong) + case (AttributeType.DOUBLE, x: Double) => Double.box(x) + case (AttributeType.DOUBLE, x: Int) => Double.box(x.toDouble) + case (AttributeType.DOUBLE, x: Long) => Double.box(x.toDouble) + case (AttributeType.BOOLEAN, x: Boolean) => Boolean.box(x) + case (AttributeType.TIMESTAMP, x: java.sql.Timestamp) => x + case (AttributeType.STRING, x) => x.toString + case (_, x) => x.toString } builder.add(schema.getAttribute(name), boxed) } @@ -443,6 +444,105 @@ object HashJoinTransformHandler extends TransformHandler { (desc, Map(PortIdentity(0) -> buildPath, PortIdentity(1) -> probePath)) } + + private type Columns = Seq[(String, AttributeType)] + + private def scenario( + testRoot: Path, + label: String, + left: (Columns, Seq[Seq[Any]]), + right: (Columns, Seq[Seq[Any]]), + build: String, + probe: String, + joinType: JoinType + ): (String, LogicalOp, Map[PortIdentity, Path]) = { + val dir = testRoot.resolve(label.replaceAll("[^A-Za-z0-9]+", "_")) + Files.createDirectories(dir) + val desc = new HashJoinOpDesc[Integer]() + desc.buildAttributeName = build + desc.probeAttributeName = probe + desc.joinType = joinType + ( + label, + desc, + Map( + PortIdentity(0) -> CuratedHandlers + .writeFixture(dir.resolve("input_port_0.jsonl"), left._1, left._2), + PortIdentity(1) -> CuratedHandlers + .writeFixture(dir.resolve("input_port_1.jsonl"), right._1, right._2) + ) + ) + } + + /** The tables the default one cannot be: names that collide once, twice, or + * with the key being set aside, and outer joins whose unmatched rows leave + * holes in integer columns or carry a column the left side also names. + * + * A missing key beside a NaN one is not among them. The script reads a double + * column into float64, where the two are already one value before the + * operator runs, so the operator's own spec asks it with a nullable column. + */ + override def extraScenarios( + testRoot: Path + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = { + import AttributeType._ + Seq( + scenario( + testRoot, + "left names a column like the right key", + (Seq("id" -> INTEGER, "key" -> STRING), Seq(Seq(1, "payload"))), + (Seq("key" -> INTEGER, "value" -> INTEGER), Seq(Seq(1, 9))), + "id", + "key", + JoinType.INNER + ), + scenario( + testRoot, + "a right column colliding twice", + (Seq("k" -> INTEGER, "x" -> INTEGER), Seq(Seq(1, 10))), + (Seq("k" -> INTEGER, "x" -> INTEGER, "x#@1" -> INTEGER), Seq(Seq(1, 20, 30))), + "k", + "k", + JoinType.INNER + ), + scenario( + testRoot, + "a probe key named like a renamed payload", + (Seq("k" -> INTEGER, "x" -> STRING), Seq(Seq(1, "left"))), + (Seq("x#@1" -> INTEGER, "x" -> STRING), Seq(Seq(1, "right"))), + "k", + "x#@1", + JoinType.INNER + ), + scenario( + testRoot, + "outer join over integers", + (Seq("k" -> INTEGER, "x" -> INTEGER), Seq(Seq(1, 10), Seq(2, 20))), + (Seq("kk" -> INTEGER, "y" -> INTEGER), Seq(Seq(1, 30), Seq(3, 40))), + "k", + "kk", + JoinType.FULL_OUTER + ), + scenario( + testRoot, + "outer join over a long past 2^53", + (Seq("k" -> LONG, "id" -> LONG), Seq(Seq(1L, 9007199254740993L))), + (Seq("kk" -> LONG, "y" -> STRING), Seq(Seq(2L, "b"))), + "k", + "kk", + JoinType.FULL_OUTER + ), + scenario( + testRoot, + "outer join over a name both sides type apart", + (Seq("k" -> INTEGER, "x" -> INTEGER), Seq(Seq(1, 10), Seq(2, 20))), + (Seq("kk" -> INTEGER, "x" -> DOUBLE), Seq(Seq(1, 1.5), Seq(3, 2.5))), + "k", + "kk", + JoinType.FULL_OUTER + ) + ) + } } /** @@ -750,7 +850,39 @@ object AggregateTransformHandler extends TransformHandler { desc.groupByKeys = List("a\"b\\c_grp") desc.aggregations = List(agg(AggregationFunction.CONCAT, "a\"b\\c_txt", "joined")) - Seq(("concat over a leading empty string", desc, Map(PortIdentity(0) -> input))) + Seq( + ("concat over a leading empty string", desc, Map(PortIdentity(0) -> input)), + engineArithmetic(testRoot) + ) + } + + /** The answers the engine's own arithmetic gives: an INTEGER sum adds as a + * Java int and wraps, a timestamp sum is a timestamp, a timestamp average is + * a DOUBLE, and CONCAT folds a boolean through Java's toString, in lower case. + */ + private def engineArithmetic(testRoot: Path): (String, LogicalOp, Map[PortIdentity, Path]) = { + val dir = testRoot.resolve("engine-arithmetic") + Files.createDirectories(dir) + val columns = Seq( + ("a\"b\\c_i", AttributeType.INTEGER), + ("a\"b\\c_t", AttributeType.TIMESTAMP), + ("a\"b\\c_b", AttributeType.BOOLEAN) + ) + val rows: Seq[Seq[Any]] = Seq( + Seq[Any](Int.MaxValue, java.sql.Timestamp.valueOf("2020-01-01 00:00:00"), true), + Seq[Any](1, java.sql.Timestamp.valueOf("2020-01-02 00:00:00"), false) + ) + val input = CuratedHandlers.writeFixture(dir.resolve("input_port_0.jsonl"), columns, rows) + + val desc = new AggregateOpDesc() + desc.groupByKeys = List.empty + desc.aggregations = List( + agg(AggregationFunction.SUM, "a\"b\\c_i", "int_total"), + agg(AggregationFunction.SUM, "a\"b\\c_t", "ts_total"), + agg(AggregationFunction.AVERAGE, "a\"b\\c_t", "ts_avg"), + agg(AggregationFunction.CONCAT, "a\"b\\c_b", "flags") + ) + ("the engine's own arithmetic", desc, Map(PortIdentity(0) -> input)) } } From af319dffeaac4c4ed6113500b3ef01c413c5cc37 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Sat, 26 Sep 2026 13:28:24 -0700 Subject: [PATCH 13/15] test(verify): join over the shared table and outer-join keys a left column names HashJoin's default run now reads the shared table, whose two ports overlap in five of their ten rows, so the swept join types part from one another and every column but the key collides. Two scenarios join keys that the left key or a left payload names, across an outer join. Co-Authored-By: Claude Opus 5.5 (1M context) --- .../translator/verify/CuratedHandlers.scala | 83 +++++++------------ 1 file changed, 32 insertions(+), 51 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index d9844f8c181..2176a0c32c9 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -381,10 +381,14 @@ object RegexTransformHandler extends TransformHandler { } } -/** HashJoin INNER on `id`. Build (port 0) and probe (port 1) intentionally - * arrive in different id orders so any probe-major / left-major mismatch - * between the JVM emit and `pd.merge` shows up. HashJoin inherits the - * unordered `LogicalOp.orderSensitive` default, so rows compare as a set. +/** HashJoin on `id` over the shared table, whose two ports overlap in five of + * their ten rows and hold them in different id orders. Each side keeps five rows + * the other lacks, so the swept join types part from one another, and every + * column but the key collides. HashJoin inherits the unordered + * `LogicalOp.orderSensitive` default, so rows compare as a set. + * + * The handler exists for its [[extraScenarios]]; the config it pins is the one + * the auto tier would pick. */ object HashJoinTransformHandler extends TransformHandler { override val opDescClass: Class[_ <: LogicalOp] = classOf[HashJoinOpDesc[_]] @@ -393,56 +397,14 @@ object HashJoinTransformHandler extends TransformHandler { * the run would be asking about an inner join that finds nothing rather than * about a null. The payload columns carry no arrangement and take the holes. */ - override def nullsKeepFilled: Option[Set[String]] = Some(Set("a\"b\\c_id")) + override def nullsKeepFilled: Option[Set[String]] = Some(Set("id")) override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { - val buildSchema = new Schema( - new Attribute("a\"b\\c_id", AttributeType.INTEGER), - new Attribute("name", AttributeType.STRING) - ) - val probeSchema = new Schema( - new Attribute("a\"b\\c_id", AttributeType.INTEGER), - new Attribute("score", AttributeType.INTEGER) - ) - - def buildTup(id: Int, name: String): Tuple = { - val b = Tuple.builder(buildSchema) - b.add(buildSchema.getAttribute("a\"b\\c_id"), Int.box(id)) - b.add(buildSchema.getAttribute("name"), name) - b.build() - } - def probeTup(id: Int, score: Int): Tuple = { - val b = Tuple.builder(probeSchema) - b.add(probeSchema.getAttribute("a\"b\\c_id"), Int.box(id)) - b.add(probeSchema.getAttribute("score"), Int.box(score)) - b.build() - } - - val buildRows = Seq( - buildTup(3, "carol"), - buildTup(1, "alice"), - buildTup(5, "eve"), - buildTup(2, "bob"), - buildTup(4, "dave") - ) - val probeRows = Seq( - probeTup(1, 95), - probeTup(2, 80), - probeTup(3, 88), - probeTup(4, 72), - probeTup(5, 91) - ) - val buildPath = testRoot.resolve("input_port_0.jsonl") - val probePath = testRoot.resolve("input_port_1.jsonl") - TupleIO.writeTuples(buildPath, buildRows.iterator, buildSchema) - TupleIO.writeTuples(probePath, probeRows.iterator, probeSchema) - val desc = new HashJoinOpDesc[Integer]() - desc.buildAttributeName = "a\"b\\c_id" - desc.probeAttributeName = "a\"b\\c_id" + desc.buildAttributeName = "id" + desc.probeAttributeName = "id" desc.joinType = JoinType.INNER - - (desc, Map(PortIdentity(0) -> buildPath, PortIdentity(1) -> probePath)) + (desc, CanonicalFixture.writeInputs(testRoot, 2)) } private type Columns = Seq[(String, AttributeType)] @@ -476,7 +438,8 @@ object HashJoinTransformHandler extends TransformHandler { /** The tables the default one cannot be: names that collide once, twice, or * with the key being set aside, and outer joins whose unmatched rows leave - * holes in integer columns or carry a column the left side also names. + * holes in integer columns or carry a column the left side also names, or + * carry a key that the left key or a left payload names. * * A missing key beside a NaN one is not among them. The script reads a double * column into float64, where the two are already one value before the @@ -540,6 +503,24 @@ object HashJoinTransformHandler extends TransformHandler { "k", "kk", JoinType.FULL_OUTER + ), + scenario( + testRoot, + "outer join over keys both sides name alike", + (Seq("k" -> INTEGER, "x" -> INTEGER), Seq(Seq(1, 10), Seq(2, 20))), + (Seq("k" -> INTEGER, "y" -> INTEGER), Seq(Seq(1, 30), Seq(3, 40))), + "k", + "k", + JoinType.FULL_OUTER + ), + scenario( + testRoot, + "outer join whose right key a left payload names", + (Seq("id" -> INTEGER, "key" -> STRING), Seq(Seq(1, "payload"), Seq(2, "kept"))), + (Seq("key" -> INTEGER, "value" -> INTEGER), Seq(Seq(1, 9), Seq(3, 8))), + "id", + "key", + JoinType.FULL_OUTER ) ) } From ca47426c9b661d43e1e6177209e62094f16ffe81 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Sat, 26 Sep 2026 14:30:01 -0700 Subject: [PATCH 14/15] test(verify): score one surviving row and two model rows, and drop two join tables that asked nothing new Co-Authored-By: Claude Opus 5.5 (1M context) --- .../translator/verify/CuratedHandlers.scala | 106 ++++++++++++------ 1 file changed, 72 insertions(+), 34 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index 2176a0c32c9..c7b31244a18 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -437,9 +437,10 @@ object HashJoinTransformHandler extends TransformHandler { } /** The tables the default one cannot be: names that collide once, twice, or - * with the key being set aside, and outer joins whose unmatched rows leave - * holes in integer columns or carry a column the left side also names, or - * carry a key that the left key or a left payload names. + * with the key being set aside, and outer joins whose unmatched rows leave a + * hole beside a long past 2^53, carry a column the left side types apart, or + * carry a key that a left payload names. A hole beside a small integer is not + * among them: the declared type writes it back as an integer either way. * * A missing key beside a NaN one is not among them. The script reads a double * column into float64, where the two are already one value before the @@ -477,15 +478,6 @@ object HashJoinTransformHandler extends TransformHandler { "x#@1", JoinType.INNER ), - scenario( - testRoot, - "outer join over integers", - (Seq("k" -> INTEGER, "x" -> INTEGER), Seq(Seq(1, 10), Seq(2, 20))), - (Seq("kk" -> INTEGER, "y" -> INTEGER), Seq(Seq(1, 30), Seq(3, 40))), - "k", - "kk", - JoinType.FULL_OUTER - ), scenario( testRoot, "outer join over a long past 2^53", @@ -504,15 +496,6 @@ object HashJoinTransformHandler extends TransformHandler { "kk", JoinType.FULL_OUTER ), - scenario( - testRoot, - "outer join over keys both sides name alike", - (Seq("k" -> INTEGER, "x" -> INTEGER), Seq(Seq(1, 10), Seq(2, 20))), - (Seq("k" -> INTEGER, "y" -> INTEGER), Seq(Seq(1, 30), Seq(3, 40))), - "k", - "k", - JoinType.FULL_OUTER - ), scenario( testRoot, "outer join whose right key a left payload names", @@ -901,7 +884,8 @@ object IfTransformHandler extends TransformHandler { } } -/** A model port holding one fitted estimator, and the table it is scored on. +/** A model port holding fitted estimators, one per row, and the table they are + * scored on. * * The model is fitted in Python, since no JVM value is an sklearn model, and * written the way Tuple.cast_to_schema writes one into a BINARY field: the @@ -940,30 +924,57 @@ private object FittedModelFixture { * @param target the column the model predicts; every other numeric column * is a feature */ - def write(dir: Path, estimator: String, target: String): Map[PortIdentity, Path] = { - val data = CuratedHandlers.writeFixture(dir.resolve("input_port_1.jsonl"), columns, rows) + def write(dir: Path, estimator: String, target: String): Map[PortIdentity, Path] = + writeModels(dir, Seq(s"$estimator(random_state=0)"), target) + + /** [[write]] with one model row per estimator, each fitted on all but the last + * of [[rows]], and port 1 holding `scored` where it is given. + * + * @param estimators constructor calls on the `sklearn.tree` classes + */ + def writeModels( + dir: Path, + estimators: Seq[String], + target: String, + scored: Option[Seq[Seq[Any]]] = None + ): Map[PortIdentity, Path] = { + val fitData = CuratedHandlers.writeFixture(dir.resolve("fit.jsonl"), columns, rows) + val data = CuratedHandlers.writeFixture( + dir.resolve("input_port_1.jsonl"), + columns, + scored.getOrElse(rows) + ) val fit = s"""import base64, pickle, sys |import pandas as pd - |from sklearn.tree import $estimator + |from sklearn.tree import DecisionTreeClassifier, DecisionTreeRegressor |table = pd.read_json(sys.argv[1], lines=True).iloc[:-1] |X = table.drop(sys.argv[2], axis=1).select_dtypes("number") - |fitted = $estimator(random_state=0).fit(X, table[sys.argv[2]]) - |print(base64.b64encode(b"pickle " + pickle.dumps(fitted)).decode("ascii")) + |for estimator in [${estimators.mkString(", ")}]: + | fitted = estimator.fit(X, table[sys.argv[2]]) + | print(base64.b64encode(b"pickle " + pickle.dumps(fitted)).decode("ascii")) |""".stripMargin val out = new java.io.ByteArrayOutputStream() val exit = scala.sys.process - .Process(Seq(PyOpExecHarness.resolvePython(), "-c", fit, data.toString, target)) + .Process(Seq(PyOpExecHarness.resolvePython(), "-c", fit, fitData.toString, target)) .#>(out) .! - require(exit == 0, s"fitting the fixture's $estimator exited with $exit") - val cell = java.util.Base64.getDecoder.decode(out.toString("US-ASCII").trim) + require(exit == 0, s"fitting the fixture's ${estimators.mkString(", ")} exited with $exit") + val cells = out + .toString("US-ASCII") + .linesIterator + .map(_.trim) + .filter(_.nonEmpty) + .map(java.util.Base64.getDecoder.decode) + .toSeq val modelSchema = new Schema(new Attribute(model, AttributeType.BINARY)) val modelPath = dir.resolve("input_port_0.jsonl") TupleIO.writeTuples( modelPath, - Iterator(Tuple.builder(modelSchema).add(modelSchema.getAttribute(model), cell).build()), + cells.iterator.map(c => + Tuple.builder(modelSchema).add(modelSchema.getAttribute(model), c).build() + ), modelSchema ) Map(PortIdentity(0) -> modelPath, PortIdentity(1) -> data) @@ -1014,16 +1025,43 @@ object SklearnTestingTransformHandler extends TransformHandler { FittedModelFixture.write(testRoot, "DecisionTreeClassifier", FittedModelFixture.label) ) + /** The regressor reads one feature, since the label is text. Beside it: a + * table the drop of missing rows leaves one row of, a shape a frame squeezed + * to one dimension would lose as the single feature would, and two model rows + * that answer differently, which the engine scores one row at a time. + */ override def extraScenarios( testRoot: Path ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = { - val dir = testRoot.resolve("regression") - Files.createDirectories(dir) + def in(name: String): Path = Files.createDirectories(testRoot.resolve(name)) + val tree = "DecisionTreeClassifier(random_state=0)" + val label = FittedModelFixture.label Seq( ( "regression", testing("petal_width", isRegression = true), - FittedModelFixture.write(dir, "DecisionTreeRegressor", "petal_width") + FittedModelFixture.write(in("regression"), "DecisionTreeRegressor", "petal_width") + ), + ( + "one row left after the drop", + testing(label, isRegression = false), + FittedModelFixture.writeModels( + in("one-row"), + Seq(tree), + label, + scored = Some( + Seq(Seq("versicolor", 5.0, 1.8), Seq("setosa", null, 0.2), Seq("virginica", 6.0, null)) + ) + ) + ), + ( + "two model rows", + testing(label, isRegression = false), + FittedModelFixture.writeModels( + in("two-models"), + Seq(tree, "DecisionTreeClassifier(max_depth=1, random_state=0)"), + label + ) ) ) } From 8d9cc7453d374c5a9c5dd22ad34635b0619878e9 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Sat, 26 Sep 2026 20:09:47 -0700 Subject: [PATCH 15/15] test(verify): cast and concatenate edge values, restore the Prediction scenarios, and read more cases from the shared table Co-Authored-By: Claude Opus 5.5 (1M context) --- .../translator/verify/CuratedHandlers.scala | 218 +++++++++++++----- 1 file changed, 156 insertions(+), 62 deletions(-) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala index c7b31244a18..2cd6ab3e8c1 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -221,12 +221,6 @@ object SpecializedFilterTransformHandler extends TransformHandler { } } -/** Handler for `DistinctOpDesc`. The canonical auto-fixture is all-distinct - * (the name column is unique by invariant), so it never exercises dedup. - * This 5-row table repeats two rows so both paths must actually drop - * duplicates, and both keep the first occurrence: JVM LinkedHashSet and pandas - * `drop_duplicates(keep="first")` agree on which of a pair survives. - */ /** * Curated CONFIG for [[ProjectionOpDesc]] over the shared table. Its `attributes` * list is not declared `required`, so the auto tier starts it empty the way the UI @@ -246,24 +240,19 @@ object ProjectionTransformHandler extends TransformHandler { } } +/** Handler for `DistinctOpDesc`. Every row of the shared table is distinct, since + * `id` is, so it never exercises dedup. Two of its columns, `name` and the 0/1 + * species, repeat whole rows, so both paths must actually drop duplicates, and + * both keep the first occurrence: JVM LinkedHashSet and pandas + * `drop_duplicates(keep="first")` agree on which of a pair survives. + */ object DistinctTransformHandler extends TransformHandler { override val opDescClass: Class[_ <: LogicalOp] = classOf[DistinctOpDesc] override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { - val columns = Seq( - ("id", AttributeType.INTEGER), - ("name", AttributeType.STRING) - ) - val rows = Seq( - Seq[Any](1, "a"), - Seq[Any](2, "b"), - Seq[Any](1, "a"), // duplicate of row 0 - Seq[Any](3, "c"), - Seq[Any](2, "b") // duplicate of row 1 - ) - val inputPath = - CuratedHandlers.writeFixture(testRoot.resolve("input_port_0.jsonl"), columns, rows) - (new DistinctOpDesc(), Map(PortIdentity(0) -> inputPath)) + val repeating = + ProjectedFixture(CanonicalFixture, Seq("name", "a\"b\\c_species"), Set.empty) + (new DistinctOpDesc(), repeating.writeInputs(testRoot, 1)) } } @@ -346,15 +335,14 @@ object RegexTransformHandler extends TransformHandler { numberRows ) - // A boolean column. The engine matches "true", where a hole leaves pandas - // holding 1.0. Case-sensitive on purpose, since case is half the difference. - val boolDir = testRoot.resolve("booleans") - Files.createDirectories(boolDir) - val boolColumn = Seq(("a\"b\\c_b", AttributeType.BOOLEAN)) - val boolRows: Seq[Seq[Any]] = - Seq(Seq[Any](true), Seq[Any](false), Seq[Any](null), Seq[Any](true)) - val boolInput = - CuratedHandlers.writeFixture(boolDir.resolve("input_port_0.jsonl"), boolColumn, boolRows) + // The shared table's boolean column, holed. The engine matches "true", where + // the hole leaves pandas holding 1.0. Case-sensitive on purpose, since case + // is half the difference. + val boolInput = CanonicalFixture.write( + Files.createDirectories(testRoot.resolve("booleans")), + 1, + withGaps = true + ) Seq( ( @@ -374,8 +362,8 @@ object RegexTransformHandler extends TransformHandler { ), ( "regex=^true$ on a boolean column", - regexOp("a\"b\\c_b", "^true$", caseInsensitive = false), - Map(PortIdentity(0) -> boolInput) + regexOp("a\"b\\c_high_score", "^true$", caseInsensitive = false), + boolInput ) ) } @@ -516,11 +504,6 @@ object HashJoinTransformHandler extends TransformHandler { * pairs a target type with a source column that type accepts, holding a value * that round-trips identically through JVM `AttributeTypeUtils` and through the * generated pandas. - * - * TIMESTAMP is left out: the two runtimes serialize one differently to JSONL, - * native as an ISO string and pandas as epoch millis, so the comparator flags a - * representation mismatch for an identical instant. That is a harness-wide gap - * rather than a TypeCasting defect. */ object TypeCastingTransformHandler extends TransformHandler { override val opDescClass: Class[_ <: LogicalOp] = classOf[TypeCastingOpDesc] @@ -539,7 +522,7 @@ object TypeCastingTransformHandler extends TransformHandler { // since Path A would end before there is anything to compare; that half // is pinned in TypeCastingOpDescSpec. ("str_to_bool", AttributeType.STRING), // "true"/"0" → BOOLEAN - // The other direction, which the five above cannot ask: the engine writes + // The other direction, which the casts above cannot ask: the engine writes // "true" where Python's str() writes "True". ("bool_to_str", AttributeType.BOOLEAN), // true/false → STRING // Named by two units at once. `tupleCasting` takes a Map, so the second @@ -551,13 +534,50 @@ object TypeCastingTransformHandler extends TransformHandler { ("lng_to_int", AttributeType.LONG), // beyond Int range → INTEGER // `new Timestamp(long)` reads the number as milliseconds; pd.to_datetime // defaults to nanoseconds and puts every one of these in 1970. - ("ms_to_ts", AttributeType.LONG) // epoch millis → TIMESTAMP + ("ms_to_ts", AttributeType.LONG), // epoch millis → TIMESTAMP + // Double.toString writes E notation from 1e7 up and below 1e-3, where + // Python's str() waits until 1e16 and 1e-4. + ("dbl_to_str", AttributeType.DOUBLE), // double → STRING + // Timestamp.toString writes every digit of the nanoseconds it holds, and + // a year past pandas' range as readily as any other. Two columns, since + // one pandas column holds nanoseconds or those years but not both. + ("ts_to_str", AttributeType.TIMESTAMP), // timestamp → STRING + ("far_ts_to_str", AttributeType.TIMESTAMP) // timestamp → STRING ) + def ts(text: String) = java.sql.Timestamp.valueOf(text) // `str_to_int` carries the text NumberFormat reads and Python's int refuses: // a decimal point, a grouping comma, trailing letters. val rows = Seq( - Seq[Any]("10", 1, 6, 11, 1, "true", true, "2024-01-07 00:00:00", 2147483648L, 1700000000000L), - Seq[Any]("6.7", 2, 7, 12, 0, "false", false, "2024-03-15 08:30:00", 2147483647L, 0L), + Seq[Any]( + "10", + 1, + 6, + 11, + 1, + "true", + true, + "2024-01-07 00:00:00", + 2147483648L, + 1700000000000L, + 1.0e20, + ts("2024-01-01 00:00:00.123456789"), + ts("2500-01-01 00:00:00.5") + ), + Seq[Any]( + "6.7", + 2, + 7, + 12, + 0, + "false", + false, + "2024-03-15 08:30:00", + 2147483647L, + 0L, + 1.0e-4, + ts("2024-03-05 14:09:07.5"), + ts("1600-06-15 08:30:00.0") + ), Seq[Any]( "1,234", 3, @@ -568,10 +588,41 @@ object TypeCastingTransformHandler extends TransformHandler { true, "2024-06-01 12:00:00", -2147483649L, - 1000000000000L + 1000000000000L, + 1.0e7, + ts("2024-06-30 23:59:59.999999999"), + ts("9999-12-31 23:59:59.0") ), - Seq[Any]("12abc", 4, 9, 14, 0, "1", false, "2024-09-20 23:59:59", 0L, 1893456000000L), - Seq[Any]("-6.7", 5, 10, 15, 1, "TRUE", true, "2024-12-31 01:02:03", -1L, 946684800000L) + Seq[Any]( + "12abc", + 4, + 9, + 14, + 0, + "1", + false, + "2024-09-20 23:59:59", + 0L, + 1893456000000L, + 0.1, + ts("2024-01-01 00:00:00.000000001"), + ts("2024-01-01 00:00:00.0") + ), + Seq[Any]( + "-6.7", + 5, + 10, + 15, + 1, + "TRUE", + true, + "2024-12-31 01:02:03", + -1L, + 946684800000L, + 3.0, + ts("2024-01-01 00:00:00.0"), + ts("1677-09-21 00:00:00.0") + ) ) val inputPath = CuratedHandlers.writeFixture(testRoot.resolve("input_port_0.jsonl"), columns, rows) @@ -594,7 +645,10 @@ object TypeCastingTransformHandler extends TransformHandler { unit("twice_cast", AttributeType.TIMESTAMP), unit("twice_cast", AttributeType.STRING), unit("lng_to_int", AttributeType.INTEGER), - unit("ms_to_ts", AttributeType.TIMESTAMP) + unit("ms_to_ts", AttributeType.TIMESTAMP), + unit("dbl_to_str", AttributeType.STRING), + unit("ts_to_str", AttributeType.STRING), + unit("far_ts_to_str", AttributeType.STRING) ) (desc, Map(PortIdentity(0) -> inputPath)) @@ -642,22 +696,17 @@ object KeywordSearchTransformHandler extends TransformHandler { override def extraScenarios( testRoot: Path ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = { - // A hole widens an integer column to float and every value grows a ".0" the - // term "0" then matches on a word boundary, so the rendering decides between - // one row and all of them. A digit term cannot: `\b6\b` finds "6.0" too. - val numberDir = testRoot.resolve("numbers") - Files.createDirectories(numberDir) - val numberColumn = Seq(("a\"b\\c_n", AttributeType.INTEGER)) - val numberRows: Seq[Seq[Any]] = - Seq(Seq[Any](6), Seq[Any](null), Seq[Any](70), Seq[Any](0)) - val numberInput = CuratedHandlers.writeFixture( - numberDir.resolve("input_port_0.jsonl"), - numberColumn, - numberRows + // A hole widens the shared table's 0/1 column to float and every value grows + // a ".0" the term "0" then matches on a word boundary, so the rendering + // decides between the 0 rows and all of them. + val numberInput = CanonicalFixture.write( + Files.createDirectories(testRoot.resolve("numbers")), + 1, + withGaps = true ) val desc = new KeywordSearchOpDesc() - desc.attribute = "a\"b\\c_n" + desc.attribute = "a\"b\\c_species" desc.keyword = "0" desc.isCaseSensitive = false @@ -684,7 +733,7 @@ object KeywordSearchTransformHandler extends TransformHandler { cased.isCaseSensitive = true Seq( - ("keyword=0 on an integer column", desc, Map(PortIdentity(0) -> numberInput)), + ("keyword=0 on an integer column", desc, numberInput), ("keyword=Love, case sensitive", cased, Map(PortIdentity(0) -> casedInput)) ) } @@ -830,11 +879,12 @@ object AggregateTransformHandler extends TransformHandler { val columns = Seq( ("a\"b\\c_i", AttributeType.INTEGER), ("a\"b\\c_t", AttributeType.TIMESTAMP), - ("a\"b\\c_b", AttributeType.BOOLEAN) + ("a\"b\\c_b", AttributeType.BOOLEAN), + ("a\"b\\c_d", AttributeType.DOUBLE) ) val rows: Seq[Seq[Any]] = Seq( - Seq[Any](Int.MaxValue, java.sql.Timestamp.valueOf("2020-01-01 00:00:00"), true), - Seq[Any](1, java.sql.Timestamp.valueOf("2020-01-02 00:00:00"), false) + Seq[Any](Int.MaxValue, java.sql.Timestamp.valueOf("2020-01-01 00:00:00"), true, 1.0e20), + Seq[Any](1, java.sql.Timestamp.valueOf("2020-01-02 00:00:00"), false, 1.0e-4) ) val input = CuratedHandlers.writeFixture(dir.resolve("input_port_0.jsonl"), columns, rows) @@ -844,7 +894,9 @@ object AggregateTransformHandler extends TransformHandler { agg(AggregationFunction.SUM, "a\"b\\c_i", "int_total"), agg(AggregationFunction.SUM, "a\"b\\c_t", "ts_total"), agg(AggregationFunction.AVERAGE, "a\"b\\c_t", "ts_avg"), - agg(AggregationFunction.CONCAT, "a\"b\\c_b", "flags") + agg(AggregationFunction.CONCAT, "a\"b\\c_b", "flags"), + agg(AggregationFunction.CONCAT, "a\"b\\c_d", "doubles"), + agg(AggregationFunction.CONCAT, "a\"b\\c_t", "moments") ) ("the engine's own arithmetic", desc, Map(PortIdentity(0) -> input)) } @@ -999,6 +1051,48 @@ object SklearnPredictionTransformHandler extends TransformHandler { FittedModelFixture.write(testRoot, "DecisionTreeClassifier", FittedModelFixture.label) ) } + + /** The ground-truth branch over rows missing a feature, which the nulls case + * does not reach since it leaves the optional knob unfilled; and two model + * rows, of which the engine predicts with the last. + */ + override def extraScenarios( + testRoot: Path + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = { + def in(name: String): Path = Files.createDirectories(testRoot.resolve(name)) + val tree = "DecisionTreeClassifier(random_state=0)" + val label = FittedModelFixture.label + def prediction(groundTruth: String): SklearnPredictionOpDesc = { + val desc = new SklearnPredictionOpDesc() + desc.model = FittedModelFixture.model + desc.resultAttribute = "prediction" + desc.groundTruthAttribute = groundTruth + desc + } + Seq( + ( + "ground truth named, a feature missing", + prediction(label), + FittedModelFixture.writeModels( + in("ground-truth-holes"), + Seq(tree), + label, + scored = Some( + Seq(Seq("versicolor", 5.0, 1.8), Seq("setosa", null, 0.2), Seq("virginica", 6.0, null)) + ) + ) + ), + ( + "two model rows", + prediction(""), + FittedModelFixture.writeModels( + in("two-models"), + Seq(tree, "DecisionTreeClassifier(max_depth=1, random_state=0)"), + label + ) + ) + ) + } } /** Testing scores a fitted model on its model port. `isRegression` has to match