kz930 commented on code in PR #8363: URL: https://github.com/apache/texera/pull/8363#discussion_r4255202756
########## 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)) Review Comment: Fixed in 303341bc8. The fit script now goes in a file and the target in an environment variable, so only paths reach the command line, and the fixture still uses the hostile `a"b\c_species` target. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
