From 1134fd8e95620c0f6c2dc09c01ae35a6e372b054 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Thu, 13 Aug 2026 16:38:17 -0700 Subject: [PATCH] feat(workflow-operator): constrain the Sklearn text column, and drop what an estimator cannot fit Two configurations ended the run with an error from inside scikit-learn or from code generation, naming neither the column nor the field to change. The text column now states what it takes. Count Vectorizer tokenizes documents, so the column is a string, and it is required exactly when that switch is on: with the switch off nothing reads it, and with the switch on a blank one reached code generation as a null and became #EXCEPTION DURING CODE GENERATION. Both are schema constraints, the second in the conditional form Aggregate already uses, so the panel refuses the configuration while it is being written. Conditional rather than a plain required, so a freshly dropped operator, whose vectorizer is off, is not flagged for a field it has no use for. The feature set drops what it cannot fit rather than ending the run. These operators take every column but the target, so a text column beside the numbers, one the user never meant as a feature, raised ValueError: could not convert string to float, and a timestamp raised DTypePromotionError. Nothing in the configuration could exclude it and the message named neither the column nor a way out. Booleans are kept, fitting as 0/1. What was left out is printed, so the choice is visible rather than silent, and this follows the rest of the codebase, where twenty-four visualization operators drop missing values before plotting. The drop is skipped under the text pipeline, where X is one string column by construction and filtering would empty it. Co-Authored-By: Claude Opus 5 (1M context) --- .../sklearn/SklearnClassifierOpDesc.scala | 3 +- .../operator/sklearn/SklearnModelOpDesc.scala | 42 +++++++++++++++++++ .../training/SklearnTrainingOpDesc.scala | 3 +- .../SklearnClassifierOpDescCodegenSpec.scala | 12 ++++-- .../SklearnTrainingOpDescCodegenSpec.scala | 12 ++++-- 5 files changed, 62 insertions(+), 10 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDesc.scala index 92aec692a6e..c947a00c377 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDesc.scala @@ -41,7 +41,8 @@ abstract class SklearnClassifierOpDesc extends SklearnModelOpDesc { | def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]: | Y = table[$target] | X = table.drop($target, axis=1) - | X = ${if (countVectorizer) pyb"X[$text]" else "X"} + | ${if (countVectorizer) pyb"X = X[$text]" + else dropNonFeatureColumns("X", " " * 8)} | if port == 0: | self.model = make_pipeline(${if (countVectorizer) "CountVectorizer()," else ""} ${if (tfidfTransformer) "TfidfTransformer()," else ""} ${getImportStatements diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnModelOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnModelOpDesc.scala index 6665477aeac..a7b5cafb840 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnModelOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnModelOpDesc.scala @@ -36,6 +36,33 @@ import org.apache.texera.amber.operator.metadata.annotations.{ HideAnnotation } +// `text` is the column Count Vectorizer tokenizes, so it takes a string and is +// required only when that switch is on. Conditional rather than plain required, +// so a freshly dropped operator is not flagged for a field it has no use for. +@JsonSchemaInject(json = """ +{ + "attributeTypeRules": { + "text": { + "enum": ["string"] + } + }, + "allOf": [ + { + "if": { + "properties": { + "countVectorizer": { "const": true } + } + }, + "then": { + "required": ["text"], + "properties": { + "text": { "pattern": "\\S" } + } + } + } + ] +} +""") abstract class SklearnModelOpDesc extends PythonOperatorDescriptor { @JsonSchemaTitle("Target Attribute") @@ -79,6 +106,21 @@ abstract class SklearnModelOpDesc extends PythonOperatorDescriptor { ) var tfidfTransformer: Boolean = false + /** Python that narrows `frame` to the columns an estimator can fit. A column the + * user did not mean as a feature, a note beside the numbers, would otherwise end + * the run from inside scikit-learn. Booleans are kept: they fit as 0/1. What was + * dropped is printed, so the choice is visible rather than silent. + * + * `indent` is the leading whitespace of the statement this replaces. + */ + @JsonIgnore + protected def dropNonFeatureColumns(frame: String, indent: String): String = + s"""_fittable = $frame.select_dtypes(include=["number", "bool"]) + |${indent}_ignored = [c for c in $frame.columns if c not in _fittable.columns] + |${indent}if _ignored: + |${indent} print("Ignoring columns an estimator cannot fit:", _ignored) + |${indent}$frame = _fittable""".stripMargin + @JsonIgnore def getImportStatements: String diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDesc.scala index 3f354cc37f5..39aa9acd461 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDesc.scala @@ -41,7 +41,8 @@ class SklearnTrainingOpDesc extends SklearnModelOpDesc { | def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]: | Y = table[$target] | X = table.drop($target, axis=1) - | X = ${if (countVectorizer) pyb"X[$text]" else "X"} + | ${if (countVectorizer) pyb"X = X[$text]" + else dropNonFeatureColumns("X", " " * 8)} | model = make_pipeline(${if (countVectorizer) "CountVectorizer()," else ""} ${if ( tfidfTransformer ) "TfidfTransformer()," diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDescCodegenSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDescCodegenSpec.scala index 9e732002eb9..874cd88e1ff 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDescCodegenSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDescCodegenSpec.scala @@ -68,8 +68,11 @@ class SklearnClassifierOpDescCodegenSpec extends AnyFlatSpec with Matchers { code should include("from sklearn.neighbors import KNeighborsClassifier") code should include(s"Y = table[${decodeExpr("label")}]") code should include(s"X = table.drop(${decodeExpr("label")}, axis=1)") - // Feature-column path: X is kept whole, the text attribute is never read. - code should include("X = X\n") + // Feature-column path: every column an estimator can fit is kept, the rest are + // named on the console, and the text attribute is never read. + code should include("""_fittable = X.select_dtypes(include=["number", "bool"])""") + code should include("""print("Ignoring columns an estimator cannot fit:", _ignored)""") + code should include("X = _fittable") code should not include decodeExpr("docs") normalized(code) should include( "self.model = make_pipeline( KNeighborsClassifier()).fit(X, Y)" @@ -82,7 +85,8 @@ class SklearnClassifierOpDescCodegenSpec extends AnyFlatSpec with Matchers { it should "select the text column and prepend CountVectorizer when countVectorizer is on" in { val code = descriptor(countVectorizer = true).generatePythonCode() code should include(s"X = X[${decodeExpr("docs")}]") - code should not include "X = X\n" + // X is one string column here, so narrowing to fittable columns would empty it. + code should not include "_fittable" normalized(code) should include( "self.model = make_pipeline(CountVectorizer(), KNeighborsClassifier()).fit(X, Y)" ) @@ -101,7 +105,7 @@ class SklearnClassifierOpDescCodegenSpec extends AnyFlatSpec with Matchers { it should "prepend only TfidfTransformer and keep all features when tfidfTransformer is on alone" in { val code = descriptor(tfidfTransformer = true).generatePythonCode() // Without countVectorizer there is no text-column selection. - code should include("X = X\n") + code should include("X = _fittable") code should not include decodeExpr("docs") normalized(code) should include( "self.model = make_pipeline( TfidfTransformer(), KNeighborsClassifier()).fit(X, Y)" diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDescCodegenSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDescCodegenSpec.scala index 0c00ca271b5..71c1f492b08 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDescCodegenSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDescCodegenSpec.scala @@ -68,8 +68,11 @@ class SklearnTrainingOpDescCodegenSpec extends AnyFlatSpec with Matchers { code should include("from sklearn.neighbors import KNeighborsClassifier") code should include(s"Y = table[${decodeExpr("label")}]") code should include(s"X = table.drop(${decodeExpr("label")}, axis=1)") - // Feature-column path: X is kept whole, the text attribute is never read. - code should include("X = X\n") + // Feature-column path: every column an estimator can fit is kept, the rest are + // named on the console, and the text attribute is never read. + code should include("""_fittable = X.select_dtypes(include=["number", "bool"])""") + code should include("""print("Ignoring columns an estimator cannot fit:", _ignored)""") + code should include("X = _fittable") code should not include decodeExpr("docs") normalized(code) should include("make_pipeline( KNeighborsClassifier()).fit(X, Y)") code should not include "CountVectorizer()" @@ -79,7 +82,8 @@ class SklearnTrainingOpDescCodegenSpec extends AnyFlatSpec with Matchers { it should "select the text column and prepend CountVectorizer when countVectorizer is on" in { val code = descriptor(countVectorizer = true).generatePythonCode() code should include(s"X = X[${decodeExpr("docs")}]") - code should not include "X = X\n" + // X is one string column here, so narrowing to fittable columns would empty it. + code should not include "_fittable" normalized(code) should include( "make_pipeline(CountVectorizer(), KNeighborsClassifier()).fit(X, Y)" ) @@ -98,7 +102,7 @@ class SklearnTrainingOpDescCodegenSpec extends AnyFlatSpec with Matchers { it should "prepend only TfidfTransformer and keep all features when tfidfTransformer is on alone" in { val code = descriptor(tfidfTransformer = true).generatePythonCode() // Without countVectorizer there is no text-column selection. - code should include("X = X\n") + code should include("X = _fittable") code should not include decodeExpr("docs") normalized(code) should include( "make_pipeline( TfidfTransformer(), KNeighborsClassifier()).fit(X, Y)"