This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/texera.git
The following commit(s) were added to refs/heads/main by this push:
new 566efb81eb feat(workflow-operator): report scan rows that fail schema
parsing (#6947)
566efb81eb is described below
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"))
+ }
}
