diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/ScanRowParseError.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/ScanRowParseError.scala new file mode 100644 index 00000000000..024f96ddb3c --- /dev/null +++ b/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 + } + } +} diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExec.scala index 147238536d8..f1ed3cbea46 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExec.scala +++ b/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) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpExec.scala index 1c9377e564e..9b994d16cc8 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpExec.scala +++ b/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,7 +47,9 @@ 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'] @@ -54,7 +57,7 @@ class ParallelCSVScanSourceOpExec private[csv] ( 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 = { diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExec.scala index e7b1e938c35..652d64f8c10 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExec.scala +++ b/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) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala index 3c47796892b..96a12dba7c6 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExec.scala +++ b/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) } } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExecSpec.scala index c5edd3b29f7..e6eb834ff57 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpExecSpec.scala +++ b/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) } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpExecSpec.scala index 51f94244540..7d3d71d7b04 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpExecSpec.scala +++ b/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 + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExecSpec.scala index f4a775fcbaf..e46cc996382 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpExecSpec.scala +++ b/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 + } } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala index 2338e48f7c4..57506ce97d8 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpExecSpec.scala +++ b/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 + } }