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

[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 exec caught the error, mapped the row to null/None, and filtered it out — silently dropping the entire row with no error, warning, skipped-row count, or log. Every downstream count, aggregate, and join then ran on a silently truncated dataset. The bug is the silence, not the rejection; no comparable engine (Spark, Polars, DuckDB, Arrow) silently drops whole rows.

Before:  row fails to parse  ->  caught -> null/None  ->  filtered out   (silent data loss)
After:   row fails to parse  ->  RuntimeException      ->  console error + run stops

Changes:

• Throw a RuntimeException (original exception kept as cause) instead of silently dropping an unparsable row, in CSVScanSourceOpExec, JSONLScanSourceOpExec, ParallelCSVScanSourceOpExec, and CSVOldScanSourceOpExec. The throw surfaces through DataProcessor.handleExecutorException as a console error and stops the run (fail-fast, matching the default of Spark FAILFAST / Polars / DuckDB / Arrow).

• Add a shared ScanRowParseError helper that identifies the offending column by re-parsing each field and builds a message that leads with the essential facts, so it stays useful even when the console title line truncates, e.g.:

  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. Fix the value in the file, or clean the data before scanning.

  A generic fallback covers rows with no single identifiable column (e.g. a malformed JSON line).

• Remove the now-dead null filter in CSV and csvOld; ParallelCSV keeps its legitimate null-skip paths (exhausted block, all-null/blank line) and its filter — only parse failures abort.

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 will now fail on the first such row. This is the intended fix (loud over silent) and matches the modern-engine default, but it is a behavior change — see the discussion on #6279. Sample photo is attached below.

Any related issues, documentation, discussions?

Closes #6279

How was this PR tested?

Scan-operator unit tests (updated to assert the thrown error and message; added a JSONL malformed-line fallback test):

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"
# => Tests: succeeded 28, failed 0

Full backend gate in one clean session (lint + format + whole suite):

sbt clean "scalafixAll --check" scalafmtCheckAll test
# scalafixAll --check: pass
# scalafmtCheckAll:    pass
# test => Total number of tests run: 1721; succeeded 1721, failed 0, aborted 0 (pending 2)

Manual, in the Texera UI: scanned a CSV whose column is integer for the first 100 rows with a non-integer value (55.5) at data row 150. The CSV File Scan operator turned red and the console showed the message above; the run stopped at row 150 instead of silently emitting a short result.

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, @Ma77Ball, @Yicong-Huang
  You can notify them by mentioning @aglinxinyuan, @Ma77Ball, @Yicong-Huang in a comment.

[codecov-commenter]

Codecov Report

❌ Patch coverage is 92.85714% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 60.79%. Comparing base (a6c85b3) to head (c43f9a9).
⚠️ Report is 32 commits behind head on main.

Files with missing lines	Patch %	Lines
...amber/operator/source/scan/ScanRowParseError.scala	89.47%	0 Missing and 2 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #6323      +/-   ##
============================================
+ Coverage     60.64%   60.79%   +0.14%     
- Complexity     3368     3378      +10     
============================================
  Files          1133     1135       +2     
  Lines         44141    44219      +78     
  Branches       4825     4841      +16     
============================================
+ Hits          26771    26883     +112     
+ Misses        15908    15852      -56     
- Partials       1462     1484      +22     

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.00% <ø> (ø)
agent-service	44.59% <ø> (ø)		Carriedforward from a6c85b3
amber	66.48% <92.85%> (+0.35%)	⬆️
computing-unit-managing-service	0.00% <ø> (ø)
config-service	52.30% <ø> (ø)
file-service	63.97% <ø> (ø)
frontend	52.40% <ø> (ø)		Carriedforward from a6c85b3
notebook-migration-service	78.94% <ø> (+0.37%)	⬆️
pyamber	91.18% <ø> (ø)		Carriedforward from a6c85b3
workflow-compiling-service	55.14% <ø> (ø)

*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 · 🔴 8 worse · ⚪ 7 noise (<±5%) · 0 without baseline

Compared against main 3e08ebb 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	376	0.229	26,891/31,412/31,412 us	🔴 +17.0% / 🔴 +121.9%
🔴	bs=100 sw=10 sl=64	796	0.486	127,291/142,499/142,499 us	🔴 +6.0% / 🔴 +38.1%
🔴	bs=1000 sw=10 sl=64	898	0.548	1,110,430/1,231,620/1,231,620 us	🔴 +6.1% / 🔴 +24.2%

Baseline details

Latest main 3e08ebb from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	376 tuples/sec	428 tuples/sec	814.14 tuples/sec	-12.1%	-53.8%
bs=10 sw=10 sl=64	MB/s	0.229 MB/s	0.261 MB/s	0.497 MB/s	-12.3%	-53.9%
bs=10 sw=10 sl=64	p50	26,891 us	22,992 us	12,120 us	+17.0%	+121.9%
bs=10 sw=10 sl=64	p95	31,412 us	29,376 us	14,792 us	+6.9%	+112.4%
bs=10 sw=10 sl=64	p99	31,412 us	29,376 us	18,232 us	+6.9%	+72.3%
bs=100 sw=10 sl=64	throughput	796 tuples/sec	831 tuples/sec	1,050 tuples/sec	-4.2%	-24.2%
bs=100 sw=10 sl=64	MB/s	0.486 MB/s	0.507 MB/s	0.641 MB/s	-4.1%	-24.2%
bs=100 sw=10 sl=64	p50	127,291 us	120,061 us	96,227 us	+6.0%	+32.3%
bs=100 sw=10 sl=64	p95	142,499 us	142,140 us	103,174 us	+0.3%	+38.1%
bs=100 sw=10 sl=64	p99	142,499 us	142,140 us	111,033 us	+0.3%	+28.3%
bs=1000 sw=10 sl=64	throughput	898 tuples/sec	917 tuples/sec	1,072 tuples/sec	-2.1%	-16.3%
bs=1000 sw=10 sl=64	MB/s	0.548 MB/s	0.559 MB/s	0.655 MB/s	-2.0%	-16.3%
bs=1000 sw=10 sl=64	p50	1,110,430 us	1,089,107 us	950,645 us	+2.0%	+16.8%
bs=1000 sw=10 sl=64	p95	1,231,620 us	1,161,156 us	991,936 us	+6.1%	+24.2%
bs=1000 sw=10 sl=64	p99	1,231,620 us	1,161,156 us	1,022,498 us	+6.1%	+20.5%

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,532.55,200,128000,376,0.229,26890.50,31412.29,31412.29
1,100,10,64,20,2511.38,2000,1280000,796,0.486,127291.10,142498.84,142498.84
2,1000,10,64,20,22270.25,20000,12800000,898,0.548,1110429.60,1231619.69,1231619.69

[aglinxinyuan]

Since it is a behavior change, can you open a discussion post for this?

[eugenegujing]

Since it is a behavior change, can you open a discussion post for this?

Yes I have raised one.