[data] Unify PhysicalOperator implementations

[raulchen]

Today, a lot of common features are implemented separately in each PhysicalOperator subclass. This makes some features missed for certain operator, or some features behave inconsistently.

A non-comprehensive list of such features includes:

• Inputs: buffering, bundling.
• Outputs: buffering, building blocks, ordering
• Meta: metrics recording.
• Execution: submitting/managing tasks.

We should refactor it with a more unified framework.

[raulchen]

The high-level idea is to abstract the above functionalities regarding to op execution in a new class OpExecutor. This class (along with its helper classes) will include most of the code that is currently in PhysicalOperator and OpState. OpState will be removed. PhysicalOperator will only need to care "how to handle an input RefBundle". The logic of handling an input RefBundle can be categorized in 2 types:

• Handle it only based on metadata: add it directly to output; put it in input buffer; or drop it.
• Submit a task to handle it.

Other notes:

• Besides the functionalities, the following ones will also be part of OpExecutor, and are orthogonal to ops.
  • Task or actor-based execution.
  • Parallelism and autoscaling.
  • Resource tracking/management.
  • Back-pressure.
• From an op's persecutive, it doesn't directly submit Ray tasks. Instead, it only passes a function of Iterable[Batch] -> Iterable[Batch] to OpExecutor. So OpExecutor is responsible for submitting and managing the tasks.
  • Based on this, it will be every easy to implement streaming generator and zero-copy op fusion.