batch: Execute root fragment task in frontend.

[liurenjie1024]

Currently the root fragment of mpp query is executed in compute node. It would be better to move it to frontend to reduce schedule and pull data latency.

[lmatz]

There are two cases:

1. the root fragment is simply a single distribution exchange operator. It will be scheduled at the frontend.
2. the root fragment has other non-exchange operators. I think we may still schedule them at the compute nodes.

[fuyufjh]

There are two cases:

1. the root fragment is simply a single distribution exchange operator. It will be scheduled at the frontend.
2. the root fragment has other non-exchange operators. I think we may still schedule them at the compute nodes.

Totally agree. By the way, the plan generated by optimizers should always satisfies

• Either, it has a singleton fragment (aka. it's ended in singleton exchange(s)) on the top e.g. in the cases of Global Agg/TopN
• Or, it has a singleton exchange on the top to gather data together

(My point is, if not so, please raise a bug :D)

[liurenjie1024]

For 2, there are two cases:

• 2.1 No table scan, for example select 1+ 2
• 2.2 Singleton table scan, for example select * from t where a = 1 and a is a key of t.

Obviously 2.1 can be scheduled on frontend, while 2.2 can't.

For 2.2 I would suggest to add a singleton exchange on top of table scan so that all three cases can be executed in same way, and we don't need to maintain another code branch for execution.

[fuyufjh]

For 2.2 I would suggest to add a singleton exchange on top of table scan so that all three cases can be executed in same way, and we don't need to maintain another code branch for execution.

+1, but we already had one.

dev=> explain select l_orderkey,l_linenumber from lineitem where l_orderkey = 10;
                               QUERY PLAN
------------------------------------------------------------------------
 BatchExchange { order: [], dist: Single }
   BatchFilter { predicate: ($0 = 10:Int32) }
     BatchScan { table: lineitem, columns: [l_orderkey, l_linenumber] }
(3 rows)