diff --git a/amber/src/test/integration/org/apache/texera/amber/engine/e2e/WorkflowParitySpec.scala b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/WorkflowParitySpec.scala new file mode 100644 index 00000000000..23ac7190d55 --- /dev/null +++ b/amber/src/test/integration/org/apache/texera/amber/engine/e2e/WorkflowParitySpec.scala @@ -0,0 +1,338 @@ +/* + * 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.engine.e2e + +import com.twitter.util.Duration +import org.apache.pekko.actor.{ActorSystem, Props} +import org.apache.pekko.testkit.{ImplicitSender, TestKit} +import org.apache.texera.amber.clustering.SingleNodeListener +import org.apache.texera.amber.core.tuple.{AttributeType, Tuple} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.engine.common.AmberRuntime +import org.apache.texera.amber.engine.e2e.TestUtils.{ + buildWorkflow, + cleanupWorkflowExecutionData, + initiateTexeraDBForTestCases, + runWorkflowAndReadTerminalResults, + setUpWorkflowExecutionData +} +import org.apache.texera.amber.operator.{LogicalOp, TestOperators} +import org.apache.texera.amber.operator.distinct.DistinctOpDesc +import org.apache.texera.amber.operator.filter.{ + ComparisonType, + FilterPredicate, + SpecializedFilterOpDesc +} +import org.apache.texera.amber.operator.limit.LimitOpDesc +import org.apache.texera.amber.operator.projection.{AttributeUnit, ProjectionOpDesc} +import org.apache.texera.amber.operator.regex.RegexOpDesc +import org.apache.texera.amber.operator.sort.{SortCriteriaUnit, SortOpDesc, SortPreference} +import org.apache.texera.amber.operator.source.scan.file.FileScanOpDesc +import org.apache.texera.amber.operator.source.scan.text.TextInputSourceOpDesc +import org.apache.texera.amber.operator.split.SplitOpDesc +import org.apache.texera.amber.operator.typecasting.{TypeCastingOpDesc, TypeCastingUnit} +import org.apache.texera.amber.operator.union.UnionOpDesc +import org.apache.texera.amber.tags.IntegrationTest +import org.apache.texera.amber.translator.WorkflowToPythonTranslator +import org.apache.texera.amber.translator.verify.{ + CanonicalSourceFixture, + Comparator, + StandaloneRunner, + TupleIO +} +import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlan, LogicalPlanPojo} +import org.apache.texera.common.compiler.{CompilationErrorHandling, WorkflowCompiler} +import org.scalatest.{BeforeAndAfterAll, BeforeAndAfterEach} +import org.scalatest.flatspec.AnyFlatSpecLike +import org.scalatest.matchers.should.Matchers + +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Path, Paths} + +/** A whole workflow, run both ways: the Texera engine is the reference, and the + * script the export writes for the same plan is what is under test. + * + * The per-operator specs feed every operator the shared table. Here an operator + * reads what the operator before it produced, in each path, which is where a + * fragment that agrees with the engine on its own can still disagree. + * + * Each leaf compares as a set of rows, as in the per-operator specs, unless the + * operator that produces it declares an order: the engine runs most operators on + * several workers and promises no order after them. + * + * Every plan reads the shared table, as the CSV the CSV source reads, each aimed + * at one way fragments can go wrong together. TranslatedPlanRunSpec keeps the two + * the engine has nothing to compare with: the files two charts write, and a chart + * that ends the script. + */ +@IntegrationTest +class WorkflowParitySpec + extends TestKit(ActorSystem("WorkflowParitySpec", AmberRuntime.pekkoConfig)) + with ImplicitSender + with AnyFlatSpecLike + with Matchers + with BeforeAndAfterAll + with BeforeAndAfterEach { + + private val specId = 7 + + override protected def beforeEach(): Unit = setUpWorkflowExecutionData(specId) + + override protected def afterEach(): Unit = cleanupWorkflowExecutionData(specId) + + override def beforeAll(): Unit = { + system.actorOf(Props[SingleNodeListener](), "cluster-info") + Class.forName("org.postgresql.Driver") + initiateTexeraDBForTestCases() + } + + override def afterAll(): Unit = TestKit.shutdownActorSystem(system) + + private def link(from: LogicalOp, to: LogicalOp): LogicalLink = + LogicalLink(from.operatorIdentifier, PortIdentity(), to.operatorIdentifier, PortIdentity()) + + private def link(from: LogicalOp, fromPort: Int, to: LogicalOp): LogicalLink = + LogicalLink( + from.operatorIdentifier, + PortIdentity(fromPort), + to.operatorIdentifier, + PortIdentity() + ) + + /** The shared table, written as the CSV the CSV source reads, so a plan starts + * from the rows every operator is verified on. Each call hands back a new + * source over the same file, for a plan that reads it twice. + */ + private def sharedTable(rows: Seq[Tuple]): (Path, () => LogicalOp) = { + val dir = Files.createTempDirectory("workflow-parity-") + val path = CanonicalSourceFixture.writeCsv(dir, StandardCharsets.UTF_8, ',', rows) + (dir, () => TestOperators.getCsvScanOpDesc(path.toString, header = true)) + } + + /** Runs the plan in the engine and its translated script, and compares every + * leaf the plan has. + */ + private def assertParity(dir: Path, ops: List[LogicalOp], links: List[LogicalLink]): Unit = { + val context = TestUtils.workflowContext(specId) + val workflow = buildWorkflow(ops, links, context) + // The schemas the export endpoint hands the translator, from the same compile. + val schemas = new WorkflowCompiler(context) + .compile( + LogicalPlanPojo(ops, links, List.empty, List.empty), + CompilationErrorHandling.Lenient + ) + .operatorIdToOutputSchemas + val leafIds = workflow.logicalPlan.getTerminalOperatorIds.toList + val engineRows = runWorkflowAndReadTerminalResults(system, workflow, Duration.fromMinutes(2)) + + val script = new WorkflowToPythonTranslator().translate(LogicalPlan(ops, links), schemas) + // The translator prints each leaf under `[] dfN:`, which is how + // the frame a leaf ends up in is found. + val printed: Map[String, String] = """print\("\\n\[([^\]]*)\] (df\d+):"\)""".r + .findAllMatchIn(script) + .map(m => m.group(1) -> m.group(2)) + .toList + .groupMap(_._1)(_._2) + .map { + case (name, Seq(v)) => name -> v + case (name, _) => fail(s"two leaves print as [$name]; name them apart") + } + val leaves = leafIds.map { id => + val name = workflow.logicalPlan.getOperator(id).operatorInfo.userFriendlyName + (id, printed.getOrElse(name, fail(s"no [$name] leaf in the script"))) + } + val leafSchemas = leaves.map { case (id, _) => schemas(id)(PortIdentity()).get } + val result = StandaloneRunner.runTranslated(script, leaves.map(_._2), leafSchemas, dir) + + leaves.zip(leafSchemas).zipWithIndex.foreach { + case (((id, _), schema), i) => + val actual = dir.resolve(s"engine_$i.jsonl") + TupleIO.writeTuples(actual, engineRows(id).iterator, schema) + withClue(s"leaf ${workflow.logicalPlan.getOperator(id).operatorInfo.userFriendlyName}: ") { + Comparator.assertEqual( + actual, + result.outputs(i + 1), + orderSensitive = workflow.logicalPlan.getOperator(id).orderSensitive + ) + } + } + } + + /** File Scan reads the paths its upstream hands it, so it cannot run in the + * per-operator specs, which give a source no input. Here Text Input names the + * shared table's own file, the same one every operator reads its rows from. + */ + "A translated workflow" should "read the files its upstream names" in { + val shared = Paths.get(getClass.getResource("/verify/canonical_fixture.json").toURI) + val dir = Files.createTempDirectory("workflow-parity-") + val paths = new TextInputSourceOpDesc + paths.textInput = shared.toString + val scan = new FileScanOpDesc + assertParity(dir, List(paths, scan), List(link(paths, scan))) + } + + // Each plan reads the shared table twice, whole and with a hole in every + // column, so the values a type reaches only at its edges, and the empty cells, + // pass through several operators rather than one. + Seq( + "the shared table" -> CanonicalSourceFixture.rows, + "the shared table with a hole in every column" -> CanonicalSourceFixture.holedRows + ).foreach { + case (table, rows) => + // Four hops, so every variable the translator assigns is read by the next + // fragment rather than by the print at the end. A two-operator plan cannot + // show a chain threaded wrong, since df1 is the only name to get right. + // + // Every score differs, so the limit after the sort takes the same rows on + // both sides. The projection comes before the sort, so the source's LONG + // column never reaches a Python operator: the engine hands one a column + // holding a null as float64, which cannot hold its values past 2^53, and + // pauses the run. Move it back to the end once the engine keeps them. + it should s"carry $table through a filter, a projection, a sort and a limit" in { + val (dir, src) = sharedTable(rows) + val filter = new SpecializedFilterOpDesc + filter.predicates = List(new FilterPredicate("score", ComparisonType.GREATER_THAN, "1.0")) + val sort = new SortOpDesc + val criterion = new SortCriteriaUnit + criterion.attributeName = "score" + criterion.sortPreference = SortPreference.DESC + sort.attributes = List(criterion) + val limit = new LimitOpDesc + limit.limit = 5 + val projection = new ProjectionOpDesc + projection.attributes ++= List("id", "score", "edge_double", "edge_ts", "digits") + .map(name => new AttributeUnit(name, name)) + + val chain = List(src(), filter, projection, sort, limit) + assertParity(dir, chain, chain.sliding(2).map { case List(a, b) => link(a, b) }.toList) + } + + // Java writes a double past 1e7 or under 1e-3 in E notation, and the regex + // after the cast keeps exactly those rows. + it should s"cast $table's edge values to text the way the engine writes them" in { + val (dir, src) = sharedTable(rows) + val cast = new TypeCastingOpDesc + cast.typeCastingUnits = List("edge_double", "edge_ts").map { name => + val unit = new TypeCastingUnit + unit.attribute = name + unit.resultType = AttributeType.STRING + unit + } + val regex = new RegexOpDesc + regex.attribute = "edge_double" + regex.regex = "E" + val source = src() + assertParity(dir, List(source, cast, regex), List(link(source, cast), link(cast, regex))) + } + + // A variadic port is the one placeholder a fragment cannot name, so the list + // the translator writes in its place is only exercised by a real plan. The + // two halves share no row, so a list that names one upstream twice and + // drops the other loses rows rather than repeating them. + it should s"union the two halves of $table and keep one copy of each row" in { + val (dir, src) = sharedTable(rows) + def half(comparison: ComparisonType): (LogicalOp, LogicalOp) = { + val filter = new SpecializedFilterOpDesc + filter.predicates = List(new FilterPredicate("score", comparison, "3.0")) + (src(), filter) + } + val (highSource, high) = half(ComparisonType.GREATER_THAN) + val (lowSource, low) = half(ComparisonType.LESS_THAN_OR_EQUAL_TO) + val union = new UnionOpDesc + val distinct = new DistinctOpDesc + assertParity( + dir, + List(highSource, high, lowSource, low, union, distinct), + List( + link(highSource, high), + link(lowSource, low), + link(high, union), + link(low, union), + link(union, distinct) + ) + ) + } + + // Every other plan here reads its upstream's port 0, so a translator that + // handed each downstream the first frame its upstream produced would pass + // all of them. A split sends different rows to each of its two ports, and + // each branch is its own leaf, so a branch reading the other port's frame + // disagrees with the engine on its rows. Split runs on one worker, so the + // engine draws the same rows the script does. + it should s"give each branch of a split over $table the port its link named" in { + val (dir, src) = sharedTable(rows) + val split = new SplitOpDesc + split.k = 50 + split.random = false + split.seed = 1 + val distinct = new DistinctOpDesc + val limit = new LimitOpDesc + limit.limit = 10 + val source = src() + assertParity( + dir, + List(source, split, distinct, limit), + List(link(source, split), link(split, 0, distinct), link(split, 1, limit)) + ) + } + + // A CSV carries no types, so the hole in `id` widens it and its 3 would be + // matched as "3.0". Only a whole plan shows it: the type belongs to the + // source, a hop away with a file in between. The hole is in the row whose + // id is 7, so 3 is a value both tables hold. + it should s"match $table's integer column on the type its source declared" in { + val (dir, src) = sharedTable(rows) + val regex = new RegexOpDesc + regex.attribute = "id" + regex.regex = "^3$" + val source = src() + assertParity(dir, List(source, regex), List(link(source, regex))) + } + + // A cast makes the column a datetime, and Python's str() then writes no + // fraction at all on a whole second where `Timestamp.toString` writes ".0". + // One operator cannot show it, since the executor casts each column once + // from its original value. Every trade date is a midnight. + it should s"render $table's cast dates the way the engine's toString does" in { + val (dir, src) = sharedTable(rows) + val cast = new TypeCastingOpDesc + val unit = new TypeCastingUnit + unit.attribute = "trade_date" + unit.resultType = AttributeType.TIMESTAMP + cast.typeCastingUnits = List(unit) + val regex = new RegexOpDesc + regex.attribute = "trade_date" + regex.regex = "\\.0$" + val source = src() + assertParity(dir, List(source, cast, regex), List(link(source, cast), link(cast, regex))) + } + + // The substitution rewrites the variable a fragment reads with, not the + // column name it asks that variable for. A fragment tested on its own never + // shows the difference, because the harness binds the placeholder to itself. + it should s"keep $table's column whose name matches a placeholder" in { + val (dir, src) = sharedTable(rows) + val projection = new ProjectionOpDesc + projection.attributes ++= List(new AttributeUnit("in1df", "kept")) + val source = src() + assertParity(dir, List(source, projection), List(link(source, projection))) + } + } +} diff --git a/bin/local-dev/main.sh b/bin/local-dev/main.sh index 9053834acbd..3c86fa08a26 100755 --- a/bin/local-dev/main.sh +++ b/bin/local-dev/main.sh @@ -785,6 +785,11 @@ _parse_sbt() { if [[ "$arg" =~ %\"?test(-\>[^\"]*)?\"? ]] || [[ "$arg" =~ %Test$ ]]; then continue fi + # A task's `.dependsOn(X / Test / compile)` names a task, not a + # project. Kept, it became a project called `X/Test/compile`, + # whose `/` turned the visited-marker lookup below into a + # `${var/pattern/repl}` and ended every command under `set -u`. + [[ "$arg" == */* ]] && continue [[ "$arg" =~ ^[A-Z] ]] || continue existing=$(amap_get SBT_DEPS "$current") case " $existing " in diff --git a/build.sbt b/build.sbt index d89f7a75080..928a24bbb31 100644 --- a/build.sbt +++ b/build.sbt @@ -255,6 +255,22 @@ lazy val WorkflowCompilingService = (project in file("workflow-compiling-service lazy val WorkflowExecutionService = (project in file("amber")) .dependsOn(WorkflowCompiler, Auth, Config) .settings(commonModuleSettings) + .settings( + // The workflow-level parity spec runs the export's translator and verify's + // script runner and comparator. Only their classes come over: depending on + // the compiling service would bring its dropwizard 4, which evicts the + // dropwizard 1 the engine is built on. + Test / unmanagedClasspath ++= Seq( + (WorkflowCompilingService / Compile / classDirectory).value, + (WorkflowCompilingService / Test / classDirectory).value, + (WorkflowCompilingService / Test / resourceDirectory).value, + (WorkflowOperator / Test / classDirectory).value, + (WorkflowOperator / Test / resourceDirectory).value + ), + Test / compile := (Test / compile) + .dependsOn(WorkflowCompilingService / Test / compile, WorkflowOperator / Test / compile) + .value + ) .settings( dependencyOverrides ++= Seq( "com.fasterxml.jackson.core" % "jackson-core" % jacksonVersion, diff --git a/workflow-compiling-service/src/test/resources/verify/canonical_fixture.json b/workflow-compiling-service/src/test/resources/verify/canonical_fixture.json index 98789459b1e..2365d5d5efb 100644 --- a/workflow-compiling-service/src/test/resources/verify/canonical_fixture.json +++ b/workflow-compiling-service/src/test/resources/verify/canonical_fixture.json @@ -34,7 +34,11 @@ "a\"b\\c_high_score": false, "a\"b\\c_species_pred": 1, "species_name": "versicolor", - "species_name_pred": "versicolor" + "species_name_pred": "versicolor", + "edge_double": 3.0, + "edge_ts": "2024-02-07 12:00:00.0", + "digits": "99", + "views": 3000000000 }, { "id": 3, @@ -71,7 +75,11 @@ "a\"b\\c_high_score": false, "a\"b\\c_species_pred": 1, "species_name": "versicolor", - "species_name_pred": "versicolor" + "species_name_pred": "versicolor", + "edge_double": 1.0E7, + "edge_ts": "2262-04-12 00:00:00.0", + "digits": "7", + "views": 2147483648 }, { "id": 11, @@ -108,7 +116,11 @@ "a\"b\\c_high_score": false, "a\"b\\c_species_pred": 0, "species_name": "setosa", - "species_name_pred": "setosa" + "species_name_pred": "setosa", + "edge_double": 0.001, + "edge_ts": "2024-02-11 12:00:00.0", + "digits": "8", + "views": 4294967296 }, { "id": 1, @@ -145,7 +157,11 @@ "a\"b\\c_high_score": false, "a\"b\\c_species_pred": 1, "species_name": "setosa", - "species_name_pred": "versicolor" + "species_name_pred": "versicolor", + "edge_double": 1.0E20, + "edge_ts": "2500-01-01 00:00:00.5", + "digits": "001", + "views": -2147483649 }, { "id": 14, @@ -182,7 +198,11 @@ "a\"b\\c_high_score": true, "a\"b\\c_species_pred": 1, "species_name": "versicolor", - "species_name_pred": "versicolor" + "species_name_pred": "versicolor", + "edge_double": 7.0, + "edge_ts": "2024-02-14 12:00:00.0", + "digits": "11", + "views": 5000000001 }, { "id": 5, @@ -219,7 +239,11 @@ "a\"b\\c_high_score": false, "a\"b\\c_species_pred": 0, "species_name": "setosa", - "species_name_pred": "setosa" + "species_name_pred": "setosa", + "edge_double": 0.1, + "edge_ts": "1677-09-21 00:00:00.0", + "digits": "3", + "views": 1 }, { "id": 9, @@ -256,7 +280,11 @@ "a\"b\\c_high_score": false, "a\"b\\c_species_pred": 0, "species_name": "versicolor", - "species_name_pred": "setosa" + "species_name_pred": "setosa", + "edge_double": 0.002, + "edge_ts": "2024-02-09 12:00:00.0", + "digits": "12", + "views": 9000000000000 }, { "id": 2, @@ -293,7 +321,11 @@ "a\"b\\c_high_score": true, "a\"b\\c_species_pred": 1, "species_name": "versicolor", - "species_name_pred": "versicolor" + "species_name_pred": "versicolor", + "edge_double": 1.0E-4, + "edge_ts": "1600-06-15 08:30:00.0", + "digits": "042", + "views": 2500000000 }, { "id": 13, @@ -330,7 +362,11 @@ "a\"b\\c_high_score": false, "a\"b\\c_species_pred": 1, "species_name": "versicolor", - "species_name_pred": "versicolor" + "species_name_pred": "versicolor", + "edge_double": 0.25, + "edge_ts": "2024-02-13 12:00:00.0", + "digits": "4", + "views": 7000000000 }, { "id": 6, @@ -367,7 +403,11 @@ "a\"b\\c_high_score": true, "a\"b\\c_species_pred": 0, "species_name": "setosa", - "species_name_pred": "setosa" + "species_name_pred": "setosa", + "edge_double": -2.5E-5, + "edge_ts": "2024-02-06 12:00:00.0", + "digits": "010", + "views": -3000000000 }, { "id": 10, @@ -404,7 +444,11 @@ "a\"b\\c_high_score": true, "a\"b\\c_species_pred": 1, "species_name": "setosa", - "species_name_pred": "versicolor" + "species_name_pred": "versicolor", + "edge_double": 999999.5, + "edge_ts": "2024-02-10 12:00:00.0", + "digits": "0005", + "views": 2147483647 }, { "id": 4, @@ -441,7 +485,11 @@ "a\"b\\c_high_score": true, "a\"b\\c_species_pred": 1, "species_name": "versicolor", - "species_name_pred": "versicolor" + "species_name_pred": "versicolor", + "edge_double": 12345678.9, + "edge_ts": "9999-12-31 23:59:59.0", + "digits": "0100", + "views": 6000000000 }, { "id": 15, @@ -478,7 +526,11 @@ "a\"b\\c_high_score": true, "a\"b\\c_species_pred": 0, "species_name": "setosa", - "species_name_pred": "setosa" + "species_name_pred": "setosa", + "edge_double": 1.2E8, + "edge_ts": "2024-02-15 12:00:00.0", + "digits": "0", + "views": 8000000000 }, { "id": 8, @@ -515,7 +567,11 @@ "a\"b\\c_high_score": true, "a\"b\\c_species_pred": 1, "species_name": "versicolor", - "species_name_pred": "versicolor" + "species_name_pred": "versicolor", + "edge_double": 1.5, + "edge_ts": "2024-02-08 12:00:00.0", + "digits": "0", + "views": 1000000000000 }, { "id": 12, @@ -552,6 +608,10 @@ "a\"b\\c_high_score": true, "a\"b\\c_species_pred": 1, "species_name": "versicolor", - "species_name_pred": "versicolor" + "species_name_pred": "versicolor", + "edge_double": 4.0E21, + "edge_ts": "2024-02-12 12:00:00.0", + "digits": "060", + "views": 3500000000 } ] diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CanonicalFixture.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CanonicalFixture.scala index 373e0596416..9c825775e11 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CanonicalFixture.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CanonicalFixture.scala @@ -123,9 +123,19 @@ object CanonicalFixture extends SharedFixture { new Attribute( "species_name_pred", AttributeType.STRING - ) // the prediction spelled out. A scorer takes a string label as readily as a + ), // the prediction spelled out. A scorer takes a string label as readily as a // numeric one and names the class after it rather than after its position, so // the pair exists a second time in text + // The values a column of each type can hold that its ordinary rows never + // reach, last so no column-picking knob comes to them before the others. + new Attribute("edge_double", AttributeType.DOUBLE), // 1.0E20 and 1.0E-4 beside + // ordinary numbers: Java writes a double past 1e7 or under 1e-3 in E notation + new Attribute("edge_ts", AttributeType.TIMESTAMP), // years 1600, 2500 and 9999, + // and 1677 and 2262 at pandas' own edges: a Timestamp holds every one of them + new Attribute("digits", AttributeType.STRING), // text made only of digits, "001" + // among them, which a reader left to infer takes for a number + new Attribute("views", AttributeType.LONG) // the one LONG column: counts past + // 2^31, an int's limit, on both sides of 2147483648, and two that fit an int ) // ── Data source ── diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CanonicalFixtureSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CanonicalFixtureSpec.scala index ba74f3df0da..0bb4e9dadd3 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CanonicalFixtureSpec.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CanonicalFixtureSpec.scala @@ -278,6 +278,23 @@ class CanonicalFixtureSpec extends AnyFlatSpec with Matchers { } } + // Each edge column exists for values the ordinary rows never reach, so both + // windows have to hold them: a year either side of pandas' nanosecond range, + // a double Java writes in E notation at each end, and digits with a leading 0. + it should "hold the edge values of each type on every port" in { + Seq(CanonicalFixture.port0Rows, CanonicalFixture.port1Rows).foreach { rows => + val years = rows.map(_.getField[java.sql.Timestamp]("edge_ts").toLocalDateTime.getYear) + years.count(_ > 2262) should be > 0 + years.count(_ < 1678) should be > 0 + val doubles = rows.map(_.getField[java.lang.Double]("edge_double").doubleValue) + doubles.count(d => math.abs(d) >= 1e7) should be > 0 + doubles.count(d => d != 0 && math.abs(d) < 1e-3) should be > 0 + val digits = rows.map(_.getField[String]("digits")) + all(digits) should fullyMatch regex "[0-9]+" + digits.count(d => d.length > 1 && d.startsWith("0")) should be > 0 + } + } + it should "write one JSONL fixture per requested input port" in { val root = Files.createTempDirectory("canonical-fixture-") val inputs = CanonicalFixture.writeInputs(root, inputPortCount = 2) diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SharedFixture.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SharedFixture.scala index b6e773e1ca1..d1152185e8c 100644 --- a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SharedFixture.scala +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/SharedFixture.scala @@ -19,7 +19,7 @@ package org.apache.texera.amber.translator.verify -import org.apache.texera.amber.core.tuple.{Schema, Tuple} +import org.apache.texera.amber.core.tuple.{AttributeType, Schema, Tuple} import org.apache.texera.amber.core.workflow.PortIdentity import java.nio.file.Path @@ -102,6 +102,18 @@ trait SharedFixture { final def writeEmptyInputs(dir: Path, inputPortCount: Int): Map[PortIdentity, Path] = writeShaped(dir, inputPortCount, _ => Seq.empty) + /** The same table with [[SharedFixture.specialDoublesPerColumn]] applied. */ + final def writeSpecialDoubles(dir: Path, inputPortCount: Int): Map[PortIdentity, Path] = + writeShaped( + dir, + inputPortCount, + rows => SharedFixture.specialDoublesPerColumn(rows, schema, keepFilled) + ) + + /** Whether [[writeSpecialDoubles]] has a column to put a value in. */ + final def hasSpecialDoubleColumn: Boolean = + SharedFixture.specialDoubleColumns(schema, keepFilled).nonEmpty + /** How many rows port 0 gets — what a row-count-sensitive knob (`limit`, * `offset`) is sized against so its value keeps some rows and drops some. */ @@ -187,6 +199,52 @@ object HostileColumn { object SharedFixture { + /** What a DOUBLE column can hold besides an ordinary number, plus a null beside + * them. The engine keeps all five apart; a float64 column holds NaN and null as + * one value and -0.0 compares equal to 0.0, so none of them reaches an operator + * through a table of ordinary numbers. + */ + val SpecialDoubles: Seq[java.lang.Double] = + Seq(Double.NaN, Double.PositiveInfinity, Double.NegativeInfinity, -0.0, null) + + /** The DOUBLE columns [[specialDoublesPerColumn]] writes into. */ + def specialDoubleColumns(schema: Schema, keepFilled: Set[String]): Seq[String] = + schema.getAttributes + .filter(a => a.getType == AttributeType.DOUBLE && !keepFilled.contains(a.getName)) + .map(_.getName) + + /** Each of [[SpecialDoubles]] once in every DOUBLE column, each column starting + * on a different row, so a row holds a special value in one column beside an + * ordinary number in another. Placement is by column position, so it is the + * same on every run. The rest of the table is untouched. + */ + def specialDoublesPerColumn( + rows: Seq[Tuple], + schema: Schema, + keepFilled: Set[String] + ): Seq[Tuple] = { + if (rows.isEmpty) return rows + val columns = specialDoubleColumns(schema, keepFilled) + val placed: Map[(Int, String), java.lang.Double] = columns.zipWithIndex.flatMap { + case (column, i) => + SpecialDoubles.zipWithIndex.map { + case (value, k) => ((i + k) % rows.size, column) -> value + } + }.toMap + rows.zipWithIndex.map { + case (t, rowIdx) => + val b = Tuple.builder(schema) + schema.getAttributes.foreach { attr => + val v: AnyRef = placed.get((rowIdx, attr.getName)) match { + case Some(special) => special + case None => t.getField[AnyRef](attr.getName) + } + b.add(attr, v) + } + b.build() + } + } + /** One empty cell per column, spread across rows so no row is wholly empty — an * operator that reads two columns should still meet a row where one is filled * and the other is not. Placement is by column position, so it is the same on diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/TransformVerificationRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/TransformVerificationRunner.scala new file mode 100644 index 00000000000..c9705b8cbd4 --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/TransformVerificationRunner.scala @@ -0,0 +1,1398 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.translator.verify + +import com.fasterxml.jackson.databind.JsonNode +import com.fasterxml.jackson.databind.node.{BooleanNode, IntNode, TextNode} +import org.apache.texera.amber.core.tuple.{AttributeType, Schema} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.util.JSONUtils.objectMapper +import org.apache.texera.amber.operator.{ + LogicalOp, + PythonOperatorDescriptor, + StandaloneCodeGenerator +} +import org.apache.texera.amber.operator.aggregate.AggregateOpDesc +import org.apache.texera.amber.operator.dummy.DummyOpDesc +import org.apache.texera.amber.operator.filter.SpecializedFilterOpDesc +import org.apache.texera.amber.operator.sleep.SleepOpDesc +import org.apache.texera.amber.operator.split.SplitOpDesc +import org.apache.texera.amber.operator.sklearn.SklearnClassifierOpDesc +import org.apache.texera.amber.operator.sklearn.SklearnModelOpDesc +import org.apache.texera.amber.operator.sklearn.SklearnGaussianNaiveBayesOpDesc +import org.apache.texera.amber.operator.sklearn.SklearnLinearRegressionOpDesc +import org.apache.texera.amber.operator.machineLearning.sklearnAdvanced.base.SklearnMLOperatorDescriptor +import org.apache.texera.amber.operator.machineLearning.Scorer.MachineLearningScorerOpDesc +import org.apache.texera.amber.operator.huggingFace.HuggingFaceSpamSMSDetectionOpDesc +import org.apache.texera.amber.operator.sklearn.training.SklearnTrainingOpDesc +import org.apache.texera.amber.operator.sklearn.training.SklearnTrainingGaussianNaiveBayesOpDesc +import org.apache.texera.amber.operator.sklearn.testing.SklearnTestingOpDesc +import org.apache.texera.amber.operator.source.scan.file.FileScanOpDesc +import org.apache.texera.amber.operator.sklearn.{ + SklearnAdaptiveBoostingOpDesc, + SklearnBaggingOpDesc, + SklearnBernoulliNaiveBayesOpDesc, + SklearnComplementNaiveBayesOpDesc, + SklearnDecisionTreeOpDesc, + SklearnExtraTreeOpDesc, + SklearnExtraTreesOpDesc, + SklearnGradientBoostingOpDesc, + SklearnKNNOpDesc, + SklearnLinearSVMOpDesc, + SklearnLogisticRegressionCVOpDesc, + SklearnLogisticRegressionOpDesc, + SklearnMultiLayerPerceptronOpDesc, + SklearnMultinomialNaiveBayesOpDesc, + SklearnNearestCentroidOpDesc, + SklearnPassiveAggressiveOpDesc, + SklearnPerceptronOpDesc, + SklearnPredictionOpDesc, + SklearnProbabilityCalibrationOpDesc, + SklearnRandomForestOpDesc, + SklearnRidgeCVOpDesc, + SklearnRidgeOpDesc, + SklearnSDGOpDesc, + SklearnSVMOpDesc +} +import org.apache.texera.amber.operator.sklearn.training.{ + SklearnTrainingAdaptiveBoostingOpDesc, + SklearnTrainingBaggingOpDesc, + SklearnTrainingBernoulliNaiveBayesOpDesc, + SklearnTrainingComplementNaiveBayesOpDesc, + SklearnTrainingDecisionTreeOpDesc, + SklearnTrainingDummyClassifierOpDesc, + SklearnTrainingExtraTreeOpDesc, + SklearnTrainingExtraTreesOpDesc, + SklearnTrainingGradientBoostingOpDesc, + SklearnTrainingKNNOpDesc, + SklearnTrainingLinearRegressionOpDesc, + SklearnTrainingLinearSVMOpDesc, + SklearnTrainingLogisticRegressionCVOpDesc, + SklearnTrainingLogisticRegressionOpDesc, + SklearnTrainingMultiLayerPerceptronOpDesc, + SklearnTrainingMultinomialNaiveBayesOpDesc, + SklearnTrainingNearestCentroidOpDesc, + SklearnTrainingPassiveAggressiveOpDesc, + SklearnTrainingPerceptronOpDesc, + SklearnTrainingProbabilityCalibrationOpDesc, + SklearnTrainingRandomForestOpDesc, + SklearnTrainingRidgeCVOpDesc, + SklearnTrainingRidgeOpDesc, + SklearnTrainingSDGOpDesc, + SklearnTrainingSVMOpDesc +} +import org.apache.texera.amber.operator.machineLearning.sklearnAdvanced.KNNTrainer.{ + SklearnAdvancedKNNClassifierTrainerOpDesc, + SklearnAdvancedKNNRegressorTrainerOpDesc +} +import org.apache.texera.amber.operator.machineLearning.sklearnAdvanced.SVCTrainer.SklearnAdvancedSVCTrainerOpDesc +import org.apache.texera.amber.operator.machineLearning.sklearnAdvanced.SVRTrainer.SklearnAdvancedSVRTrainerOpDesc +import org.apache.texera.amber.operator.substringSearch.SubstringSearchOpDesc +import org.apache.texera.amber.operator.typecasting.TypeCastingOpDesc +import org.apache.texera.amber.operator.unneststring.UnnestStringOpDesc +import org.apache.texera.amber.operator.visualization.wordCloud.WordCloudOpDesc +import org.apache.texera.amber.operator.visualization.DotPlot.DotPlotOpDesc +import org.apache.texera.amber.operator.visualization.barChart.BarChartOpDesc +import org.apache.texera.amber.operator.visualization.boxViolinPlot.BoxViolinPlotOpDesc +import org.apache.texera.amber.operator.visualization.ImageViz.ImageVisualizerOpDesc +import org.apache.texera.amber.operator.visualization.IcicleChart.IcicleChartOpDesc +import org.apache.texera.amber.operator.visualization.bubbleChart.BubbleChartOpDesc +import org.apache.texera.amber.operator.visualization.bulletChart.BulletChartOpDesc +import org.apache.texera.amber.operator.visualization.candlestickChart.CandlestickChartOpDesc +import org.apache.texera.amber.operator.visualization.carpetPlot.CarpetPlotOpDesc +import org.apache.texera.amber.operator.visualization.choroplethMap.ChoroplethMapOpDesc +import org.apache.texera.amber.operator.visualization.continuousErrorBands.ContinuousErrorBandsOpDesc +import org.apache.texera.amber.operator.visualization.contourPlot.ContourPlotOpDesc +import org.apache.texera.amber.operator.visualization.dendrogram.DendrogramOpDesc +import org.apache.texera.amber.operator.visualization.dumbbellPlot.DumbbellPlotOpDesc +import org.apache.texera.amber.operator.visualization.ecdfPlot.ECDFPlotOpDesc +import org.apache.texera.amber.operator.visualization.figureFactoryTable.FigureFactoryTableOpDesc +import org.apache.texera.amber.operator.visualization.filledAreaPlot.FilledAreaPlotOpDesc +import org.apache.texera.amber.operator.visualization.funnelPlot.FunnelPlotOpDesc +import org.apache.texera.amber.operator.visualization.ganttChart.GanttChartOpDesc +import org.apache.texera.amber.operator.visualization.gaugeChart.GaugeChartOpDesc +import org.apache.texera.amber.operator.visualization.ScatterMatrixChart.ScatterMatrixChartOpDesc + +import org.apache.texera.amber.operator.visualization.heatMap.HeatMapOpDesc +import org.apache.texera.amber.operator.visualization.hierarchychart.HierarchyChartOpDesc +import org.apache.texera.amber.operator.visualization.histogram2d.Histogram2DOpDesc +import org.apache.texera.amber.operator.visualization.histogram.HistogramChartOpDesc +import org.apache.texera.amber.operator.visualization.lineChart.LineChartOpDesc +import org.apache.texera.amber.operator.visualization.nestedTable.NestedTableOpDesc +import org.apache.texera.amber.operator.visualization.networkGraph.NetworkGraphOpDesc +import org.apache.texera.amber.operator.visualization.parallelCoordinatesPlot.ParallelCoordinatesPlotOpDesc +import org.apache.texera.amber.operator.visualization.pieChart.PieChartOpDesc +import org.apache.texera.amber.operator.visualization.polarChart.PolarChartOpDesc +import org.apache.texera.amber.operator.visualization.quiverPlot.QuiverPlotOpDesc +import org.apache.texera.amber.operator.visualization.radarChart.RadarChartOpDesc +import org.apache.texera.amber.operator.visualization.radarPlot.RadarPlotOpDesc +import org.apache.texera.amber.operator.visualization.rangeSlider.RangeSliderOpDesc +import org.apache.texera.amber.operator.visualization.sankeyDiagram.SankeyDiagramOpDesc +import org.apache.texera.amber.operator.visualization.scatter3DChart.Scatter3dChartOpDesc +import org.apache.texera.amber.operator.visualization.scatterplot.ScatterplotOpDesc +import org.apache.texera.amber.operator.visualization.stripChart.StripChartOpDesc +import org.apache.texera.amber.operator.visualization.tablesChart.TablesPlotOpDesc +import org.apache.texera.amber.operator.visualization.ternaryContour.TernaryContourOpDesc +import org.apache.texera.amber.operator.visualization.ternaryPlot.TernaryPlotOpDesc +import org.apache.texera.amber.operator.visualization.timeSeriesplot.TimeSeriesOpDesc +import org.apache.texera.amber.operator.visualization.treeplot.TreePlotOpDesc +import org.apache.texera.amber.operator.visualization.volcanoPlot.VolcanoPlotOpDesc +import org.apache.texera.amber.operator.visualization.waterfallChart.WaterfallChartOpDesc +import org.apache.texera.amber.operator.visualization.windRoseChart.WindRoseChartOpDesc +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Path} +import scala.jdk.CollectionConverters._ +import scala.util.{Failure, Success, Try} + +/** + * Unified verification runner for non-source operators implementing + * [[StandaloneCodeGenerator]]. Resolves, per operator: + * - Path A engine: [[PyOpExecHarness]] for PythonOperatorDescriptor, + * [[OpExecHarness]] otherwise; Path B is always [[StandaloneRunner]]. + * - Config + fixture: curated handler ([[CuratedHandlers]]) if registered, + * else [[ConfigGenerator]] against the [[CanonicalFixture]] schemas. + * - Comparison: order-insensitive by default (parallel output order isn't a + * contract); strict positional only when the operator declares + * `orderSensitive = true` (the sort family). All output ports are compared. + * Operators that can't be run are Flagged with a reason — never silently + * skipped. + */ +object TransformVerificationRunner { + + /** + * A value this operator's variants must carry, and where it applies: what the + * generator has to be told about a knob that its metadata does not say. + * + * `Pinned` holds a knob at one value and keeps it out of the sweep, for a + * knob whose other value selects non-determinism rather than another + * behaviour to check. Split's "Auto-Generate Seed" is the case: with it on + * the executor seeds from the clock, so that run agrees with nothing, its own + * previous run included. The value reaches the test name via + * [[pinnedTierNote]], so the run does not read as full coverage. + * + * `WithOptionals` sets a knob inside the `optionals` variant, for a branch + * that needs a switch and the field it governs together. Ternary Plot colours + * its points only when both are set, and the sweep and the optional fill each + * supply one, so neither variant reached the coloured branch. + * + * Named per operator rather than applied wholesale, because switches are not + * generally independent: turning every Boolean on at once paired Sklearn's + * two mutually exclusive text pipelines, among others. + */ + sealed trait KnobScope + object KnobScope { + case object Pinned extends KnobScope + case object WithOptionals extends KnobScope + } + + final case class Knob(field: String, value: JsonNode, scope: KnobScope) + + val knobOverrides: Map[Class[_], Seq[Knob]] = Map( + classOf[SplitOpDesc] -> Seq(Knob("random", BooleanNode.FALSE, KnobScope.Pinned)), + // `SleepOpExec` sleeps this many seconds per tuple, and the generator fills a + // required Int with half the row count, so the fixture would spend tens of + // seconds asleep for nothing: the delay never reaches the output being + // compared, and the standalone translation is a passthrough by design. + classOf[SleepOpDesc] -> Seq(Knob("sleepTime", IntNode.valueOf(0), KnobScope.Pinned)), + classOf[TernaryPlotOpDesc] -> Seq( + Knob("colorEnabled", BooleanNode.TRUE, KnobScope.WithOptionals) + ), + // Both text switches are pinned off on the numeric table: the pipeline they + // build reads a column that table does not have, and `tfidfTransformer` has + // no meaning at all outside a CountVectorizer pipeline (the schema hides it + // when the vectorizer is off). Their branches are generated against the text + // table by the [[AltScenario]]s instead. + classOf[SklearnClassifierOpDesc] -> Seq( + Knob("countVectorizer", BooleanNode.FALSE, KnobScope.Pinned), + Knob("tfidfTransformer", BooleanNode.FALSE, KnobScope.Pinned) + ), + classOf[SklearnTrainingOpDesc] -> Seq( + Knob("countVectorizer", BooleanNode.FALSE, KnobScope.Pinned), + Knob("tfidfTransformer", BooleanNode.FALSE, KnobScope.Pinned) + ) + ) + + /** This operator's overrides for one scope, as the generator takes them. + * Exact class first, then base class, so a family can be named once instead + * of per estimator. + */ + private def knobsFor(opClass: Class[_ <: LogicalOp], scope: KnobScope): Map[String, JsonNode] = + knobOverrides + .get(opClass) + .orElse(knobOverrides.collectFirst { + case (base, knobs) if base.isAssignableFrom(opClass) => knobs + }) + .getOrElse(Seq.empty) + .filter(_.scope == scope) + .map(k => k.field -> k.value) + .toMap + + /** A second generation pass for a branch the base config cannot reach. Usually + * that is a branch needing a DIFFERENT table: a swept variant cannot switch + * tables, since [[ConfigGenerator]] resolves every column picker against ONE + * schema, so the branch is generated separately against the table it needs + * with its switch pinned on. + * + * It also reaches a branch whose own knobs the base config leaves empty. The + * sweep offers the values a config already holds, so a list filled only on the + * far side of a switch has nothing to offer until the switch is pinned — and + * then the second pass may well take the same table as the first. + * + * The auto-tier twin of [[TransformHandler.extraScenarios]]: it names only the + * table and the pins, and the generator writes the config. + */ + /** @param withGaps one empty cell per column, as [[nullsCase]] writes the base + * table. That case takes the base config, so a hole only ever + * lands in a column the operator carries through, never in the + * one a pin points it at. + */ + final case class AltScenario( + label: String, + fixture: SharedFixture, + pinned: Map[String, JsonNode], + withGaps: Boolean = false + ) + + /** Every kind of run a [[variantsNotRun]] row can name: the derived variants, + * plus the [[AltScenario]] labels (an alt scenario IS one kind of run). Named + * so a row cannot misspell one and silently stop suppressing anything. + */ + object RunKind { + val Nulls = "nulls" + val EmptyTable = "emptyTable" + val SpecialDoubles = "specialDoubles" + val EnumSweep = "enumSweep" + val HostileText = "hostileText" + val CountVectorizerText = "countVectorizer_text" + val TfidfText = "tfidf_text" + val NonFeatureColumn = "nonFeatureColumn" + val HoledIntegerColumn = "holedIntegerColumn" + val BooleanColumn = "booleanColumn" + val NumericTextColumn = "numericTextColumn" + val TextLabels = "textLabels" + val RegressionBranch = "regressionBranch" + } + + /** The [[RunKind]] a generated variant's label names. A `merged` variant labels + * itself `kind(fields…)`, and the fields it happened to move are not part of + * what is being withheld. + */ + private def kindOf(label: String): String = label.takeWhile(_ != '(') + + /** Keyed by BASE class, not by concrete operator: a newly registered sklearn + * estimator is covered with no entry of its own, matching how the families + * themselves are discovered. + */ + val altFixtureScenarios: Map[Class[_], Seq[AltScenario]] = { + // One scenario per text pipeline rather than a sweep inside one: which + // pipeline is built is the branch under test, and the two are alternatives, + // not a knob crossed with everything else. + val countVectorizerText = AltScenario( + label = RunKind.CountVectorizerText, + fixture = CanonicalFixture.sklearnText, + pinned = Map( + "countVectorizer" -> BooleanNode.TRUE, + "tfidfTransformer" -> BooleanNode.FALSE + ) + ) + val tfidfText = countVectorizerText.copy( + label = RunKind.TfidfText, + pinned = Map( + "countVectorizer" -> BooleanNode.TRUE, + "tfidfTransformer" -> BooleanNode.TRUE + ) + ) + // The vectorizer stays off: what this scenario covers is the branch that + // narrows `X` for an estimator, and Count Vectorizer replaces it rather than + // feeding it, naming the text columns the narrowing would otherwise drop. + val nonFeatureColumn = AltScenario( + label = RunKind.NonFeatureColumn, + fixture = CanonicalFixture.sklearnNumericWithText, + pinned = Map( + "countVectorizer" -> BooleanNode.FALSE, + "tfidfTransformer" -> BooleanNode.FALSE + ) + ) + // The advanced trainers are the ones left out: they name the feature columns + // themselves rather than taking every column but the target, so a column an + // estimator cannot fit is not reachable for them. + // A holed integer column, which the `@SampleColumn` text one cannot show: it + // widens to float in the script and grows a decimal point. The species label + // rather than `id`, which is held filled so joins keep pairing. + val holedIntegerColumn = AltScenario( + label = RunKind.HoledIntegerColumn, + fixture = CanonicalFixture, + pinned = Map("Attribute" -> TextNode.valueOf("a\"b\\c_species")), + withGaps = true + ) + // The one boolean column, holed for the same reason: pandas then reads it as + // 1.0 and 0.0 where the engine wrote "true". Filled it renders right on its + // own, so the hole is what makes this ask something. + val booleanColumn = AltScenario( + label = RunKind.BooleanColumn, + fixture = CanonicalFixture, + pinned = Map("Attribute" -> TextNode.valueOf("a\"b\\c_high_score")), + withGaps = true + ) + // The same two columns. The substring is pinned because a generated one would + // not tell the renderings apart: "0" is in both "0" and the widened "1.0". + val substringScenarios = Seq( + holedIntegerColumn.copy( + pinned = Map( + "attribute" -> TextNode.valueOf("a\"b\\c_species"), + "substring" -> TextNode.valueOf("0") + ) + ), + booleanColumn.copy( + pinned = Map( + "attribute" -> TextNode.valueOf("a\"b\\c_high_score"), + "substring" -> TextNode.valueOf("true") + ) + ) + ) + // Word Cloud pointed at a column of numbers. Its type rule only warns, so a + // user reaches this, and `.str` raises on anything but text. + val numericTextColumn = AltScenario( + label = RunKind.NumericTextColumn, + fixture = CanonicalFixture, + pinned = Map("textColumn" -> TextNode.valueOf("id")) + ) + Map( + classOf[WordCloudOpDesc] -> Seq(numericTextColumn), + classOf[UnnestStringOpDesc] -> Seq(holedIntegerColumn, booleanColumn), + classOf[SubstringSearchOpDesc] -> substringScenarios, + classOf[SklearnClassifierOpDesc] -> Seq(countVectorizerText, tfidfText, nonFeatureColumn), + classOf[SklearnTrainingOpDesc] -> Seq(countVectorizerText, tfidfText, nonFeatureColumn), + // No text scenarios, and nothing pinned: this operator declares neither + // switch, and a pin is set on the config whether or not the field exists, + // so pinning one here would hand it a property it cannot read back. + classOf[SklearnLinearRegressionOpDesc] -> Seq(nonFeatureColumn.copy(pinned = Map.empty)), + // A scorer reads a text label as readily as a numeric one, and names the + // class after the label rather than after its position. Regression is + // pinned off rather than swept: a regression metric puts both columns + // through `float()`, so on this table the sweep would generate the one + // configuration the operator is right to refuse. The two columns are + // pinned because the operator's `@SampleColumn`s name the numeric pair, + // and an annotation naming a column the table does not hold ends the run + // rather than falling back — which is what catches a misspelling. + classOf[MachineLearningScorerOpDesc] -> Seq( + AltScenario( + label = RunKind.TextLabels, + fixture = CanonicalFixture.scorerTextLabels, + pinned = Map( + "isRegression" -> BooleanNode.FALSE, + "actualValueColumn" -> TextNode.valueOf("species_name"), + "predictValueColumn" -> TextNode.valueOf("species_name_pred") + ) + ), + // The regression metrics are unreachable from the base config: the sweep + // reads the sites the config already holds, and on the classification + // branch the regression list is empty, so it offers none. Pinned on, the + // list is filled before the sweep looks, and the other three metrics + // become variants like any other enum. Same table as the default runs — + // this scenario is here for the branch, not for a different set of rows. + AltScenario( + label = RunKind.RegressionBranch, + fixture = CanonicalFixture, + pinned = Map("isRegression" -> BooleanNode.TRUE) + ) + ) + ) + } + + /** The alternate-table scenarios this operator takes, resolved by family and + * minus any [[variantsNotRun]] names. + */ + private def altScenariosFor(opClass: Class[_ <: LogicalOp]): Seq[AltScenario] = + altFixtureScenarios + .collectFirst { case (base, scenarios) if base.isAssignableFrom(opClass) => scenarios } + .getOrElse(Seq.empty) + .filterNot(alt => notRun(opClass, alt.label)) + + /** How a pinned operator's tier reads in the report, e.g. `auto, random=false`. */ + private def pinnedTierNote(opClass: Class[_ <: LogicalOp]): String = { + val pinned = knobsFor(opClass, KnobScope.Pinned) + if (pinned.isEmpty) "" + else pinned.map { case (field, value) => s"$field=${value.asText}" }.mkString(", ", ", ", "") + } + + /** One kind of run an operator is not compared on, and why. + * + * `op` matches its subclasses, so one row covers a family, and `except` names + * the members of that family the row does not cover, for the one that behaves + * otherwise. `kind` is a [[RunKind]]. + */ + final case class RunRow( + op: Class[_], + kind: String, + why: String, + except: Seq[Class[_]] = Seq.empty + ) + + /** The runs an operator does not get, because they cannot be set up, and why. + * + * One table rather than one per kind: every row makes the same statement, so + * the next exemption has an obvious home instead of arriving as another set + * somewhere else. + * + * A curated handler's own [[TransformHandler.nullsKeepFilled]] stays where it + * is: it describes the table that handler wrote rather than the operator, and + * changes when the fixture is rewritten. + */ + val variantsNotRun: Seq[RunRow] = { + // The operator refuses the text pipeline in `getOutputSchemas`, so there is no + // configuration to compare: neither path is generated. An invalid configuration + // rather than a translation gap. + // + // Not sklearn raising, which is what the estimator's own limitation would look + // like. This fixture's word counts repeat enough that `ColumnTransformer` stays + // above its 0.3 sparse threshold and hands over a dense array, which GaussianNB + // fits without complaint. Only a wider vocabulary would reach the limitation + // the operator is guarding against. + val dense = "the operator refuses Count Vectorizer at compile time" + val denseOnly = for { + op <- Seq( + classOf[SklearnGaussianNaiveBayesOpDesc], + classOf[SklearnTrainingGaussianNaiveBayesOpDesc] + ) + label <- Seq(RunKind.CountVectorizerText, RunKind.TfidfText) + } yield RunRow(op, label, dense) + + Seq( + // An enum whose legal values depend on a sibling field: flipping it alone + // builds a config the curated fixture already covers properly. + RunRow( + classOf[TypeCastingOpDesc], + RunKind.EnumSweep, + "resultType is legal only for certain source column types, and the native " + + "executor throws on an illegal cast; the fixture pairs each type with a " + + "compatible column already" + ), + RunRow( + classOf[SklearnTestingOpDesc], + RunKind.EnumSweep, + "isRegression has to match the model the model port carries, and the " + + "fixture's is a classifier; a regressor on a numeric target covers " + + "the other branch in its own scenario" + ), + RunRow( + classOf[AggregateOpDesc], + RunKind.EnumSweep, + "aggFunction is cross-constrained with its attribute's type and with " + + "COUNT(*)'s empty attribute; the fixture pairs each function with a " + + "compatible column already" + ), + RunRow( + classOf[FileScanOpDesc], + "attributeType=large binary", + "a large binary is written to object storage, and the harness gives the " + + "executor no storage to write it to" + ), + // The one estimator that fits on nothing: it never reads a feature. Both + // paths hand back a model, but one with no classes, which raises on any row + // it is asked to predict, so there is no answer to compare it by. + RunRow( + classOf[SklearnTrainingDummyClassifierOpDesc], + RunKind.EmptyTable, + "the dummy classifier fits on nothing into a model with no classes, which " + + "cannot predict anything, so there is no answer to compare; both paths " + + "do return one" + ) + ) ++ denseOnly + } + + /** The runs both paths must refuse, and why. + * + * Each still runs, and passes only when the engine and the script both raise: + * a script that answers where the engine refuses is as wrong as a different + * answer. Two errors pass only for a run named here. Anywhere else they fail + * the run, so a broken environment, which makes both paths raise on + * everything, cannot turn the suite green. + */ + val variantsBothRefuse: Seq[RunRow] = { + // Fitting a model on nothing has no answer to compare, and both paths say so + // the same way: sklearn refuses outright ("Found array with 0 sample(s) while + // a minimum of 1 is required"). Listed across both estimator hierarchies + // rather than one row each, since the reason is the estimator's, not the + // operator's. The dummy classifier is left out: it does fit on nothing, and + // [[variantsNotRun]] says why it is not compared either. + val nothingToFit = + "an estimator has no rows to fit on, and sklearn refuses rather than returning a model" + val noRowsToFit = Seq( + RunRow( + classOf[SklearnModelOpDesc], + RunKind.EmptyTable, + nothingToFit, + except = Seq(classOf[SklearnTrainingDummyClassifierOpDesc]) + ), + // Named on its own because it is the one estimator outside that hierarchy: + // it mixes in SklearnFittableColumns directly. Keying the row on the trait + // instead would also cover SklearnTestingOpDesc, which answers. + RunRow(classOf[SklearnLinearRegressionOpDesc], RunKind.EmptyTable, nothingToFit), + RunRow(classOf[SklearnMLOperatorDescriptor[_]], RunKind.EmptyTable, nothingToFit), + RunRow( + classOf[MachineLearningScorerOpDesc], + RunKind.EmptyTable, + "the operator refuses a table where no row carries an actual value, and " + + "an empty one carries none; the refusal is the operator's own and both " + + "paths raise it" + ) + ) + + Seq( + // Stated about the operator's own fixture rather than about the operator: a + // predicate over a string column takes the hostile value fine, so this row goes + // the day that fixture filters on one. + RunRow( + classOf[SpecializedFilterOpDesc], + RunKind.HostileText, + "`id > 8` compares against an INTEGER column, and the platform parses the " + + "predicate value as that column's type, so the number parser refuses the " + + "hostile string before any escaping could matter" + ), + RunRow( + classOf[SklearnMLOperatorDescriptor[_]], + RunKind.HostileText, + "what a hyperparameter's value may hold is decided by the parameter beside " + + "it, and every one of those is a number or a word from a fixed set, so a " + + "spliced a\"b fails at the conversion rather than at any escaping" + ), + RunRow( + classOf[FileScanOpDesc], + "extract=true", + "the file it is handed is the shared table's JSON, not an archive, so the " + + "engine finds no archiver for it and Python's zipfile refuses it alike" + ) + ) ++ fileScanTypedLines ++ noRowsToFit ++ refusesNonFinite + } + + // The sweep reads the shared table's JSON as each type a line can be cast to, + // and its first line is `[`, which no parser takes. File Scan's own scenarios + // read files whose lines are of the type, so these only ask that both refuse. + private def fileScanTypedLines: Seq[RunRow] = + Seq("integer", "long", "double", "boolean", "timestamp").map { t => + RunRow( + classOf[FileScanOpDesc], + s"attributeType=$t", + s"the shared table's JSON lines are text, so parsing the first as $t fails on both sides" + ) + } + + // sklearn checks its input for finite values before it fits or predicts, and + // raises on a NaN or an infinity alike. Both paths call the same library, so + // both refuse the same table; which value it names first depends on how the + // column arrived, and the run fails either way. Listed one by one rather than + // by family: the two dummy classifiers never read a feature, and they pass. + private def refusesNonFinite: Seq[RunRow] = { + val notFinite = + "the library refuses a NaN or an infinity in what it fits, predicts or draws, " + + "and both paths call it, so both raise on this table" + for { + op <- Seq[Class[_ <: LogicalOp]]( + classOf[SklearnAdaptiveBoostingOpDesc], + classOf[SklearnAdvancedKNNClassifierTrainerOpDesc], + classOf[SklearnAdvancedKNNRegressorTrainerOpDesc], + classOf[SklearnAdvancedSVCTrainerOpDesc], + classOf[SklearnAdvancedSVRTrainerOpDesc], + classOf[SklearnBaggingOpDesc], + classOf[SklearnBernoulliNaiveBayesOpDesc], + classOf[SklearnComplementNaiveBayesOpDesc], + classOf[SklearnDecisionTreeOpDesc], + classOf[SklearnExtraTreeOpDesc], + classOf[SklearnExtraTreesOpDesc], + classOf[SklearnGaussianNaiveBayesOpDesc], + classOf[SklearnGradientBoostingOpDesc], + classOf[SklearnKNNOpDesc], + classOf[SklearnLinearRegressionOpDesc], + classOf[SklearnLinearSVMOpDesc], + classOf[SklearnLogisticRegressionCVOpDesc], + classOf[SklearnLogisticRegressionOpDesc], + classOf[SklearnMultiLayerPerceptronOpDesc], + classOf[SklearnMultinomialNaiveBayesOpDesc], + classOf[SklearnNearestCentroidOpDesc], + classOf[SklearnPassiveAggressiveOpDesc], + classOf[SklearnPerceptronOpDesc], + classOf[SklearnPredictionOpDesc], + classOf[SklearnProbabilityCalibrationOpDesc], + classOf[SklearnRandomForestOpDesc], + classOf[SklearnRidgeCVOpDesc], + classOf[SklearnRidgeOpDesc], + classOf[SklearnSDGOpDesc], + classOf[SklearnSVMOpDesc], + classOf[SklearnTestingOpDesc], + classOf[SklearnTrainingAdaptiveBoostingOpDesc], + classOf[SklearnTrainingBaggingOpDesc], + classOf[SklearnTrainingBernoulliNaiveBayesOpDesc], + classOf[SklearnTrainingComplementNaiveBayesOpDesc], + classOf[SklearnTrainingDecisionTreeOpDesc], + classOf[SklearnTrainingExtraTreeOpDesc], + classOf[SklearnTrainingExtraTreesOpDesc], + classOf[SklearnTrainingGaussianNaiveBayesOpDesc], + classOf[SklearnTrainingGradientBoostingOpDesc], + classOf[SklearnTrainingKNNOpDesc], + classOf[SklearnTrainingLinearRegressionOpDesc], + classOf[SklearnTrainingLinearSVMOpDesc], + classOf[SklearnTrainingLogisticRegressionCVOpDesc], + classOf[SklearnTrainingLogisticRegressionOpDesc], + classOf[SklearnTrainingMultiLayerPerceptronOpDesc], + classOf[SklearnTrainingMultinomialNaiveBayesOpDesc], + classOf[SklearnTrainingNearestCentroidOpDesc], + classOf[SklearnTrainingPassiveAggressiveOpDesc], + classOf[SklearnTrainingPerceptronOpDesc], + classOf[SklearnTrainingProbabilityCalibrationOpDesc], + classOf[SklearnTrainingRandomForestOpDesc], + classOf[SklearnTrainingRidgeCVOpDesc], + classOf[SklearnTrainingRidgeOpDesc], + classOf[SklearnTrainingSDGOpDesc], + classOf[SklearnTrainingSVMOpDesc], + // The charts whose library refuses one: Bubble's marker size, the + // Contour fit and the Dendrogram distance matrix. + classOf[BubbleChartOpDesc], + classOf[ContourPlotOpDesc], + classOf[DendrogramOpDesc] + ) + } yield RunRow(op, RunKind.SpecialDoubles, notFinite) + } + + /** Why `rows` names this kind of run for this operator, if it does. A family + * row and an operator row for the same kind are the same statement twice, and + * the first one wins. + */ + private def reasonIn( + rows: Seq[RunRow], + opClass: Class[_ <: LogicalOp], + kind: String + ): Option[String] = + rows.collectFirst { + case RunRow(op, k, why, except) + if k == kind && op.isAssignableFrom(opClass) && + !except.exists(_.isAssignableFrom(opClass)) => + why + } + + private def notRun(opClass: Class[_ <: LogicalOp], kind: String): Boolean = + reasonIn(variantsNotRun, opClass, kind).isDefined + + /** Why both paths must refuse this kind of run, if they must. */ + private def refusal(opClass: Class[_ <: LogicalOp], kind: String): Option[String] = + reasonIn(variantsBothRefuse, opClass, kind) + + /** Visualization operators with deterministic Plotly JSON validation. */ + val visualizationJsonOps: Set[Class[_]] = Set( + // Its layout is seeded, so both paths place the nodes identically and the two + // figures can be compared number by number. + classOf[NetworkGraphOpDesc], + classOf[RangeSliderOpDesc], + classOf[HeatMapOpDesc], + classOf[HierarchyChartOpDesc], + classOf[HistogramChartOpDesc], + classOf[Histogram2DOpDesc], + classOf[LineChartOpDesc], + classOf[ParallelCoordinatesPlotOpDesc], + classOf[PieChartOpDesc], + classOf[PolarChartOpDesc], + classOf[QuiverPlotOpDesc], + classOf[RadarChartOpDesc], + classOf[RadarPlotOpDesc], + classOf[SankeyDiagramOpDesc], + classOf[Scatter3dChartOpDesc], + classOf[ScatterplotOpDesc], + classOf[StripChartOpDesc], + classOf[TablesPlotOpDesc], + classOf[TernaryContourOpDesc], + classOf[TernaryPlotOpDesc], + classOf[TimeSeriesOpDesc], + classOf[TreePlotOpDesc], + classOf[VolcanoPlotOpDesc], + classOf[WaterfallChartOpDesc], + classOf[WindRoseChartOpDesc], + classOf[BarChartOpDesc], + classOf[BulletChartOpDesc], + classOf[CandlestickChartOpDesc], + classOf[CarpetPlotOpDesc], + classOf[ChoroplethMapOpDesc], + classOf[ContinuousErrorBandsOpDesc], + classOf[ContourPlotOpDesc], + classOf[DendrogramOpDesc], + classOf[DumbbellPlotOpDesc], + classOf[ECDFPlotOpDesc], + classOf[FigureFactoryTableOpDesc], + classOf[FilledAreaPlotOpDesc], + classOf[FunnelPlotOpDesc], + classOf[GanttChartOpDesc], + classOf[GaugeChartOpDesc], + classOf[DotPlotOpDesc], + classOf[IcicleChartOpDesc], + classOf[BubbleChartOpDesc], + classOf[ScatterMatrixChartOpDesc], + classOf[BoxViolinPlotOpDesc] + ) + + /** Visualization operators with deterministic HTML validation. */ + val visualizationHtmlOps: Set[Class[_]] = Set( + classOf[ImageVisualizerOpDesc], + // A word cloud is a picture, not a figure with values to read, so the two + // paths are compared as the HTML they emit. Its placement is seeded, which + // is what makes that comparison mean anything. + classOf[WordCloudOpDesc], + classOf[NestedTableOpDesc] + ) + + /** Triaged, explicitly-not-run operators: class → honest reason, shown in + * the test report and coverage table. + */ + val knownIssues: Map[Class[_], String] = Map( + classOf[DummyOpDesc] -> + ("harness gap: placeholder operator with no physical execution — " + + "LogicalOp.getPhysicalOp throws NotImplementedError") + ) + + sealed trait Disposition + final case class Runnable(tier: String) extends Disposition // "auto" | "curated" + final case class Flagged(reason: String) extends Disposition + + /** When `VERIFY_FORCE_AUTO=1`, ignore CuratedHandlers so every operator is + * exercised through the shared-table auto path instead. Lets us measure how + * much of the hand-written curated set the auto tier can now replace: an op + * that stays RUNNABLE/passes under force-auto no longer needs its curated + * handler. + */ + private def forceAuto: Boolean = sys.env.get("VERIFY_FORCE_AUTO").contains("1") + + /** The shared table an operator runs on in the AUTO tier. Which table an + * operator takes is its own axis (see [[SharedFixture]]); the auto tier used + * to be pinned to the whole of [[CanonicalFixture]], which is why an operator + * needing a narrower table had to be curated just to name one. sklearn cannot + * fit canonical's string columns — `X = table.drop(target)` feeds every + * remaining column to `fit` — so its families take the petal-and-label view + * of that same table. + */ + private[verify] def fixtureFor(opClass: Class[_ <: LogicalOp]): SharedFixture = + if (CuratedHandlers.sklearnNumericClasses.contains(opClass)) CanonicalFixture.sklearnNumeric + else if (opClass == classOf[HuggingFaceSpamSMSDetectionOpDesc]) CanonicalFixture.withoutScore + else CanonicalFixture + + /** Static classification — cheap (reflection only, no subprocesses), called + * at spec construction time to decide test-vs-ignore. + */ + def disposition(opClass: Class[_ <: LogicalOp]): Disposition = + knownIssues.get(opClass) match { + case Some(reason) => Flagged(s"known issue: $reason") + case None => + Try(opClass.getDeclaredConstructor().newInstance()) match { + case Failure(e) => Flagged(s"cannot instantiate: ${e.getMessage}") + case Success(op: StandaloneCodeGenerator) => + if (!op.producesDataFrame()) + if (visualizationJsonOps.contains(opClass) || visualizationHtmlOps.contains(opClass)) + Runnable("visualization") + else Flagged("visualization: no DataFrame output to compare") + else if (!forceAuto && CuratedHandlers.byClass.contains(opClass)) + Runnable("curated") + else + ConfigGenerator.generate(opClass, fixtureFor(opClass).schemasByPort) match { + case Left(reason) => Flagged(s"cannot auto-configure: $reason") + case Right(configured) => + Try(configured.operatorInfo.inputPorts.size) match { + case Failure(e) => + Flagged(s"operatorInfo failed on generated config: ${e.getMessage}") + case Success(n) if n < 1 || n > 2 => + Flagged(s"unsupported input port count: $n") + case Success(_) + if outputHasBinaryColumn(configured, fixtureFor(opClass)) && + fixtureFor(opClass) == CanonicalFixture => + // A trained-model (BINARY) output cannot be fit on the + // canonical table, whose string columns reach `fit`. The + // model itself is not byte-comparable either, but that is + // handled for every tier alike (see modelColumns in run). + // An op that names a numeric fixture is fine here. + Flagged( + "model output: emits a BINARY (trained-model) column; " + + "requires a numeric fixture, not the canonical table" + ) + case Success(_) => Runnable(s"auto${pinnedTierNote(opClass)}") + } + } + case Success(_) => + Flagged("does not implement StandaloneCodeGenerator") + } + } + + /** True if the configured operator declares a BINARY output column (e.g. a + * serialized trained model). Best-effort: only Python descriptors expose + * getOutputSchemas, and a throw (schema needs real inputs) reads as "no + * detectable BINARY column" so the op falls through to its normal tier. + */ + private def outputHasBinaryColumn(configured: LogicalOp, fixture: SharedFixture): Boolean = + configured match { + case p: PythonOperatorDescriptor => + val inputSchemas = fixture.schemasByPort.map { + case (port, schema) => PortIdentity(port) -> schema + } + Try(p.getOutputSchemas(inputSchemas)).toOption + .exists(_.values.exists(_.getAttributes.exists(_.getType == AttributeType.BINARY))) + case _ => false + } + + /** Execute both paths and assert parity on every declared output port. + * Precondition: disposition(opClass) returned Runnable. + */ + def run(opClass: Class[_ <: LogicalOp]): Unit = { + val testRoot = Files.createTempDirectory(s"verify-${opClass.getSimpleName}-") + + // Resolve the run list: each entry is (label, configured op, its inputs). + // Both tiers yield the base config PLUS one variant per enum value, so each + // enum branch (e.g. a line chart's mode = line / dots / line+dots) is + // exercised, not just the default, PLUS the `optionals` and `hostileText` + // variants. Variants of one fixture share input files; a curated handler's + // extraScenarios carry their own (structurally different) inputs. + val runs: Seq[(String, LogicalOp, Map[PortIdentity, Path])] = + (if (forceAuto) None else CuratedHandlers.byClass.get(opClass)) match { + case Some(handler) => + val (op, in) = handler.fixture(testRoot) + // The variants are derived against the handler's OWN fixture, not the + // canonical one — a curated handler writes the table its operator needs, + // so that is what an optional column knob has to resolve against. + // + // An enum-sweep-exempt op still gets the fills: what is cross-constrained + // with a sibling field is its ENUM values, so a blind sweep produces invalid + // configs — filling an optional knob or splicing a quote does not. + // + // Fall back to the single curated config if it can't be varied at all. + val primary = + ConfigGenerator + .fullVariantsOf( + op, + schemasOf(in), + rowCountOf(in), + sweepEnums = !notRun(opClass, RunKind.EnumSweep) + ) + .fold(_ => Seq("default" -> op), identity) + // A variant this operator does not get, named in [[variantsNotRun]]. + .filterNot { case (label, _) => notRun(opClass, kindOf(label)) } + primary.map { case (label, o) => (label, o, in) } ++ + handler.extraScenarios(testRoot) ++ + handler.nullsKeepFilled.toSeq.flatMap( + curatedNullsCase(opClass, op, in, testRoot, _) + ) ++ + handler.nullsKeepFilled.toSeq.flatMap( + curatedSpecialDoublesCase(opClass, op, in, testRoot, _) + ) ++ + curatedEmptyTableCase(opClass, op, in, testRoot) + case None => + val fixture = fixtureFor(opClass) + val vs = ConfigGenerator + .generateVariants( + opClass, + fixture.schemasByPort, + fixture.port0RowCount, + knobsFor(opClass, KnobScope.Pinned), + knobsFor(opClass, KnobScope.WithOptionals) + ) + .fold( + reason => throw new IllegalStateException(s"cannot auto-configure: $reason"), + identity + ) + val inputPortCount = vs.head._2.operatorInfo.inputPorts.size + val in = fixture.writeInputs(testRoot, inputPortCount) + // A variant the operator itself cannot take, named in [[variantsNotRun]]: + // for a swept hyperparameter that is one row per parameter, so the sweep + // keeps covering the rest. + vs.filterNot { case (label, _) => notRun(opClass, kindOf(label)) } + .map { case (label, o) => (label, o, in) } ++ + nullsCase(opClass, vs.head._2, testRoot, fixture) ++ + specialDoublesCase(opClass, vs.head._2, testRoot, fixture) ++ + emptyTableCase(opClass, vs.head._2, testRoot, fixture) ++ + altScenariosFor(opClass).flatMap { alt => + // Each scenario writes under its own directory: two tables in one + // testRoot would otherwise both claim input_port_0.jsonl. + val dir = testRoot.resolve(alt.label) + Files.createDirectories(dir) + ConfigGenerator + .generateVariants( + opClass, + alt.fixture.schemasByPort, + alt.fixture.port0RowCount, + pinned = alt.pinned, + switches = knobsFor(opClass, KnobScope.WithOptionals) + ) + .fold( + reason => + throw new IllegalStateException( + s"cannot auto-configure ${alt.label}: $reason" + ), + identity + ) + // The base variant carries the branch's own column knob: the pins + // are visible while the config is built, so the knob the schema + // requires under them is filled like any other required field. + .map { + case (label, o) => + ( + s"${alt.label}/$label", + o, + alt.fixture.write(dir, inputPortCount, withGaps = alt.withGaps) + ) + } + } + } + + runs.foreach { + case (label, opDesc, inputs) => + val workDir = + if (runs.size == 1) testRoot + else testRoot.resolve(label.replaceAll("[^A-Za-z0-9]+", "_")) + Files.createDirectories(workDir) + try refusal(opClass, kindOf(label)) match { + case Some(why) => runExpectingRefusal(opClass, opDesc, inputs, workDir, why) + case None => runVariant(opClass, opDesc, inputs, workDir) + } catch { + case e: Throwable => + throw new AssertionError(s"[variant: $label] ${e.getMessage}", e) + } + } + } + + /** One extra run per operator, on `fixture` with one empty cell per column (see + * [[SharedFixture.emptyOneCellPerColumn]]). It takes the base config rather than + * crossing with the other variants: what an operator does with a null is a + * property of the operator, and multiplying it across every knob would buy more + * runtime than signal. + * + * The auto tier's form: the table is the shared one [[fixtureFor]] resolves, so + * the holes come from the fixture itself. See [[curatedNullsCase]] for the other + * tier, which has no fixture object to ask. + */ + private def nullsCase( + opClass: Class[_ <: LogicalOp], + base: LogicalOp, + testRoot: Path, + fixture: SharedFixture + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = + if (notRun(opClass, RunKind.Nulls)) Seq.empty + else { + val dir = testRoot.resolve("nulls-input") + Files.createDirectories(dir) + val in = fixture.write(dir, base.operatorInfo.inputPorts.size, withGaps = true) + Seq(("nulls", base, in)) + } + + /** [[nullsCase]] for the curated tier, where there is no fixture object to write + * a second time: the handler's own files are read back, holed, and rewritten. + * So a handler opts in by naming its load-bearing columns and nothing else, and + * [[TransformHandler.fixture]] keeps returning paths. + */ + private def curatedNullsCase( + opClass: Class[_ <: LogicalOp], + base: LogicalOp, + inputs: Map[PortIdentity, Path], + testRoot: Path, + keepFilled: Set[String] + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = + if (notRun(opClass, RunKind.Nulls)) Seq.empty + else { + val dir = testRoot.resolve("nulls-input") + Files.createDirectories(dir) + val holed = inputs.map { + case (portId, path) => + val schema = TupleIO.readSchemaSidecar(path) + val rows = TupleIO.readTuples(path, schema).toSeq + val out = dir.resolve(path.getFileName.toString) + TupleIO.writeTuples( + out, + SharedFixture.emptyOneCellPerColumn(rows, schema, keepFilled).iterator, + schema + ) + portId -> out + } + Seq(("nulls", base, holed)) + } + + /** One more run per operator, on the same columns with no rows under them (see + * [[SharedFixture.writeEmptyInputs]]). Like [[nullsCase]] it takes the base + * config: an empty table is a property of what arrived, not of any knob. + * + * It is a different question from the one [[nullsCase]] asks. That table still + * has a value in every column somewhere, so code that reads a column's range or + * its quantiles still finds one. Here there is nothing to read, and an operator + * that assumed otherwise raises instead of passing the emptiness through. + */ + private def emptyTableCase( + opClass: Class[_ <: LogicalOp], + base: LogicalOp, + testRoot: Path, + fixture: SharedFixture + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = + if (notRun(opClass, RunKind.EmptyTable)) Seq.empty + else { + val dir = testRoot.resolve("empty-input") + Files.createDirectories(dir) + val in = fixture.writeEmptyInputs(dir, base.operatorInfo.inputPorts.size) + Seq((RunKind.EmptyTable, base, in)) + } + + /** [[emptyTableCase]] for the curated tier: the handler's own files are read for + * their schema alone and rewritten with no rows. Unlike [[curatedNullsCase]] + * there is nothing for a handler to opt into, since emptying a table cannot + * single out a load-bearing column. + */ + private def curatedEmptyTableCase( + opClass: Class[_ <: LogicalOp], + base: LogicalOp, + inputs: Map[PortIdentity, Path], + testRoot: Path + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = + if (notRun(opClass, RunKind.EmptyTable)) Seq.empty + else { + val dir = testRoot.resolve("empty-input") + Files.createDirectories(dir) + val emptied = inputs.map { + case (portId, path) => + val out = dir.resolve(path.getFileName.toString) + TupleIO.writeTuples(out, Iterator.empty, TupleIO.readSchemaSidecar(path)) + portId -> out + } + Seq((RunKind.EmptyTable, base, emptied)) + } + + /** One more run per operator, on the same table with NaN, both infinities, -0.0 + * and a null in every DOUBLE column (see [[SharedFixture.specialDoublesPerColumn]]). + * Like [[nullsCase]] it takes the base config. A table with no DOUBLE column the + * values could go in gets no run, since it would be the base table again. + */ + private def specialDoublesCase( + opClass: Class[_ <: LogicalOp], + base: LogicalOp, + testRoot: Path, + fixture: SharedFixture + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = + if (notRun(opClass, RunKind.SpecialDoubles) || !fixture.hasSpecialDoubleColumn) Seq.empty + else { + val dir = testRoot.resolve("special-doubles-input") + Files.createDirectories(dir) + val in = fixture.writeSpecialDoubles(dir, base.operatorInfo.inputPorts.size) + Seq((RunKind.SpecialDoubles, base, in)) + } + + /** [[specialDoublesCase]] for the curated tier, read back and rewritten the way + * [[curatedNullsCase]] is, and opted into by the same `nullsKeepFilled`: a + * column the handler holds filled holds its ordinary numbers too. + */ + private def curatedSpecialDoublesCase( + opClass: Class[_ <: LogicalOp], + base: LogicalOp, + inputs: Map[PortIdentity, Path], + testRoot: Path, + keepFilled: Set[String] + ): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = { + val schemas = inputs.map { case (portId, path) => portId -> TupleIO.readSchemaSidecar(path) } + if ( + notRun(opClass, RunKind.SpecialDoubles) || + schemas.values.forall(SharedFixture.specialDoubleColumns(_, keepFilled).isEmpty) + ) Seq.empty + else { + val dir = testRoot.resolve("special-doubles-input") + Files.createDirectories(dir) + val rewritten = inputs.map { + case (portId, path) => + val schema = schemas(portId) + val rows = TupleIO.readTuples(path, schema).toSeq + val out = dir.resolve(path.getFileName.toString) + TupleIO.writeTuples( + out, + SharedFixture.specialDoublesPerColumn(rows, schema, keepFilled).iterator, + schema + ) + portId -> out + } + Seq((RunKind.SpecialDoubles, base, rewritten)) + } + } + + /** The schema of each input file, keyed by port index — what a curated handler + * actually wrote, read back off the sidecar its writer drops. A file without one + * contributes no schema, so a column knob resolved against that port simply finds + * nothing to fill and the variant is skipped rather than built on a guess. + */ + private def schemasOf(inputs: Map[PortIdentity, Path]): Map[Int, Schema] = + inputs.flatMap { + case (portId, path) => Try(TupleIO.readSchemaSidecar(path)).toOption.map(portId.id -> _) + } + + /** How many rows port 0 holds — the hint a numeric knob's fill is scaled against + * (a `limit` worth running is one that keeps some rows and drops some). + */ + private def rowCountOf(inputs: Map[PortIdentity, Path]): Int = + inputs + .get(PortIdentity(0)) + .flatMap(path => Try(Files.readAllLines(path).asScala.count(_.trim.nonEmpty)).toOption) + .filter(_ > 0) + .getOrElse(ConfigGenerator.DefaultRowCount) + + /** Run a variant both paths must refuse, and pass only when each raises. + * + * What each says is not compared: the engine's refusal is a Java or a Python + * exception and the script's is a Python traceback, so their wording differs + * even when the cause is the same. That one of them answered is the finding. + */ + private def runExpectingRefusal( + opClass: Class[_ <: LogicalOp], + opDesc: LogicalOp, + inputs: Map[PortIdentity, Path], + workDir: Path, + why: String + ): Unit = { + val actualDir = workDir.resolve("actual") + Files.createDirectories(actualDir) + // A copy for the same reason as in [[runVariant]]: Path A may rewrite it. + val opDescForPathA = + objectMapper + .readValue(objectMapper.writeValueAsString(opDesc), opClass) + .asInstanceOf[LogicalOp] + val pathAIsPython = classOf[PythonOperatorDescriptor].isAssignableFrom(opClass) + val pathA = Try { + if (pathAIsPython) + PyOpExecHarness.execute(opDescForPathA, inputs = inputs, outputDir = actualDir) + else OpExecHarness.execute(opDescForPathA, inputs = inputs, outputDir = actualDir) + } + val standaloneInputs: Map[Int, Path] = + inputs.toSeq + .sortBy(_._1.id) + .zipWithIndex + .map { case ((_, path), idx) => (idx + 1) -> path } + .toMap + val pathB = Try { + StandaloneRunner.run( + opDesc = opDesc, + inputs = standaloneInputs, + outputPortCount = opDesc.operatorInfo.outputPorts.size, + workDir = workDir, + exactIntegers = !pathAIsPython + ) + } + // A script that does not compile, or is never generated, has not refused + // the input: only a run that ends in an error counts. + val scriptRefused = pathB match { + case Failure(_: StandaloneExecutionException) => true + case _ => false + } + if (pathA.isSuccess || !scriptRefused) { + def said(t: Try[_]): String = + t match { + case Success(_) => "answered" + case Failure(e) => s"raised ${e.getClass.getSimpleName}: ${e.getMessage}" + } + throw new AssertionError( + s"both paths should refuse this run ($why), but Texera ${said(pathA)}, " + + s"and the script ${said(pathB)}" + ) + } + } + + /** Run one configured variant of `opDesc` through both paths against `inputs`, + * writing all intermediate/output files under `workDir`, and assert parity on + * every declared output port. + */ + private def runVariant( + opClass: Class[_ <: LogicalOp], + opDesc: LogicalOp, + inputs: Map[PortIdentity, Path], + workDir: Path + ): Unit = { + val outputPortCount = opDesc.operatorInfo.outputPorts.size + val actualDir = workDir.resolve("actual") + Files.createDirectories(actualDir) + + if (!opDesc.asInstanceOf[StandaloneCodeGenerator].producesDataFrame()) { + runVisualization(opClass, opDesc, inputs, outputPortCount, actualDir, workDir) + return + } + + // Path A's getPhysicalPlan/getPhysicalOp may mutate the OpDesc in place — + // AggregateOpDesc rewrites its `aggregations` to the final stage (COUNT→SUM) + // via getFinal. Run Path A on an isolated deep copy (same JSON round-trip the + // executor itself uses) so the shared instance stays pristine for Path B, + // whose generateStandaloneCode reads the original fields directly. + val opDescForPathA = + objectMapper + .readValue(objectMapper.writeValueAsString(opDesc), opClass) + .asInstanceOf[LogicalOp] + // A Python Path A builds its table with pandas, so a holed integer column is + // a float on that side too and the script must match it. A JVM Path A holds + // the tuple, and there the script has to be given the value back exactly. + val pathAIsPython = classOf[PythonOperatorDescriptor].isAssignableFrom(opClass) + val (pathAOutputs, pathAOutputSchemas): (Map[PortIdentity, Path], Map[PortIdentity, Schema]) = + if (pathAIsPython) { + val r = PyOpExecHarness.execute(opDescForPathA, inputs = inputs, outputDir = actualDir) + (r.outputs, r.outputSchemas) + } else { + val r = OpExecHarness.execute(opDescForPathA, inputs = inputs, outputDir = actualDir) + (r.outputs, r.outputSchemas) + } + + // StandaloneRunner keys inputs by 1-based port index (the inNdf convention). + val standaloneInputs: Map[Int, Path] = + inputs.toSeq + .sortBy(_._1.id) + .zipWithIndex + .map { + case ((_, path), idx) => (idx + 1) -> path + } + .toMap + + val pathB = StandaloneRunner.run( + opDesc = opDesc, + inputs = standaloneInputs, + outputPortCount = outputPortCount, + workDir = workDir, + exactIntegers = !pathAIsPython, + // What the run declared its output columns to be, so the script writes an + // integral column the way the engine's writer wrote it. + outputSchemas = pathAOutputSchemas + ) + + // The operator declares whether its output row order is meaningful via + // LogicalOp.orderSensitive (true only for the sort family); default unordered. + val orderSensitive = opDesc.orderSensitive + (0 until outputPortCount).foreach { port => + val actual = pathAOutputs.getOrElse( + PortIdentity(port), + throw new AssertionError(s"Texera path produced no output for port $port") + ) + val expected = pathB.outputs.getOrElse( + port + 1, + throw new AssertionError(s"standalone path produced no output for port $port") + ) + // A BINARY column holds a trained model: the two paths produce + // behaviorally-equivalent but not bit-identical models, so the comparator + // unpickles both and asserts their predictions on the training features + // (the probe) match — verifying behavior, not just completion. + val modelColumns: Seq[String] = pathAOutputSchemas + .get(PortIdentity(port)) + .map(_.getAttributes.filter(_.getType == AttributeType.BINARY).map(_.getName)) + .getOrElse(Seq.empty) + // The features are on the first port without a BINARY column. An operator + // that reads a model on port 0 has its table on port 1, and a probe of + // model cells has no feature to predict on. + val probePath: Option[Path] = + if (modelColumns.isEmpty) None + else { + val byPort = inputs.toSeq.sortBy(_._1.id).map(_._2) + byPort + .find(path => + !TupleIO + .readSchemaSidecar(path) + .getAttributes + .exists(_.getType == AttributeType.BINARY) + ) + .orElse(byPort.headOption) + } + Comparator.assertEqual( + actual, + expected, + orderSensitive = orderSensitive, + modelColumns = modelColumns, + probePath = probePath + ) + } + } + + private def runVisualization( + opClass: Class[_ <: LogicalOp], + opDesc: LogicalOp, + inputs: Map[PortIdentity, Path], + outputPortCount: Int, + actualDir: Path, + testRoot: Path + ): Unit = { + require( + visualizationJsonOps.contains(opClass) || visualizationHtmlOps.contains(opClass), + s"${opClass.getSimpleName} is not registered for visualization validation" + ) + require( + outputPortCount == 1, + "visualization JSON validation currently supports one output port" + ) + require( + classOf[PythonOperatorDescriptor].isAssignableFrom(opClass), + "visualization JSON validation currently supports Python visualization operators" + ) + + val actual = PyOpExecHarness + .execute(opDesc, inputs = inputs, outputDir = actualDir) + .outputs + .getOrElse( + PortIdentity(0), + throw new AssertionError("Texera path produced no visualization output for port 0") + ) + + val standaloneInputs: Map[Int, Path] = + inputs.toSeq + .sortBy(_._1.id) + .zipWithIndex + .map { + case ((_, path), idx) => (idx + 1) -> path + } + .toMap + + StandaloneRunner.run( + opDesc = opDesc, + inputs = standaloneInputs, + outputPortCount = outputPortCount, + workDir = testRoot + ) + + // A JSON-compared operator can still legitimately render its own error page + // instead of a figure (a non-numeric threshold, no non-null rows). There is + // then no Plotly payload to compare on either path, so compare what the user + // actually sees — the HTML. + if (visualizationJsonOps.contains(opClass) && hasPlotlyFigure(actual)) { + val expected = testRoot.resolve("output.json") + if (!Files.exists(expected)) { + throw new AssertionError(s"standalone visualization path did not produce $expected") + } + VisualizationJsonComparator.assertEqual(actual, expected) + } else { + val expected = testRoot.resolve("output.html") + if (!Files.exists(expected)) { + throw new AssertionError(s"standalone visualization path did not produce $expected") + } + VisualizationHtmlComparator.assertEqual(actual, expected) + } + } + + /** True if the runtime path's visualization output carries a Plotly figure — + * either a `json-content` payload or an `html-content` holding a + * `Plotly.newPlot(...)` call. False for an operator's own error page. + */ + private def hasPlotlyFigure(visualizationJsonl: Path): Boolean = { + val line = Files + .readAllLines(visualizationJsonl, StandardCharsets.UTF_8) + .asScala + .find(_.trim.nonEmpty) + .getOrElse(throw new AssertionError(s"$visualizationJsonl is empty")) + val node = objectMapper.readTree(line) + val json = node.get("json-content") + if (json != null && !json.isNull && json.asText().nonEmpty) true + else { + val html = node.get("html-content") + html != null && !html.isNull && html.asText().contains("Plotly.newPlot(") + } + } +} diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/TransformVerificationRunnerSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/TransformVerificationRunnerSpec.scala new file mode 100644 index 00000000000..6ccb5828c23 --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/TransformVerificationRunnerSpec.scala @@ -0,0 +1,160 @@ +/* + * 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 + +// This spec pins the tier-routing logic (disposition). Per-operator end-to-end +// runs are NOT duplicated here: OperatorBehaviorSpec auto-discovers every +// registered operator and runs TransformVerificationRunner.run on each, and a +// single operator can be run in isolation with e.g. +// sbt "WorkflowCompilingService/testOnly *OperatorBehaviorSpec -- -z LimitOpDesc" +// (the auto-generated test name starts with the operator's simple name). What +// disposition asserts — which tier an operator routes to — is the one thing +// OperatorBehaviorSpec does not check, so it lives here. + +import org.apache.texera.amber.operator.dummy.DummyOpDesc +import org.apache.texera.amber.operator.hashJoin.HashJoinOpDesc +import org.apache.texera.amber.operator.limit.LimitOpDesc +import org.apache.texera.amber.operator.udf.python.PythonUDFOpDescV2 +import org.apache.texera.amber.operator.union.UnionOpDesc +import org.apache.texera.amber.operator.machineLearning.Scorer.MachineLearningScorerOpDesc +import org.apache.texera.amber.operator.machineLearning.sklearnAdvanced.SVCTrainer.SklearnAdvancedSVCTrainerOpDesc +import org.apache.texera.amber.operator.sklearn.SklearnPredictionOpDesc +import org.apache.texera.amber.operator.sklearn.training.SklearnTrainingLogisticRegressionOpDesc +import org.apache.texera.amber.operator.visualization.DotPlot.DotPlotOpDesc +import org.apache.texera.amber.operator.visualization.IcicleChart.IcicleChartOpDesc +import org.apache.texera.amber.operator.visualization.ImageViz.ImageVisualizerOpDesc +import org.apache.texera.amber.operator.visualization.ScatterMatrixChart.ScatterMatrixChartOpDesc +import org.apache.texera.amber.operator.visualization.barChart.BarChartOpDesc +import org.apache.texera.amber.operator.visualization.boxViolinPlot.BoxViolinPlotOpDesc +import org.apache.texera.amber.operator.visualization.bubbleChart.BubbleChartOpDesc +import org.apache.texera.amber.operator.visualization.bulletChart.BulletChartOpDesc +import org.apache.texera.amber.operator.visualization.candlestickChart.CandlestickChartOpDesc +import org.apache.texera.amber.operator.visualization.carpetPlot.CarpetPlotOpDesc +import org.apache.texera.amber.operator.visualization.choroplethMap.ChoroplethMapOpDesc +import org.apache.texera.amber.operator.visualization.continuousErrorBands.ContinuousErrorBandsOpDesc +import org.apache.texera.amber.operator.visualization.contourPlot.ContourPlotOpDesc +import org.apache.texera.amber.operator.visualization.dendrogram.DendrogramOpDesc +import org.apache.texera.amber.operator.visualization.dumbbellPlot.DumbbellPlotOpDesc +import org.apache.texera.amber.operator.visualization.ecdfPlot.ECDFPlotOpDesc +import org.apache.texera.amber.operator.visualization.figureFactoryTable.FigureFactoryTableOpDesc +import org.apache.texera.amber.operator.visualization.filledAreaPlot.FilledAreaPlotOpDesc +import org.apache.texera.amber.operator.visualization.funnelPlot.FunnelPlotOpDesc +import org.apache.texera.amber.operator.visualization.ganttChart.GanttChartOpDesc +import org.apache.texera.amber.operator.visualization.gaugeChart.GaugeChartOpDesc +import org.apache.texera.amber.operator.visualization.networkGraph.NetworkGraphOpDesc +import org.apache.texera.amber.operator.visualization.wordCloud.WordCloudOpDesc +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +class TransformVerificationRunnerSpec extends AnyFlatSpec with Matchers { + import TransformVerificationRunner._ + + "disposition" should "flag knownIssues operators with the triage reason" in { + // A placeholder with no physical execution, so the harness has nothing to + // run either path against. + disposition(classOf[DummyOpDesc]) match { + case Flagged(reason) => reason should include("known issue") + case other => fail(s"expected Flagged, got $other") + } + } + + // It was withheld because a JSONL fixture could not carry a fitted model. Both + // paths now unpickle a model cell the way the worker does, so its fixture can. + it should "run the prediction op on a fitted model its fixture carries" in { + disposition(classOf[SklearnPredictionOpDesc]) shouldBe Runnable("curated") + } + + it should "run the union now that its code names every upstream" in { + // It used to be flagged for naming exactly two, which was wrong in both + // directions: a third link was dropped and a lone link left the second + // frame unbound. The runner draws one link per port, so what runs here is + // the one-upstream case — the one the old code got wrong. + disposition(classOf[UnionOpDesc]) shouldBe Runnable("auto") + } + + it should "route auto-configurable operators to the auto tier" in { + disposition(classOf[LimitOpDesc]) shouldBe Runnable("auto") + } + + // Both were withheld for drawing a different picture on every run, which + // stopped being true once their placement was seeded. + it should "run the two seeded visualizations" in { + disposition(classOf[WordCloudOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[NetworkGraphOpDesc]) shouldBe Runnable("visualization") + } + + it should "route the visualizations to the visualization tier" in { + disposition(classOf[BarChartOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[BoxViolinPlotOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[BubbleChartOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[BulletChartOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[CandlestickChartOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[CarpetPlotOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[ChoroplethMapOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[ContinuousErrorBandsOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[ContourPlotOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[DendrogramOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[DotPlotOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[DumbbellPlotOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[ECDFPlotOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[FigureFactoryTableOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[FilledAreaPlotOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[FunnelPlotOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[GanttChartOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[GaugeChartOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[IcicleChartOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[ImageVisualizerOpDesc]) shouldBe Runnable("visualization") + disposition(classOf[ScatterMatrixChartOpDesc]) shouldBe Runnable("visualization") + } + + it should "route genuine one-off curated ops to the curated tier" in { + disposition(classOf[HashJoinOpDesc[_]]) shouldBe Runnable("curated") + } + + it should "route the scorer to the auto tier now the table holds a label pair" in { + // What kept it curated was the canonical table, not the operator: scoring reads + // one label through two columns, and until a prediction column joined the label + // was no such pair for @SampleColumn to name. + disposition(classOf[MachineLearningScorerOpDesc]) shouldBe Runnable("auto") + } + + it should "route a sklearn estimator to the auto tier on the numeric projection" in { + disposition(classOf[SklearnTrainingLogisticRegressionOpDesc]) shouldBe + Runnable("auto, countVectorizer=false, tfidfTransformer=false") + fixtureFor(classOf[SklearnTrainingLogisticRegressionOpDesc]) shouldBe + CanonicalFixture.sklearnNumeric + } + + it should "route an advanced trainer to the auto tier on the same projection" in { + // Its `paraList` holds a row whose `parameter` is the operator's own enum, named + // only on the generic supertype. The generator resolves it, so the hand-written + // handler these four used to need is gone. + disposition(classOf[SklearnAdvancedSVCTrainerOpDesc]) shouldBe Runnable("auto") + fixtureFor(classOf[SklearnAdvancedSVCTrainerOpDesc]) shouldBe CanonicalFixture.sklearnNumeric + } + + // A UDF's body is written by whoever drops the operator, so there is nothing + // for a generator to emit. It stands here for the shape of the report: an + // operator that cannot be exported is carried as a row, not passed over. + it should "flag an operator that has no standalone generator" in { + disposition(classOf[PythonUDFOpDescV2]) shouldBe + Flagged("does not implement StandaloneCodeGenerator") + } +} diff --git a/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/TranslatedPlanRunSpec.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/TranslatedPlanRunSpec.scala new file mode 100644 index 00000000000..5f0a1846e6d --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/TranslatedPlanRunSpec.scala @@ -0,0 +1,180 @@ +/* + * 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.virtualidentity.WorkflowIdentity +import org.apache.texera.amber.core.workflow.{PortIdentity, WorkflowContext} +import org.apache.texera.amber.operator.LogicalOp +import org.apache.texera.amber.operator.distinct.DistinctOpDesc +import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc +import org.apache.texera.amber.operator.visualization.contourPlot.{ + ContourPlotColoringFunction, + ContourPlotOpDesc +} +import org.apache.texera.amber.operator.visualization.pieChart.PieChartOpDesc +import org.apache.texera.amber.translator.WorkflowToPythonTranslator +import org.apache.texera.amber.translator.verify.tags.IntegrationTest +import org.apache.texera.common.compiler.model.{LogicalLink, LogicalPlan, LogicalPlanPojo} +import org.apache.texera.common.compiler.{CompilationErrorHandling, WorkflowCompiler} +import org.scalatest.flatspec.AnyFlatSpec +import org.scalatest.matchers.should.Matchers + +import java.nio.charset.StandardCharsets +import java.nio.file.{Files, Path} +import scala.collection.mutable.ArrayBuffer +import scala.sys.process.{Process, ProcessLogger} + +/** The translated script itself, run as one program. + * + * Every other verify spec runs a single operator's fragment with the harness + * binding its placeholders, so a fault that appears only once fragments share + * one script and one directory has nowhere to show. The ones the engine can + * answer too run in amber's WorkflowParitySpec, against the engine. The two + * left here are the script's own: two charts writing one file, and a branch + * that ends the process the other branches still need. + * + * Tagged @IntegrationTest with the rest of the specs that fork Python. + */ +@IntegrationTest +class TranslatedPlanRunSpec extends AnyFlatSpec with Matchers { + + private val CsvName = "input.csv" + + // The same chain StandaloneRunner uses: the env var CI and the shared venv + // set, then the conventional name. + private val python = + sys.env.get("UDF_PYTHON_PATH").filter(_.nonEmpty).getOrElse("python3.12") + + private def csvSource(id: String, file: String = CsvName): CSVScanSourceOpDesc = { + val op = new CSVScanSourceOpDesc + // The generated read strips this to its basename, and run() below gives the + // script the temp directory as its cwd, so the plain name resolves. + op.fileName = Some(file) + op.setOperatorId(id) + op + } + + private def link(from: LogicalOp, to: LogicalOp): LogicalLink = + LogicalLink(from.operatorIdentifier, PortIdentity(0), to.operatorIdentifier, PortIdentity(0)) + + private def run(csv: String, ops: List[LogicalOp], links: List[LogicalLink]): (Path, String) = + run(Map(CsvName -> csv), ops, links) + + /** Translates the plan, writes it next to the CSVs it reads, and runs it. + * Returns the directory it ran in, which holds whatever files it wrote, and + * what it printed. + */ + private def run( + csvs: Map[String, String], + ops: List[LogicalOp], + links: List[LogicalLink] + ): (Path, String) = { + val dir = Files.createTempDirectory("translated-plan-") + csvs.foreach { + case (name, content) => + Files.write(dir.resolve(name), content.getBytes(StandardCharsets.UTF_8)) + } + + // The service compiles before translating, so a column's declared type + // reaches the operator that needs it. The compiler reads the source from + // here, so it takes the full path; the generated read strips it back. + ops.foreach { + case scan: CSVScanSourceOpDesc => scan.fileName = scan.fileName.map(dir.resolve(_).toString) + case _ => () + } + val schemas = new WorkflowCompiler(new WorkflowContext(workflowId = WorkflowIdentity(0))) + .compile( + LogicalPlanPojo(ops, links, opsToViewResult = List.empty, opsToReuseResult = List.empty), + CompilationErrorHandling.Lenient + ) + .operatorIdToOutputSchemas + val script = new WorkflowToPythonTranslator().translate(LogicalPlan(ops, links), schemas) + val scriptPath = dir.resolve("script.py") + Files.write(scriptPath, script.getBytes(StandardCharsets.UTF_8)) + + val out = ArrayBuffer.empty[String] + val err = ArrayBuffer.empty[String] + val exit = Process(Seq(python, scriptPath.toString), Some(dir.toFile)) + .!(ProcessLogger(line => out += line, line => err += line)) + val stdout = out.mkString("\n") + withClue( + s"script: $scriptPath\n--- stdout ---\n$stdout\n--- stderr ---\n${err.mkString("\n")}\n" + ) { + exit shouldBe 0 + } + (dir, stdout) + } + + // The files the plan wrote, script and inputs aside. + private def written(dir: Path): Seq[String] = + Option(dir.toFile.list()).toSeq.flatten + .filterNot(name => name.endsWith(".csv") || name == "script.py") + .sorted + + /** Two charts run in one directory, so a name either had chosen for itself + * would leave one file where the plan drew two pictures. + */ + it should "give two charts in one plan a file each" in { + val source = csvSource("source") + val charts = List("pie1", "pie2").map { id => + val op = new PieChartOpDesc + op.name = "label" + op.value = "amount" + op.setOperatorId(id) + op + } + val (dir, _) = run( + "label,amount\na,1\nb,2\n", + source :: charts, + charts.map(chart => link(source, chart)) + ) + written(dir) shouldBe Seq( + "pie_chart_1.html", + "pie_chart_1.json", + "pie_chart_2.html", + "pie_chart_2.json" + ) + } + + /** A chart with nothing to draw writes its reason and stops there. Ending the + * process would take the sibling branches with it, and the frames the plan + * prints at the end are the last thing to run. + */ + it should "let the rest of a plan finish when one chart cannot be drawn" in { + val source = csvSource("source") + val contour = new ContourPlotOpDesc + contour.setOperatorId("contour") + contour.x = "x" + contour.y = "y" + contour.z = "z" + contour.coloringMethod = ContourPlotColoringFunction.HEATMAP + val distinct = new DistinctOpDesc + distinct.setOperatorId("distinct") + // Collinear points: the contour has no area to interpolate over. + val (dir, stdout) = run( + "x,y,z\n0,0,1\n1,1,2\n2,2,3\n", + List(source, contour, distinct), + List(link(source, contour), link(source, distinct)) + ) + stdout should include("[Distinct]") + // The error page and nothing else: no chart was drawn, so no JSON figure. + written(dir) shouldBe Seq("contour_plot_1.html") + } +}