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..5e7736af46e 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,10 @@ "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" }, { "id": 3, @@ -71,7 +74,10 @@ "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" }, { "id": 11, @@ -108,7 +114,10 @@ "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" }, { "id": 1, @@ -145,7 +154,10 @@ "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" }, { "id": 14, @@ -182,7 +194,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": 7.0, + "edge_ts": "2024-02-14 12:00:00.0", + "digits": "11" }, { "id": 5, @@ -219,7 +234,10 @@ "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" }, { "id": 9, @@ -256,7 +274,10 @@ "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" }, { "id": 2, @@ -293,7 +314,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": 1.0E-4, + "edge_ts": "1600-06-15 08:30:00.0", + "digits": "042" }, { "id": 13, @@ -330,7 +354,10 @@ "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" }, { "id": 6, @@ -367,7 +394,10 @@ "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" }, { "id": 10, @@ -404,7 +434,10 @@ "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" }, { "id": 4, @@ -441,7 +474,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": 12345678.9, + "edge_ts": "9999-12-31 23:59:59.0", + "digits": "0100" }, { "id": 15, @@ -478,7 +514,10 @@ "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" }, { "id": 8, @@ -515,7 +554,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": 1.5, + "edge_ts": "2024-02-08 12:00:00.0", + "digits": "0" }, { "id": 12, @@ -552,6 +594,9 @@ "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" } ] 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..3309fe338d7 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,17 @@ 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 ) // ── 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/TransformVerificationRunner.scala b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/TransformVerificationRunner.scala new file mode 100644 index 00000000000..832e815e462 --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/TransformVerificationRunner.scala @@ -0,0 +1,1093 @@ +/* + * 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.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 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(", ", ", ", "") + } + + /** Why one kind of run is left out for an operator. The distinction is whether + * anyone should be waiting for it: [[PendingFix]] is a debt someone closes, + * [[ByDesign]] is an answer that will not change. + */ + sealed trait NotRunReason + final case class PendingFix(issue: String) extends NotRunReason + final case class ByDesign(why: String) extends NotRunReason + + final case class NotRun(op: Class[_], kind: String, reason: NotRunReason) + + /** The runs an operator does not get, 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. + * + * `op` matches its subclasses, so one row covers a family. `kind` is a + * [[RunKind]]. + * + * 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[NotRun] = { + // 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 = ByDesign("the operator refuses Count Vectorizer at compile time") + val denseOnly = for { + op <- Seq( + classOf[SklearnGaussianNaiveBayesOpDesc], + classOf[SklearnTrainingGaussianNaiveBayesOpDesc] + ) + label <- Seq(RunKind.CountVectorizerText, RunKind.TfidfText) + } yield NotRun(op, label, dense) + + // 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"), and where an estimator does fit on nothing, + // the row a comparison would predict on to tell two models apart is missing + // too. Withheld across both estimator hierarchies rather than one row each, + // since the reason is the estimator's, not the operator's. + val nothingToFit = ByDesign( + "an estimator has no rows to fit on, and sklearn refuses rather than " + + "returning a model; where one does fit on nothing, no row is left to " + + "predict on, so two models cannot be told apart either" + ) + val noRowsToFit = Seq( + NotRun(classOf[SklearnModelOpDesc], RunKind.EmptyTable, nothingToFit), + // 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 withhold SklearnTestingOpDesc, which passes. + NotRun(classOf[SklearnLinearRegressionOpDesc], RunKind.EmptyTable, nothingToFit), + NotRun(classOf[SklearnMLOperatorDescriptor[_]], RunKind.EmptyTable, nothingToFit), + NotRun( + classOf[MachineLearningScorerOpDesc], + RunKind.EmptyTable, + ByDesign( + "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( + // An enum whose legal values depend on a sibling field: flipping it alone + // builds a config the curated fixture already covers properly. + NotRun( + classOf[TypeCastingOpDesc], + RunKind.EnumSweep, + ByDesign( + "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" + ) + ), + NotRun( + classOf[SklearnTestingOpDesc], + RunKind.EnumSweep, + ByDesign( + "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" + ) + ), + NotRun( + classOf[AggregateOpDesc], + RunKind.EnumSweep, + ByDesign( + "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" + ) + ), + // 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. + NotRun( + classOf[SpecializedFilterOpDesc], + RunKind.HostileText, + ByDesign( + "`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" + ) + ), + NotRun( + classOf[SklearnMLOperatorDescriptor[_]], + RunKind.HostileText, + ByDesign( + "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" + ) + ) + ) ++ denseOnly ++ noRowsToFit + } + + /** Every kind of run withheld from this operator, with why. One entry per kind: + * a family row and an operator row for the same kind are the same statement + * twice, and the first one wins. + */ + private def withheldRunsFor(opClass: Class[_ <: LogicalOp]): Seq[(String, NotRunReason)] = + variantsNotRun + .collect { case NotRun(op, kind, reason) if op.isAssignableFrom(opClass) => kind -> reason } + .distinctBy(_._1) + + private def notRun(opClass: Class[_ <: LogicalOp], kind: String): Boolean = + withheldRunsFor(opClass).exists(_._1 == 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, _) + ) ++ + 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) ++ + 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 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)) + } + + /** 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 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..671fccd0dfc --- /dev/null +++ b/workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/TranslatedPlanRunSpec.scala @@ -0,0 +1,372 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.translator.verify + +import org.apache.texera.amber.core.tuple.AttributeType +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.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.split.SplitOpDesc +import org.apache.texera.amber.operator.typecasting.{TypeCastingOpDesc, TypeCastingUnit} +import org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpDesc +import org.apache.texera.amber.operator.union.UnionOpDesc +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. Four of them are pinned + * here: a column named after a placeholder, two charts writing one file, a + * branch that ends the process the other branches still need, and a branch + * handed the wrong one of its upstream's two output ports. + * + * 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 link(from: LogicalOp, fromPort: Int, to: LogicalOp): LogicalLink = + LogicalLink( + from.operatorIdentifier, + PortIdentity(fromPort), + 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 frame the plan printed under one operator's heading, up to the next + * heading, so a plan that prints two of them can be read a branch at a time. + */ + private def printed(stdout: String, operator: String): String = { + val heading = s"[$operator]" + val start = stdout.indexOf(heading) + withClue(s"no [$operator] section in:\n$stdout") { start should be >= 0 } + val section = stdout.substring(start + heading.length) + val next = section.indexOf("\n[") + if (next < 0) section else section.take(next) + } + + // 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 + + /** 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. + */ + it should "carry a frame through a chain of operators" in { + val source = csvSource("source") + val filter = new SpecializedFilterOpDesc + filter.setOperatorId("filter") + filter.predicates = List(new FilterPredicate("amount", ComparisonType.GREATER_THAN, "1")) + val sort = new SortOpDesc + sort.setOperatorId("sort") + val criterion = new SortCriteriaUnit + criterion.attributeName = "amount" + criterion.sortPreference = SortPreference.DESC + sort.attributes = List(criterion) + val limit = new LimitOpDesc + limit.setOperatorId("limit") + limit.limit = 2 + val projection = new ProjectionOpDesc + projection.setOperatorId("projection") + projection.attributes ++= List(new AttributeUnit("label", "label")) + + val chain = List(source, filter, sort, limit, projection) + val (_, stdout) = run( + "label,amount\nant,3\nbee,1\ncat,5\ndog,2\n", + chain, + chain.sliding(2).map { case Seq(from, to) => link(from, to) }.toList + ) + // bee is filtered out, dog falls outside the limit, and cat sorts above ant. + stdout should not include "bee" + stdout should not include "dog" + stdout.indexOf("cat") should be < stdout.indexOf("ant") + } + + /** A CSV carries no types, so a blank cell widens an integer column and its 6 + * would be matched as "6.0". Only a whole plan shows it: the type belongs to + * the source, a hop away with a file in between. + */ + it should "match an integer column on the type its source declared" in { + val source = csvSource("source") + val regex = new RegexOpDesc + regex.setOperatorId("regex") + regex.attribute = "n" + regex.regex = "^6$" + val (_, stdout) = run( + "label,n\nant,6\nbee,\ncat,70\n", + List(source, regex), + List(link(source, regex)) + ) + stdout should include("ant") + stdout should not include "cat" + } + + /** 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. + */ + it should "render a cast timestamp the way the engine's toString does" in { + val source = csvSource("source") + val cast = new TypeCastingOpDesc + cast.setOperatorId("cast") + val unit = new TypeCastingUnit + unit.attribute = "ts" + unit.resultType = AttributeType.TIMESTAMP + cast.typeCastingUnits = List(unit) + val regex = new RegexOpDesc + regex.setOperatorId("regex") + regex.attribute = "ts" + regex.regex = "\\.0$" + + val (_, stdout) = run( + "label,ts\nant,2024-01-07 00:00:00\nbee,2024-03-15 08:30:00.123\n", + List(source, cast, regex), + List(link(source, cast), link(cast, regex)) + ) + // ant's whole second grows the ".0" the engine writes; bee's fraction keeps + // three digits rather than the six Python pads to, so it does not match. + stdout should include("ant") + stdout should not include "bee" + } + + /** 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. + */ + it should "hand a variadic port the frames its upstreams produced" in { + val left = csvSource("left", "left.csv") + val right = csvSource("right", "right.csv") + val union = new UnionOpDesc + union.setOperatorId("union") + val (_, stdout) = run( + // A row only one of them reads, so a list that names one upstream twice + // and drops the other loses a row rather than repeating one. + Map("left.csv" -> "label,amount\nant,3\n", "right.csv" -> "label,amount\nbee,1\n"), + List(left, right, union), + List(link(left, union), link(right, union)) + ) + // Read the union's own frame: an upstream the list dropped is a leaf of the + // plan, and printing that leaf would put its rows in stdout anyway. + val united = printed(stdout, "Union") + united should include("ant") + united should include("bee") + } + + /** 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 prints on its own, so a branch reading the other port's frame + * is the rows it shows. + */ + it should "give each branch of a split the port its link named" in { + val source = csvSource("source") + val split = new SplitOpDesc + split.setOperatorId("split") + split.k = 50 + split.random = false + split.seed = 1 + val distinct = new DistinctOpDesc + distinct.setOperatorId("distinct") + val limit = new LimitOpDesc + limit.setOperatorId("limit") + limit.limit = 10 + + val (_, stdout) = run( + "label,amount\nant,1\nbee,2\ncat,3\ndog,4\neel,5\nfox,6\n", + List(source, split, distinct, limit), + List(link(source, split), link(split, 0, distinct), link(split, 1, limit)) + ) + // Seeded at 1, the draw that decides each row is 85, 88, 47, 13, 54, 4, so + // the three rows under 50 leave by port 0 and the other three by port 1. + val upper = printed(stdout, "Distinct") + val lower = printed(stdout, "Limit") + Seq("cat", "dog", "fox").foreach { row => + upper should include(row) + lower should not include row + } + Seq("ant", "bee", "eel").foreach { row => + lower should include(row) + upper should not include row + } + } + + /** 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 "keep a column whose name matches a placeholder" in { + val source = csvSource("source") + val projection = new ProjectionOpDesc + projection.setOperatorId("projection") + projection.attributes ++= List(new AttributeUnit("in1df", "kept")) + val (_, stdout) = run( + "in1df,other\n1,a\n2,b\n", + List(source, projection), + List(link(source, projection)) + ) + stdout should include("kept") + } + + /** 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") + } +}