(seq dag.Seq, n int, sortExprs []dag.SortExpr, replicas int)
| 114 | } |
| 115 | |
| 116 | func parallelizeHead(seq dag.Seq, n int, sortExprs []dag.SortExpr, replicas int) dag.Seq { |
| 117 | head := seq[:n] |
| 118 | tail := seq[n:] |
| 119 | scatter := &dag.ScatterOp{ |
| 120 | Kind: "ScatterOp", |
| 121 | Paths: make([]dag.Seq, replicas), |
| 122 | } |
| 123 | for k := range replicas { |
| 124 | scatter.Paths[k] = dag.CopySeq(head) |
| 125 | } |
| 126 | var merge dag.Op |
| 127 | if len(sortExprs) > 0 { |
| 128 | // At this point, we always insert a merge as we don't know if the |
| 129 | // downstream DAG requires the sort order. A later step will look at |
| 130 | // the fanin from this parallel structure and see if the merge can be |
| 131 | // removed while also pushing additional ops from the output segment up into |
| 132 | // the parallel branches to enhance concurrency. |
| 133 | merge = &dag.MergeOp{Kind: "MergeOp", Exprs: sortExprs} |
| 134 | } else { |
| 135 | merge = &dag.CombineOp{Kind: "CombineOp"} |
| 136 | } |
| 137 | return append(dag.Seq{scatter, merge}, tail...) |
| 138 | } |
| 139 | |
| 140 | func (o *Optimizer) optimizeParallels(seq dag.Seq) { |
| 141 | Walk(seq, func(seq dag.Seq) dag.Seq { |
no test coverage detected