concurrentPath returns the largest path within seq from front to end that can be parallelized and run concurrently while preserving its semantics where the input to seq is known to have an order defined by sortKey (or order.Nil if unknown). The length of the concurrent path is returned and the sort
(seq dag.Seq, sortKeys order.SortKeys)
| 261 | // concurrent path is allowed to include operators that do not guarantee |
| 262 | // an output order. |
| 263 | func (o *Optimizer) concurrentPath(seq dag.Seq, sortKeys order.SortKeys) (length int, outputSortExprs []dag.SortExpr, orderRequired bool, err error) { |
| 264 | for k := range seq { |
| 265 | switch op := seq[k].(type) { |
| 266 | // This should be a boolean in op.go that defines whether |
| 267 | // function can be parallelized... need to think through |
| 268 | // what the meaning is here exactly. This is all still a bit |
| 269 | // of a heuristic. See #2660 and #2661. |
| 270 | case *dag.AggregateOp: |
| 271 | // We want input sorted when we are preserving order into |
| 272 | // aggregate so we can release values incrementally which is really |
| 273 | // important when doing a head on the aggregate results |
| 274 | if isKeyOfAggregate(op, sortKeys) { |
| 275 | // Keep the input ordered so we can incrementally release |
| 276 | // results from the aggregate as a streaming operation. |
| 277 | return k, sortExprsForSortKeys(sortKeys), true, nil |
| 278 | } |
| 279 | return k, nil, false, nil |
| 280 | case *dag.CountOp: |
| 281 | return k, nil, true, nil |
| 282 | case *dag.SortOp: |
| 283 | if len(op.Exprs) == 0 { |
| 284 | // No analysis for sort without expression since we can't |
| 285 | // parallelize the heuristic. We should revisit these semantics |
| 286 | // and define a global order across types. |
| 287 | return 0, nil, false, nil |
| 288 | } |
| 289 | return k, op.Exprs, false, nil |
| 290 | case *dag.TopOp: |
| 291 | if len(op.Exprs) == 0 { |
| 292 | // No analysis for top without expression since we can't |
| 293 | // parallelize the heuristic. |
| 294 | return 0, nil, false, nil |
| 295 | } |
| 296 | return k, op.Exprs, false, nil |
| 297 | case *dag.LoadOp: |
| 298 | // XXX At some point Load should have an optimization where if the |
| 299 | // upstream sort is the same as the Load destination sort we |
| 300 | // request a merge and set the Load operator to do a sorted write. |
| 301 | return k, nil, false, nil |
| 302 | case *dag.ForkOp, *dag.ScatterOp, *dag.HeadOp, *dag.TailOp, *dag.UniqOp, *dag.FuseOp, |
| 303 | *dag.HashJoinOp, *dag.InferOp, *dag.JoinOp, *dag.OutputOp: |
| 304 | return k, sortExprsForSortKeys(sortKeys), true, nil |
| 305 | default: |
| 306 | next, err := o.analyzeSortKeys(op, sortKeys) |
| 307 | if err != nil { |
| 308 | return 0, nil, false, err |
| 309 | } |
| 310 | if !sortKeys.IsNil() && next.IsNil() { |
| 311 | return k, sortExprsForSortKeys(sortKeys), true, nil |
| 312 | } |
| 313 | sortKeys = next |
| 314 | } |
| 315 | } |
| 316 | return len(seq), sortExprsForSortKeys(sortKeys), true, nil |
| 317 | } |
| 318 | |
| 319 | func sortExprsForSortKeys(keys order.SortKeys) []dag.SortExpr { |
| 320 | var exprs []dag.SortExpr |
no test coverage detected