Method OptimizeDeleter

compiler/optimizer/optimizer.go:146–199

(main *dag.Main, replicas int)

}

func (o *Optimizer) OptimizeDeleter(main *dag.Main, replicas int) error {
	seq := main.Body
	if len(seq) != 3 {
		return errors.New("internal error: bad deleter structure")
	}
	scan, ok := seq[0].(*dag.DeleteScan)
	if !ok {
		return errors.New("internal error: bad deleter structure")
	}
	filter, ok := seq[1].(*dag.FilterOp)
	if !ok {
		return errors.New("internal error: bad deleter structure")
	}
	output, ok := seq[2].(*dag.OutputOp)
	if !ok {
		return errors.New("internal error: bad deleter structure")
	}
	lister := &dag.ListerScan{
		Kind:   "ListerScan",
		Pool:   scan.ID,
		Commit: scan.Commit,
	}
	sortKeys, err := o.sortKeysOfSource(lister)
	if err != nil {
		return err
	}
	deleter := &dag.DeleterScan{
		Kind:  "DeleterScan",
		Pool:  scan.ID,
		Where: filter.Expr,
		//XXX KeyPruner?
	}
	lister.KeyPruner = maybeNewRangePruner(filter.Expr, sortKeys)
	scatter := &dag.ScatterOp{Kind: "ScatterOp"}
	for range replicas {
		scatter.Paths = append(scatter.Paths, dag.CopySeq(dag.Seq{deleter}))
	}
	var merge dag.Op
	if sortKeys.IsNil() {
		merge = &dag.CombineOp{Kind: "CombineOp"}
	} else {
		sortKey := sortKeys.Primary()
		merge = &dag.MergeOp{
			Kind: "MergeOp",
			Exprs: []dag.SortExpr{{
				Key:   dag.NewThis(sortKey.Key),
				Order: sortKey.Order,
				Nulls: sortKey.Order.NullsMax(true),
			}},
		}
	}
	main.Body = dag.Seq{lister, scatter, merge, output}
	return nil
}

func (o *Optimizer) optimizeSourcePaths(seq dag.Seq) (dag.Seq, error) {
	return walkEntries(seq, func(seq dag.Seq) (dag.Seq, error) {