fix(workflow-operator): fail loudly on unparsable scan rows

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/ScanRowParseError.scala

@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.texera.amber.operator.source.scan
+
+import org.apache.texera.amber.core.tuple.{Attribute, AttributeTypeUtils, Schema}
+
+import scala.util.Try
+
+/**
+  * Builds actionable errors for scan sources when a row's values do not parse into
+  * the inferred schema. The console title line in the UI truncates, so the essential
+  * facts (row number, offending value, column name, expected type) come first.
+  */
+object ScanRowParseError {
+
+  /**
+    * Builds a RuntimeException describing why a row failed to parse.
+    *
+    * The failing column is identified by re-parsing each raw field individually;
+    * this only runs on a row that has already failed, so the extra cost is irrelevant.
+    * If no single column can be identified (e.g. a malformed JSON line or a structural
+    * row error), a generic fallback message carrying the original reason is used.
+    *
+    * @param rawFields      raw field values of the failing row, in schema order
+    *                       (may be empty or shorter than the schema)
+    * @param schema         the inferred schema of the scan
+    * @param inferReadLimit number of rows used for type inference (desc.INFER_READ_LIMIT)
+    * @param rowNumber      1-based row number, if cheaply available
+    * @param cause          the original parse exception
+    */
+  def build(
+      rawFields: Seq[Any],
+      schema: Schema,
+      inferReadLimit: Int,
+      rowNumber: Option[Int],
+      cause: Throwable
+  ): RuntimeException = {
+    val message = findFailingColumn(rawFields, schema) match {
+      case Some((attribute, value)) =>
+        val prefix = rowNumber.map(n => s"Row $n: value").getOrElse("Value")
+        s"$prefix '$value' in column '${attribute.getName}' cannot be read as " +
+          s"${attribute.getType.name()}. " +
+          s"Column types were inferred from the first $inferReadLimit rows of the file, " +
+          "and this value does not match. " +
+          "Fix the value in the file, or clean the data before scanning."
+      case None =>
+        val reason = Option(cause.getMessage).getOrElse(cause.getClass.getSimpleName)
+        val prefix = rowNumber.map(n => s"Row $n").getOrElse("A row")
+        s"$prefix could not be parsed into the inferred schema: $reason. " +
+          s"Column types were inferred from the first $inferReadLimit rows of the file. " +
+          "Fix the row in the file, or clean the data before scanning."
+    }
+    new RuntimeException(message, cause)
+  }
+
+  /**
+    * Re-parses each raw field against its attribute type; the first failure identifies
+    * the offending column. Missing trailing fields are treated as null (as the scan does)
+    * and thus never fail.
+    */
+  private def findFailingColumn(
+      rawFields: Seq[Any],
+      schema: Schema
+  ): Option[(Attribute, Any)] = {
+    schema.getAttributes.iterator
+      .zip(rawFields.iterator)
+      .find {
+        case (attribute, value) =>
+          Try(AttributeTypeUtils.parseField(value, attribute.getType)).isFailure
+      }
+  }
+}

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExec.scala

@@ -24,6 +24,7 @@ import com.univocity.parsers.csv.{CsvFormat, CsvParser, CsvParserSettings}
 import org.apache.texera.amber.core.executor.SourceOperatorExecutor
 import org.apache.texera.amber.core.storage.DocumentFactory
 import org.apache.texera.amber.core.tuple.{AttributeTypeUtils, Schema, TupleLike}
+import org.apache.texera.amber.operator.source.scan.ScanRowParseError
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 import org.apache.texera.dao.SiteSettings
 
@@ -70,10 +71,16 @@ class CSVScanSourceOpExec private[csv] (descString: String) extends SourceOperat
             ): _*
           )
         } catch {
-          case _: Throwable => null
+          case e: Throwable =>
+            throw ScanRowParseError.build(
+              row.toSeq,
+              schema,
+              desc.INFER_READ_LIMIT,
+              Some(numRowGenerated),
+              e
+            )
         }
       })
-      .filter(t => t != null)
 
     if (desc.limit.isDefined) tupleIterator = tupleIterator.take(desc.limit.get)
 

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpExec.scala

@@ -23,6 +23,7 @@ import org.apache.texera.amber.core.executor.SourceOperatorExecutor
 import org.apache.texera.amber.core.storage.DocumentFactory
 import org.apache.texera.amber.core.tuple.{Attribute, AttributeTypeUtils, TupleLike}
 import org.apache.texera.amber.operator.source.BufferedBlockReader
+import org.apache.texera.amber.operator.source.scan.ScanRowParseError
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 import org.tukaani.xz.SeekableFileInputStream
 
@@ -46,15 +47,17 @@ class ParallelCSVScanSourceOpExec private[csv] (
       override def hasNext: Boolean = reader.hasNext
 
       override def next(): TupleLike = {
-
+        // raw field values of the current line, hoisted out of the try block so the
+        // failure handler below can report the offending column and value
+        var fields: Array[AnyRef] = null
         try {
           // obtain String representation of each field
           // a null value will present if omit in between fields, e.g., ['hello', null, 'world']
           val line = reader.readLine
           if (line == null) {
             return null
           }
-          var fields: Array[AnyRef] = line.toArray
+          fields = line.toArray
 
           if (fields == null || util.Arrays.stream(fields).noneMatch(s => s != null)) {
             // discard tuple if it's null or it only contains null
@@ -81,10 +84,19 @@ class ParallelCSVScanSourceOpExec private[csv] (
           )
           TupleLike(ArraySeq.unsafeWrapArray(parsedFields): _*)
         } catch {
-          case _: Throwable => null
+          case e: Throwable =>
+            throw ScanRowParseError.build(
+              Option(fields).map(_.toSeq).getOrElse(Seq.empty),
+              schema,
+              desc.INFER_READ_LIMIT,
+              None,
+              e
+            )
         }
       }
 
+      // null marks intentionally skipped lines (exhausted block or all-null/blank rows),
+      // not parse failures; parse failures now abort the scan above.
     }.filter(tuple => tuple != null)
 
   override def open(): Unit = {

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExec.scala

@@ -23,6 +23,7 @@ import com.github.tototoshi.csv.{CSVReader, DefaultCSVFormat}
 import org.apache.texera.amber.core.executor.SourceOperatorExecutor
 import org.apache.texera.amber.core.storage.DocumentFactory
 import org.apache.texera.amber.core.tuple.{Attribute, AttributeTypeUtils, Schema, TupleLike}
+import org.apache.texera.amber.operator.source.scan.ScanRowParseError
 import org.apache.texera.amber.util.JSONUtils.objectMapper
 
 import java.net.URI
@@ -49,10 +50,10 @@ class CSVOldScanSourceOpExec private[csvOld] (
           )
           TupleLike(ArraySeq.unsafeWrapArray(parsedFields): _*)
         } catch {
-          case _: Throwable => null
+          case e: Throwable =>
+            throw ScanRowParseError.build(fields, schema, desc.INFER_READ_LIMIT, None, e)
         }
       )
-      .filter(tuple => tuple != null)
 
     if (desc.limit.isDefined)
       tuples.take(desc.limit.get)

common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala

@@ -23,6 +23,7 @@ import org.apache.texera.amber.core.executor.SourceOperatorExecutor
 import org.apache.texera.amber.core.storage.DocumentFactory
 import org.apache.texera.amber.core.tuple.AttributeTypeUtils.parseField
 import org.apache.texera.amber.core.tuple.TupleLike
+import org.apache.texera.amber.operator.source.scan.ScanRowParseError
 import org.apache.texera.amber.util.JSONUtils.{JSONToMap, objectMapper}
 
 import java.io.{BufferedReader, InputStreamReader}
@@ -42,16 +43,22 @@ class JSONLScanSourceOpExec private[json] (
   private val schema = desc.sourceSchema()
 
   override def produceTuple(): Iterator[TupleLike] = {
-    rows.flatMap { line =>
+    rows.map { line =>
+      // raw values extracted from the JSON line, in schema order; kept visible to the
+      // failure handler so it can report the offending column and value. Stays empty
+      // when the line itself is malformed JSON (readTree fails before extraction).
+      var rawFields: Seq[Any] = Seq.empty
       Try {
         val data = JSONToMap(objectMapper.readTree(line), desc.flatten).withDefaultValue(null)
-        val fields = schema.getAttributeNames.map { fieldName =>
-          parseField(data(fieldName), schema.getAttribute(fieldName).getType)
+        rawFields = schema.getAttributeNames.map(fieldName => data(fieldName))
+        val fields = rawFields.zip(schema.getAttributes).map {
+          case (value, attribute) => parseField(value, attribute.getType)
         }
         TupleLike(fields: _*)
       } match {
-        case Success(tuple) => Some(tuple)
-        case Failure(_)     => None
+        case Success(tuple) => tuple
+        case Failure(e) =>
+          throw ScanRowParseError.build(rawFields, schema, desc.INFER_READ_LIMIT, None, e)
       }
     }
   }

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExecSpec.scala

@@ -185,27 +185,30 @@ class CSVScanSourceOpExecSpec extends AnyFlatSpec with BeforeAndAfterAll {
     assert(firstCol == List("2", "3"))
   }
 
-  it should "silently drop rows that cannot be parsed into the inferred schema" in {
-    // No header, so every line is data. The schema is inferred from the first
-    // `limit` rows only (INFER_READ_LIMIT is capped by limit); those are integers,
-    // so the column is inferred as INTEGER. `offset` then shifts the output window
-    // past that inference sample onto a row whose value ("oops") is not an integer,
-    // so parseFields throws and produceTuple filters that row out instead of failing.
-    val exec =
-      execOver(
-        writeTempCsv("1\n2\noops\n3\n4\n"),
-        hasHeader = false,
-        offset = Some(2),
-        limit = Some(2)
-      )
+  it should "fail loudly when a row cannot be parsed into the inferred schema" in {
+    // No header, so every line is data. Type inference samples only the first
+    // INFER_READ_LIMIT (=100) rows; here they are all integers, so the single
+    // column (auto-named "column-1") is inferred as INTEGER. Row 101 holds a
+    // non-integer ("oops"), which does not match the inferred schema. The scan
+    // must abort loudly on that row (surfacing to the UI via
+    // DataProcessor.handleExecutorException) rather than silently dropping it.
+    val content = (1 to 100).mkString("\n") + "\noops\n"
+    val exec = execOver(writeTempCsv(content), hasHeader = false)
     exec.open()
-    val tuples =
+    val ex = intercept[RuntimeException] {
       try exec.produceTuple().toList
       finally exec.close()
+    }
 
-    // Output window is rows 3,4,5 ("oops","3","4"); the bad row is skipped, so we
-    // get the two good ones rather than an exception. Count is below the raw 5 rows.
-    assert(tuples.size == 2)
-    assert(tuples.map(_.getFields(0).toString) == List("3", "4"))
+    // The message must lead with the essentials — row number, offending value,
+    // column name, expected type — then the actionable fix.
+    assert(ex.getMessage.startsWith("Row 101: value")) // 1-based row of the bad value
+    assert(ex.getMessage.contains("'oops'")) // the offending value
+    assert(ex.getMessage.contains("'column-1'")) // the offending column's name
+    assert(ex.getMessage.contains("cannot be read as"))
+    assert(ex.getMessage.contains("INTEGER")) // the inferred/expected type
+    assert(ex.getMessage.contains("clean the data before scanning")) // actionable fix
+    // The original parse exception is preserved as the cause for debugging.
+    assert(ex.getCause != null)
   }
 }

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpExecSpec.scala

@@ -103,4 +103,25 @@ class ParallelCSVScanSourceOpExecSpec extends AnyFlatSpec {
     assert(keys.size == 4) // each row read exactly once
     assert(rows0.nonEmpty && rows1.nonEmpty)
   }
+
+  it should "fail loudly when a row cannot be parsed into the inferred schema" in {
+    // Type inference samples only the first INFER_READ_LIMIT (=100) data rows;
+    // those are integers, so column "v" is inferred as INTEGER. Row 101 holds a
+    // non-integer, which the inferred schema cannot accept, so the scan must
+    // abort loudly rather than dropping the row. (This is distinct from the
+    // legitimate null returns for exhausted blocks / all-null lines, which the
+    // trailing .filter still drops.)
+    val content = "v\n" + (0 until 100).mkString("\n") + "\noops\n"
+    val exec = new ParallelCSVScanSourceOpExec(descString(writeCsv(content)))
+    val ex = intercept[RuntimeException](drain(exec))
+
+    // Column-identified message (no row number for this exec): value, column,
+    // expected type, then the actionable fix.
+    assert(ex.getMessage.startsWith("Value 'oops'")) // the offending value, up front
+    assert(ex.getMessage.contains("'v'")) // the offending column's name
+    assert(ex.getMessage.contains("cannot be read as"))
+    assert(ex.getMessage.contains("INTEGER")) // the inferred/expected type
+    assert(ex.getMessage.contains("clean the data before scanning")) // actionable fix
+    assert(ex.getCause != null) // original parse exception preserved as the cause
+  }
 }

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExecSpec.scala

@@ -82,4 +82,23 @@ class CSVOldScanSourceOpExecSpec extends AnyFlatSpec {
     val exec = new CSVOldScanSourceOpExec(descString("a\n1\n"))
     exec.close() // reader is still null -> guarded, no exception
   }
+
+  it should "fail loudly when a row cannot be parsed into the inferred schema" in {
+    // Type inference samples only the first INFER_READ_LIMIT (=100) data rows;
+    // those are integers, so column "v" is inferred as INTEGER. Row 101 holds a
+    // non-integer, which the inferred schema cannot accept, so the scan must
+    // abort loudly rather than dropping the row.
+    val content = "v\n" + (0 until 100).mkString("\n") + "\noops\n"
+    val exec = new CSVOldScanSourceOpExec(descString(content, header = true))
+    val ex = intercept[RuntimeException](drain(exec))
+
+    // Column-identified message (no row number for this exec): value, column,
+    // expected type, then the actionable fix.
+    assert(ex.getMessage.startsWith("Value 'oops'")) // the offending value, up front
+    assert(ex.getMessage.contains("'v'")) // the offending column's name
+    assert(ex.getMessage.contains("cannot be read as"))
+    assert(ex.getMessage.contains("INTEGER")) // the inferred/expected type
+    assert(ex.getMessage.contains("clean the data before scanning")) // actionable fix
+    assert(ex.getCause != null) // original parse exception preserved as the cause
+  }
 }

common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala

@@ -81,4 +81,39 @@ class JSONLScanSourceOpExecSpec extends AnyFlatSpec {
     val exec = new JSONLScanSourceOpExec(descString(uri, limit = Some(2)))
     assert(drain(exec).map(_.head) == Seq(0, 1))
   }
+
+  it should "fail loudly when a row cannot be parsed into the inferred schema" in {
+    // Type inference samples only the first INFER_READ_LIMIT (=100) rows; those
+    // have integer "v" values, so "v" is inferred as INTEGER. Row 101 holds a
+    // non-integer, which the inferred schema cannot accept, so the scan must
+    // abort loudly rather than dropping the row.
+    val cleanRows = (0 until 100).map(i => s"""{"v":$i}""")
+    val badRow = """{"v":"oops"}"""
+    val exec = new JSONLScanSourceOpExec(descString(writeJsonl((cleanRows :+ badRow): _*)))
+    val ex = intercept[RuntimeException](drain(exec))
+
+    // Column-identified message (no row number for this exec): value, column,
+    // expected type, then the actionable fix.
+    assert(ex.getMessage.startsWith("Value 'oops'")) // the offending value, up front
+    assert(ex.getMessage.contains("'v'")) // the offending column's name
+    assert(ex.getMessage.contains("cannot be read as"))
+    assert(ex.getMessage.contains("INTEGER")) // the inferred/expected type
+    assert(ex.getMessage.contains("clean the data before scanning")) // actionable fix
+    assert(ex.getCause != null) // original parse exception preserved as the cause
+  }
+
+  it should "fall back to a generic row error when a line is not valid JSON" in {
+    // A syntactically malformed line yields no fields at all (readTree throws
+    // before extraction), so no single offending column can be identified and
+    // the generic fallback message must be used. The bad line sits after the
+    // first 100 rows so schema inference (which also parses lines) never sees it.
+    val cleanRows = (0 until 100).map(i => s"""{"v":$i}""")
+    val badRow = """{not valid json"""
+    val exec = new JSONLScanSourceOpExec(descString(writeJsonl((cleanRows :+ badRow): _*)))
+    val ex = intercept[RuntimeException](drain(exec))
+
+    assert(ex.getMessage.contains("could not be parsed into the inferred schema"))
+    assert(ex.getMessage.contains("clean the data before scanning"))
+    assert(ex.getCause != null) // original parse exception preserved as the cause
+  }
 }