(seq dag.Seq)
| 8 | ) |
| 9 | |
| 10 | func liftFiltersIntoJoins(seq dag.Seq) dag.Seq { |
| 11 | var filter *dag.FilterOp |
| 12 | var i int |
| 13 | for i = range len(seq) - 3 { |
| 14 | _, isfork := seq[i].(*dag.ForkOp) |
| 15 | _, isjoin := seq[i+1].(*dag.JoinOp) |
| 16 | var isfilter bool |
| 17 | filter, isfilter = seq[i+2].(*dag.FilterOp) |
| 18 | if isfork && isjoin && isfilter { |
| 19 | break |
| 20 | } |
| 21 | } |
| 22 | if filter == nil { |
| 23 | return seq |
| 24 | } |
| 25 | in := splitPredicate(filter.Expr) |
| 26 | var exprs []dag.Expr |
| 27 | for _, e := range in { |
| 28 | if b, ok := e.(*dag.BinaryExpr); ok && b.Op == "==" && liftFilterIntoJoin(seq[i:], b.LHS, b.RHS) { |
| 29 | continue |
| 30 | } |
| 31 | exprs = append(exprs, e) |
| 32 | } |
| 33 | if len(exprs) == 0 { |
| 34 | seq.Delete(i+2, i+3) |
| 35 | } else if len(exprs) != len(in) { |
| 36 | seq[i+2] = dag.NewFilterOp(buildConjunction(exprs)) |
| 37 | } |
| 38 | return seq |
| 39 | } |
| 40 | |
| 41 | func liftFilterIntoJoin(seq dag.Seq, lhs, rhs dag.Expr) bool { |
| 42 | fork, isfork := seq[0].(*dag.ForkOp) |
no test coverage detected