(seq dag.Seq)
| 199 | } |
| 200 | |
| 201 | func (o *Optimizer) optimizeSourcePaths(seq dag.Seq) (dag.Seq, error) { |
| 202 | return walkEntries(seq, func(seq dag.Seq) (dag.Seq, error) { |
| 203 | if len(seq) == 0 { |
| 204 | return nil, errors.New("internal error: optimizer encountered empty sequential operator") |
| 205 | } |
| 206 | chain := seq[1:] |
| 207 | if len(chain) == 0 { |
| 208 | // Nothing to push down. |
| 209 | return seq, nil |
| 210 | } |
| 211 | o.propagateSortKey(seq, []order.SortKeys{nil}) |
| 212 | // See if we can lift a filtering predicate into the source op. |
| 213 | // Filter might be nil in which case we just put the chain back |
| 214 | // on the source op and zero out the source's filter. |
| 215 | filter, chain := matchFilter(chain) |
| 216 | switch op := seq[0].(type) { |
| 217 | case *dag.PoolScan: |
| 218 | o.nent++ |
| 219 | // Here we transform a PoolScan into a Lister followed by one or more chains |
| 220 | // of slicers and sequence scanners. We'll eventually choose other configurations |
| 221 | // here based on metadata and availability of CSUP. |
| 222 | lister := &dag.ListerScan{ |
| 223 | Kind: "ListerScan", |
| 224 | Pool: op.ID, |
| 225 | Commit: op.Commit, |
| 226 | } |
| 227 | // Check to see if we can add a range pruner when the pool key is used |
| 228 | // in a normal filtering operation. |
| 229 | sortKeys, err := o.sortKeysOfSource(op) |
| 230 | if err != nil { |
| 231 | return nil, err |
| 232 | } |
| 233 | lister.KeyPruner = maybeNewRangePruner(filter, sortKeys) |
| 234 | seq = dag.Seq{lister} |
| 235 | _, _, orderRequired, err := o.concurrentPath(chain, sortKeys) |
| 236 | if err != nil { |
| 237 | return nil, err |
| 238 | } |
| 239 | if orderRequired { |
| 240 | seq = append(seq, &dag.SlicerOp{Kind: "SlicerOp"}) |
| 241 | } |
| 242 | seq = append(seq, &dag.SeqScan{ |
| 243 | Kind: "SeqScan", |
| 244 | Pool: op.ID, |
| 245 | Commit: op.Commit, |
| 246 | Filter: filter, |
| 247 | KeyPruner: lister.KeyPruner, |
| 248 | }) |
| 249 | seq = append(seq, chain...) |
| 250 | case *dag.FileScan: |
| 251 | o.nent++ |
| 252 | if o.env.UseVAM() { |
| 253 | // Here, we install the filter without a projection. |
| 254 | // The demand pass comes subsequently and will add |
| 255 | // the projection. |
| 256 | op.Pushdown.MetaFilter = newMetaFilter(filter) |
| 257 | // Vector file readers don't support DataFilter pushdown yet so no need |
| 258 | // to install the filter here. But we will eventually and this is |
no test coverage detected