From 94cfb7734016ab1256b29292aa18469b81ab7f18 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Wed, 2 Sep 2026 16:30:43 -0700 Subject: [PATCH 1/4] 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 2/4] 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 3/4] 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 4/4] 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))