Cleanly Enforce Input Port Dependencies

[Xiao-zhen-Liu]

This PR redesigns our mechanisms for enforcing an operator's input port dependency relationships.

Background and Motivation: Input Port Dependency Relationships

Besides blocking output ports that must be materialized (which creates a dependency relationship on the outgoing edges of a blocking output port, a.k.a., a blocking edge), which is precisely explained in our Pasta paper, many Texera operators also contain another kind of dependency relationship, which are those between two input ports of the same operator. Examples include :

• HashJoinProbe: Build Port (a Dependee port, i.e., a port that is depended on by another port) -> Probe Port (a Depender port, i.e., a port that depends on another port)
• ML Inference UDFs: Model Port -> Data Port

A few more notations:

• In our context, Dependee/Depender ports are two names used specifically for the input port dependency relationship.
• A dependee edge is the incoming edge of a dependee port.

So far we have been using a hack to enforce this kind of dependency relationship. We can take HashJoinProbe as a running example. In terms of scheduling, we have been treating a dependee edge the same as a blocking edge in the sense that it also converts to a region edge. But different from a blocking edge, we do not materialize a dependee edge.

Instead, we:

• Let HashJoinProbe belong to both the dependee regions (the regions of the output port of the incoming edges of a dependee port) and depender regions (similarly defined).
• We let a dependee port belong to the dependee region, and let the depender port belong to the depender region.
• We execute the dependee region that only includes the dependee port, first.
• Then we execute the depender region that includes the depender port and the output port(s).

This hack design has been causing many issues. The most critical issue is that it prevents a worker/region to be terminated when it is finished because we might need it for the next region.

New Design

In this PR, we use a new and cleaner design:

• We materialize a dependee edge. Essentially, in the scheduling phase, a dependee edge is now exactly the same as a blocking edge.
• In the execution phase, we use a two-phase execution scheme, with the 1st including only the dependee ports and the 2nd including all other ports (depender ports, input ports with no dependencies, output ports). This essentially is splitting a region into 2 mini-regions.

Note we have these assumption:

• We only allow input port dependencies where the input ports of a region can be grouped as two layers, with one layer of “dependee” ports and another layer of “depender” ports. We do not allow the case where an input port can both be a dependee and a depender.
• We only allow depender ports to send data to output ports. Depenee input ports cannot send data to output ports.

This is a temporary design that fits all the current use cases and can unblock region separation. We can generalize it in the future.

Code Changes

• Hash Join Probe is modified to require HashPartition for its builder port.
• Both schedulers are modified to include materializations on dependee edges.
• RegionExecutionCoordinator and WorkflowExecutionCoordinator are updated with the two-phase execution scheme.

More details can be found in the code documentations.

Testing

The same testing workflow from #3425 can be used for testing in this PR.

Future TODOs

• Ports can be removed from Region's internal structures as they are now redundant and can be derived from Operators.
• Need to replace mutable.HashMap with concurrent.TrieMap in InputManager.scala and OutputManager.scala for thread safety.

[Yicong-Huang]

Looks much better! Some minor comments are not resolved.