From 17e5144118d0ccb49fe99810d91c2b56eb6498d5 Mon Sep 17 00:00:00 2001 From: kary zheng Date: Tue, 11 Aug 2026 16:25:46 -0700 Subject: [PATCH] feat(sklearn): skip rows with missing values instead of ending the run Every Sklearn operator ended the execution when a cell it reads was empty, with an error raised by scikit-learn's own input validation rather than by the operator: ValueError: Input X contains NaN. Nothing in the family looked at missing values, so the message named X, a variable inside generated code, and pointed at neither the column nor the row. An empty value is ordinary input here. A blank CSV cell arrives as null, since univocity returns null for an empty field and AttributeTypeUtils.parseField passes it through by design. Skipping is what the rest of the codebase does with a value that is not there: twenty-four visualization operators open their generated Python with dropna, COUNT counts only non-null rows, CONCAT and MIN pass over them, and FilterPredicate answers false for every condition but IS_NULL / IS_NOT_NULL. This family was the only one with no answer at all. What skipping means follows what each operator emits. The training, classifier and testing operators produce a model, or scores over one, so a dropped row costs nothing the user can see. Training and classifier read every column but the target, so they drop on the whole table; with Count Vectorizer on they read only the text and the target and drop on those two, since a blank in a column they never touch should not cost the row. The advanced trainers read a named list of features and drop on that list plus the ground truth. Prediction is the exception. It adds a result column to each input row, so dropping would take the user's row out of the output along with the value the model had nothing to say about. It keeps the row and leaves the result empty, which is what the Hugging Face inference operators already do with a row they cannot process. Fitting on fewer rows changes the model, the way COUNT over a column with nulls changes the count. That trade is one this codebase has already made everywhere else, and it beats ending the run. Closes #7582 Generated-by: Claude Code (Claude Opus 5) Co-Authored-By: Claude Opus 5 (1M context) --- .../base/SklearnAdvancedBaseDesc.scala | 5 +++-- .../sklearn/SklearnClassifierOpDesc.scala | 2 ++ .../sklearn/SklearnPredictionOpDesc.scala | 3 +++ .../sklearn/testing/SklearnTestingOpDesc.scala | 2 +- .../training/SklearnTrainingOpDesc.scala | 2 ++ .../base/SklearnAdvancedBaseDescSpec.scala | 7 +++++++ .../SklearnBernoulliNaiveBayesOpDescSpec.scala | 8 ++++++++ .../sklearn/SklearnPredictionOpDescSpec.scala | 11 +++++++++++ .../testing/SklearnTestingOpDescSpec.scala | 9 +++++++++ ...TrainingBernoulliNaiveBayesOpDescSpec.scala | 18 ++++++++++++++++++ 10 files changed, 64 insertions(+), 3 deletions(-) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDesc.scala index 3127fa91232..8fd06b79e45 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDesc.scala +++ b/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): 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..f7a2181976c 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 @@ -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"} diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDesc.scala index 6e894fccd90..d7c17eaac44 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDesc.scala +++ b/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]) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDesc.scala index e262bb29531..f4204d48bdb 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDesc.scala +++ b/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()) 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..2bba59e2537 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 @@ -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"} diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDescSpec.scala index ba620af298a..f151fc3b3ff 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/sklearnAdvanced/base/SklearnAdvancedBaseDescSpec.scala +++ b/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() diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnBernoulliNaiveBayesOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnBernoulliNaiveBayesOpDescSpec.scala index b34d9b2e1b9..c86117419a8 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnBernoulliNaiveBayesOpDescSpec.scala +++ b/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 diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDescSpec.scala index 2b5a76284a7..c5a12eaccee 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/SklearnPredictionOpDescSpec.scala +++ b/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 diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDescSpec.scala index 8c932005003..68ec31a8d71 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/testing/SklearnTestingOpDescSpec.scala +++ b/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 diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingBernoulliNaiveBayesOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingBernoulliNaiveBayesOpDescSpec.scala index 7bfe9c0905b..8ffb8691df4 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/sklearn/training/SklearnTrainingBernoulliNaiveBayesOpDescSpec.scala +++ b/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"