Robust and extensible resource metadata in Daft

[xcharleslin]

Issue to consider

Currently, there can only be a single ResourceRequest for a logical operator. This might be an "impedance mismatch", since a logical operator can have many partitions of different sizes.

For instance:

• A single Scan operator can read many files, each of which can vary dramatically in size.
• However, the single Scan operator can expose only a single ResourceRequest.
• At execution time, this single ResourceRequest is used for each file.

Solution discussion

The most rigorous solution to this is probably to:

1. associate ResourceRequests with physical operators (dispatchable tasks) instead of logical plans, and
2. compute resource requests individually for each physical operator at the time it is dispatched.

I think this is fundamentally aligned, because resource requests are a property of performing a physical computation on a physical set of data.

That is, we can say "My hash join on two physical partitions totalling 10MB will require 20MB memory", but we cannot say "My abstract logical join on two dataframes of 1GB will require ___ memory", because we know neither how big the partitions are nor what algorithm we will use for the join.

To build this, we will need to:

• Implement size metadata for vPartitions.
• Identify a "Physical Operation" concept and add a ResourceRequest to it.
• Modify runners to dispatch Physical Operations instead of a list of LogicalPlans.

Physical Operation concept

It seems today our closest analogue to a "Physical Operation" is LogicalOpRunner.

However, this is a post-dispatch concept; that is, what ends up physically running is not determined until after the task is dispatched. That is, today, we have

-------- Pre-Dispatch --------

        LogicalPlans
    ->  Runner, e.g. RayRunner  
    ->  (dispatch LogicalPlans to nodes)


-------- Post-dispatch --------

    ->  (receive dispatched LogicalPlans)
    ->  LogicalOpRunner
    ->  (figure out what physical operations to run)
    ->  (run the physical operations)

The suggestion is to "figure out what physical operations to run" at pre-dispatch time. This would give us the pre-execution Physical Operation concept that we need.

Dispatching Physical Operations

Calling resource_request() on a Physical Operation would then be the central source of truth for scheduling.

The semantics of this function are "Given this particular Physical Operation and its specific input partitions, produce a best-effort estimate of the resources this operation needs to succeed." As previously discussed offline, this is the thing that can start simple (e.g. concatenating input partition sizes) and be arbitrarily fancier later. This is also the level it needs to be at to be maximally expressive (e.g. we can output size estimation based on distribution metadata within individual input partitions).

Implementation

It is possible to implement this now (by implementing the refactor outlined above). It may also be worth deferring and integrating with cost-based optimization, since the code concepts this RFC introduces (partition metadata, physical operations) are exactly the concepts that the cost-based optimizer would ingest.