[C++][Compute] Replace ExecNode::InputReceived with ::MakeTask (Part 2)

[asfimport]

In ARROW-13576 we made possible to defer computation of exec nodes but only for some implementations (e.g. Filter and Project) 

This ticket is to continue that effort and make all the nodes submit their tasks at the moment they are operating the ExecNode::InputReceived

Related tasks:

https://issues.apache.org/jira/browse/ARROW-13576

https://issues.apache.org/jira/browse/ARROW-14330

 

Reporter: Percy Camilo Triveño Aucahuasi / @aucahuasi

Related issues:

• [C++] Investigate potential performance improvements for the filter node (relates to)

PRs and other links:

• GitHub Pull Request #11923

_{Note: This issue was originally created as ARROW-14970. Please see the migration documentation for further details.}

[asfimport]

Weston Pace / @westonpace:
We talked about this offline but I'll add it as a comment here for visibility. I think this is a good idea but I'd like to avoid task-per-node (although that could be an acceptable stepping stone). One way we can do this (inspired by work from @aocsa) is change InputReceived from:

void InputReceived(ExecNode* input, ExecBatch batch)

to something like...

void InputReceived(ExecNode* input, std::function<void(ExecBatch)> task)

A "terminating node" (e.g. sink node / pipeline breaker) could then submit the task. A "non-terminating node" (e.g. filter/project/etc.) could wrap the task with its own work (function composition) and call InputReceived on the downstream node.

There are probably other ways to do this as well.

[asfimport]

Todd Farmer / @toddfarmer:
This issue was last updated over 90 days ago, which may be an indication it is no longer being actively worked. To better reflect the current state, the issue is being unassigned. Please feel free to re-take assignment of the issue if it is being actively worked, or if you plan to start that work soon.