This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6947-98d3d4b1ec8b02f611801d035c190fa31f56fa47 in repository https://gitbox.apache.org/repos/asf/texera.git
commit 566efb81eb33cdcaab1a759ee26951c5514f1b77 Author: Eugene Gu <[email protected]> AuthorDate: Fri Oct 2 00:23:25 2026 +0000 feat(workflow-operator): report scan rows that fail schema parsing (#6947) ### What changes were proposed in this PR? CSV/JSONL scan sources infer each column's type from only the first `INFER_READ_LIMIT` (=100) rows. When a later row held a value that did not parse as the inferred type, the execs caught the error, mapped the row to `null`/`None`, and filtered it out — silently dropping the entire row with no error, warning, count, or log. Every downstream count, aggregate, and join then ran on a silently truncated dataset. The bug is the silence, not the rejection. Following the consensus in discussion #6324 (skip + surface, rather than the fail-fast approach originally proposed in #6323), a scan now still skips the unparsable row — the run completes and existing workflows keep working — but each skipped row is surfaced to the user as a console warning naming the row number, offending value, column, and expected type, for example: > WARNING: skipped row 150 — value '55.5' in column 'age' cannot be read as INTEGER. Column types were inferred from an initial sample of 100 rows, and this value does not match. <img width="1137" height="788" alt="Screenshot 2026-08-05 at 2 10 49 PM" src="https://github.com/user-attachments/assets/171f741e-b11b-4567-8c62-d6939b9faa58"; /> Changes: - Add `ScanRowParseError`, which builds the per-row warning by re-parsing the failing row's fields to identify the offending column, with a generic fallback message when no single column can be identified (e.g. a malformed JSON line). The message names the inference sample size actually used — `INFER_READ_LIMIT` capped by the user-set row limit (`ScanSourceOpDesc.inferSampleSize`) — rather than a hard-coded 100. - Add `SkippedRowReporter`, shared by all scan variants: it reports the first 100 skipped rows individually and then appends a single summary line carrying the true total, so a heavily malformed file cannot flood the console or exhaust memory. - Record-and-skip in `CSVScanSourceOpExec`, `JSONLScanSourceOpExec`, `ParallelCSVScanSourceOpExec`, and `CSVOldScanSourceOpExec`. The legitimate null paths (an empty cell parsed to null, a blank/all-null line, an exhausted block) are untouched and are not reported. Row numbers: CSV reuses its existing counter, JSONL and csvOld gain absolute line/data-row numbers (a configured `offset` is added back in, so the reported number always matches the file), and ParallelCSV reports none because it is byte-partitioned across workers. - Add `OperatorExecutor.getWarnings` (defaults to empty), a generic non-fatal warning channel that any executor can opt into. - Emit the warnings at `FinalizeExecutor` in `DataProcessor` as PRINT console messages via the new `ErrorUtils.mkPrintConsoleMessage`. The titles keep the `WARNING: ` prefix the UI keys on, and, unlike the executor exception path, this does not pause the run. Inference logic is unchanged (the 100-row sampling is untouched). Scala only — no UI change. Behavior-change note for reviewers: workflows that previously ran to completion while silently discarding malformed rows now still complete, but they surface a console warning for each skipped row (capped at 100 detailed entries plus a summary). One case worth calling out: a csvOld scan over a file with an empty cell in a numeric column previously dropped that row silently and will now report it — this is exactly the silent loss this PR exists to surface, not a regression. Known limitations: warnings are collected in the executor and emitted only when it finalizes — a run that fails or is killed mid-scan does not surface them, and nothing is shown while the scan is still running. Streaming emission as rows are skipped is possible follow-up work; the finalize-only design follows the consensus in #6324. ### Any related issues, documentation, discussions? Closes #6279. Design discussion: #6324. This supersedes the fail-fast approach in #6323, which will be closed in favor of this PR. ### How was this PR tested? Scan-operator unit tests (skip-and-report naming row/value/column/type, the 100-detail cap with its summary line, empty-cell and blank-line non-reporting, the malformed-JSON fallback, absolute line numbers when an `offset` is set, and the warning naming the actual inference sample size when a row `limit` caps it): ``` sbt "WorkflowOperator/testOnly \ org.apache.texera.amber.operator.source.scan.csv.CSVScanSourceOpExecSpec \ org.apache.texera.amber.operator.source.scan.json.JSONLScanSourceOpExecSpec \ org.apache.texera.amber.operator.source.scan.csv.ParallelCSVScanSourceOpExecSpec \ org.apache.texera.amber.operator.source.scan.csvOld.CSVOldScanSourceOpExecSpec" ``` Engine-side emission path (`DataProcessorSpec`): a new test asserts the warnings reach the coordinator as PRINT console messages at finalize and that the run is not paused. Full local gate (lint + format + backend unit tests): ``` sbt "scalafixAll --check" # pass sbt scalafmtCheckAll # pass AMBER_TEST_FILTER=skip-integration sbt WorkflowExecutionService/test # pass sbt WorkflowOperator/test WorkflowCore/test WorkflowCompilingService/test # pass ``` Manual, in the Texera UI, over CSV and JSONL datasets derived from a real 403-row file whose `age` column is inferred as INTEGER: verified a single bad row (skipped, one warning naming row/value/column/type), many bad rows (100 detailed warnings plus a summary line), the clean file (all rows, no warning), an empty cell (kept as a null row, no warning), and a malformed JSONL line (skipped, generic fallback warning). In every case the scan completed instead of failing. Rebased onto main after #7247 (the JSONL worker-slice offset fix) landed ### Was this PR authored or co-authored using generative AI tooling? Co-authored by: Claude Code (Fable 5) --------- Signed-off-by: Eugene Gu <[email protected]> Co-authored-by: Copilot Autofix powered by AI <[email protected]> --- .../engine/architecture/worker/DataProcessor.scala | 11 ++- .../org/apache/texera/amber/error/ErrorUtils.scala | 18 +++- .../architecture/worker/DataProcessorSpec.scala | 85 +++++++++++++++++++ .../amber/core/executor/OperatorExecutor.scala | 5 ++ .../operator/source/scan/ScanRowParseError.scala | 99 ++++++++++++++++++++++ .../operator/source/scan/ScanSourceOpDesc.scala | 6 ++ .../operator/source/scan/SkippedRowReporter.scala | 66 +++++++++++++++ .../source/scan/csv/CSVScanSourceOpDesc.scala | 2 +- .../source/scan/csv/CSVScanSourceOpExec.scala | 13 ++- .../scan/csv/ParallelCSVScanSourceOpDesc.scala | 2 +- .../scan/csv/ParallelCSVScanSourceOpExec.scala | 26 +++++- .../scan/csvOld/CSVOldScanSourceOpDesc.scala | 2 +- .../scan/csvOld/CSVOldScanSourceOpExec.scala | 47 +++++++--- .../source/scan/json/JSONLScanSourceOpDesc.scala | 2 +- .../source/scan/json/JSONLScanSourceOpExec.scala | 47 +++++++--- .../source/scan/ScanSourceOpDescSpec.scala | 1 + .../source/scan/csv/CSVScanSourceOpExecSpec.scala | 64 +++++++++++++- .../scan/csv/ParallelCSVScanSourceOpExecSpec.scala | 21 ++++- .../scan/csvOld/CSVOldScanSourceOpExecSpec.scala | 30 +++++++ .../scan/json/JSONLScanSourceOpExecSpec.scala | 40 +++++++++ 20 files changed, 550 insertions(+), 37 deletions(-) diff --git a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala index 618fa73427..bbe1ed84e7 100644 --- a/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala +++ b/amber/src/main/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessor.scala @@ -58,7 +58,7 @@ import org.apache.texera.amber.engine.architecture.worker.statistics.WorkerStati import org.apache.texera.amber.engine.common.ambermessage._ import org.apache.texera.amber.engine.common.statetransition.WorkerStateManager import org.apache.texera.amber.engine.common.virtualidentity.util.COORDINATOR -import org.apache.texera.amber.error.ErrorUtils.{mkConsoleMessage, safely} +import org.apache.texera.amber.error.ErrorUtils.{mkConsoleMessage, mkPrintConsoleMessage, safely} import java.util.concurrent.LinkedBlockingQueue @@ -167,6 +167,15 @@ class DataProcessor( outputTuple match { case FinalizeExecutor() => sendECMToDataChannels(METHOD_END_CHANNEL, PORT_ALIGNMENT) + // Surface non-fatal warnings the executor accumulated (e.g. rows a scan + // skipped) as console messages. Unlike handleExecutorException, this must + // not pause the run — the messages are emitted and processing continues. + executor.getWarnings.foreach { warning => + asyncRPCClient.coordinatorInterface.consoleMessageTriggered( + ConsoleMessageTriggeredRequest(mkPrintConsoleMessage(actorId, warning)), + asyncRPCClient.mkContext(COORDINATOR) + ) + } // Send Completed signal to worker actor. executor.close() adaptiveBatchingMonitor.stopAdaptiveBatching() diff --git a/amber/src/main/scala/org/apache/texera/amber/error/ErrorUtils.scala b/amber/src/main/scala/org/apache/texera/amber/error/ErrorUtils.scala index 139f7b41a7..c1ac5090dc 100644 --- a/amber/src/main/scala/org/apache/texera/amber/error/ErrorUtils.scala +++ b/amber/src/main/scala/org/apache/texera/amber/error/ErrorUtils.scala @@ -22,7 +22,10 @@ package org.apache.texera.amber.error import com.google.protobuf.timestamp.Timestamp import org.apache.texera.amber.core.virtualidentity.ActorVirtualIdentity import org.apache.texera.amber.engine.architecture.rpc.controlcommands.ConsoleMessage -import org.apache.texera.amber.engine.architecture.rpc.controlcommands.ConsoleMessageType.ERROR +import org.apache.texera.amber.engine.architecture.rpc.controlcommands.ConsoleMessageType.{ + ERROR, + PRINT +} import org.apache.texera.amber.engine.architecture.rpc.controlreturns.{ControlError, ErrorLanguage} import org.apache.texera.amber.util.{StackTraceUtils, VirtualIdentityUtils} @@ -57,6 +60,19 @@ object ErrorUtils { ConsoleMessage(actorId.name, Timestamp(Instant.now), ERROR, source, title, message) } + /** + * Builds a PRINT console message carrying a plain informational/warning line (no + * stack trace). A title starting with "WARNING: " is collected into the `warnings` + * field of the sync-execution API (see SyncExecutionResource), so callers should + * keep that prefix intact. + */ + def mkPrintConsoleMessage( + actorId: ActorVirtualIdentity, + title: String, + source: String = "" + ): ConsoleMessage = + ConsoleMessage(actorId.name, Timestamp(Instant.now), PRINT, source, title, "") + def mkControlError(err: Throwable): ControlError = { // Format each stack trace element with "at " prefix val stacktrace = err.getStackTrace.map(element => s"at ${element}").mkString("\n") diff --git a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorSpec.scala b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorSpec.scala index 53f167c4ea..ab1b0ad0b3 100644 --- a/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorSpec.scala +++ b/amber/src/test/scala/org/apache/texera/amber/engine/architecture/worker/DataProcessorSpec.scala @@ -30,6 +30,9 @@ import org.apache.texera.amber.engine.architecture.logreplay.{ReplayLogManager, import org.apache.texera.amber.engine.architecture.messaginglayer.WorkerTimerService import org.apache.texera.amber.engine.architecture.rpc.controlcommands.{ AsyncRPCContext, + ConsoleMessageTriggeredRequest, + ConsoleMessageType, + ControlInvocation => ControlInvocationMessage, EmbeddedControlMessage, EmbeddedControlMessageType, EmptyRequest @@ -143,6 +146,7 @@ class DataProcessorSpec extends AnyFlatSpec with MockFactory with Matchers with ) (adaptiveBatchingMonitor.startAdaptiveBatching _).expects().anyNumberOfTimes() (adaptiveBatchingMonitor.stopAdaptiveBatching _).expects().once() + (() => executor.getWarnings).expects().returning(Seq.empty) (executor.close _).expects().once() (outputHandler.apply _).expects(*).anyNumberOfTimes() dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty) @@ -237,6 +241,7 @@ class DataProcessorSpec extends AnyFlatSpec with MockFactory with Matchers with dp.continueDataProcessing() } (adaptiveBatchingMonitor.stopAdaptiveBatching _).expects().once() + (() => executor.getWarnings).expects().returning(Seq.empty) (executor.close _).expects().once() dp.processECM( ChannelIdentity(senderWorkerId, testWorkerId, isControl = false), @@ -458,4 +463,84 @@ class DataProcessorSpec extends AnyFlatSpec with MockFactory with Matchers with dp.outputManager.hasUnfinishedOutput shouldBe false } + "data processor" should "emit executor warnings as PRINT console messages at finalize without pausing" in { + val dp = mkDataProcessor + dp.executor = executor + dp.stateManager.transitTo(READY) + val emitted = scala.collection.mutable.ArrayBuffer[WorkflowFIFOMessage]() + (outputHandler.apply _) + .expects(*) + .onCall { (m: Either[MainThreadDelegateMessage, WorkflowFIFOMessage]) => + m.foreach(emitted += _); () + } + .anyNumberOfTimes() + (executor.open _).expects().once() + ( + ( + tuple: Tuple, + input: Int + ) => executor.processTupleMultiPort(tuple, input) + ) + .expects(tuples.head, 0) + ( + ( + input: Int + ) => executor.produceStateOnFinish(input) + ) + .expects(0) + .returning(None) + ( + ( + input: Int + ) => executor.onFinishMultiPort(input) + ) + .expects(0) + val warningText = "WARNING: skipped row 3 - value 'oops' cannot be read as INTEGER" + (() => executor.getWarnings).expects().returning(Seq(warningText)) + (adaptiveBatchingMonitor.startAdaptiveBatching _).expects().anyNumberOfTimes() + (adaptiveBatchingMonitor.stopAdaptiveBatching _).expects().once() + (executor.close _).expects().once() + dp.inputManager.addPort(inputPortId, schema, List.empty, List.empty) + dp.inputGateway + .getChannel(ChannelIdentity(senderWorkerId, testWorkerId, isControl = false)) + .setPortId(inputPortId) + dp.outputManager.addPort(outputPortId, schema, None) + dp.processDCM( + ChannelIdentity(COORDINATOR, testWorkerId, isControl = true), + ControlInvocation( + METHOD_OPEN_EXECUTOR, + EmptyRequest(), + AsyncRPCContext(COORDINATOR, testWorkerId), + 0 + ) + ) + dp.processDataPayload( + ChannelIdentity(senderWorkerId, testWorkerId, isControl = false), + DataFrame(Array(tuples.head)) + ) + while (dp.inputManager.hasUnfinishedInput || dp.outputManager.hasUnfinishedOutput) { + dp.continueDataProcessing() + } + dp.processECM( + ChannelIdentity(senderWorkerId, testWorkerId, isControl = false), + endChannelPayload, + logManager + ) + while (dp.inputManager.hasUnfinishedInput || dp.outputManager.hasUnfinishedOutput) { + dp.continueDataProcessing() + } + + // The warning must reach the coordinator as a PRINT console message carrying the + // executor's warning line as its title, and it must not pause the run. + val consoleMessages = emitted.flatMap(_.payload match { + case ControlInvocationMessage(_, req: ConsoleMessageTriggeredRequest, _, _) => + Some(req.consoleMessage) + case _ => None + }) + assert( + consoleMessages.exists(m => m.msgType == ConsoleMessageType.PRINT && m.title == warningText) + ) + dp.pauseManager.isPaused shouldBe false + } + } diff --git a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala index 9837213abb..86412fae3b 100644 --- a/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala +++ b/common/workflow-core/src/main/scala/org/apache/texera/amber/core/executor/OperatorExecutor.scala @@ -48,6 +48,11 @@ trait OperatorExecutor { def onFinish(port: Int): Iterator[TupleLike] = Iterator.empty + /** Non-fatal warnings the worker surfaces as console messages once the executor + * completes (e.g. scan rows skipped for not matching the inferred schema). + */ + def getWarnings: Seq[String] = Seq.empty + def close(): Unit = {} } 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 0000000000..f2aa2c0839 --- /dev/null +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/ScanRowParseError.scala @@ -0,0 +1,99 @@ +/* + * 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, per-row warnings for scan sources when a row's values do not + * parse into the inferred schema. The scan skips such a row (rather than failing the + * whole run) and surfaces one of these warnings so the user knows exactly which row + * was dropped, and why. Messages lead with the essential facts (row number, offending + * value, column, expected type) because the console title line truncates. + */ +object ScanRowParseError { + + /** Prefix that makes a PRINT console message surface as a warning in the UI. */ + private val WarningPrefix = "WARNING: " + + /** + * Builds the warning for a single skipped row. + * + * 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 carrying the original reason is used instead. + * + * @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 inferSampleSize cap on the rows read for type inference + * (desc.inferSampleSize); a shorter file uses fewer + * @param rowNumber 1-based row number, if cheaply available + * @param cause the original parse exception + */ + def skipWarning( + rawFields: Seq[Any], + schema: Schema, + inferSampleSize: Int, + rowNumber: Option[Int], + cause: Throwable + ): String = { + val where = rowNumber.map(n => s"row $n").getOrElse("a row") + findFailingColumn(rawFields, schema) match { + case Some((attribute, value)) => + s"${WarningPrefix}skipped $where — value '$value' in column '${attribute.getName}' " + + s"cannot be read as ${attribute.getType.name()}. " + + s"Column types were inferred from an initial sample of up to $inferSampleSize rows, " + + "and this value does not match." + case None => + val reason = Option(cause.getMessage).getOrElse(cause.getClass.getSimpleName) + s"${WarningPrefix}skipped $where — could not be parsed into the inferred schema: $reason. " + + s"Column types were inferred from an initial sample of up to $inferSampleSize rows." + } + } + + /** + * Summary warning appended when the per-row detail list is capped, so the user still + * learns the true total even though only the first rows are listed individually. + */ + def moreSkipped(hidden: Int, total: Int): String = + s"${WarningPrefix}...and $hidden more row(s) skipped ($total total). " + + "Fix the values in the file, or clean the data before scanning." + + /** + * Re-parses each raw field against its attribute type; the first failure identifies + * the offending column. Missing trailing fields are ignored (equivalent to treating + * them as null for failure detection) 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/ScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDesc.scala index b0a6ee5e20..20e59f751e 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDesc.scala @@ -80,6 +80,12 @@ abstract class ScanSourceOpDesc extends SourceOperatorDescriptor { @JsonIgnore def windowLimit: Option[Int] = limit.map(_.max(0)) + /** Rows actually used for type inference: INFER_READ_LIMIT, capped by `windowLimit` + * when smaller. + */ + @JsonIgnore + def inferSampleSize: Int = windowLimit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT) + override def sourceSchema(): Schema = null override def operatorInfo: OperatorInfo = { diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/SkippedRowReporter.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/SkippedRowReporter.scala new file mode 100644 index 0000000000..f84ed9b6e7 --- /dev/null +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/SkippedRowReporter.scala @@ -0,0 +1,66 @@ +/* + * 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 scala.collection.mutable.ArrayBuffer + +/** + * Accumulates the warnings for rows a scan source skipped because they did not parse + * into the inferred schema. Shared by all scan variants (CSV, JSONL, ParallelCSV, + * csvOld) so they surface skipped rows identically. + * + * Only the first `maxDetailed` rows are listed individually; the total is always + * tracked, and when it exceeds the cap a single summary line reports how many more + * were skipped. This keeps a pathological file (e.g. hundreds of thousands of bad + * rows) from exhausting memory or flooding the console. + * + * @param maxDetailed maximum number of rows reported individually + */ +class SkippedRowReporter(maxDetailed: Int = SkippedRowReporter.DefaultMaxDetailed) { + + private val detailed = ArrayBuffer.empty[String] + private var total = 0 + + /** Records one skipped row. `warning` is evaluated lazily so building the (relatively + * expensive) per-row message is skipped once the detail cap is reached. + */ + def record(warning: => String): Unit = { + total += 1 + if (detailed.size < maxDetailed) detailed += warning + } + + /** Total number of rows skipped so far. */ + def count: Int = total + + /** + * The warnings to surface: one line per detailed row, plus a trailing summary line + * when more rows were skipped than the cap allows. Empty when nothing was skipped. + */ + def warnings: Seq[String] = { + if (total == 0) Seq.empty + else if (total > detailed.size) + detailed.toSeq :+ ScanRowParseError.moreSkipped(total - detailed.size, total) + else detailed.toSeq + } +} + +object SkippedRowReporter { + val DefaultMaxDetailed: Int = 100 +} diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala index 3ec9bf9834..38a3d18736 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/CSVScanSourceOpDesc.scala @@ -115,7 +115,7 @@ class CSVScanSourceOpDesc extends ScanSourceOpDesc { parser.beginParsing(inputReader) var data: Array[Array[String]] = Array() - val readLimit = windowLimit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT) + val readLimit = inferSampleSize for (_ <- 0 until readLimit) { val row = CSVScanSourceOpExec.parseNextRow(parser, maxColumns) if (row != null) { 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 4fe6f72318..26155bbc69 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, SkippedRowReporter} import org.apache.texera.amber.util.JSONUtils.objectMapper import org.apache.texera.dao.SiteSettings @@ -40,6 +41,9 @@ class CSVScanSourceOpExec private[csv] (descString: String) extends SourceOperat var numRowGenerated = 0 private var maxColumns: Int = CSVScanSourceOpExec.DEFAULT_MAX_COLUMNS private val schema: Schema = desc.sourceSchema() + private val skippedRows = new SkippedRowReporter() + + override def getWarnings: Seq[String] = skippedRows.warnings override def produceTuple(): Iterator[TupleLike] = { @@ -70,7 +74,14 @@ class CSVScanSourceOpExec private[csv] (descString: String) extends SourceOperat ): _* ) } catch { - case _: Throwable => null + case e: Throwable => + // Skip the unparsable row but surface it as a warning instead of + // dropping it silently. numRowGenerated is the 1-based data-row number. + skippedRows.record( + ScanRowParseError + .skipWarning(row.toSeq, schema, desc.inferSampleSize, Some(numRowGenerated), e) + ) + null } }) .filter(t => t != null) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala index c011a76cdf..dda5cd3eec 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csv/ParallelCSVScanSourceOpDesc.scala @@ -110,7 +110,7 @@ class ParallelCSVScanSourceOpDesc extends ScanSourceOpDesc { val attributeTypeList: Array[AttributeType] = inferSchemaFromRows( reader.iterator - .take(windowLimit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT)) + .take(inferSampleSize) .map(seq => seq.toArray) ) 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 1c9377e564..15017db41d 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, SkippedRowReporter} import org.apache.texera.amber.util.JSONUtils.objectMapper import org.tukaani.xz.SeekableFileInputStream @@ -40,13 +41,17 @@ class ParallelCSVScanSourceOpExec private[csv] ( objectMapper.readValue(descString, classOf[ParallelCSVScanSourceOpDesc]) private var reader: BufferedBlockReader = _ private val schema = desc.sourceSchema() + private val skippedRows = new SkippedRowReporter() + + override def getWarnings: Seq[String] = skippedRows.warnings override def produceTuple(): Iterator[TupleLike] = new Iterator[TupleLike]() { override def hasNext: Boolean = reader.hasNext override def next(): TupleLike = { - + // hoisted so the catch block can report the raw values of a failing row + 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 +59,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,7 +86,22 @@ class ParallelCSVScanSourceOpExec private[csv] ( ) TupleLike(ArraySeq.unsafeWrapArray(parsedFields): _*) } catch { - case _: Throwable => null + case e: Throwable => + // Skip the unparsable row but surface it as a warning instead of + // dropping it silently. The legitimate silent skips above (exhausted + // block, all-null/blank line) return early and never reach here. + // rowNumber is None: this scan is byte-partitioned across workers, so a + // per-worker row count would not match the file's actual line numbers. + skippedRows.record( + ScanRowParseError.skipWarning( + Option(fields).map(_.toSeq).getOrElse(Seq.empty), + schema, + desc.inferSampleSize, + None, + e + ) + ) + null } } diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala index a9eff8534f..b8ec56e4df 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/csvOld/CSVOldScanSourceOpDesc.scala @@ -104,7 +104,7 @@ class CSVOldScanSourceOpDesc extends ScanSourceOpDesc { val startOffset = windowOffset + (if (hasHeader) 1 else 0) val endOffset = - startOffset + windowLimit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT) + startOffset + inferSampleSize val attributeTypeList: Array[AttributeType] = inferSchemaFromRows( reader.iterator .slice(startOffset, endOffset) 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 87ac8976e1..8f6535572a 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, SkippedRowReporter} import org.apache.texera.amber.util.JSONUtils.objectMapper import java.net.URI @@ -36,22 +37,40 @@ class CSVOldScanSourceOpExec private[csvOld] ( var reader: CSVReader = _ var rows: Iterator[Seq[String]] = _ val schema: Schema = desc.sourceSchema() + private val skippedRows = new SkippedRowReporter() + + override def getWarnings: Seq[String] = skippedRows.warnings + override def produceTuple(): Iterator[TupleLike] = { - val tuples = rows - .map(fields => - try { - val parsedFields: Array[Any] = AttributeTypeUtils.parseFields( - fields.toArray, - schema.getAttributes - .map((attr: Attribute) => attr.getType) - .toArray - ) - TupleLike(ArraySeq.unsafeWrapArray(parsedFields): _*) - } catch { - case _: Throwable => null - } - ) + val tuples = rows.zipWithIndex + .map { + case (fields, index) => + try { + val parsedFields: Array[Any] = AttributeTypeUtils.parseFields( + fields.toArray, + schema.getAttributes + .map((attr: Attribute) => attr.getType) + .toArray + ) + TupleLike(ArraySeq.unsafeWrapArray(parsedFields): _*) + } catch { + case e: Throwable => + // Skip the unparsable row but surface it as a warning instead of + // dropping it silently. `rows` already dropped header and offset, so + // the absolute 1-based data-row number adds the offset back. + skippedRows.record( + ScanRowParseError.skipWarning( + fields, + schema, + desc.inferSampleSize, + Some(desc.windowOffset + index + 1), + e + ) + ) + null + } + } .filter(tuple => tuple != null) if (desc.windowLimit.isDefined) diff --git a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala index 1b0c928929..88e9ae57d1 100644 --- a/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala +++ b/common/workflow-operator/src/main/scala/org/apache/texera/amber/operator/source/scan/json/JSONLScanSourceOpDesc.scala @@ -88,7 +88,7 @@ class JSONLScanSourceOpDesc extends ScanSourceOpDesc { val startOffset = windowOffset val endOffset = - startOffset + windowLimit.getOrElse(INFER_READ_LIMIT).min(INFER_READ_LIMIT) + startOffset + inferSampleSize reader .lines() .iterator() 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 9b732e73c3..b6992ed795 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, SkippedRowReporter} import org.apache.texera.amber.util.JSONUtils.{JSONToMap, objectMapper} import java.io.{BufferedReader, InputStreamReader} @@ -39,20 +40,43 @@ class JSONLScanSourceOpExec private[json] ( objectMapper.readValue(descString, classOf[JSONLScanSourceOpDesc]) private var rows: Iterator[String] = _ private var reader: BufferedReader = _ + // 0-based line offset of this worker's first row, captured in open() so skipped + // rows can be reported with their absolute 1-based line number. + private var startLine: Int = 0 private val schema = desc.sourceSchema() + private val skippedRows = new SkippedRowReporter() + + override def getWarnings: Seq[String] = skippedRows.warnings override def produceTuple(): Iterator[TupleLike] = { - rows.flatMap { line => - Try { - val data = JSONToMap(objectMapper.readTree(line), desc.flatten).withDefaultValue(null) - val fields = schema.getAttributeNames.map { fieldName => - parseField(data(fieldName), schema.getAttribute(fieldName).getType) + rows.zipWithIndex.flatMap { + case (line, index) => + // Hoisted out of the Try so the failure handler can name the offending + // field; stays empty when the JSON line itself is malformed. + var rawFields: Seq[Any] = Seq.empty + Try { + val data = JSONToMap(objectMapper.readTree(line), desc.flatten).withDefaultValue(null) + rawFields = schema.getAttributeNames.map(fieldName => data(fieldName)) + val fields = schema.getAttributeNames.map { fieldName => + parseField(data(fieldName), schema.getAttribute(fieldName).getType) + } + TupleLike(fields: _*) + } match { + case Success(tuple) => Some(tuple) + case Failure(e) => + // Skip the unparsable line but surface it as a warning instead of + // dropping it silently. + skippedRows.record( + ScanRowParseError.skipWarning( + rawFields, + schema, + desc.inferSampleSize, + Some(startLine + index + 1), + e + ) + ) + None } - TupleLike(fields: _*) - } match { - case Success(tuple) => Some(tuple) - case Failure(_) => None - } } } @@ -74,6 +98,9 @@ class JSONLScanSourceOpExec private[json] ( val endOffset: Int = if (idx != workerCount - 1) count / workerCount * (idx + 1) else count + // `startOffset` is relative to the post-offset iterator; add the offset back + // so skipped-row warnings report absolute 1-based file line numbers. + startLine = offsetValue + startOffset rows = it2.iterator.slice(startOffset, endOffset) } diff --git a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDescSpec.scala b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDescSpec.scala index df016eb5ab..120ec07475 100644 --- a/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDescSpec.scala +++ b/common/workflow-operator/src/test/scala/org/apache/texera/amber/operator/source/scan/ScanSourceOpDescSpec.scala @@ -206,6 +206,7 @@ class ScanSourceOpDescSpec extends AnyFlatSpec with Matchers { tree.has("fileTypeName") shouldBe false tree.has("windowLimit") shouldBe false tree.has("windowOffset") shouldBe false + tree.has("inferSampleSize") shouldBe false } it should "omit unset optional fields entirely, the shape saved workflows store" in { 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 329f6e9f07..210914d40c 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 @@ -209,12 +209,12 @@ class CSVScanSourceOpExecSpec extends AnyFlatSpec with BeforeAndAfterAll { assert(all.map(_.getFields(0).toString) == List("1", "2", "3")) } - it should "silently drop rows that cannot be parsed into the inferred schema" in { + it should "skip rows that cannot be parsed into the inferred schema and report them" 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. + // so parseFields throws and produceTuple skips that row instead of failing. val exec = execOver( writeTempCsv("1\n2\noops\n3\n4\n"), @@ -231,5 +231,65 @@ class CSVScanSourceOpExecSpec extends AnyFlatSpec with BeforeAndAfterAll { // 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 skip is no longer silent: the row is surfaced as a warning naming the + // offending value and its absolute data-row number (offset rows included). + assert(exec.getWarnings.size == 1) + assert(exec.getWarnings.head.contains("row 3")) + assert(exec.getWarnings.head.contains("'oops'")) + // The warning names the real inference sample size (limit-capped), not the + // default INFER_READ_LIMIT of 100. + assert(exec.getWarnings.head.contains("sample of up to 2 rows")) + } + + it should "skip a post-inference row that does not parse and report row, value, column, type" in { + // 100 clean integer rows fix the column type at INTEGER (INFER_READ_LIMIT=100); + // data row 101 holds a non-integer and must be skipped but reported. + val clean = (1 to 100).map(_.toString).mkString("\n") + val exec = execOver(writeTempCsv(s"a\n$clean\noops\n"), hasHeader = true) + exec.open() + val tuples = + try exec.produceTuple().toList + finally exec.close() + + assert(tuples.size == 100) + val warnings = exec.getWarnings + assert(warnings.size == 1) + assert(warnings.head.startsWith("WARNING: ")) + assert(warnings.head.contains("row 101")) + assert(warnings.head.contains("'oops'")) + assert(warnings.head.contains("column 'a'")) + assert(warnings.head.contains("INTEGER")) + } + + it should "cap detailed warnings at 100 and append a summary with the true total" in { + val clean = (1 to 100).map(_.toString).mkString("\n") + val bad = (1 to 150).map(i => s"bad$i").mkString("\n") + val exec = execOver(writeTempCsv(s"a\n$clean\n$bad\n"), hasHeader = true) + exec.open() + val tuples = + try exec.produceTuple().toList + finally exec.close() + + assert(tuples.size == 100) + val warnings = exec.getWarnings + assert(warnings.size == 101) // 100 detail lines + 1 summary line + assert(warnings.take(100).forall(_.contains("INTEGER"))) + assert(warnings.last.contains("50 more")) + assert(warnings.last.contains("150 total")) + } + + it should "keep a row whose empty cell parses to null instead of skipping it" in { + // An empty INTEGER cell parses to null, which is a legitimate value: the row + // must survive (with the null) and must not be reported as skipped. + val exec = execOver(writeTempCsv("a,b\n1,x\n2,y\n,z\n"), hasHeader = true) + exec.open() + val tuples = + try exec.produceTuple().toList + finally exec.close() + + assert(tuples.size == 3) + assert(tuples(2).getFields(0) == null) + assert(tuples(2).getFields(1).toString == "z") + assert(exec.getWarnings.isEmpty) } } 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 51f9424454..8016ec40c8 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 @@ -75,12 +75,31 @@ class ParallelCSVScanSourceOpExecSpec extends AnyFlatSpec { assert(rows(1)(2) == null) } - it should "discard all-null (blank) lines" in { + it should "discard all-null (blank) lines without reporting them as parse failures" in { val exec = new ParallelCSVScanSourceOpExec(descString(writeCsv("name,city\nAlice,NYC\n\nBob,LA\n"))) val rows = drain(exec) assert(rows.size == 2) assert(rows.map(_.head) == Seq("Alice", "Bob")) + // A blank line is a legitimate silent skip, not a parse failure to surface. + assert(exec.getWarnings.isEmpty) + } + + it should "skip a post-inference row that does not parse and report value and column" in { + // 100 clean integer rows fix the column type at INTEGER (INFER_READ_LIMIT=100); + // the later non-integer row must be skipped but reported. This scan is + // byte-partitioned across workers, so no row number is expected in the message. + val clean = (1 to 100).map(_.toString).mkString("\n") + val exec = new ParallelCSVScanSourceOpExec(descString(writeCsv(s"a\n$clean\noops\n"))) + val rows = drain(exec) + + assert(rows.size == 100) + val warnings = exec.getWarnings + assert(warnings.size == 1) + assert(warnings.head.startsWith("WARNING: ")) + assert(warnings.head.contains("'oops'")) + assert(warnings.head.contains("column 'a'")) + assert(warnings.head.contains("INTEGER")) } it should "partition a headerless file across workers by byte range" in { 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 f4a775fcba..a683b3a1e7 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,34 @@ class CSVOldScanSourceOpExecSpec extends AnyFlatSpec { val exec = new CSVOldScanSourceOpExec(descString("a\n1\n")) exec.close() // reader is still null -> guarded, no exception } + + it should "skip a post-inference row that does not parse and report row, value, column, type" in { + // 100 clean integer rows fix the column type at INTEGER (INFER_READ_LIMIT=100); + // data row 101 holds a non-integer and must be skipped but reported. + val clean = (1 to 100).map(_.toString).mkString("\n") + val exec = new CSVOldScanSourceOpExec(descString(s"a\n$clean\noops\n")) + val rows = drain(exec) + + assert(rows.size == 100) + val warnings = exec.getWarnings + assert(warnings.size == 1) + assert(warnings.head.startsWith("WARNING: ")) + assert(warnings.head.contains("row 101")) + assert(warnings.head.contains("'oops'")) + assert(warnings.head.contains("column 'a'")) + assert(warnings.head.contains("INTEGER")) + } + + it should "report the absolute data-row number when an offset is set" in { + // offset=2 shifts both inference and reading to data row 3; the inference sample + // (rows 3..102) is all integers, so "oops" at data row 103 is skipped. The + // reported number must count from the start of the data, not from the offset. + val clean = (1 to 102).map(_.toString).mkString("\n") + val exec = new CSVOldScanSourceOpExec(descString(s"a\n$clean\noops\n", offset = Some(2))) + val rows = drain(exec) + + assert(rows.size == 100) + assert(exec.getWarnings.size == 1) + assert(exec.getWarnings.head.contains("row 103")) + } } 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 a90dcd3640..207b56aa31 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 @@ -120,4 +120,44 @@ class JSONLScanSourceOpExecSpec extends AnyFlatSpec { (0 until 3).map(i => new JSONLScanSourceOpExec(desc, idx = i, workerCount = 3)) assert(workers.map(drain(_).map(_.head)) == Seq(Seq(2), Seq(3), Seq(4, 5))) } + + it should "skip a post-inference line whose value does not match the inferred type and report it" in { + // 100 clean integer lines fix "v" at INTEGER (INFER_READ_LIMIT=100); line 101 + // holds a string and must be skipped but reported with its absolute line number. + val clean = (1 to 100).map(i => s"""{"v":$i}""") + val exec = new JSONLScanSourceOpExec(descString(writeJsonl(clean :+ """{"v":"oops"}""": _*))) + val rows = drain(exec) + + assert(rows.size == 100) + val warnings = exec.getWarnings + assert(warnings.size == 1) + assert(warnings.head.startsWith("WARNING: ")) + assert(warnings.head.contains("row 101")) + assert(warnings.head.contains("'oops'")) + assert(warnings.head.contains("column 'v'")) + assert(warnings.head.contains("INTEGER")) + } + + it should "skip a malformed JSON line and report it with the generic fallback" in { + val clean = (1 to 100).map(i => s"""{"v":$i}""") + val exec = new JSONLScanSourceOpExec(descString(writeJsonl(clean :+ """{"v": not json""": _*))) + val rows = drain(exec) + + assert(rows.size == 100) + val warnings = exec.getWarnings + assert(warnings.size == 1) + assert(warnings.head.contains("row 101")) + assert(warnings.head.contains("could not be parsed into the inferred schema")) + } + + it should "report the absolute line number when an offset is set" in { + // offset=2 skips lines 1-2; inference then samples lines 3-102 (100 clean + // integers), and the bad value sits at file line 103. + val clean = (1 to 102).map(i => s"""{"v":$i}""") + val exec = new JSONLScanSourceOpExec( + descString(writeJsonl(clean :+ """{"v":"oops"}""": _*), offset = Some(2)) + ) + assert(drain(exec).size == 100) + assert(exec.getWarnings.head.contains("row 103")) + } }
