diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDesc.scala
index e43c3f3947f..d0039f552ba 100644
--- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDesc.scala
+++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDesc.scala
@@ -33,6 +33,36 @@ import org.apache.texera.amber.operator.PythonOperatorDescriptor
 import org.apache.texera.amber.operator.metadata.annotations.{AutofillAttributeName, HideAnnotation}
 import org.apache.texera.amber.operator.metadata.{OperatorGroupConstants, OperatorInfo}
 
+// Whichever metric list the task shows must hold at least one metric: an empty
+// one renders as `metric_list = ['']` and the metric lookup fails on the empty
+// name. The rule is conditional because only one of the two lists is visible at
+// a time, and a flat `required` on both can never be satisfied.
+@JsonSchemaInject(json = """
+{
+  "allOf": [
+    {
+      "if": {
+        "required": ["isRegression"],
+        "properties": { "isRegression": { "const": false } }
+      },
+      "then": {
+        "required": ["classificationFlag"],
+        "properties": { "classificationFlag": { "minItems": 1 } }
+      }
+    },
+    {
+      "if": {
+        "required": ["isRegression"],
+        "properties": { "isRegression": { "const": true } }
+      },
+      "then": {
+        "required": ["regressionFlag"],
+        "properties": { "regressionFlag": { "minItems": 1 } }
+      }
+    }
+  ]
+}
+""")
 class MachineLearningScorerOpDesc extends PythonOperatorDescriptor {
   @JsonProperty(required = true, defaultValue = "false")
   @JsonSchemaTitle("Regression")
@@ -85,9 +115,61 @@ class MachineLearningScorerOpDesc extends PythonOperatorDescriptor {
       inputPorts = List(InputPort()),
       outputPorts = List(OutputPort())
     )
+
+  /** The two scored columns hold one quantity measured twice, so they have to be
+    * comparable. Left to sklearn, a mismatched pair either raises `Mix of label
+    * input types` from deep inside a metric or, for Accuracy, quietly scores 0 —
+    * neither of which points back at the two columns the user picked.
+    *
+    * Types only, not values: a DOUBLE label column of 0.0 / 1.0 is a legitimate
+    * classification target even though a continuous one is not, and the schema
+    * cannot tell the two apart.
+    */
+  private def validateScoredColumns(inputSchemas: Map[PortIdentity, Schema]): Unit = {
+    val numeric = Set(AttributeType.INTEGER, AttributeType.LONG, AttributeType.DOUBLE)
+    // Blank names are what the form reports before the user has picked; the
+    // schema's own `required` already says so, and saying it twice would put a
+    // second message on a half-filled operator.
+    if (actualValueColumn.isEmpty || predictValueColumn.isEmpty) return
+    inputSchemas.get(operatorInfo.inputPorts.head.id).foreach { schema =>
+      Seq(("Actual Value", actualValueColumn), ("Predicted Value", predictValueColumn))
+        .foreach {
+          case (label, column) =>
+            if (!schema.containsAttribute(column))
+              throw new RuntimeException(s"$label column '$column' is not in the input table")
+        }
+      val actualType = schema.getAttribute(actualValueColumn).getType
+      val predictType = schema.getAttribute(predictValueColumn).getType
+      if (isRegression) {
+        Seq(
+          ("Actual Value", actualValueColumn, actualType),
+          ("Predicted Value", predictValueColumn, predictType)
+        )
+          .foreach {
+            case (label, column, attributeType) =>
+              if (!numeric.contains(attributeType))
+                throw new RuntimeException(
+                  s"A regression metric needs a numeric $label column, but '$column' is " +
+                    s"${attributeType.getName}"
+                )
+          }
+      } else {
+        val comparable = (numeric.contains(actualType) && numeric.contains(predictType)) ||
+          actualType == predictType
+        if (!comparable)
+          throw new RuntimeException(
+            s"Actual Value '$actualValueColumn' (${actualType.getName}) and Predicted Value " +
+              s"'$predictValueColumn' (${predictType.getName}) hold different kinds of label, " +
+              "so a classification metric cannot compare them"
+          )
+      }
+    }
+  }
+
   override def getOutputSchemas(
       inputSchemas: Map[PortIdentity, Schema]
   ): Map[PortIdentity, Schema] = {
+    validateScoredColumns(inputSchemas)
     val metrics = if (isRegression) {
       regressionMetrics.map(_.getName())
     } else {
diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDescSpec.scala
index 09f916a51d0..a9a46a8b9f0 100644
--- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDescSpec.scala
+++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/machineLearning/Scorer/MachineLearningScorerOpDescSpec.scala
@@ -21,6 +21,7 @@ package org.apache.texera.amber.operator.machineLearning.Scorer
 
 import com.fasterxml.jackson.databind.node.ObjectNode
 import org.apache.texera.amber.core.tuple.{Attribute, AttributeType, Schema}
+import org.apache.texera.amber.core.workflow.PortIdentity
 import org.apache.texera.amber.operator.LogicalOp
 import org.apache.texera.amber.operator.metadata.OperatorGroupConstants
 import org.apache.texera.amber.util.JSONUtils.objectMapper
@@ -109,6 +110,67 @@ class MachineLearningScorerOpDescSpec extends AnyFlatSpec with Matchers {
       List(("MSE", AttributeType.DOUBLE), ("R2", AttributeType.DOUBLE))
   }
 
+  /** An input table holding one column of each type the scored-column check reasons about. */
+  private def inputSchemas(d: MachineLearningScorerOpDesc): Map[PortIdentity, Schema] =
+    Map(
+      d.operatorInfo.inputPorts.head.id -> Schema(
+        List(
+          new Attribute("y_int", AttributeType.INTEGER),
+          new Attribute("pred_long", AttributeType.LONG),
+          new Attribute("label_str", AttributeType.STRING),
+          new Attribute("pred_str", AttributeType.STRING)
+        )
+      )
+    )
+
+  private def scorer(
+      regression: Boolean,
+      actual: String,
+      predict: String
+  ): MachineLearningScorerOpDesc = {
+    val d = new MachineLearningScorerOpDesc
+    d.isRegression = regression
+    d.actualValueColumn = actual
+    d.predictValueColumn = predict
+    d
+  }
+
+  it should "accept a classification pair whose types are both numeric" in {
+    // INTEGER against LONG is one label domain read through two widths, which is
+    // what an upstream predictor emitting a wider column looks like.
+    val d = scorer(regression = false, "y_int", "pred_long")
+    d.classificationMetrics = List(classificationMetricsFnc.accuracy)
+    noException should be thrownBy d.getOutputSchemas(inputSchemas(d))
+  }
+
+  it should "reject a classification pair that mixes a number with a string" in {
+    val d = scorer(regression = false, "y_int", "pred_str")
+    d.classificationMetrics = List(classificationMetricsFnc.accuracy)
+    the[RuntimeException] thrownBy d.getOutputSchemas(inputSchemas(d)) should have message
+      "Actual Value 'y_int' (integer) and Predicted Value 'pred_str' (string) hold different " +
+        "kinds of label, so a classification metric cannot compare them"
+  }
+
+  it should "reject a non-numeric column once the task is regression" in {
+    val d = scorer(regression = true, "label_str", "pred_str")
+    d.regressionMetrics = List(regressionMetricsFnc.mse)
+    the[RuntimeException] thrownBy d.getOutputSchemas(inputSchemas(d)) should have message
+      "A regression metric needs a numeric Actual Value column, but 'label_str' is string"
+  }
+
+  it should "reject a column the input table does not hold" in {
+    val d = scorer(regression = false, "y_int", "absent")
+    the[RuntimeException] thrownBy d.getOutputSchemas(inputSchemas(d)) should have message
+      "Predicted Value column 'absent' is not in the input table"
+  }
+
+  it should "say nothing about the pair while a column is still unpicked" in {
+    // Half-filled is the state every operator passes through on the way to a
+    // valid one; the empty field already carries its own required marker.
+    val d = scorer(regression = false, "y_int", "")
+    noException should be thrownBy d.getOutputSchemas(inputSchemas(d))
+  }
+
   "MachineLearningScorerOpDesc.generatePythonCode" should "emit the scorer table operator" in {
     val d = new MachineLearningScorerOpDesc
     d.actualValueColumn = "y"