Method LocalSort

flow/dataset_sort.go:16–44

f(V, V) bool : less than function New Dataset contains K,V

(f interface{})

// f(V, V) bool : less than function
// New Dataset contains K,V
func (d *Dataset) LocalSort(f interface{}) *Dataset {
	if f == nil && d.IsKeyLocalSorted {
		return d
	}
	ret, step := add1ShardTo1Step(d, d.Type)
	ret.IsKeyPartitioned = d.IsKeyPartitioned
	if f == nil {
		ret.IsKeyLocalSorted = true
	}
	step.Name = "LocalSort"
	step.Function = func(task *Task) {
		outChan := task.Outputs[0].WriteChan
		var kvs []interface{}
		for input := range task.InputChan() {
			kvs = append(kvs, input.Interface())
		}
		if len(kvs) == 0 {
			return
		}
		comparator := getLessThanComparator(d.Type, reflect.ValueOf(kvs[0]), f)
		timsort.Sort(kvs, comparator)

		for _, kv := range kvs {
			outChan.Send(reflect.ValueOf(kv))
			// println(task.Name(), "sent kv index:", i)
		}
	}
	return ret
}

func (d *Dataset) MergeSorted(f interface{}) (ret *Dataset) {
	ret = d.context.newNextDataset(1, d.Type)