Matthew-DWM opened a new pull request, #9008:
URL: https://github.com/apache/texera/pull/9008
### What changes were proposed in this PR?
A Python worker reports the size of its input queue to the JVM, which uses
it as the worker's queued credit for flow control. The queue sized each item
with `sys.getsizeof(item)`. For a data batch, that measures only the 48-byte
`DataElement` wrapper, not the Arrow data it holds. Upstream senders therefore
never ran out of credit, and a slow Python UDF could buffer its whole input in
memory.
```mermaid
sequenceDiagram
participant U as Upstream worker (JVM)
participant J as Python worker, JVM side
participant P as Python worker (InternalQueue)
U->>J: data batch
J-->>U: NetworkAck(queuedCredit = JVM buffer + last Python in_mem_size)
J->>P: Arrow Flight put
P-->>J: ack carrying in_mem_size()
Note over U: not enough credit: stash the batch and backpressure
```
```
Before: 400-row batch (~400 KB) queued -> reported as 48 B -> sender
never backpressured
After: 400-row batch (~400 KB) queued -> reported as ~400 KB -> sender
backpressured at the credit limit
```
This PR:
- lets `LinkedBlockingMultiQueue` take a keyword-only `size_of` (default
`sys.getsizeof`), applied once per item at `put`;
- has `InternalQueue` size a `DataFrame` by `Table.get_total_buffer_size()`,
and keep `sys.getsizeof` for every other item.
It uses `get_total_buffer_size()` rather than the issue's `nbytes`-first
order because a slice keeps its parent's whole buffers alive. A `DataFrame`
always holds a `pyarrow.Table`, so the other fallbacks aren't needed.
There are no Scala, configuration or protocol changes. The visible effect is
that backpressure now engages for a Python worker once its queue reaches the
per-channel credit limit (1.6 GB by default).
### Any related issues, documentation, discussions?
Closes #4313. Follows up #4281, which the stale bot closed.
### How was this PR tested?
The tests were written first; 9 of the 11 new tests fail on `main`.
- `test_internal_queue.py`: frames are sized by their Arrow buffers
(including multi-chunk, empty and sliced frames), and other items keep the
wrapper size.
- `test_linked_blocking_multi_queue.py`: default and custom `size_of`, a
zero size, a `size_of` that raises, and `None`.
- `test_network_receiver.py`: batches put over Arrow Flight are
acknowledged, and polled, with the queue's total of Arrow bytes, which drops to
0 once the queue drains. On `main` the acknowledgements are `[48, 96]`.
```
cd amber
ruff check src/main/python src/test/python
ruff format --check src/main/python src/test/python
LOGURU_LEVEL=WARNING pytest -m "not integration" # 1345 passed, 1 xfailed
```
End to end, I ran Python source → Python UDF sleeping 0.5 ms per tuple →
count, with 100,000 rows and the credit limit lowered to 2 MB. All rows arrived
in both runs:
| | `main` | This PR |
|---|---|---|
| Largest queue size reported to the JVM | 9,120 B | 1,612,864 B |
| Backpressure activations | 0 | 245 |
| Slow worker's peak memory | 201 MB | 129 MB |
| Elapsed | 119.1 s | 116.8 s |
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
