A control-variable port on every operator: design and prototype

[aglinxinyuan]

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 existing operator can then run once per loop iteration on a value computed upstream. Today File Scan needs a second "from input" operator for this.

A prototype that runs end to end is on aglinxinyuan/texera:control-variable-port. It is a prototype for discussion, so there is no PR.

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. Its only change is that it resolves the file name at run time, once $file is bound. In each iteration, Loop Start sends {i, file} into File Scan's port. File Scan binds $file, opens that file, and its rows flow on to Loop End.

Design

1. One port per operator, no new edge type. Any output port may link to it, and the port does the conversion.
2. No schema. A control message merges into the operator's control variables. A tuple's columns become variables of the same name.
3. Binding is by name. A name that never arrives is a run-time error.
4. The port's input ends before the operator starts. The edge into the port is materialized. The receiver's data edges can still be pipelined.
5. The later value wins. Several edges may enter the port. Everywhere, if two values of one name reach an operator at the same loop depth before it first uses them, the later one replaces the earlier one. This replaces today's conflict error.
6. Port-fed values travel with the block. They are forwarded through the rest of the loop block, like any other control variable.

Prototype

• The port is the reserved PortIdentity(-1). The compiler adds it only when a link targets it.
• The worker starts the port's reader first and holds data channels until the port's input ends.
• Loop Start's init() can read D.
• The frontend draws the port and its links in purple and sends each link as port -1.

PASTA's search needs no change. The prototype adds the port link to getBlockingAndDependeeLinks, so the sender's region runs before the receiver's. The receiver's own data edge can still be pipelined.

End-to-end test. ControlVariablePortIntegrationSpec runs the example over three files that hold the numbers 1 to 2, 3 to 5 and 6, one per line. Loop End receives 1 through 6. It passes locally.

Limits

• Among file readers, only File Scan. The CSV reader infers its schema from the file at compile time, and with $file there is no file yet.
• No receivers that run code (UDFs, Python-based operators, Loop Start, Loop End). They are rejected at compile time.
• Only the first tuple that reaches the port is converted. Later tuples, from any edge, are dropped.
• One output port cannot feed both the port and a data port of one operator. The compiler rejects this.
• The branch includes open PRs feat(frontend): accept $variable references in loop blocks #8640, feat(pyamber): register the state before process_state #8642 and fix(amber): reject $i in settings the plan is built from #8750, plus the merged feat(amber): bind $i loop-variable references in operators #8639. The prototype itself is these commits.

[sshiv012]

How do you decide which operator properties can resolve at runtime, especially when properties like CSV file paths affect compile time schema inference? And how are the bound values checked against the property’s expected type?

[aglinxinyuan]

1. A property can take $name unless the plan is built from it, because the plan is compiled before any variable has a value. fix(amber): reject $i in settings the plan is built from #8750 marks those properties with @NoLoopVariable (Projection's columns, group-by keys, join keys), and a $name there is a compile error. File Scan works with $file because its schema comes from its properties, not from the file. CSV infers its schema by reading the file at compile time, so $file doesn't work there yet. The plan is to take the schema from an example file or a declared schema instead.

2. Values are checked at run time, right before the operator first sees data. StateReferenceBinding converts each value to the field's Java type: an integer field accepts "7" but rejects 7.5. The value then goes through the descriptor's own Jackson deserialization. A wrong type or a missing name shows up as an operator error, and the operator pauses.