Yicong-Huang commented on code in PR #8699:
URL: https://github.com/apache/texera/pull/8699#discussion_r4365121548
##########
amber/src/main/python/core/storage/runnables/input_port_materialization_reader_runnable.py:
##########
@@ -110,10 +110,12 @@ def __init__(
}
the_partitioning: Partitioning = get_one_of(partitioning)
partitioner = self._partitioning_to_partitioner[type(the_partitioning)]
+ # A one-to-one partitioner picks the channel whose sender is the given
+ # worker; this reader sends as its virtual actor, not as the worker.
self.partitioner: Partitioner = (
partitioner(the_partitioning)
if partitioner != OneToOnePartitioner
- else partitioner(the_partitioning, self.worker_actor_id)
+ else partitioner(the_partitioning, from_actor_id.name)
Review Comment:
**Advisory:**
Does the Scala reader need the same fix?
`InputPortMaterializationReaderThread.scala:76` still builds its partitioner
with `workerActorId`, and Scala `OneToOnePartitioner.allReceivers` filters
channels by that sender and calls `.head` per tuple (line 133). With the
coordinator's reader-actor senders, that should throw on the first tuple when a
JVM worker replays a materialized port with SinglePartition or OneToOne. Fold
it in here or file a follow-up?
##########
amber/src/main/python/core/architecture/sendsemantics/one_to_one_partitioner.py:
##########
@@ -54,6 +54,8 @@ def add_tuple_to_batch(
def flush(
self, to: ActorVirtualIdentity, ecm: EmbeddedControlMessage
) -> Iterator[typing.Union[EmbeddedControlMessage, typing.List[Tuple]]]:
+ if to != self.receiver:
Review Comment:
**Advisory:**
`BroadcastPartitioner.flush` (`broad_cast_partitioner.py:57-61`) calls
`self.reset()` for any `to`, so a broadcast link's tail goes only to the first
channel flushed. Nothing declares `BroadcastPartition` today, so it is not
reachable yet. A `to in self.receivers` guard there would close the same gap
this line closes.
--
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]
