jrt1899 opened a new pull request, #8862:
URL: https://github.com/apache/texera/pull/8862
### What changes were proposed in this PR?
The first of three PRs adding opt-in least-loaded partitioning. Nothing
selects the new partitioner yet and round robin is untouched, so behaviour is
unchanged.
- `LeastLoadedPartitioning` / `LeastLoadedPartitioner` (Scala and Python):
the sender fills **one** batch at a time and sends it to a single preferred
receiver, so the first batch ships after `batchSize` tuples instead of senders
× receivers × `batchSize`.
- `received_tuple_count` in `WorkerStatistics`: counted when a batch
arrives. Received minus processed gives each receiver's backlog, which PR 2
ranks receivers by.
- Each sender's starting receiver is taken from its position in the channel
list, so batches that fill before the first ranking arrives are spread across
receivers instead of all going to receiver 0.
```mermaid
flowchart LR
subgraph RR["Round robin (today): n × m buffers"]
direction LR
S1["UDF1 worker"] --> B1["buffer → r1"] --> R1["UDF2 r1"]
S1 --> B2["buffer → r2"] --> R2["UDF2 r2"]
S1 --> B3["buffer → rm"] --> R3["UDF2 rm"]
end
subgraph LL["Least-loaded (proposed): n buffers"]
direction LR
S2["UDF1 worker"] --> B4["single buffer"] -->|"least backlog"| R4["UDF2
r*"]
end
```
| | Round robin (today) | Least-loaded (proposed) |
|---|---|---|
| Buffers per sender | m (one per receiver) | 1 |
| Buffers on the link | n × m | n |
| Tuples before the first batch ships | up to n × m × batchSize | batchSize
per sender |
| Receiver chosen | fixed rotation | smallest backlog, updated each poll |
| Batch size shipped | batchSize | batchSize (unchanged) |
### Any related issues, documentation, discussions?
Part of: #8858
Discussion: #8519
Next: PR 2 (routing RPC + coordinator), PR 3 (opt-in switch)
### How was this PR tested?
- `sbt "testOnly *PartitionersSpec"`: every tuple goes to one receiver, the
preference redirects later tuples, an out-of-range index is ignored, senders
start on different receivers, and receivers are deduplicated.
- `sbt "testOnly *WorkerExecutionSpec *OperatorExecutionSpec"`: updated for
the new statistics field.
### Was this PR authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Sonnet 5.5, Claude Opus 5.5)
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
