[Data] Link PhysicalOperator to its LogicalOperator

[scottjlee]

Why are these changes needed?

It can be useful to know the LogicalOperator which was translated to the particular PhysicalOperator. For example, if we are trying to determine the exact type of Read operation, the PhysicalOperator will be a Map operator, which is ambiguous (we can only tell by examining the operator name, which is not reliable).

This PR links each PhysicalOperator to its originating LogicalOperator upon creation in the execution plan optimizer, so that we can access this information during execution.

Related issue number

Checks

• I've signed off every commit(by using the -s flag, i.e., git commit -s) in this PR.
• I've run scripts/format.sh to lint the changes in this PR.
• I've included any doc changes needed for https://docs.ray.io/en/master/.
  • I've added any new APIs to the API Reference. For example, if I added a
    method in Tune, I've added it in doc/source/tune/api/ under the
    corresponding .rst file.
• I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
• Testing Strategy
  • Unit tests
  • Release tests
  • This PR is not tested :(

[bveeramani]

How does this work for logical operators that create multiple physical operators (e.g., Read which creates an InputDataBuffer and a TaskPoolOperator)? Would PhysicalOperator._logical_operator be None for everything but the last physical operator created by a single logical operator?

[scottjlee]

Good point. yeah, the current behavior would be exactly as you describe above. alternatively, i could add it to each physical operator created, but that would require a bigger change to PhysicalOperator.__init__() to accept the source LogicalOperator as a parameter.

i can see that also being useful, but requires a bigger change. thoughts on if it's worth the additional complexity here?

[scottjlee]

the Read -> InputDataBuffer->TaskPoolMapOperator was the only case i could come up with where one LogicalOperator produces multiple PhysicalOperators. so that's why i'm wondering if it's worth the extra change.

[scottjlee]

actually, i can just manually call set_logical_operator() in plan_read_op, so no need to add it as a constructor arg.

[raulchen]

another approach is that you iterate the DAG here and set logical operator for those don't have it.

[scottjlee]

if the physical DAG is already generated, how would we know which logical operator to set?

for example (ignoring operator fusion), if we have Read -> Map, which gets translated to `InputDataBufer -> TaskPoolMapOperator_1 -> TaskPoolMapOperator_2, and the mapping of physical -> logical operator is:

{
    Read: TaskPoolMapOperator_1,
    Map: TaskPoolMapOperator_2
}

how would we know that the InputDataBuffer also originated from the Read op, unless we explicitly set it during the translation from logical to physical op?

[raulchen]

This _plan function generates the DAG in a post-order traverse order.
So at this line, all nodes without logical operators must be created by the current logical operator.

[raulchen]

this should be a list for fused physical operators.

[scottjlee]

good point. i was looking at this, and realized that during operator fusion, we actually construct a new LogicalOperator to represent the fused PhysicalOperators. For example for fusing two Map->Map ops, we construct a new logical Map op: https://github.com/ray-project/ray/blob/master/python/ray/data/_internal/logical/rules/operator_fusion.py#L355-L360

so i was thinking we should link this newly constructed logical op, instead of combining the list of linked logical ops from the physical operators being fused, since this most closely reflects what's going on under the hood. let me know if you think we should still maintain the full list of logical operators (i can still see that being useful for keeping track of where linked logical operators end up).