feat(workflow-operator): constrain the Sklearn text column, and drop what an estimator cannot fit

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

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
 

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(),"

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)"

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)"