func MergeResultStreamsUnique(ctx context.Context, args QueryArgs, pkeyFields []string, vshardResults []*ResultStream, resultStream stream.ServerStream) {
	defer resultStream.Close()
	pkeyFieldParts := make([][]string, len(pkeyFields))
	for i, pkeyField := range pkeyFields {
		pkeyFieldParts[i] = strings.Split(pkeyField, ".")
	}
	getPkeyID := func(r record.Record) uint64 {
		pkeyFields := make([]interface{}, len(pkeyFieldParts))
		var ok bool
		for i, pkeyField := range pkeyFieldParts {
			pkeyFields[i], ok = r.Get(pkeyField)
			if !ok {
				panic("Missing pkey in response!!!")
			}
		}
		pkey, err := (sharding.HashMethod(sharding.SHA256).Get())(sharding.CombineKeys(pkeyFields))
		if err != nil {
			panic(fmt.Sprintf("MergeResult doesn't know how to hash pkey: %v", err))
		}
		return pkey
	}
	ids := make(map[uint64]struct{})
	offset := args.Offset
	if args.Sort != nil {
		for _, vshardResult := range vshardResults {
			defer vshardResult.Close()
			if vshardResult.Errors != nil && len(vshardResult.Errors) > 0 {
				resultStream.SendError(fmt.Errorf("errors in thing: %v", vshardResult.Errors))
				return
			}
			if vshardResult.Stream == nil {
				resultStream.SendError(fmt.Errorf("no stream in resultStream"))
				return
			}
		}
		if args.SortReverse == nil {
			sortReverseList := make([]bool, len(args.Sort))
			for i := range sortReverseList {
				sortReverseList[i] = false
			}
			args.SortReverse = sortReverseList
		}
		splitSortKeys := make([][]string, len(args.Sort))
		for i, sortKey := range args.Sort {
			splitSortKeys[i] = strings.Split(sortKey, ".")
		}
		h := record.NewRecordHeap(splitSortKeys, args.SortReverse)
		heap.Init(h)
		for i, vshardResult := range vshardResults {
			v, err := vshardResult.Recv()
			if err == io.EOF {
				continue
			}
			if err != nil {
				resultStream.SendError(err)
				return
			}
			item := record.RecordItem{
				Record: v,
				Source: i,
			}
			h.PushDirect(item)
		}
		resultsSent := uint64(0)
		for h.Len() > 0 {
			item := h.PopDirect()
			pkeyID := getPkeyID(item.Record)
			if _, ok := ids[pkeyID]; !ok {
				ids[pkeyID] = struct{}{}
				if offset > 0 {
					offset--
				} else {
					if err := resultStream.SendResult(item.Record); err != nil {
						resultStream.SendError(err)
						return
					}
					resultsSent++
					if args.Limit > 0 && resultsSent >= args.Limit {
						return
					}
				}
			}
			v, err := vshardResults[item.Source].Recv()
			if err == io.EOF {
				continue
			}
			if err != nil {
				resultStream.SendError(err)
				return
			}
			newItem := record.RecordItem{
				Record: v,
				Source: item.Source,
			}
			h.PushDirect(newItem)
		}
	} else {
		resultsSent := uint64(0)
		for _, vshardResult := range vshardResults {
			defer vshardResult.Close()
			if vshardResult.Errors != nil && len(vshardResult.Errors) > 0 {
				resultStream.SendError(fmt.Errorf("errors in thing: %v", vshardResult.Errors))
				return
			}
			if vshardResult.Stream == nil {
				resultStream.SendError(fmt.Errorf("no stream in resultStream"))
				return
			}
		}
		s := mergeStreams(ctx, vshardResults)
		for item := range s {
			if item.err != nil {
				resultStream.SendError(item.err)
				return
			} else {
				pkeyID := getPkeyID(item.item)
				if _, ok := ids[pkeyID]; !ok {
					ids[pkeyID] = struct{}{}
					if offset > 0 {
						offset--
					} else {
						if err := resultStream.SendResult(item.item); err != nil {
							resultStream.SendError(err)
							return
						}
						resultsSent++
						if args.Limit > 0 && resultsSent >= args.Limit {
							return
						}
					}
				}
			}
		}
	}
}
MergeResultStreams is responsible to (1) merge result streams uniquely based on pkey and (2) maintain sort order
(if sorted) from the streams (each stream is assumed to be in-order already)
package query
import (
	"container/heap"
	"context"
	"fmt"
	"io"
	"strings"
	"sync"
	"github.com/jacksontj/dataman/record"
	"github.com/jacksontj/dataman/routernode/sharding"
	"github.com/jacksontj/dataman/stream"
)
type ResultStreamItemTransformation func(record.Record) (record.Record, error)
type ResultStream struct {
	Stream stream.ClientStream `json:"-"`
	Errors []string `json:"errors,omitempty"`
	Meta map[string]interface{} `json:"meta,omitempty"`
	started bool
	transformations []ResultStreamItemTransformation
}
func (r *ResultStream) Err() error {
	if r.Errors == nil {
		return nil
	} else {
		return fmt.Errorf(strings.Join(r.Errors, "\n"))
	}
}
func (r *ResultStream) AddTransformation(t ResultStreamItemTransformation) error {
	if r.started {
		return fmt.Errorf("cannot add transformation after stream has started consuming")
	}
	if r.transformations == nil {
		r.transformations = []ResultStreamItemTransformation{t}
	} else {
		r.transformations = append(r.transformations, t)
	}
	return nil
}
func (r *ResultStream) Recv() (record.Record, error) {
	result, err := r.Stream.Recv()
	if err != nil {
		return nil, err
	}
	r.started = true
	for _, t := range r.transformations {
		if transformedRecord, err := t(result); err != nil {
			return result, err
		} else if transformedRecord != nil {
			result = transformedRecord
		}
	}
	return result, nil
}
func (r *ResultStream) Close() error {
	if r.Stream != nil {
		return r.Stream.Close()
	}
	return nil
}
type streamItem struct {
	item record.Record
	err  error
}
func mergeStreams(ctx context.Context, streams []*ResultStream) chan *streamItem {
	c := make(chan *streamItem, 1000)
	wg := &sync.WaitGroup{}
	for _, stream := range streams {
		wg.Add(1)
		go func(stream *ResultStream) {
			defer wg.Done()
			for {
				v, err := stream.Recv()
				if err == io.EOF {
					return
				}
				select {
				case c <- &streamItem{v, err}:
				case <-ctx.Done():
					select {
					case c <- &streamItem{err: ctx.Err()}:
					default:
					}
					return
				}
			}
		}(stream)
	}
	go func() {
		wg.Wait()
		close(c)
	}()
	return c
}