fix(amber): reject duplicate worker initialization

[carloea2]

What changes were proposed in this PR?

OperatorExecution.initWorkerExecution now checks whether the worker ID is already a key in the execution map. This prevents a repeated initialization from silently replacing the existing execution.

The old characterization and pending tests are replaced with one active regression test. It verifies both that the repeated initialization fails and that the original execution remains registered.

Before: A repeated worker ID replaced the original execution.

After: A repeated worker ID raises AssertionError and preserves the original execution.

This change removes 19 lines overall.

Any related issues, documentation, discussions?

Closes #8082

How was this PR tested?

The regression test failed before the production change with 13 tests passing and 1 test failing. It passed after the change with all 14 tests passing.

sbt "set DAO / Compile / sourceGenerators := Seq.empty" "WorkflowExecutionService / Test / testOnly org.apache.texera.amber.engine.architecture.coordinator.execution.OperatorExecutionSpec"
sbt "set DAO / Compile / sourceGenerators := Seq.empty" "scalafixAll --check"
sbt scalafmtCheckAll

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

Generated-by: OpenAI Codex, GPT-5

[github-actions]

Backport auto-label report

This fix: PR was checked against each actively-supported release branch. release/* labels drive the post-merge backport, so add or remove one to change where this fix lands.

Release branch

Analysis

⚠️ release/v1.2

Not labeled automatically — none of the files this PR modifies exist on this branch (amber/src/main/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/OperatorExecution.scala, amber/src/test/scala/org/apache/texera/amber/engine/architecture/coordinator/execution/OperatorExecutionSpec.scala). The fix may target code that isn't on this release, or the files were moved/renamed after the branch was cut. Please check and add release/v1.2 by hand if this fix should be backported here.

Auto-label run.

[github-actions]

Automated Reviewer Suggestions

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

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

[github-actions]

⚠️ Benchmark changes need a look

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

Compared against main d9c0e19 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	778	0.475	11,820/18,591/18,591 us	🔴 +7.6% / 🔴 +30.7%
🔴	bs=100 sw=10 sl=64	1,755	1.071	54,146/88,861/88,861 us	🔴 +21.3% / 🟢 +51.6%
🔴	bs=1000 sw=10 sl=64	2,076	1.267	471,246/599,022/599,022 us	🔴 +18.5% / 🟢 +74.8%

Baseline details

Latest main d9c0e19 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	778 tuples/sec	803 tuples/sec	894.27 tuples/sec	-3.1%	-13.0%
bs=10 sw=10 sl=64	MB/s	0.475 MB/s	0.49 MB/s	0.546 MB/s	-3.1%	-13.0%
bs=10 sw=10 sl=64	p50	11,820 us	12,378 us	11,164 us	-4.5%	+5.9%
bs=10 sw=10 sl=64	p95	18,591 us	17,281 us	14,219 us	+7.6%	+30.7%
bs=10 sw=10 sl=64	p99	18,591 us	17,281 us	16,863 us	+7.6%	+10.2%
bs=100 sw=10 sl=64	throughput	1,755 tuples/sec	1,843 tuples/sec	1,158 tuples/sec	-4.8%	+51.6%
bs=100 sw=10 sl=64	MB/s	1.071 MB/s	1.125 MB/s	0.707 MB/s	-4.8%	+51.5%
bs=100 sw=10 sl=64	p50	54,146 us	51,352 us	88,711 us	+5.4%	-39.0%
bs=100 sw=10 sl=64	p95	88,861 us	73,236 us	95,062 us	+21.3%	-6.5%
bs=100 sw=10 sl=64	p99	88,861 us	73,236 us	102,275 us	+21.3%	-13.1%
bs=1000 sw=10 sl=64	throughput	2,076 tuples/sec	2,153 tuples/sec	1,188 tuples/sec	-3.6%	+74.8%
bs=1000 sw=10 sl=64	MB/s	1.267 MB/s	1.314 MB/s	0.725 MB/s	-3.6%	+74.8%
bs=1000 sw=10 sl=64	p50	471,246 us	464,661 us	874,204 us	+1.4%	-46.1%
bs=1000 sw=10 sl=64	p95	599,022 us	505,539 us	916,845 us	+18.5%	-34.7%
bs=1000 sw=10 sl=64	p99	599,022 us	505,539 us	947,186 us	+18.5%	-36.8%

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,257.01,200,128000,778,0.475,11820.49,18590.52,18590.52
1,100,10,64,20,1139.57,2000,1280000,1755,1.071,54146.37,88861.03,88861.03
2,1000,10,64,20,9633.68,20000,12800000,2076,1.267,471245.94,599021.62,599021.62

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 93.51%. Comparing base (d9c0e19) to head (591a50f).
✅ All tests successful. No failed tests found.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #8083      +/-   ##
============================================
- Coverage     93.51%   93.51%   -0.01%     
+ Complexity     4708     4707       -1     
============================================
  Files          1195     1195              
  Lines         48467    48467              
  Branches       5378     5378              
============================================
- Hits          45326    45322       -4     
- Misses         1693     1696       +3     
- Partials       1448     1449       +1     

Flag	Coverage Δ		*Carryforward flag
access-control-service	81.00% <ø> (ø)		Carriedforward from d9c0e19
agent-service	99.32% <ø> (ø)		Carriedforward from d9c0e19
amber	89.56% <100.00%> (-0.03%)	⬇️
computing-unit-managing-service	73.67% <ø> (ø)		Carriedforward from d9c0e19
config-service	86.86% <ø> (ø)		Carriedforward from d9c0e19
file-service	87.91% <ø> (ø)		Carriedforward from d9c0e19
frontend	95.86% <ø> (ø)		Carriedforward from d9c0e19
notebook-migration-service	79.31% <ø> (ø)		Carriedforward from d9c0e19
pyamber	98.54% <ø> (ø)		Carriedforward from d9c0e19
workflow-compiling-service	77.19% <ø> (ø)		Carriedforward from d9c0e19

*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.

[Yicong-Huang]

🟡 0 must-fix · 1 advisory · 0 polish — correct, minimal and complete; the one finding is an optional shape suggestion.

Simplifications (1)

• OperatorExecution.scala:55 — putIfAbsent would collapse the check-then-act into one atomic call (advisory, see inline)

Verification trace

Checked that the now-live assertion cannot fire on an existing path. The single production caller is ExecutorDeployment.createWorkers, and all three routes into it register only fresh maps: RegionExecutionManager calls it exclusively under if (!existOpExecution) (the inherited/deep-cloned branch never deploys), restartRegionExecution replaces the RegionExecution wholesale before the restart path rebuilds it, and Coordinator.loadFromCheckpoint passes a freshly constructed OperatorExecution() per operator. Within one call the worker ids come from (0 until workerCount), so they cannot repeat either.

Also checked the fix is complete: no other contains-on-a-Java-map call exists in the repo. Hashtable is unused, and the three other ConcurrentHashMap fields never call contains. The replacement test genuinely fails against the pre-change tree.

On the backport bot's open question for release/v1.2: nothing needs backporting. The corrected assertion is unreachable on every production path, so visible behavior on a release branch would be unchanged.

[carloea2]

Ready to merge when you have a chance.