[jk-shah]

When a batched DoFn's input must be rebatched from another batch type, GeneralPurposeConsumerSet.receive_batch passed a generator to BatchConverter.produce_batch, which is declared to take a sequence. The pyarrow Table converter reads its input once per column, so every column after the first was empty and the bundle failed with ArrowInvalid.

Fixes #41042.

[github-actions]

📋 Review Summary

This PR fixes a bug in GeneralPurposeConsumerSet.receive_batch where the iterator returned by explode_batch was passed directly to BatchConverter.produce_batch without being materialized into a sequence. Because PyarrowBatchConverter.produce_batch iterates over its input once per column, passing a generator caused subsequent columns to be empty and fail with ArrowInvalid. The fix and accompanying regression test are clean, minimal, and well-targeted.

🔍 General Feedback

• Materializing explode_batch(...) into a list in GeneralPurposeConsumerSet.receive_batch properly satisfies the Sequence[E] contract of BatchConverter.produce_batch.
• Consider hoisting the batch conversion outside the inner for consumer in consumers: loop if multiple consumers share the same consumer_batch_converter (or even hoisting explode_batch if there are multiple other_batch_consumers entries), though this is a minor optimization on an already-warned fallback path.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

Assigning reviewers:

R: @damccorm for label python.
Backup reviewer: claudevdm (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

[damccorm]

Thanks, nice find! I'll wait for tests to run, but will merge if they pass

[damccorm]

I was nervous about this until I did more digging. For future readers, I think that materializing list(explode_batch(...)) is safe from a memory standpoint because:

1. windowed_batch.values is already a single in-memory batch, and produce_batch must return a single materialized in-memory batch (pa.Table, pd.DataFrame, np.ndarray, etc.) for the downstream consumer anyway. So we're going to do this sort of materialization immediately anyways.
2. As mentioned in the description, BatchConverter.produce_batch(elements: Sequence[E]) expects a Sequence, which all other call sites already pass. Passing a list here also avoids issues when rebatching into converters like NumpyBatchConverter or PytorchBatchConverter that reject generators. So I think it is generically required, and we can't just push this into a consumer.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 57.44%. Comparing base (8970af1) to head (2aac5df).
⚠️ Report is 4 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #41045      +/-   ##
============================================
- Coverage     57.44%   57.44%   -0.01%     
  Complexity     2492     2492              
============================================
  Files          1203     1203              
  Lines        171109   171109              
  Branches       2103     2103              
============================================
- Hits          98294    98287       -7     
- Misses        70161    70168       +7     
  Partials       2654     2654              

Flag	Coverage Δ
python	80.17% <ø> (-0.01%)	⬇️

Flags with carried forward coverage won't be shown. 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:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.