code: func (this *Dataset) CoGroupPartitionedSorted(name string, that *Dataset, indexes []int) (ret *Dataset) {
  	ret = this.Flow.NewNextDataset(len(this.Shards))
  	ret.IsPartitionedBy = indexes
  
  	inputs := []*Dataset{this, that}
  	step := this.Flow.MergeDatasets1ShardTo1Step(inputs, ret)
  	step.SetInstruction(name, instruction.NewCoGroupPartitionedSorted(indexes))
  	return ret
  }
docs: // CoGroupPartitionedSorted joins 2 datasets that are sharded
  // by the same key and already locally sorted within each shard.