You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@texera.apache.org by gi...@apache.org on 2026/07/07 16:39:30 UTC
(texera) branch gh-readonly-queue/main/pr-5989-40526b8e1783e7782ce6043e7ca72840b7fbe04d created (now 3ebbfeb2d1)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-5989-40526b8e1783e7782ce6043e7ca72840b7fbe04d
in repository https://gitbox.apache.org/repos/asf/texera.git
at 3ebbfeb2d1 refactor(pyamber): share receiver-batch construction in partitioners (#5989)
This branch includes the following new commits:
new 3ebbfeb2d1 refactor(pyamber): share receiver-batch construction in partitioners (#5989)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(texera) 01/01: refactor(pyamber): share receiver-batch construction in partitioners (#5989)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-5989-40526b8e1783e7782ce6043e7ca72840b7fbe04d
in repository https://gitbox.apache.org/repos/asf/texera.git
commit 3ebbfeb2d18b863e21255ab2851fc6841aaf6e4f
Author: Matthew B. <mg...@uci.edu>
AuthorDate: Tue Jul 7 09:39:07 2026 -0700
refactor(pyamber): share receiver-batch construction in partitioners (#5989)
### What changes were proposed in this PR?
- Added `Partitioner.build_receiver_batches(channels)` to the base
class, the single home for the ordered `(receiver, batch)` list built
via `dict.fromkeys` (which preserves channel order where a set literal
would not).
- Replaced the three identical inline copies of that construction in
`RoundRobinPartitioner`, `HashBasedShufflePartitioner`, and
`RangeBasedShufflePartitioner` with a call to the shared helper,
dropping the drifted copy-pasted comments.
- No behavior change: each partitioner still builds the same ordered,
deduplicated receiver list.
### Any related issues, documentation, discussions?
Closes: #5988
### How was this PR tested?
- Run `cd amber && PYTHONPATH=src/main/python python -m pytest
src/test/python/core/architecture/sendsemantics/test_partitioners.py
-q`, expect all 35 tests to pass.
- The order/dedup behavior owned by the new helper is covered by the
existing `TestRoundRobinPartitioner::test_init_preserves_channel_order`
and `test_init_dedupes_duplicate_channels_preserving_first_seen_order`;
confirm they still pass against the refactored construction.
### Was this PR authored or co-authored using generative AI tooling?
Co-authored with Claude Opus 4.8 in compliance with ASF
---
.../sendsemantics/hash_based_shuffle_partitioner.py | 14 ++------------
.../python/core/architecture/sendsemantics/partitioner.py | 12 ++++++++++++
.../sendsemantics/range_based_shuffle_partitioner.py | 14 ++------------
.../architecture/sendsemantics/round_robin_partitioner.py | 15 ++-------------
4 files changed, 18 insertions(+), 37 deletions(-)
diff --git a/amber/src/main/python/core/architecture/sendsemantics/hash_based_shuffle_partitioner.py b/amber/src/main/python/core/architecture/sendsemantics/hash_based_shuffle_partitioner.py
index de018e5a3c..ceee764c1b 100644
--- a/amber/src/main/python/core/architecture/sendsemantics/hash_based_shuffle_partitioner.py
+++ b/amber/src/main/python/core/architecture/sendsemantics/hash_based_shuffle_partitioner.py
@@ -37,18 +37,8 @@ class HashBasedShufflePartitioner(Partitioner):
super().__init__(set_one_of(Partitioning, partitioning))
logger.debug(f"got {partitioning}")
self.batch_size = partitioning.batch_size
- # Partitioning contains an ordered list of downstream worker ids.
- # Currently we are using the index of such an order to choose
- # a downstream worker to send tuples to.
- # Must use dict.fromkeys to ensure the order of receiver workers
- # from partitioning is preserved (using `{}` to create a set
- # does not preserve order and will not work correctly.)
- self.receivers = [
- (rid, [])
- for rid in dict.fromkeys(
- channel.to_worker_id for channel in partitioning.channels
- )
- ]
+ # Indexed by hash_code to choose the downstream worker to send to.
+ self.receivers = self.build_receiver_batches(partitioning.channels)
self.hash_attribute_names = partitioning.hash_attribute_names
@overrides
diff --git a/amber/src/main/python/core/architecture/sendsemantics/partitioner.py b/amber/src/main/python/core/architecture/sendsemantics/partitioner.py
index c4aac57cbe..ed2332d5cb 100644
--- a/amber/src/main/python/core/architecture/sendsemantics/partitioner.py
+++ b/amber/src/main/python/core/architecture/sendsemantics/partitioner.py
@@ -32,6 +32,18 @@ class Partitioner(ABC):
def __init__(self, partitioning: Message):
self.partitioning: Partitioning = get_one_of(partitioning)
+ @staticmethod
+ def build_receiver_batches(
+ channels,
+ ) -> typing.List[typing.Tuple[ActorVirtualIdentity, typing.List[Tuple]]]:
+ # An ordered (receiver, batch) pair per distinct downstream worker.
+ # dict.fromkeys preserves the channel order; a set literal would not,
+ # which breaks input-port materialization reader threads.
+ return [
+ (rid, [])
+ for rid in dict.fromkeys(channel.to_worker_id for channel in channels)
+ ]
+
def add_tuple_to_batch(
self, tuple_: Tuple
) -> Iterator[typing.Tuple[ActorVirtualIdentity, typing.List[Tuple]]]:
diff --git a/amber/src/main/python/core/architecture/sendsemantics/range_based_shuffle_partitioner.py b/amber/src/main/python/core/architecture/sendsemantics/range_based_shuffle_partitioner.py
index 28aff35935..3f8f83c504 100644
--- a/amber/src/main/python/core/architecture/sendsemantics/range_based_shuffle_partitioner.py
+++ b/amber/src/main/python/core/architecture/sendsemantics/range_based_shuffle_partitioner.py
@@ -37,18 +37,8 @@ class RangeBasedShufflePartitioner(Partitioner):
super().__init__(set_one_of(Partitioning, partitioning))
logger.info(f"got {partitioning}")
self.batch_size = partitioning.batch_size
- # Partitioning contains an ordered list of downstream worker ids.
- # Currently we are using the index of such an order to choose
- # a downstream worker to send tuples to.
- # Must use dict.fromkeys to ensure the order of receiver workers
- # from partitioning is preserved (using `{}` to create a set
- # does not preserve order and will not work correctly.)
- self.receivers = [
- (rid, [])
- for rid in dict.fromkeys(
- channel.to_worker_id for channel in partitioning.channels
- )
- ]
+ # Indexed by get_receiver_index to choose the downstream worker to send to.
+ self.receivers = self.build_receiver_batches(partitioning.channels)
self.range_attribute_names = partitioning.range_attribute_names
self.range_min = partitioning.range_min
self.range_max = partitioning.range_max
diff --git a/amber/src/main/python/core/architecture/sendsemantics/round_robin_partitioner.py b/amber/src/main/python/core/architecture/sendsemantics/round_robin_partitioner.py
index 87c3fee87d..21c1a3ca0f 100644
--- a/amber/src/main/python/core/architecture/sendsemantics/round_robin_partitioner.py
+++ b/amber/src/main/python/core/architecture/sendsemantics/round_robin_partitioner.py
@@ -35,19 +35,8 @@ class RoundRobinPartitioner(Partitioner):
def __init__(self, partitioning: RoundRobinPartitioning):
super().__init__(set_one_of(Partitioning, partitioning))
self.batch_size = partitioning.batch_size
- # Partitioning contains an ordered list of downstream worker ids.
- # Currently we are using the index of such an order to choose
- # a downstream worker to send tuples to.
- # Must use dict.fromkeys to ensure the order of receiver workers
- # from partitioning is preserved (using `{}` to create a set
- # does not preserve order and will not work with input-port
- # materialization reader threads.)
- self.receivers = [
- (rid, [])
- for rid in dict.fromkeys(
- channel.to_worker_id for channel in partitioning.channels
- )
- ]
+ # Indexed by round_robin_index to choose the downstream worker to send to.
+ self.receivers = self.build_receiver_batches(partitioning.channels)
self.round_robin_index = 0
@overrides