feat(workflow-operator): skip and report scan rows that do not match the inferred schema

[eugenegujing]

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 the first 100 rows of the file, and this value does not match.

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).
• 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, 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.

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, and the malformed-JSON fallback):

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   # 1197 succeeded, 0 failed
sbt WorkflowOperator/test WorkflowCore/test WorkflowCompilingService/test   # 1760 / 581 / 11 succeeded, 0 failed

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.

Was this PR authored or co-authored using generative AI tooling?

Co-authored by: Claude Code (Fable 5)

[github-actions]

Automated Reviewer Suggestions

Based on the git blame history of the changed files, we recommend the following reviewers:

• Contributors with relevant context: @aglinxinyuan, @Yicong-Huang, @Ma77Ball
  You can notify them by mentioning @aglinxinyuan, @Yicong-Huang, @Ma77Ball in a comment.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 89.47368% with 6 lines in your changes missing coverage. Please review.
✅ Project coverage is 76.49%. Comparing base (42027e8) to head (95bcefd).
⚠️ Report is 138 commits behind head on main.

Files with missing lines	Patch %	Lines
...ber/engine/architecture/worker/DataProcessor.scala	66.66%	0 Missing and 2 partials ⚠️
...amber/operator/source/scan/ScanRowParseError.scala	88.88%	0 Missing and 2 partials ⚠️
.../texera/amber/core/executor/OperatorExecutor.scala	0.00%	1 Missing ⚠️
...mber/operator/source/scan/SkippedRowReporter.scala	91.66%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #6947      +/-   ##
============================================
+ Coverage     74.46%   76.49%   +2.02%     
- Complexity     3441     3805     +364     
============================================
  Files          1157     1158       +1     
  Lines         45621    45659      +38     
  Branches       5031     5039       +8     
============================================
+ Hits          33972    34925     +953     
+ Misses         9994     8985    -1009     
- Partials       1655     1749      +94     

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.00% <ø> (ø)
agent-service	76.76% <ø> (ø)		Carriedforward from 3ba7130
amber	72.26% <89.47%> (+5.42%)	⬆️
computing-unit-managing-service	20.49% <ø> (+2.77%)	⬆️
config-service	66.66% <ø> (ø)
file-service	67.21% <ø> (+0.40%)	⬆️
frontend	78.18% <ø> (ø)		Carriedforward from 3ba7130
notebook-migration-service	78.94% <ø> (ø)
pyamber	91.79% <ø> (ø)		Carriedforward from 3ba7130
workflow-compiling-service	26.31% <ø> (-28.84%)	⬇️

*This pull request uses carry forward flags. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

⚠️ Benchmark changes need a look

🟢 0 better · 🔴 5 worse · ⚪ 10 noise (<±5%) · 0 without baseline

Compared against main e9bf133 benchmarked on this same runner, so the delta is largely free of cross-runner hardware noise. The "7d avg" column still reflects the gh-pages dashboard. Treat <±5% as noise unless repeated.

Dashboard · Run

	config	throughput	MB/s	latency	max Δ latest / 7d
🔴	bs=10 sw=10 sl=64	366	0.224	26,173/41,677/41,677 us	🔴 +38.7% / 🔴 +156.7%
🔴	bs=100 sw=10 sl=64	818	0.499	121,564/148,889/148,889 us	🔴 +5.5% / 🔴 +37.0%
⚪	bs=1000 sw=10 sl=64	921	0.562	1,069,485/1,177,830/1,177,830 us	⚪ within ±5% / 🔴 +12.5%

Baseline details

Latest main e9bf133 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	366 tuples/sec	408 tuples/sec	767.9 tuples/sec	-10.3%	-52.3%
bs=10 sw=10 sl=64	MB/s	0.224 MB/s	0.249 MB/s	0.469 MB/s	-10.0%	-52.2%
bs=10 sw=10 sl=64	p50	26,173 us	25,015 us	12,502 us	+4.6%	+109.4%
bs=10 sw=10 sl=64	p95	41,677 us	30,055 us	16,234 us	+38.7%	+156.7%
bs=10 sw=10 sl=64	p99	41,677 us	30,055 us	18,919 us	+38.7%	+120.3%
bs=100 sw=10 sl=64	throughput	818 tuples/sec	854 tuples/sec	974.8 tuples/sec	-4.2%	-16.1%
bs=100 sw=10 sl=64	MB/s	0.499 MB/s	0.521 MB/s	0.595 MB/s	-4.2%	-16.1%
bs=100 sw=10 sl=64	p50	121,564 us	115,262 us	102,449 us	+5.5%	+18.7%
bs=100 sw=10 sl=64	p95	148,889 us	149,352 us	108,652 us	-0.3%	+37.0%
bs=100 sw=10 sl=64	p99	148,889 us	149,352 us	116,310 us	-0.3%	+28.0%
bs=1000 sw=10 sl=64	throughput	921 tuples/sec	938 tuples/sec	1,004 tuples/sec	-1.8%	-8.3%
bs=1000 sw=10 sl=64	MB/s	0.562 MB/s	0.572 MB/s	0.613 MB/s	-1.7%	-8.3%
bs=1000 sw=10 sl=64	p50	1,069,485 us	1,058,902 us	999,606 us	+1.0%	+7.0%
bs=1000 sw=10 sl=64	p95	1,177,830 us	1,191,908 us	1,046,770 us	-1.2%	+12.5%
bs=1000 sw=10 sl=64	p99	1,177,830 us	1,191,908 us	1,076,937 us	-1.2%	+9.4%

Raw CSV

config_idx,batch_size,schema_width,string_len,num_batches,total_ms,total_tuples,total_bytes,tuples_per_sec,mb_per_sec,lat_p50_us,lat_p95_us,lat_p99_us
0,10,10,64,20,545.70,200,128000,366,0.224,26172.54,41677.41,41677.41
1,100,10,64,20,2445.93,2000,1280000,818,0.499,121563.69,148888.77,148888.77
2,1000,10,64,20,21720.26,20000,12800000,921,0.562,1069484.78,1177829.86,1177829.86