aglinxinyuan opened a new pull request, #8917: URL: https://github.com/apache/texera/pull/8917
### What changes were proposed in this PR? Every operator can take 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`. With it, File Scan can read a different file in each loop iteration: ``` Text Input ──► Loop Start ══(purple)══► File Scan (fileName = $file) ──► Loop End ``` - **The port.** It uses the reserved port id `PortIdentity(-1)`, so the proto does not change. The compiler adds the port to an operator only when a link targets it, and it never counts as a data input. - **Scheduling.** A link into the port counts as blocking in `getBlockingAndDependeeLinks`. The cost-based scheduler (`CostBasedScheduleGenerator`) therefore always places the sender's region before the receiver's, and its search code does not change. The receiver's data edges can still be pipelined. - **Reading the port first.** A worker starts the port's reader before its other inputs and holds its data channels until the port's input ends. An operator with a dependee input, such as Cartesian Product, reads the port in that earlier phase, still first. A source fed through the port, such as File Scan, starts producing only after the port's input ends. Everything that arrives on the port goes to every worker of the receiver. - **Turning input into variables.** A control message on the port is handled like one on a data port: each variable a property refers to is written in, and the message is forwarded through the rest of the loop block. A tuple on the port turns each of its columns into a variable of the same name; the first tuple is used. - **The later value wins.** When 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. Before, this failed with "got two different values in one iteration". - **File names bound at run time.** The compiler leaves a `$name` file name unresolved, and File Scan resolves it when the variable arrives. - **Loop Start.** Its Initialization code now runs after its input `D` is complete, so it can read `D`. `D` joins `table` as a reserved name. - **Frontend.** Every operator shows a purple port at its bottom, and links into it 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 a reader is fed through the port, its file-name field takes a variable such as `$file` instead of the file picker. ### Any related issues, documentation, discussions? Closes #8916. Design discussion: https://github.com/apache/texera/discussions/8915 ### How was this PR tested? New and updated tests cover the port, the compiler, the worker, the scheduler, resource allocation, Loop Start and the frontend. The end-to-end test `ControlVariablePortIntegrationSpec` runs the example above in MATERIALIZED mode: Text Input lists three files that hold 1–2, 3–5 and 6, and Loop End receives `1` through `6`. ``` sbt "WorkflowCore/testOnly org.apache.texera.amber.core.workflow.ControlVariablePortSpec org.apache.texera.amber.core.workflow.PhysicalOpSpec org.apache.texera.amber.core.workflow.PhysicalPlanSpec org.apache.texera.amber.core.executor.StateReferenceBindingSpec" \ "WorkflowCompiler/testOnly org.apache.texera.common.compiler.WorkflowCompilerSpec" \ "WorkflowExecutionService/testOnly org.apache.texera.amber.engine.architecture.messaginglayer.InputManagerSpec org.apache.texera.amber.engine.architecture.worker.promisehandlers.EndChannelHandlerSpec org.apache.texera.amber.engine.architecture.worker.DataProcessorSpec org.apache.texera.amber.engine.architecture.scheduling.CostBasedScheduleGeneratorSpec org.apache.texera.amber.engine.architecture.scheduling.resourcePolicies.ResourcePoliciesSpec org.apache.texera.amber.engine.architecture.scheduling.RegionExecutionManagerSpec" AMBER_TEST_FILTER=integration-only sbt "WorkflowExecutionService/testOnly *ControlVariablePortIntegrationSpec" (cd amber && python -m pytest src/test/python/core/models/test_loop_operators.py src/test/python/core/runnables/test_data_processor.py) (cd frontend && yarn ng test --watch=false \ --include=src/app/workspace/util/loop-variable.util.spec.ts \ --include=src/app/workspace/service/workflow-graph/model/workflow-graph.spec.ts \ --include=src/app/workspace/service/execute-workflow/execute-workflow.service.spec.ts \ --include=src/app/workspace/service/joint-ui/joint-ui.service.spec.ts \ --include=src/app/workspace/service/validation/validation-workflow.service.spec.ts \ --include=src/app/workspace/service/compile-workflow/workflow-compiling.service.spec.ts \ --include=src/app/workspace/component/property-editor/operator-property-edit-frame/operator-property-edit-frame.component.spec.ts \ --include=src/app/workspace/component/workflow-editor/workflow-editor.component.spec.ts) ``` All of these pass locally. ### Was this PR authored or co-authored using generative AI tooling? Generated-by: Claude Code (Opus 5.5) -- 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]
