Optimize transforms the DAG by attempting to lift stateless operators from the downstream sequence into the trunk of each data source in the From operator at the entry point of the DAG. Once these paths are lifted, it also attempts to move any candidate filtering operations into the source's pushdo
(main *dag.Main)
| 120 | // source's pushdown predicate. This should be called before ParallelizeScan(). |
| 121 | // TBD: we need to do pushdown for search/cut to optimize columnar extraction. |
| 122 | func (o *Optimizer) Optimize(main *dag.Main) error { |
| 123 | seq := main.Body |
| 124 | seq = liftFilterOps(seq) |
| 125 | seq = mergeFilters(seq) |
| 126 | seq = mergeValuesOps(seq) |
| 127 | inlineRecordExprSpreads(seq) |
| 128 | seq = liftFiltersIntoJoins(seq) |
| 129 | replaceJoinWithHashJoin(seq) |
| 130 | seq = joinFilterPullup(seq) |
| 131 | seq = removePassOps(seq) |
| 132 | seq = replaceSortAndHeadOrTailWithTop(seq) |
| 133 | o.optimizeParallels(seq) |
| 134 | seq = mergeFilters(seq) |
| 135 | seq, err := o.optimizeSourcePaths(seq) |
| 136 | if err != nil { |
| 137 | return err |
| 138 | } |
| 139 | seq = removePassOps(seq) |
| 140 | DemandForSeq(seq, demand.All()) |
| 141 | setPushdownUnordered(seq, false) |
| 142 | main.Body = seq |
| 143 | return nil |
| 144 | } |
| 145 | |
| 146 | func (o *Optimizer) OptimizeDeleter(main *dag.Main, replicas int) error { |
| 147 | seq := main.Body |
no test coverage detected