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..2cd6ab3e8c1 --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala @@ -0,0 +1,1162 @@ +/* + * 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.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} +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. + */ +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, + SklearnPredictionTransformHandler, + SklearnTestingTransformHandler + ) + + 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.TIMESTAMP, x: java.sql.Timestamp) => 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. + */ +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)) + } +} + +/** + * 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 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, CanonicalFixture.writeInputs(testRoot, 1)) + } +} + +/** 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 repeating = + ProjectedFixture(CanonicalFixture, Seq("name", "a\"b\\c_species"), Set.empty) + (new DistinctOpDesc(), repeating.writeInputs(testRoot, 1)) + } +} + +/** + * Curated handler for [[RegexOpDesc]]. The auto tier only ever feeds it the + * pattern `"1"`, which asks nothing of either regex engine. Three real ones do: + * + * - `[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. + * + * 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] + + 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(("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("a\"b\\c_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) + + // 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 + ) + + // 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( + ( + "regex=\\d+", + regexOp("a\"b\\c_text", "\\d+", caseInsensitive = false), + Map(PortIdentity(0) -> digitInput) + ), + ( + "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_high_score", "^true$", caseInsensitive = false), + boolInput + ) + ) + } +} + +/** 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[_]] + + /** `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 desc = new HashJoinOpDesc[Integer]() + desc.buildAttributeName = "id" + desc.probeAttributeName = "id" + desc.joinType = JoinType.INNER + (desc, CanonicalFixture.writeInputs(testRoot, 2)) + } + + 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 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 + * 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 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 + ), + 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 + ) + ) + } +} + +/** + * 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. + */ +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 + // 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 + // 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 + // 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 + // 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, + 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, + 8, + 13, + 1, + "0", + true, + "2024-06-01 12:00:00", + -2147483649L, + 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, + 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) + + 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), + 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), + unit("dbl_to_str", AttributeType.STRING), + unit("ts_to_str", AttributeType.STRING), + unit("far_ts_to_str", AttributeType.STRING) + ) + + (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, 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. + * + * 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] + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val columns = Seq(("a\"b\\c_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 = "a\"b\\c_txt" + desc.keyword = "love day" + desc.isCaseSensitive = false + + (desc, Map(PortIdentity(0) -> inputPath)) + } + + override def extraScenarios( + testRoot: Path + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = { + // 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_species" + 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. 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[Any]("everything was absolutely Love.") + ) + 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, numberInput), + ("keyword=Love, case sensitive", cased, Map(PortIdentity(0) -> casedInput)) + ) + } +} + +/** 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("a\"b\\c_image_bytes", AttributeType.BINARY)) + + def tup(bytes: Array[Byte]): Tuple = { + val builder = Tuple.builder(schema) + builder.add(schema.getAttribute("a\"b\\c_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 = "a\"b\\c_image_bytes" + + (desc, Map(PortIdentity(0) -> inputPath)) + } +} + +/** 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 + * 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] + + 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)) + } + + /** 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)), + 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), + ("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, 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) + + 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"), + 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)) + } +} + +/** 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] + + 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)) + } +} + +/** 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 + * 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] = + 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 DecisionTreeClassifier, DecisionTreeRegressor + |table = pd.read_json(sys.argv[1], lines=True).iloc[:-1] + |X = table.drop(sys.argv[2], axis=1).select_dtypes("number") + |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, fitData.toString, target)) + .#>(out) + .! + 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, + cells.iterator.map(c => + Tuple.builder(modelSchema).add(modelSchema.getAttribute(model), c).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) + ) + } + + /** 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 + * 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) + ) + + /** 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])] = { + 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(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 + ) + ) + ) + } +}