GitHub user sshiv012 edited a discussion: Proposal: operators that emit on more
than one output port
## Why
A common step in an analysis produces more than one thing: a cleaned table and
a chart of it, a train split and a test split, matched rows and rejects. Today
one operator shows one result. To get a table and a plot you chain a second
operator off the same link and look at two result tabs. That works, but it adds
another operator for each additional output and a Python UDF that already has
both in hand has no way to hand them out separately.
The port model already supports this. Split and If emit on two ports today.
What is missing is the plumbing between a Python UDF and the result panel. This
proposal is to finish that plumbing, in small PRs, and to agree on the API
decisions up front.
## Current architecture
Ports are first-class from the descriptor down to storage:
```mermaid
flowchart LR
D[OperatorInfo<br/>outputPorts: List OutputPort<br/>id, displayName, mode]
--> P[PhysicalOp<br/>outputPorts: Map port -> schema, links]
P --> C[WorkflowCompiler<br/>storage for every external port<br/>of a viewed
op]
C --> S[Scheduler<br/>AssignPort per output port<br/>with that port's schema]
S --> W[Worker]
W --> ST[(Storage<br/>URI keyed by op + port)]
```
The Scala worker routes by port. An executor returns `(tuple, Some(port))` from
`processTupleMultiPort`, and `OutputManager.passTupleToDownstream` sends it
only on links whose `fromPortId` matches. Split and If use this.
Two paths stop honoring the port.
**Python worker.** The UDF yields tuples with no port. pyamber finalizes each
tuple against the first port's schema, then hands it to every partitioner, so
it goes out on every link of every port and into every port's storage.
```mermaid
flowchart LR
U[UDF yields tuple] --> F[finalize with<br/>port 0 schema]
F --> O[OutputManager.tuple_to_batch<br/>all partitioners]
O --> L0[links off port 0]
O --> L1[links off port 1]
F --> SS[storage of every port]
```
**Result delivery.** Result delivery, pagination, export, and sync execution
then select port 0. The websocket result event is keyed by operator, and the
result panel builds one Result tab per operator. The output mode (table vs
HTML) is taken from the first external port encountered, so one operator is
either all table or all HTML.
```mermaid
flowchart LR
ST[(storage<br/>op + port 0)] --> R[ExecutionResultService<br/>getResultUri
op, PortIdentity 0]
ST1[(storage<br/>op + port 1)] -. not delivered independently .-> R
R --> E[WebResultUpdateEvent<br/>Map opId -> update]
E --> UI[result panel<br/>one Result tab per operator]
```
Summary of what exists and what does not:
| Layer | Per-port today |
| --- | --- |
| Port model, schema propagation, compiler, scheduler, storage | yes |
| Scala executor API and output manager | yes (Split, If) |
| Frontend: add/remove ports, persistence, links by port, per-port row counts
on the canvas | yes |
| Python UDF descriptor: ports declared by the user | yes |
| Python UDF descriptor: output schema for ports after the first | no, port 0
only |
| pytexera: a way for code to name a port | no |
| pyamber: routing, schema, stats, storage by port | no, port 0 / all |
| Result service, export, sync API, websocket event, result panel | no, port 0
only |
Related: #8274 and #8276 (pyamber `get_port` ignores the id; fixes the lookup
only, nothing calls it with a port yet), #2078 (UDF port validation).
## What would change
Four incremental PRs: the first delivers multi-port results for Scala
operators; the remaining PRs build toward multi-port Python UDF output.
### 1. Results per port
Key the result store by operator plus port. Send the websocket update per port,
add a `portId` (default 0) to the pagination, export and sync requests, and
read the output mode from the port being served. The result panel shows one tab
per external output port when an operator has more than one, labeled by the
port's display name.
```mermaid
flowchart LR
ST0[(op + port 0)] --> R[ExecutionResultService<br/>keyed by op + port]
ST1[(op + port 1)] --> R
R --> E[WebResultUpdateEvent<br/>per op + port]
E --> UI[result panel<br/>Result: table / Result: chart]
```
Single port workflows retain their behavior and presentation, so existing
workflows should see no difference. This alone makes both ports of Split and If
viewable.
### 2. pyamber honors a port
`tuple_to_batch(tuple, port)` filters partitioners by the link's source port.
`data_processor` finalizes against that port's schema. Statistics and storage
writes take the port. Lands on top of #8276. Arrow batch construction uses that
port’s schema, including buffered batches flushed at completion.
### 3. pytexera: naming the port
Add an explicit PortOutput wrapper that carries a value and its destination
port. Bare yields continue to target port 0. The SDK will introduce a return
type covering both bare and wrapped outputs.
```python
class ProcessTableOperator(UDFTableOperator):
@overrides
def process_table(self, table: Table, port: int) -> Iterator[UDFOutput]:
yield table # port 0, unchanged
yield PortOutput({"html-content": fig.to_html()}, port=1)
```
A bare yield targets port 0. all_output_to_tuple learns to unwrap PortOutput,
normalize its value into tuples, and preserve the destination port for each
tuple.
### 4. Per-port output schema and mode on the Python UDF
The descriptor propagates a schema for port 0 only, and the saved port
description carries no schema or mode. Add a per-port `kind`:
| kind | schema | mode |
| --- | --- | --- |
| `table` (default) | retain input columns + extra output columns, as today |
SET_SNAPSHOT |
| `visualization` | `html-content: STRING` | SINGLE_SNAPSHOT |
Workflows saved before this default every port to `table`. The port editor in
the property panel gets the kind field for output ports; today it edits input
ports only.
Nothing changes in the scheduler, storage layout, Arrow data path (maybe Arrow
serialization code?), or the `PythonDataHeader` proto: partitioning happens on
the Python side, so Scala never needs the port.
## Where I want feedback
1. **Unnamed output on a multi-port Python UDF.** Scala treats "no port" as
"all ports" for sending and asserts a single port for schema. I propose Python
defaults to port 0, so a chart row is never validated against the table schema.
Should Scala be aligned, or is "all ports" a semantic anyone relies on?
2. ** Explicit PortOutput wrapper vs. pair vs. emit.** I propose yield
PortOutput(output, port=1) because it preserves the generator contract and
makes the destination explicit. Would a raw (output, port) pair or
self.emit(output, port=1) be preferable?
3. **Per-port `kind` vs. per-port column list.** `kind` covers the table+chart
case with one field. A per-port extra-columns list is more general but a bigger
form change. Start with `kind` and add columns later?
4. **Websocket shape.** Nested `Map[opId, Map[portId, update]]` vs. a composite
key `"opId:portId"`. The composite key is the smaller frontend diff; the nested
map keeps one shape for every operator and avoids key parsing in every consumer
## Acceptance Criteria
Validate a UDF with at least three outputs, including different table/HTML
schemas, an empty output port, and buffered output flushed at completion.
Verify independent results, pagination, export, invalid-port errors, and
compatibility with existing single-port workflows.
Note - This supports any number of output ports; table plus visualization is
the initial use case. Independently configurable table schemas are future work.
GitHub link: https://github.com/apache/texera/discussions/8914
----
This is an automatically sent email for [email protected].
To unsubscribe, please send an email to: [email protected]
