(seq dag.Seq)
| 481 | } |
| 482 | |
| 483 | func joinFilterPullup(seq dag.Seq) dag.Seq { |
| 484 | seq = mergeFilters(seq) |
| 485 | for i := 0; i <= len(seq)-3; i++ { |
| 486 | fork, isfork := seq[i].(*dag.ForkOp) |
| 487 | leftAlias, rightAlias, isjoin := isJoin(seq[i+1]) |
| 488 | filter, isfilter := seq[i+2].(*dag.FilterOp) |
| 489 | if !isfork || !isjoin || !isfilter { |
| 490 | continue |
| 491 | } |
| 492 | if len(fork.Paths) != 2 { |
| 493 | panic(seq[i]) |
| 494 | } |
| 495 | var remaining []dag.Expr |
| 496 | for _, e := range splitPredicate(filter.Expr) { |
| 497 | if pullup, ok := pullupExpr(leftAlias, e); ok { |
| 498 | fork.Paths[0] = append(fork.Paths[0], dag.NewFilterOp(pullup)) |
| 499 | continue |
| 500 | } |
| 501 | if pullup, ok := pullupExpr(rightAlias, e); ok { |
| 502 | fork.Paths[1] = append(fork.Paths[1], dag.NewFilterOp(pullup)) |
| 503 | continue |
| 504 | } |
| 505 | remaining = append(remaining, e) |
| 506 | } |
| 507 | if len(remaining) == 0 { |
| 508 | // Filter has been fully pulled up and can be removed. |
| 509 | seq.Delete(i+2, i+3) |
| 510 | } else { |
| 511 | seq[i+2] = dag.NewFilterOp(buildConjunction(remaining)) |
| 512 | } |
| 513 | fork.Paths[0] = joinFilterPullup(fork.Paths[0]) |
| 514 | fork.Paths[1] = joinFilterPullup(fork.Paths[1]) |
| 515 | } |
| 516 | return seq |
| 517 | } |
| 518 | |
| 519 | func isJoin(op dag.Op) (string, string, bool) { |
| 520 | switch op := op.(type) { |
no test coverage detected