test(pyamber): add unit test coverage for InputManager

[eugenegujing]

What changes were proposed in this PR?

Adds test_input_manager.py (new, 20 tests) for the Python worker's InputManager (amber/src/main/python/core/architecture/packaging/input_manager.py), which had no direct unit tests — the Scala InputManager is covered by #5451, but the Python counterpart was only exercised indirectly through MainLoop. Uses a real InternalQueue, in-memory Schema/pyarrow tables, and stubbed reader runnables, so no engine or storage is needed.

The suite covers:

• Registration — add_input_port / register_input: None id/internal fields are normalized to 0/False, re-adding an existing port is ignored, and channel-to-port bindings are checked in both directions.
• Completion tracking — completing a channel marks its port; all_ports_completed is vacuously True with zero ports; completing one channel completes the whole port even if its other channels are still open (safe today because the EndChannel ECM is port-aligned — documented in a test comment).
• Channel filtering — get_all_data_channel_ids excludes control channels.
• Materialization readers — mismatched uris/partitionings raise; one reader is created per URI with the right pairing; a second setup replaces the readers instead of appending; finished readers are not restarted, and unfinished ones run on daemon threads.
• process_data_payload — a DataFrame becomes Tuples carrying the port's schema; an empty table yields nothing; a StateFrame passes through unchanged; a DataFrame from an unregistered channel fails only when the generator is consumed; unknown payload types raise NotImplementedError.

Any related issues, documentation, discussions?

Closes #6486

How was this PR tested?

cd amber/src/main/python
PYTHONPATH=. python -m pytest ../../test/python/core/architecture/packaging/ -v
# 28 passed (20 new InputManager tests + 8 existing OutputManager tests)

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:

• No candidates found from git blame history.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 70.32%. Comparing base (8df38af) to head (ea32d07).
⚠️ Report is 9 commits behind head on main.

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #6491      +/-   ##
============================================
+ Coverage     70.27%   70.32%   +0.05%     
  Complexity     3395     3395              
============================================
  Files          1142     1142              
  Lines         44844    44844              
  Branches       4948     4948              
============================================
+ Hits          31514    31537      +23     
+ Misses        11683    11660      -23     
  Partials       1647     1647              

Flag	Coverage Δ		*Carryforward flag
access-control-service	70.00% <ø> (ø)		Carriedforward from 8df38af
agent-service	76.76% <ø> (ø)		Carriedforward from 8df38af
amber	66.65% <ø> (ø)		Carriedforward from 8df38af
computing-unit-managing-service	17.72% <ø> (ø)		Carriedforward from 8df38af
config-service	52.30% <ø> (ø)		Carriedforward from 8df38af
file-service	65.63% <ø> (ø)		Carriedforward from 8df38af
frontend	69.02% <ø> (ø)		Carriedforward from 8df38af
notebook-migration-service	78.94% <ø> (ø)		Carriedforward from 8df38af
pyamber	91.74% <ø> (+0.54%)	⬆️
workflow-compiling-service	55.14% <ø> (ø)		Carriedforward from 8df38af

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

Compared against main f722c26 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	374	0.228	26,046/36,582/36,582 us	🔴 -5.4% / 🔴 +154.8%
🔴	bs=100 sw=10 sl=64	774	0.473	127,847/150,440/150,440 us	🔴 +7.8% / 🔴 +51.8%
⚪	bs=1000 sw=10 sl=64	910	0.555	1,092,701/1,173,054/1,173,054 us	⚪ within ±5% / 🔴 +23.4%

Baseline details

Latest main f722c26 from same runner

config	metric	PR	latest main	7d avg	Δ latest	Δ 7d
bs=10 sw=10 sl=64	throughput	374 tuples/sec	395 tuples/sec	831.47 tuples/sec	-5.3%	-55.0%
bs=10 sw=10 sl=64	MB/s	0.228 MB/s	0.241 MB/s	0.507 MB/s	-5.4%	-55.1%
bs=10 sw=10 sl=64	p50	26,046 us	25,030 us	11,844 us	+4.1%	+119.9%
bs=10 sw=10 sl=64	p95	36,582 us	36,492 us	14,359 us	+0.2%	+154.8%
bs=10 sw=10 sl=64	p99	36,582 us	36,492 us	18,888 us	+0.2%	+93.7%
bs=100 sw=10 sl=64	throughput	774 tuples/sec	815 tuples/sec	1,089 tuples/sec	-5.0%	-28.9%
bs=100 sw=10 sl=64	MB/s	0.473 MB/s	0.497 MB/s	0.665 MB/s	-4.8%	-28.8%
bs=100 sw=10 sl=64	p50	127,847 us	118,564 us	92,503 us	+7.8%	+38.2%
bs=100 sw=10 sl=64	p95	150,440 us	154,809 us	99,121 us	-2.8%	+51.8%
bs=100 sw=10 sl=64	p99	150,440 us	154,809 us	112,422 us	-2.8%	+33.8%
bs=1000 sw=10 sl=64	throughput	910 tuples/sec	906 tuples/sec	1,117 tuples/sec	+0.4%	-18.5%
bs=1000 sw=10 sl=64	MB/s	0.555 MB/s	0.553 MB/s	0.682 MB/s	+0.4%	-18.6%
bs=1000 sw=10 sl=64	p50	1,092,701 us	1,104,243 us	910,046 us	-1.0%	+20.1%
bs=1000 sw=10 sl=64	p95	1,173,054 us	1,155,472 us	950,716 us	+1.5%	+23.4%
bs=1000 sw=10 sl=64	p99	1,173,054 us	1,155,472 us	985,832 us	+1.5%	+19.0%

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,535.43,200,128000,374,0.228,26046.47,36582.31,36582.31
1,100,10,64,20,2583.02,2000,1280000,774,0.473,127847.49,150439.52,150439.52
2,1000,10,64,20,21983.92,20000,12800000,910,0.555,1092700.80,1173054.46,1173054.46

[Copilot]

Pull request overview

Adds direct unit-test coverage for the Python worker’s InputManager (Amber’s Python-side input bookkeeping), ensuring its core behaviors are pinned independently of higher-level MainLoop integration paths.

Changes:

• Introduces a new pytest suite (test_input_manager.py) covering registration/binding, completion tracking, channel filtering, reader runnable setup/start semantics, and process_data_payload behavior.
• Uses in-memory Arrow tables + a real InternalQueue, and stubs InputPortMaterializationReaderRunnable to avoid storage/partitioner dependencies while still asserting construction/threading behavior.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Yicong-Huang]

🤖 texera-dev review

🚦 1 must-fix · 0 advisory · 0 polish — requesting changes only on the commit scope; the test code itself is a clean characterization suite.

📝 Title & commit scope (1)

• 🔴 must-fix — retitle test(amber) → test(pyamber): add unit test coverage for InputManager. The PR title becomes the squash-merge commit message, and the tested code is the Python worker engine (pyamber), not the Scala engine (amber). AGENTS.md scopes the Python engine pyamber (its test example is test(pyamber): …), the pyamber codecov flag moved (+0.54%, not amber), and the near-identical merged PR is test(pyamber): add unit test coverage for InternalQueue.

[eugenegujing]

🤖 texera-dev review

🚦 1 must-fix · 0 advisory · 0 polish — requesting changes only on the commit scope; the test code itself is a clean characterization suite.

📝 Title & commit scope (1)

• 🔴 must-fix — retitle test(amber) → test(pyamber): add unit test coverage for InputManager. The PR title becomes the squash-merge commit message, and the tested code is the Python worker engine (pyamber), not the Scala engine (amber). AGENTS.md scopes the Python engine pyamber (its test example is test(pyamber): …), the pyamber codecov flag moved (+0.54%, not amber), and the near-identical merged PR is test(pyamber): add unit test coverage for InternalQueue.

@Yicong-Huang fixed

[Yicong-Huang]

🟢 1 resolved · 0 open · 0 new — the prior retitle blocker (test(amber) → test(pyamber)) is applied; title now conforms to the pyamber scope and the description is complete.

0 must-fix · 0 advisory · 0 polish — clean characterization suite. No new findings.

Verification trace

Traced all 20 assertions against input_manager.py; each matches the source path. Confirmed the None-field normalization the suite pins is genuinely reachable — PortIdentity(id=None, internal=None) keeps .id is None/.internal is None under betterproto (default PortIdentity() is 0/False, but explicit None stays None), so test_none_port_fields_are_normalized exercises the branch rather than passing vacuously. Ran the suite locally: 20 passed, consistent with codecov's pyamber +0.54% (all lines covered). Placement and stub fidelity mirror the sibling test_output_manager.py.