Method AddOneToAllStep

flow/context.go:80–94

the task should run on the source dataset shard input is nil for initial source dataset

(input *Dataset, output *Dataset)

// the task should run on the source dataset shard
// input is nil for initial source dataset
func (f *FlowContext) AddOneToAllStep(input *Dataset, output *Dataset) (step *Step) {
	step = f.NewStep()
	FromStepToDataset(step, output)
	FromDatasetToStep(input, step)

	// setup the network
	task := step.NewTask()
	if input != nil {
		FromDatasetShardToTask(input.GetShards()[0], task)
	}
	for _, shard := range output.GetShards() {
		FromTaskToDatasetShard(task, shard)
	}
	return
}

func (f *FlowContext) AddOneToEveryNStep(input *Dataset, n int, output *Dataset) (step *Step) {
	step = f.NewStep()