GitHub user aglinxinyuan created a discussion: A control-variable port on every operator: design and prototype
## Summary Every operator gets one more input port, the **control-variable port**, drawn in purple. Whatever arrives on it becomes the operator's control variables, and a property that holds `$name` takes the value of the variable `name`. An unchanged operator can then run once per loop iteration on a value computed upstream. File Scan is the example: today it needs a second "from input" version of the operator for this. A prototype that runs end to end is on [`aglinxinyuan/texera:control-variable-port`](https://github.com/aglinxinyuan/texera/tree/control-variable-port). It is a prototype for discussion, so there is no PR. Feedback on the design and on the open questions at the end is very welcome. ## Motivating example: read every file of a folder ``` Text Input ──► Loop Start ══(purple)══► File Scan ──► Loop End (one path init: i = 0 fileName = $file update: i += 1 per row) file = D.line[0] file = D.line.get(i) output: None condition: i < len(D) ``` File Scan is a source: it has no data input, and its code does not change. In each iteration, Loop Start sends `{i, file}` into File Scan's control-variable port. File Scan binds `$file`, opens that file, and its rows flow on to Loop End. Today the same workflow needs File Scan From Input, which reads the file name from a tuple field. ## Design 1. **One port per operator, no new edge type.** An edge from any output port may end at the control-variable port. Edges carry data and control variables as they do today; the port is where the conversion happens. 2. **The port has no schema.** A control message (the variables that Loop Start sends) is merged into the operator's control variables. A tuple is converted column by column, and each column name becomes a variable name. 3. **Binding is by name.** `$name` in a property takes the value of the variable `name`. A name that never arrives is a run-time error. 4. **The port's input ends before the operator starts.** Every edge into the port is a dependency, so the scheduler has to materialize it. The receiver's data edges can still be pipelined. 5. **Several edges may enter one port, and the later value wins.** The rule now applies everywhere: if two values of one name reach an operator at the same loop depth, the later one replaces the earlier one. This replaces today's conflict error. 6. **A value fed through the port travels with the block.** It joins the operator's control variables and is forwarded downstream through the rest of the loop block, like any other control variable. ## How the prototype implements it | Layer | Change | |---|---| | core | The port uses a reserved port id, `ControlVariablePort.Id = PortIdentity(-1)`, so the proto does not change. `PhysicalPlan.getBlockingAndDependeeLinks` treats a link into the port as blocking. Schema propagation ignores a schema that arrives on the port. | | compiler | The port is added only to an operator that a link targets (`PhysicalOp.withControlVariablePort`). It is not part of `inputPorts`, so it never counts as a data input. When File Scan's file name is `$file`, the compiler keeps it for run time instead of resolving it. | | scheduler (PASTA) | No change; see the next section. | | worker | A source with the port starts its input readers instead of the source starter. The reader for the port starts first. Data channels are paused (`ControlVariablePortPause`) until the port's EndChannel arrives. A tuple on port -1 becomes control variables (first row only). They are registered and emitted downstream with the envelope of the port's last control message. | | binding | `StateReferenceBinding` lets the later value win at the same depth and no longer throws on a conflict. | | pyamber | Loop Start's `init()` runs once its input `D` is complete, and it can read `D`. `D` and `table` are reserved names. `provideData` may return `None`. Values that arrive on the back edge overlay what `init()` set. | | File Scan | Resolves a file name that is bound at run time. | | frontend | Every operator has a purple port at its bottom, and links into that port are purple. The link is sent to the backend as port -1. The property panel offers the names that arrive on the port: Loop Start's variables, or the sender's output columns. When File Scan is fed through the port, its file name becomes a variable input. Validation skips the undeclared-name check for an operator fed through the port, because its names arrive at run time. | ## What PASTA does with the port edge `CostBasedScheduleGenerator` already treats the port edge as a dependency, with no scheduler change, because `getBlockingAndDependeeLinks` includes it: - the search never treats the port link as a pipelining candidate; - the sender's region always runs before the receiver's; - the receiver's own data edge can still be pipelined with its upstream operator. `CostBasedScheduleGeneratorSpec` has two new cases: 1. CSV A feeds Keyword's data port and CSV C feeds Keyword's control-variable port. A and Keyword end up in one region, and C's region has an edge to Keyword's region. 2. A diamond in which `first` feeds `second` both directly and through the port. The two operators end up in different regions, and `second`'s region comes after `first`'s. ## Tests - **Unit tests.** The touched suites pass locally: - core: `ControlVariablePortSpec`, `PhysicalOpSpec`, `PhysicalPlanSpec` - compiler: `WorkflowCompilerSpec` - worker: `InputManagerSpec`, `EndChannelHandlerSpec`, `DataProcessorSpec`, `StateReferenceBindingSpec` - scheduler: `CostBasedScheduleGeneratorSpec` - pyamber: loop operators and the data processor - the eight touched frontend specs - **End to end.** `ControlVariablePortIntegrationSpec` runs the folder example above over three files that contain `1,2`, `3,4,5` and `6`. It checks that Loop End receives `1` through `6`, and it passes locally. - **Rejected at compile time:** - an operator that runs generated or user code (for example, a Python UDF) as the port's receiver; - one output port that feeds both the control-variable port and a data port of the same operator. ## Limits of the prototype - **Only File Scan is supported.** The CSV reader infers its schema by reading the file at compile time, and with `$file` there is no file yet. One answer is an example file that fixes the schema; it is not built. - **Python operators cannot receive on the port yet.** They are rejected at compile time because the Python worker has no path for port -1. - **Only the first row of a tuple input is converted.** The column and row picker on the port is not built. - **One output port cannot feed both the port and a data port** of the same operator; the compiler rejects this. - **The branch includes open PRs.** It is based on #8640, #8642 and #8750 as well as the merged #8639, so its diff against `main` includes those PRs. The prototype itself is [these eight commits](https://github.com/aglinxinyuan/texera/compare/3d524b4d9f08edbfdf541afcac97627a717dad1a...control-variable-port). ## Open questions 1. **Port id.** Is a reserved port id (`-1`) acceptable long term, or should the port be a real port in `OperatorInfo` and the proto? 2. **Where the port exists.** The UI shows the port on every operator, but the backend adds it only when a link targets it. Is that split right? 3. **Several tuples on the port.** Is "pick the columns and a row index on the port" the right UI, or should more than one row be an error? 4. **CSV and other readers.** Should the schema come from an example file per `$file` property, or from a schema the user declares? 5. **Python receivers.** Should port -1 go through the same tuple-to-variables conversion in pyamber, or should Python receivers stay rejected? GitHub link: https://github.com/apache/texera/discussions/8915 ---- This is an automatically sent email for [email protected]. To unsubscribe, please send an email to: [email protected]
