feat(sklearn): skip rows with missing values instead of ending the run

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDesc.scala

@@ -117,8 +117,9 @@ abstract class SklearnMLOperatorDescriptor[T <: ParamClass] extends PythonOperat
          |      self.dataset = table
          |
          |    if port == 1 :
-         |      y_train = self.dataset[$groundTruthAttribute]
-         |      X_train = self.dataset[features]
+         |      dataset = self.dataset.dropna(subset=features + [$groundTruthAttribute]) #remove missing values
+         |      y_train = dataset[$groundTruthAttribute]
+         |      X_train = dataset[features]
          |      loop_times = ${getLoopTimes(paraList)}
          |
          |      for i in range(loop_times):

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnClassifierOpDesc.scala

@@ -39,6 +39,8 @@ abstract class SklearnClassifierOpDesc extends SklearnModelOpDesc {
        |class ProcessTableOperator(UDFTableOperator):
        |    @overrides
        |    def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]:
+       |        table = ${if (countVectorizer) pyb"table.dropna(subset=[$text, $target])"
+    else "table.dropna()"} #remove missing values
        |        Y = table[$target]
        |        X = table.drop($target, axis=1)
        |        X = ${if (countVectorizer) pyb"X[$text]" else "X"}

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDesc.scala

@@ -62,6 +62,9 @@ class SklearnPredictionOpDesc extends PythonOperatorDescriptor {
        |            input_features = tuple_
        |            if $groundTruthAttribute != "":
        |                input_features = input_features.get_partial_tuple([col for col in tuple_.get_field_names() if col != $groundTruthAttribute])
+       |            if Table.from_tuple_likes([tuple_]).isna().any(axis=None):
+       |                tuple_[$resultAttribute] = None #keep the row, leave the result empty
+       |            elif $groundTruthAttribute != "":
        |                tuple_[$resultAttribute] = type(tuple_[$groundTruthAttribute])(self.model.predict(Table.from_tuple_likes([input_features]))[0])
        |            else:
        |                tuple_[$resultAttribute] = str(self.model.predict(Table.from_tuple_likes([input_features]))[0])

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDesc.scala

@@ -66,7 +66,7 @@ class SklearnTestingOpDesc extends PythonOperatorDescriptor {
          |            self.data.append(tuple_)
          |        else:
          |            model = tuple_[$model]
-         |            table = Table(self.data)
+         |            table = Table(self.data).dropna() #remove missing values
          |            Y = table[$target]
          |            X = table.drop($target, axis=1)
          |            predictions = model.predict(X.squeeze())

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingOpDesc.scala

@@ -39,6 +39,8 @@ class SklearnTrainingOpDesc extends SklearnModelOpDesc {
        |class ProcessTableOperator(UDFTableOperator):
        |    @overrides
        |    def process_table(self, table: Table, port: int) -> Iterator[Optional[TableLike]]:
+       |        table = ${if (countVectorizer) pyb"table.dropna(subset=[$text, $target])"
+    else "table.dropna()"} #remove missing values
        |        Y = table[$target]
        |        X = table.drop($target, axis=1)
        |        X = ${if (countVectorizer) pyb"X[$text]" else "X"}

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDescSpec.scala

@@ -103,6 +103,13 @@ class SklearnAdvancedBaseDescSpec extends AnyFlatSpec with Matchers {
     code should include("yield df")
   }
 
+  // This family reads a named list of features rather than every column, so the
+  // drop names those columns: a blank anywhere else must not cost the row.
+  it should "drop rows missing a selected feature or the ground truth" in {
+    val d = newOp(List(hyperParam("n_neighbors", "int", fromWorkflow = false, value = "5")))
+    d.generatePythonCode() should include("self.dataset.dropna(subset=features + [")
+  }
+
   it should "loop once when no parameter is sourced from the workflow" in {
     val d = newOp(List(hyperParam("n_neighbors", "int", fromWorkflow = false, value = "5")))
     val code = d.generatePythonCode()

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnBernoulliNaiveBayesOpDescSpec.scala

@@ -65,6 +65,14 @@ class SklearnBernoulliNaiveBayesOpDescSpec extends AnyFlatSpec with Matchers {
     code should include("Bernoulli Naive Bayes")
   }
 
+  // The same table statement serves both ports, so training and scoring skip a
+  // row with a missing value alike.
+  it should "drop rows with missing values before fitting and before scoring" in {
+    val d = new SklearnBernoulliNaiveBayesOpDesc
+    d.target = "y"
+    d.generatePythonCode() should include("table.dropna()")
+  }
+
   "SklearnBernoulliNaiveBayesOpDesc" should
     "round-trip its config fields through the polymorphic base" in {
     val d = new SklearnBernoulliNaiveBayesOpDesc

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDescSpec.scala

@@ -92,6 +92,17 @@ class SklearnPredictionOpDescSpec extends AnyFlatSpec with Matchers {
     code should include("yield tuple_")
   }
 
+  // This operator adds a column to the user's rows, so a row it cannot predict
+  // on keeps its place with an empty result rather than disappearing.
+  it should "keep a row with a missing value and leave its result empty" in {
+    val d = new SklearnPredictionOpDesc
+    d.model = "model"
+    d.resultAttribute = "prediction"
+    val code = d.generatePythonCode()
+    code should include("isna().any(axis=None)")
+    code should include("] = None")
+  }
+
   "SklearnPredictionOpDesc" should
     "round-trip its config fields through the polymorphic base" in {
     val d = new SklearnPredictionOpDesc

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDescSpec.scala

@@ -80,6 +80,15 @@ class SklearnTestingOpDescSpec extends AnyFlatSpec with Matchers {
     code should include(".predict(")
   }
 
+  // The scores are computed over the rows the model can be applied to, the way
+  // COUNT and MIN are computed over the rows that have a value.
+  it should "drop rows with missing values before scoring" in {
+    val d = new SklearnTestingOpDesc
+    d.model = "model"
+    d.target = "y"
+    d.generatePythonCode() should include("Table(self.data).dropna()")
+  }
+
   "SklearnTestingOpDesc" should
     "round-trip its config fields through the polymorphic base" in {
     val d = new SklearnTestingOpDesc

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingBernoulliNaiveBayesOpDescSpec.scala

@@ -64,6 +64,24 @@ class SklearnTrainingBernoulliNaiveBayesOpDescSpec extends AnyFlatSpec with Matc
     code should include("Training: Bernoulli Naive Bayes")
   }
 
+  // Every column but the target is a feature here, so a row missing any value is
+  // one the estimator cannot be fitted on.
+  it should "drop rows with missing values before fitting" in {
+    val d = new SklearnTrainingBernoulliNaiveBayesOpDesc
+    d.target = "y"
+    d.generatePythonCode() should include("table.dropna()")
+  }
+
+  // With Count Vectorizer on, only the text and target columns are read, so a
+  // blank in any other column must not cost the row.
+  it should "drop on the text and target columns only when vectorizing text" in {
+    val d = new SklearnTrainingBernoulliNaiveBayesOpDesc
+    d.target = "y"
+    d.countVectorizer = true
+    d.text = "note"
+    d.generatePythonCode() should include("table.dropna(subset=[")
+  }
+
   "SklearnTrainingBernoulliNaiveBayesOpDesc" should "round-trip its config fields through the polymorphic base" in {
     val d = new SklearnTrainingBernoulliNaiveBayesOpDesc
     d.target = "label"