code: func (d *Dataset) OnDisk(fn func(*Dataset) *Dataset) *Dataset {
  	ret := fn(d)
  
  	var parents, currents []*Dataset
  	currents = append(currents, ret)
  	for {
  		for _, t := range currents {
  			isFirstDataset, isLastDataset := false, false
  			if t == ret {
  				isLastDataset = true
  			} else if t.Step.OutputDataset != nil {
  				t.Step.OutputDataset.Meta.OnDisk = ModeOnDisk
  			}
  			for _, p := range t.Step.InputDatasets {
  				if p != d {
  					parents = append(parents, p)
  				} else {
  					isFirstDataset = true
  				}
  			}
  			if !isFirstDataset && !isLastDataset {
  				t.Step.Meta.IsRestartable = true
  			}
  		}
  		if len(parents) == 0 {
  			break
  		}
  		currents = parents
  		parents = nil
  	}
  
  	return ret
  }
docs: // OnDisk ensure the intermediate dataset are persisted to disk.
  // This allows executors to run not in parallel if executors are limited.