aglinxinyuan opened a new issue, #8698:
URL: https://github.com/apache/texera/issues/8698
### What happened?
`MainLoop._send_ecm_to_data_channels` builds one ECM per output channel and
sends each with its own `OutputManager.emit_ecm(to, ecm)` call, and `emit_ecm`
calls `flush(to, ecm)` on the partitioner of every output link. The
round-robin, hash, range and broadcast partitioners yield only when `to` is one
of their receivers. `OneToOnePartitioner.flush` ignores `to` and yields its
pending batch (if any) and the ECM on every call. So whichever channel is
flushed first takes the one-to-one link's pending tuples, and every channel
that is not the link's receiver gets an extra copy of its ECM. This applies to
StartChannel, EndChannel, and the ECMs a worker forwards from upstream
(checkpoint, state-retrieval and reconfiguration ECMs), which go through the
same `emit_ecm` call one channel at a time.
Default settings get there. Stable Merge Sort is a `manyToOnePhysicalOp` and
keeps its `SinglePartition` input requirement. A Python UDF's output is
`UnknownPartition`, which does not satisfy it, so the link from a Python UDF
into a Stable Merge Sort is a `OneToOnePartitioning`. A Python UDF (one worker,
the default) whose output port feeds a Stable Merge Sort and any other operator
therefore has a one-to-one link next to another output link. With a round-robin
link to R1 and R2 added before a one-to-one link to C, and tuples 1, 2, 3
pending at EndChannel:
```
Now Expected
R1 [1, 3] END [1, 2, 3] END [1, 3] END
R2 [2] END END [2] END
C END [1, 2, 3] END
```
With the links added the other way round, C gets its tuples, and R1 and R2
each get an extra END ahead of their own tuples and END. The order follows the
order in which the worker receives its links' `AddPartitioning` requests, and
the coordinator sends those concurrently.
The extra END is not harmless. A receiver's alignment keeps, per ECM id, the
set of channels the ECM has arrived on, and both ENDs carry the id
`EndChannel`, so a repeat on the same channel looks like a fresh arrival. A
receiver whose port has only this channel aligns on both ENDs, and the
misrouted tuples between them arrive after its port has completed. A Python
receiver whose last input port this was hangs: `complete()` stopped its
`DataProcessor` at the first END, so the next tuple or END leaves `MainLoop`
waiting in `_switch_context`, and the worker stops answering control messages
such as `EndWorker`. By the code, a Scala receiver instead processes those
tuples and runs its end-of-input handling a second time: `onFinish` again, a
second `PortCompleted`, and, if that was its last input port, a second END to
every downstream channel. With a second upstream worker, the early END plus the
other worker's END complete the port before this worker's data behind the early
END is read
.
This worked while end of input was one pass in which every partitioner sent
its own payloads to its own receivers. #3245 added the per-channel loop, at
first only for the markers a worker forwards. #3445 moved its receiver check
from `OutputManager` into each partitioner's `flush(to, ...)`, and the
one-to-one partitioner's `flush` never got it. #3432 then sent StartChannel and
EndChannel through the same loop.
The input-port materialization reader depends on `flush` ignoring the
receiver. It builds its `OneToOnePartitioner` with the worker's
`ActorVirtualIdentity`, and the partitioner looks for the channel whose
sender's name equals the id it is given. No channel matches: the id is not a
name string, and every channel the coordinator gives a reader is sent by the
reader's virtual actor (`getFromActorIdForInputPortStorage`), not by the
worker. So the partitioner never sets `receiver`. A replay shorter than one
batch still finishes, because its tuples go out with the EndChannel flush. Once
a replay reaches one batch (the workflow's data transfer batch size, 400 by
default), `add_tuple_to_batch` raises `AttributeError` and `run()` only logs
it. None of the tuples reach the operator, EndChannel is never sent, nothing
restarts the reader, and the execution stays running. Default settings get
there too: a one-worker 2-in Python UDF requires a single partition on its
`model` port, and that por
t always reads from materialized storage because the `tuples` port depends on
it, so a `model` input of 400 tuples or more leaves the run hanging.
### How to reproduce?
Run from `amber/src/main/python`, with the generated `proto` package in
place:
```python
from core.architecture.packaging.output_manager import OutputManager
from core.models import Schema, Tuple
from core.models.payload import DataFrame
from core.util import set_one_of
from proto.org.apache.texera.amber.core import (
ActorVirtualIdentity,
ChannelIdentity,
PhysicalLink,
PortIdentity,
)
from proto.org.apache.texera.amber.engine.architecture.rpc import
EmbeddedControlMessage
from proto.org.apache.texera.amber.engine.architecture.sendsemantics import (
OneToOnePartitioning,
Partitioning,
RoundRobinPartitioning,
)
me = "Worker:WF0-S-main-0"
def channel(to):
return ChannelIdentity(ActorVirtualIdentity(me),
ActorVirtualIdentity(to), False)
om = OutputManager(me)
om.add_output_port(PortIdentity(0, False), Schema(raw_schema={"k":
"INTEGER"}))
om.add_partitioning( # a link to a two-worker operator
PhysicalLink(to_port_id=PortIdentity(1)),
set_one_of(
Partitioning,
RoundRobinPartitioning(batch_size=10, channels=[channel("R1"),
channel("R2")]),
),
)
om.add_partitioning( # a link to a Stable Merge Sort
PhysicalLink(to_port_id=PortIdentity(2)),
set_one_of(
Partitioning, OneToOnePartitioning(batch_size=10,
channels=[channel("C")])
),
)
for k in (1, 2, 3):
list(om.tuple_to_batch(Tuple({"k": k})))
# What MainLoop._send_ecm_to_data_channels does at EndChannel:
ecm = EmbeddedControlMessage()
for ch in om.get_output_channel_ids():
sent = [
p.frame.to_pydict()["k"] if isinstance(p, DataFrame) else "END"
for p in om.emit_ecm(ch.to_worker_id, ecm)
]
print(ch.to_worker_id.name, sent)
```
It prints
```
R1 [[1, 3], 'END', [1, 2, 3], 'END']
R2 [[2], 'END', 'END']
C ['END']
```
where each receiver should get its own tuples and one END: `R1 [[1, 3],
'END']`, `R2 [[2], 'END']`, `C [[1, 2, 3], 'END']`.
For the reader, replay 400 tuples through a one-to-one partitioning shaped
like the one the coordinator sends (one channel from the reader's virtual actor
to the worker):
```python
from unittest.mock import MagicMock, patch
from core.models import Schema, Tuple
from core.models.internal_queue import ECMElement
from core.storage.runnables.input_port_materialization_reader_runnable
import (
InputPortMaterializationReaderRunnable,
)
from core.util import set_one_of
from core.util.virtual_identity import
get_from_actor_id_for_input_port_storage
from proto.org.apache.texera.amber.core import ActorVirtualIdentity,
ChannelIdentity
from proto.org.apache.texera.amber.engine.architecture.sendsemantics import (
OneToOnePartitioning,
Partitioning,
)
uri = "vfs:///wf/0/exec/0/result/op-a"
worker = ActorVirtualIdentity("Worker:WF0-udf-main-0")
# The partitioning ResourceAllocator.scala sends a reader of a one-worker
# operator: one channel from the reader's virtual actor to the worker.
partitioning = OneToOnePartitioning(
batch_size=400,
channels=[
ChannelIdentity(
get_from_actor_id_for_input_port_storage(uri, worker), worker,
False
)
],
)
queue = MagicMock()
reader = InputPortMaterializationReaderRunnable(
uri, queue, worker, set_one_of(Partitioning, partitioning)
)
schema = Schema(raw_schema={"k": "INTEGER"})
result, states = MagicMock(), MagicMock()
result.get.return_value = iter(Tuple({"k": k}) for k in range(400))
states.get.return_value = iter([])
with patch(
"core.storage.runnables.input_port_materialization_reader_runnable.DocumentFactory"
) as factory:
factory.open_document.side_effect = [(result, schema), (states, None)]
reader.run()
print("finished:", reader.finished())
sent = [
c.args[0].payload.id.id if isinstance(c.args[0], ECMElement) else "data"
for c in queue.put.call_args_list
]
print("sent:", sent)
```
It prints `finished: False` and `sent: ['StartChannel']`: no data and no
EndChannel.
### Version/Branch
1.4.0-incubating-SNAPSHOT (main)
### Commit Hash (Optional)
d102e26978cf0a377992b7d3ded03eaccd5d5c75
### What browsers are you seeing the problem on?
_No response_
### Relevant log output
```shell
# The reader repro, in the Python worker's stderr:
ERROR |
core.storage.runnables.input_port_materialization_reader_runnable:run:183 -
'OneToOnePartitioner' object has no attribute 'receiver'
```
--
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]
