Use Output Ports of an Operator to Write Storage

[Xiao-zhen-Liu]

Currently the Amber engine creates and uses sink operators to write the results of output ports of an operator. This design causes many problems, mainly because it alters the physical plan. Ideally a physical plan should not be changed once it is compiled.

This PR updates the design to use output ports of an operator instead of sink operators to write port results to storage. The changes implemented in this PR include:

• The logic to add sink operators during compilation and scheduling are not removed in this PR. However, all the execution logic in the sink operator are removed. A sink operator will not do anything during execution. We will remove sink operators in the next PR.
• GlobalPortIdentity is moved from Region to workflow-core so that it is accessible by all the modules.
• The compiler does not create storage objects anymore. Instead, it produces a set of GlobalPortIdentity that need view results. This set is passed along to the scheduler as part of WorkflowContext.workflowSettings. In the future, this information will be directly produced by the frontend instead of by the compiler.
• The scheduler combines the ports that need view result and materialized ports needed by the scheduler as part of a region. Ideally this information should be fixed once a region is created by the SchedulerGenerator. However, as the physical plan still needs to be changed currently (because of additional cache read operators), additional logic is implemented in the ScheduleGenerator to make sure this information is correct for all the regions.
• The scheduler and resource allocator do not create storage objects directly and only assign storage URIs for each region as part of resourceConfig.
• When a region is scheduled to execute, during the initialization of a region, these URIs are used to create storage objects.
• AssignPortRequest is used to indicate whether an output port of a worker needs storage and to pass the storage URI information to a worker. Note this request is used for both input ports and output ports, and this PR only updates output ports. As a result, for input ports, empty storage URIs will be provided in AssignPortRequest. In the future, after we also use input ports to read storage, we will also update and use these storage URIs.
• Note that since currently operators with dependee inputs belong to multiple regions, and AssignPortRequest is only used once for each worker, I had to implement additional logic in the to make sure all the regions that such an operator belongs to have the proper storage information (specifically, the output port connecting a dependee link belongs to two regions, and both regions need to have storageURI for this port)
• Inside a worker (both Java and Python), the OutputManager is used to create writer threads for each output port that needs storage. The writing does not block the data processor, but will block the completion status of the operator/port.

TODOs:

• Completely remove sink op
• Use GlobalPortIdentity for storage URIs
• Remove the mention of "result" in the storage layer
• Let the frontend specify view results on the port level
• Remove cache read op and use input port for reading storage

[Yicong-Huang]

In general looks good! left code comments.

[Yicong-Huang]

LGTM!