carloea2 commented on code in PR #8363: URL: https://github.com/apache/texera/pull/8363#discussion_r4357201398
########## workflow-compiling-service/src/test/scala/org/apache/texera/amber/translator/verify/CuratedHandlers.scala: ########## @@ -0,0 +1,1225 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.texera.amber.translator.verify + +import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema, Tuple} +import org.apache.texera.amber.core.workflow.PortIdentity +import org.apache.texera.amber.operator.LogicalOp +import org.apache.texera.amber.operator.distinct.DistinctOpDesc +import org.apache.texera.amber.operator.aggregate.{ + AggregateOpDesc, + AggregationFunction, + AggregationOperation +} +import org.apache.texera.amber.operator.filter.{ + ComparisonType, + FilterPredicate, + SpecializedFilterOpDesc +} +import org.apache.texera.amber.operator.hashJoin.{HashJoinOpDesc, JoinType} +import org.apache.texera.amber.operator.keywordSearch.KeywordSearchOpDesc +import org.apache.texera.amber.operator.projection.{AttributeUnit, ProjectionOpDesc} +import org.apache.texera.amber.operator.regex.RegexOpDesc +import org.apache.texera.amber.operator.typecasting.{TypeCastingOpDesc, TypeCastingUnit} +import org.apache.texera.amber.operator.visualization.ImageViz.ImageVisualizerOpDesc + +import org.apache.texera.amber.operator.visualization.dumbbellPlot.{ + DumbbellDotConfig, + DumbbellPlotOpDesc +} +import org.apache.texera.amber.operator.sklearn.training.SklearnTrainingOpDesc +import org.apache.texera.amber.operator.sklearn.SklearnClassifierOpDesc +import org.apache.texera.amber.operator.sklearn.SklearnLinearRegressionOpDesc +import org.apache.texera.amber.operator.sklearn.SklearnPredictionOpDesc +import org.apache.texera.amber.operator.sklearn.testing.SklearnTestingOpDesc +import org.apache.texera.amber.operator.machineLearning.sklearnAdvanced.base.SklearnMLOperatorDescriptor +import org.apache.texera.amber.operator.ifStatement.IfOpDesc +import org.apache.texera.amber.operator.source.scan.file.FileScanOpDesc +import org.apache.texera.amber.util.JSONUtils.objectMapper +import java.nio.file.{Files, Path, Paths} +import java.util + +/** + * A curated handler ships a configured OpDesc and the input fixtures it + * needs, written once into `testRoot`. Register it in [[CuratedHandlers.all]] + * to override the auto-config tier for that operator. + */ +trait TransformHandler { + def opDescClass: Class[_ <: LogicalOp] + def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) + + /** Extra independent scenarios beyond [[fixture]], each a self-contained + * (label, configured op, its own inputs). The runner runs each as a PINNED + * config (no enum sweep), in its own work subdir. Default: none. + * + * Used where one operator needs structurally different inputs per config + * branch that a single swept fixture can't cover — e.g. the sklearn + * `countVectorizer=true` text path, whose feature column must be text and so + * is incompatible with the numeric default fixture (`X = table.drop(target)` + * would feed a string column to a numeric estimator). Each scenario must + * write its input files somewhere unique (e.g. a `testRoot` subdir) so it + * does not clobber the primary fixture's files. + */ + def extraScenarios(testRoot: Path): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = + Seq.empty + + /** Opts this fixture into the `nulls` case, naming the columns it must never + * empty because their VALUE is what the fixture was built to arrange rather + * than data under test: a join key that has to pair, a grouping key that has + * to group. Emptying one of those changes what the test asks instead of asking + * what the operator does with a null. + * + * The default is `Some(Set.empty)`: most curated tables arrange nothing that a + * hole would disturb, so taking part is the normal case and a fixture that + * cannot afford a hole says so. `None` sits the case out entirely, for a table + * whose every column is load-bearing. + */ + def nullsKeepFilled: Option[Set[String]] = Some(Set.empty) +} + +/** + * The curated override tier of the config/fixture resolution chain: an + * operator listed here is verified with its hand-written fixture instead of + * the auto-generated one. + */ +object CuratedHandlers { + + /** Concrete `LogicalOp` classes discovered from the `@JsonSubTypes` registry + * on [[LogicalOp]] — the same source [[ConfigGenerator]] enumerates. The + * sklearn handler families below are auto-derived from this list, so a newly + * registered sklearn estimator is picked up with zero per-operator + * boilerplate here. + */ + private val registeredOps: Seq[Class[_ <: LogicalOp]] = + Option(classOf[LogicalOp].getAnnotation(classOf[com.fasterxml.jackson.annotation.JsonSubTypes])) + .map(_.value().toSeq.map(_.value().asInstanceOf[Class[_ <: LogicalOp]])) + .getOrElse(Seq.empty) + + private def isConcrete(cls: Class[_]): Boolean = + !java.lang.reflect.Modifier.isAbstract(cls.getModifiers) + + /** The concrete leaf ops under one sklearn base, excluding the base itself. + * + * No hard-coded baseline: a new sklearn operator is picked up automatically + * the moment it is registered in LogicalOp's @JsonSubTypes — zero per-op code + * here. The test suite (ConfigCoverageSpec / TransformVerificationRunnerSpec) + * is the safety net: a mis-discovered or misbehaving op fails its own parity + * check rather than being frozen by an assertion. + */ + private def sklearnFamily(base: Class[_]): Seq[Class[_ <: LogicalOp]] = + registeredOps.filter(c => base.isAssignableFrom(c) && c != base && isConcrete(c)) + + private def trainingOps = sklearnFamily(classOf[SklearnTrainingOpDesc]) + private def classifierOps = sklearnFamily(classOf[SklearnClassifierOpDesc]) + private def advancedOps = sklearnFamily(classOf[SklearnMLOperatorDescriptor[_]]) + + /** Every sklearn op, whichever tier serves it. `X = table.drop(target)` feeds + * each remaining column to `fit`, so these take canonical's petal-and-label + * projection rather than the whole table, whose string columns end the fit. + * + * Linear Regression is named on its own because it descends from + * `PythonOperatorDescriptor` directly rather than from one of the three + * bases, so no family picks it up. + */ + val sklearnNumericClasses: Set[Class[_ <: LogicalOp]] = + (trainingOps ++ classifierOps ++ advancedOps).toSet + classOf[SklearnLinearRegressionOpDesc] + + val all: Seq[TransformHandler] = Seq( + AggregateTransformHandler, + SpecializedFilterTransformHandler, + DistinctTransformHandler, + ProjectionTransformHandler, + HashJoinTransformHandler, + TypeCastingTransformHandler, + KeywordSearchTransformHandler, + DumbbellPlotVisualizationHandler, + ImageVisualizerVisualizationHandler, + IfTransformHandler, + RegexTransformHandler, + SklearnPredictionTransformHandler, + SklearnTestingTransformHandler, + FileScanTransformHandler + ) + + val byClass: Map[Class[_ <: LogicalOp], TransformHandler] = + all.map(h => h.opDescClass -> h).toMap + + /** Generic fixture writer: builds a JSONL file with the given typed columns + * and rows, boxing each value per its declared [[AttributeType]]. Lets a + * curated handler declare bespoke per-operator input data in one call + * instead of hand-rolling a Schema + Tuple.builder loop. + */ + def writeFixture( + path: Path, + columns: Seq[(String, AttributeType)], + rows: Seq[Seq[Any]] + ): Path = { + val schema = new Schema(columns.map { case (n, t) => new Attribute(n, t) }: _*) + val tuples = rows.map { row => + val builder = Tuple.builder(schema) + columns.zip(row).foreach { + case ((name, attrType), value) => + val boxed: AnyRef = (attrType, value) match { + case (_, null) => null + case (AttributeType.INTEGER, x: Int) => Int.box(x) + case (AttributeType.INTEGER, x: Long) => Int.box(x.toInt) + case (AttributeType.INTEGER, x: Double) => Int.box(x.toInt) + case (AttributeType.LONG, x: Long) => Long.box(x) + case (AttributeType.LONG, x: Int) => Long.box(x.toLong) + case (AttributeType.DOUBLE, x: Double) => Double.box(x) + case (AttributeType.DOUBLE, x: Int) => Double.box(x.toDouble) + case (AttributeType.DOUBLE, x: Long) => Double.box(x.toDouble) + case (AttributeType.BOOLEAN, x: Boolean) => Boolean.box(x) + case (AttributeType.TIMESTAMP, x: java.sql.Timestamp) => x + case (AttributeType.STRING, x) => x.toString + case (_, x) => x.toString + } + builder.add(schema.getAttribute(name), boxed) + } + builder.build() + } + TupleIO.writeTuples(path, tuples.iterator, schema) + path + } + +} + +/** Curated INPUT for [[FileScanOpDesc]], which reads the files its input names + * rather than a table. Its one row names the shared table's own file, so the + * scan reads the data every other operator reads, and no file is written for it. + */ +object FileScanTransformHandler extends TransformHandler { + override val opDescClass: Class[_ <: LogicalOp] = classOf[FileScanOpDesc] + + override def fixture(testRoot: Path): (LogicalOp, Map[PortIdentity, Path]) = { + val shared = Paths.get(getClass.getResource(CanonicalFixture.fixtureResource).toURI) + val input = CuratedHandlers.writeFixture( + testRoot.resolve("input_port_0.jsonl"), + Seq("filename" -> AttributeType.STRING), + Seq(Seq(shared.toString)) + ) + (new FileScanOpDesc(), Map(PortIdentity(0) -> input)) + } + + /** File Scan Source's files, each read the way that scenario reads it, from the + * path this input names: the two operators share the engine's reader and differ + * only in where the path comes from. The shared table's JSON lines are text, so + * these are what exercise a typed line, an encoding the bytes are in, and an + * archive. + */ + override def extraScenarios(testRoot: Path): Seq[(String, LogicalOp, Map[PortIdentity, Path])] = + FileScanSourceHandler.files.map { Review Comment: This now needs FileScanSourceHandler and the public fixtureResource from #8364. But #8364's OperatorBehaviorSpec needs #8362, and #8362 needs CuratedHandlers from this PR, so there is no merge order that compiles. The local build confirms both missing references. Please move the shared file fixtures into an earlier prerequisite or split these calls out so the cycle is broken. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
