Merge pull request #13 from bgokden/refactor

Fix known bugs

Author: bgokden

README.md

@@ -15,7 +15,7 @@ Veri also supports creating sub sample spaces of data by default.
 
 Veri works as a cluster that can hold a Vector Space with fixed dimension and allows easy querying of k nearest neighbour search queries and also querying a sample space to be used in a machine learning algorithm.
 
-Veri is currently in Alpha Stage
+Veri is currently in Beta Stage
 
 *Veri means data in Turkish.*
 
@@ -25,7 +25,7 @@ In machine learning, data scientist usually convert data into a feature label ve
 
 I have worked in different roles as a Data Engineer, Data scientist and a Software Developer. In many projects, I wanted a scalable approach to vector space search which is not available. I wanted to optimise the data ingestion and data querying into one tool.
 
-Veri is meant to be scale. Each Veri instance tries to synchronise its data with other peers and keep a statistically identical subset of the general vector space.
+Veri is meant to scale. Each Veri instance tries to synchronise its data with other peers and keep a statistically identical subset of the general vector space.
 
 ## What does statistically identical mean?
 
@@ -34,11 +34,12 @@ Every instance continue, exchanging data as long as their average and histogram
 
 ## Knn querying
 
-Veri internally has a kd-tree, but it also queries its neighbours and merges the result. It is very similar to map-reduce process done on the fly without planning.
+Veri internally has an internal key-value store, but it also queries its neighbours and merges the result. 
+It is very similar to map-reduce process done on the fly without planning.
 
-When a knn query is stated, veri creates a unique id,
+When a knn query is stated, veri creates a unique hash,
 Starts a timer,
-Then do a local kd-tree search,
+Then do a local knn search locally,
 Then calls its peers to do the same with a smaller timeout,
 Merges results into a map,
 Waits for timeout and then do a refine process on the result map,
@@ -50,19 +51,14 @@ Every knn query has a timeout and timeout defines the precision of the result. U
 
 ## High Availability
 
-Veri has a different way of approaching high availability.
-Veri as a cluster try to use all the memory it is allowed to use.
-If there is enough memory, all the data is replicated to every instance.
-If there is not enough memory, data is split within instances using histogram balancing.
-If memory is nearly full, Veri will reject insertion requests.
-So if you want more high availability, use more instances.
-Currently, it is recommended to use another database for long term storage. Usually vector spaces, change over time and only the original data is kept. So I didn't implement a direct backend into it. Instead, you can regularly insert new data and evict old data. So you will keep your vector space up to date. Veri can respond queries while data being inserted or deleted, unlike most knn search systems.
+Veri replicates the data to its peers periodically and data is persisted to the disk for crahes.
 
 TODO:
-- Add Dump data function to allow machine learning algorithms to get a Sample Space.
-- Add Query Caching and Return Cached Result instead of rejecting result.
-- Add Internal classification endpoint.
+- Test multinode syncranization
 - Authentication.
 - Documentation.
 
+### Note:
+Veri uses [badger](https://github.com/dgraph-io/badger) internally. Many functions are made possible thanks to badger.
+
 Contact me for any questions: berkgokden@gmail.com

data/aggregator.go

@@ -6,6 +6,7 @@ import (
 
 	"github.com/bgokden/go-cache"
 	pb "github.com/bgokden/veri/veriservice"
+	"github.com/jinzhu/copier"
 )
 
 type AggregatorInterface interface {
@@ -48,7 +49,7 @@ func (a *Aggregator) IsNewScoredBetter(old, new float64) bool {
 }
 
 func (a *Aggregator) BestScore(scoredDatum *pb.ScoredDatum) float64 {
-	if a.Context != nil || len(a.Context.GetDatum()) > 0 {
+	if a.Context != nil && len(a.Context.GetDatum()) > 0 {
 		var isSet = false
 		var current float64
 		// When Context is crioritized search score is ignored.
@@ -102,7 +103,10 @@ func (a *Aggregator) Insert(scoredDatum *pb.ScoredDatum) error {
 			aGroupAggregator := aGroupAggregatorInterface.(AggregatorInterface)
 			return aGroupAggregator.Insert(scoredDatum)
 		} else {
-			aGroupAggregator := NewAggrator(a.Config, false, nil)
+			var aConfig pb.SearchConfig
+			copier.Copy(&aConfig, a.Config)
+			aConfig.Limit = a.Config.GroupLimit // TODO: find a better solution.
+			aGroupAggregator := NewAggrator(&aConfig, false, nil)
 			a.DeDuplicationMap.Set(keyString, aGroupAggregator, cache.NoExpiration)
 			return aGroupAggregator.Insert(scoredDatum)
 		}

data/data.go

@@ -66,7 +66,8 @@ func NewPreData(config *pb.DataConfig, dataPath string) *Data {
 func (dt *Data) InitData() error {
 	log.Printf("Init Data %v\n", dt.Config)
 	if dt.Initialized == false {
-		db, err := badger.Open(badger.DefaultOptions(dt.DBPath))
+		options := badger.DefaultOptions(dt.DBPath).WithKeepL0InMemory(true)
+		db, err := badger.Open(options)
 		if err != nil {
 			return err
 		}

data/search.go

@@ -198,18 +198,15 @@ func GetSearchKey(datum *pb.Datum, config *pb.SearchConfig) string {
 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)
-	// log.Printf("DatumKey: %v\n", datum.GetKey())
 	queryKey := GetSearchKey(datum, config)
 	if dt.QueryCache == nil {
 		dt.InitData()
 	}
 	if result, ok := dt.QueryCache.Get(queryKey); ok {
-		log.Printf("Found in cache\n")
 		cachedResult := result.([]*pb.ScoredDatum)
 		for _, i := range cachedResult {
 			scoredDatumStreamOutput <- i
 		}
-		log.Printf("Returned in cache\n")
 		if upperWaitGroup != nil {
 			upperWaitGroup.Done()
 		}
@@ -271,6 +268,7 @@ func (dt *Data) AggregatedSearch(datum *pb.Datum, scoredDatumStreamOutput chan<-
 	}
 	cacheDuration := time.Duration(config.CacheDuration) * time.Second
 	dt.QueryCache.Set(queryKey, result, cacheDuration)
+	dt.QueryCache.IncrementExpiration(queryKey, cacheDuration)
 	log.Printf("AggregatedSearch: finished. Cache Duration: %v\n", cacheDuration)
 	return nil
 }

go.mod

@@ -8,6 +8,7 @@ require (
 	github.com/dgraph-io/badger/v2 v2.2007.2
 	github.com/golang/protobuf v1.4.1
 	github.com/gorilla/mux v1.7.2
+	github.com/jinzhu/copier v0.0.0-20201025035756-632e723a6687
 	github.com/magneticio/go-common v0.0.1
 	github.com/mitchellh/go-homedir v1.1.0
 	github.com/pkg/errors v0.8.1

go.sum

@@ -141,6 +141,8 @@ github.com/imdario/mergo v0.3.7 h1:Y+UAYTZ7gDEuOfhxKWy+dvb5dRQ6rJjFSdX2HZY1/gI=
 github.com/imdario/mergo v0.3.7/go.mod h1:2EnlNZ0deacrJVfApfmtdGgDfMuh/nq6Ok1EcJh5FfA=
 github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
+github.com/jinzhu/copier v0.0.0-20201025035756-632e723a6687 h1:bWXum+xWafUxxJpcXnystwg5m3iVpPYtrGJFc1rjfLc=
+github.com/jinzhu/copier v0.0.0-20201025035756-632e723a6687/go.mod h1:24xnZezI2Yqac9J61UC6/dG/k76ttpq0DdJI3QmUvro=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=
 github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
 github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=