func (dt *Data) AggregatedSearch(datum *pb.Datum, scoredDatumStreamOutput chan<- *pb.ScoredDatum, upperWaitGroup *sync.WaitGroup, config *pb.SearchConfig) error {
	duration := time.Duration(config.Timeout) * time.Millisecond
	timeLimit := time.After(duration)
	queryKey := GetSearchKey(datum, config)
	config.Timeout = uint64(float64(config.Timeout) * 0.9)  
	if dt.QueryCache == nil {
		dt.InitData()
	}
	if config.CacheDuration > 0 {
		if result, ok := dt.QueryCache.Get(queryKey); ok {
			cachedResult := result.([]*pb.ScoredDatum)
			resultCopy := CloneResult(cachedResult)
			for _, i := range resultCopy {
				scoredDatumStreamOutput <- i
			}
			if upperWaitGroup != nil {
				upperWaitGroup.Done()
			}
			cacheDuration := time.Duration(config.CacheDuration) * time.Second
			dt.QueryCache.IncrementExpiration(queryKey, cacheDuration)
			return nil
		}
	}
	scoredDatumStream := make(chan *pb.ScoredDatum, 100)
	var queryWaitGroup sync.WaitGroup
	waitChannel := make(chan struct{})
	queryWaitGroup.Add(1)
	go func() {
		dt.StreamSearch(datum, scoredDatumStream, &queryWaitGroup, config)
	}()
	dt.RunOnRandomSources(5, func(source DataSource) error {
		queryWaitGroup.Add(1)
		go source.StreamSearch(datum, scoredDatumStream, &queryWaitGroup, config)
		return nil
	})
	go func() {
		defer close(waitChannel)
		queryWaitGroup.Wait()
	}()
	temp := NewAggrator(config, false, nil)
	dataAvailable := true
	for dataAvailable {
		select {
		case scoredDatum := <-scoredDatumStream:
			temp.Insert(scoredDatum)
		case <-waitChannel:
			close(scoredDatumStream)
			for scoredDatum := range scoredDatumStream {
				temp.Insert(scoredDatum)
			}
			dataAvailable = false
			break
		case <-timeLimit:
			log.Printf("timeout")
			dataAvailable = false
			break
		}
	}
	result := temp.Result()
	resultCopy := CloneResult(result)
	for _, i := range result {
		scoredDatumStreamOutput <- i
	}
	if upperWaitGroup != nil {
		upperWaitGroup.Done()
	}
	if config.CacheDuration > 0 {
		cacheDuration := time.Duration(config.CacheDuration) * time.Second
		dt.QueryCache.Set(queryKey, resultCopy, cacheDuration)
	}
	return nil
}
AggregatedSearch searches and merges other resources
package data
import (
	"encoding/json"
	"log"
	"sort"
	"strings"
	"sync"
	"time"
	"github.com/jinzhu/copier"
	"github.com/bgokden/veri/util"
	pb "github.com/bgokden/veri/veriservice"
	"github.com/tidwall/gjson"
)
func DefaultSearchConfig() *pb.SearchConfig {
	return &pb.SearchConfig{
		ScoreFuncName: "VectorDistance",
		HigherIsBetter: false,
		Limit:          10,
		Timeout:        1000,  
	}
}
func EncodeSearchConfig(sc *pb.SearchConfig) []byte {
	var config pb.SearchConfig
	copier.Copy(&config, sc)
	config.Uuid = ""
	marshalled, _ := json.Marshal(&config)
	return marshalled
}
type Collector struct {
	List           []*pb.ScoredDatum
	ScoreFunc      func(arr1 []float32, arr2 []float32) float64
	MaxScore       float64
	DatumKey       *pb.DatumKey
	N              uint32
	HigherIsBetter bool
	Filters        []string
	GroupFilters   []string
}
func (c *Collector) Insert(scoredDatum *pb.ScoredDatum) error {
	itemAdded := false
	if uint32(len(c.List)) < c.N {
		c.List = append(c.List, scoredDatum)
		itemAdded = true
	} else if (c.HigherIsBetter && scoredDatum.Score > c.List[len(c.List)-1].Score) ||
		(!c.HigherIsBetter && scoredDatum.Score < c.List[len(c.List)-1].Score) {
		c.List[len(c.List)-1] = scoredDatum
		itemAdded = true
	}
	if itemAdded {
		if c.HigherIsBetter {
			sort.Slice(c.List, func(i, j int) bool {
				return c.List[i].Score > c.List[j].Score
			})
		} else {
			sort.Slice(c.List, func(i, j int) bool {
				return c.List[i].Score < c.List[j].Score
			})
		}
	}
	return nil
}
func (c *Collector) PassesFilters(datum *pb.Datum) bool {
	if len(c.GroupFilters) > 0 {
		jsonLabel := string(datum.Key.GroupLabel)
		for _, filter := range c.GroupFilters {
			value := gjson.Get(jsonLabel, filter)
			if !value.Exists() {
				return false
			}
		}
	}
	if len(c.Filters) > 0 {
		jsonLabel := string(datum.Value.Label)
		for _, filter := range c.Filters {
			value := gjson.Get(jsonLabel, filter)
			if !value.Exists() {
				return false
			}
		}
	}
	return true
}
var vectorComparisonFuncs = map[string]func(arr1 []float32, arr2 []float32) float64{
	"AnnoyVectorDistance":   VectorDistance,
	"AnnoyCosineSimilarity": CosineSimilarity,
	"VectorDistance":        VectorDistance,
	"VectorMultiplication":  VectorMultiplication,
	"CosineSimilarity":      CosineSimilarity,
	"QuickVectorDistance":   QuickVectorDistance,
	"AngularDistance":       AngularDistance,
	"AnnoyAngularDistance":  AngularDistance,
}
func GetVectorComparisonFunction(name string) func(arr1 []float32, arr2 []float32) float64 {
	if function, ok := vectorComparisonFuncs[name]; ok {
		return function
	}
	return VectorDistance
}
func (dt *Data) StreamSearch(datum *pb.Datum, scoredDatumStream chan<- *pb.ScoredDatum, queryWaitGroup *sync.WaitGroup, config *pb.SearchConfig) error {
	var collector *Collector
	defer queryWaitGroup.Done()
	if config == nil {
		config = DefaultSearchConfig()
	}
	if strings.HasPrefix(config.ScoreFuncName, "Annoy") {
		collector = dt.SearchAnnoy(datum, config)
	} else {
		collector = dt.Search(datum, config, nil)
	}
	if collector != nil {
		for _, i := range collector.List {
			scoredDatumStream <- i
		}
	}
	return nil
}
func GetSearchKey(datum *pb.Datum, config *pb.SearchConfig) string {
	keyByte, err := GetKeyAsBytes(datum)
	signature := EncodeSearchConfig(config)
	if err != nil {
		return util.EncodeToString(signature)
	}
	return util.EncodeToString(append(keyByte, signature...))
}