Merge pull request #86 from weaviate/iteration_id

Store iteration ID and make use of Hfresh term

Author: robbespo00

benchmarker/README.md

@@ -48,7 +48,7 @@ Flags:
   -h, --help                         help for ann-benchmark
       --httpOrigin string            The http origin for Weaviate (only used if grpc enabled) (default "localhost:8080")
       --httpScheme string            The http scheme (http or https) (default "http")
-      --indexType string             Index type (hnsw, flat, spfresh) (default "hnsw")
+      --indexType string             Index type (hnsw, flat, hfresh) (default "hnsw")
       --labels string                Labels of format key1=value1,key2=value2,...
   -l, --limit int                    Set the query limit / k (default 10) (default 10)
       --maxConnections int           Set Weaviate efConstruction parameter (default 16) (default 16)

benchmarker/cmd/ann_benchmark.go

@@ -64,6 +64,7 @@ type ResultsJSONBenchmark struct {
 	Limit            int     `json:"limit"`
 	ImportTime       float64 `json:"importTime"`
 	RunID            string  `json:"run_id"`
+	IterationRunID   string  `json:"iteration"`
 	Dataset          string  `json:"dataset_file"`
 	Recall           float64 `json:"recall"`
 	NDCG             float64 `json:"ndcg"`
@@ -713,7 +714,7 @@ func parseEfValues(s string) ([]int, error) {
 }
 
 func runQueries(cfg *Config, importTime time.Duration, testData [][]float32, neighbors [][]int, filters []int) {
-	baseRunID := strconv.FormatInt(time.Now().Unix(), 10)
+	runID := strconv.FormatInt(time.Now().Unix(), 10)
 
 	efCandidates, err := parseEfValues(cfg.EfArray)
 	if err != nil {
@@ -742,13 +743,9 @@ func runQueries(cfg *Config, importTime time.Duration, testData [][]float32, nei
 		}
 
 		iteration++
-		runID := fmt.Sprintf("%s-%d", baseRunID, iteration)
+		iterationRunID := fmt.Sprintf("%d", iteration)
 		isFinalIteration := !cfg.WaitForBackground || shouldStop
 
-		if isFinalIteration {
-			runID = fmt.Sprintf("%s-true", runID)
-		}
-
 		benchmarkResultsMap := make([]map[string]interface{}, 0, len(efCandidates))
 		for _, ef := range efCandidates {
 			updateEf(ef, cfg, client)
@@ -797,6 +794,7 @@ func runQueries(cfg *Config, importTime time.Duration, testData [][]float32, nei
 				Limit:            cfg.Limit,
 				ImportTime:       importTime.Seconds(),
 				RunID:            runID,
+				IterationRunID:   iterationRunID,
 				Dataset:          dataset,
 				NDCG:             result.NDCG,
 				Recall:           result.Recall,
@@ -862,9 +860,9 @@ func shouldStopRunQueries(iteration int, cfg *Config) bool {
 		return false
 	}
 
-	metrics, err := readSPFreshMetrics(cfg)
+	metrics, err := readHFreshMetrics(cfg)
 	if err != nil {
-		log.WithError(err).Warn("Failed to read SPFresh pending operations metrics")
+		log.WithError(err).Warn("Failed to read HFresh pending operations metrics")
 		return false
 	}
 
@@ -875,16 +873,16 @@ func shouldStopRunQueries(iteration int, cfg *Config) bool {
 	if noPendingOps {
 		log.WithFields(log.Fields{
 			"iteration": iteration,
-		}).Info("All SPFresh background operations complete")
+		}).Info("All HFresh background operations complete")
 		return true
 	}
 
 	secs := 30
 
 	for {
-		metrics, err := readSPFreshMetrics(cfg)
+		metrics, err := readHFreshMetrics(cfg)
 		if err != nil {
-			log.WithError(err).Warn("Failed to read SPFresh pending operations metrics")
+			log.WithError(err).Warn("Failed to read HFresh pending operations metrics")
 			return false
 		}
 
@@ -893,7 +891,7 @@ func shouldStopRunQueries(iteration int, cfg *Config) bool {
 			"pendingSplitOperations":    metrics.PendingSplitOperations,
 			"pendingMergeOperations":    metrics.PendingMergeOperations,
 			"pendingReassignOperations": metrics.PendingReassignOperations,
-		}).Info("SPFresh background operations still running, checking again in ", secs, " seconds")
+		}).Info("HFresh background operations still running, checking again in ", secs, " seconds")
 		noPendingOps := metrics.PendingSplitOperations == 0 &&
 			metrics.PendingMergeOperations == 0 &&
 			metrics.PendingReassignOperations == 0
@@ -1078,7 +1076,7 @@ func initAnnBenchmark() {
 	annBenchmarkCommand.PersistentFlags().StringVar(&globalConfig.EfArray,
 		"efArray", "16,24,32,48,64,96,128,256,512", "Array of ef parameters as comma separated list")
 	annBenchmarkCommand.PersistentFlags().StringVar(&globalConfig.IndexType,
-		"indexType", "hnsw", "Index type (hnsw, flat or spfresh)")
+		"indexType", "hnsw", "Index type (hnsw, flat or hfresh)")
 	annBenchmarkCommand.PersistentFlags().IntVar(&globalConfig.MaxConnections,
 		"maxConnections", 16, "Set Weaviate efConstruction parameter (default 16)")
 	annBenchmarkCommand.PersistentFlags().IntVar(&globalConfig.Shards,
@@ -1142,13 +1140,13 @@ func initAnnBenchmark() {
 	annBenchmarkCommand.PersistentFlags().StringVar(&globalConfig.Dataset,
 		"dataset", "", "Dataset name e.g. dbpedia-openai-ada002-1536-float32-angular-100k")
 	annBenchmarkCommand.PersistentFlags().IntVar(&globalConfig.MaxPostingSize,
-		"maxPostingSize", 0, "Max posting size for SPFresh index (default 0)")
+		"maxPostingSize", 0, "Max posting size for HFresh index (default 0)")
 	annBenchmarkCommand.PersistentFlags().IntVar(&globalConfig.MinPostingSize,
-		"minPostingSize", 10, "Min posting size for SPFresh index (default 10)")
+		"minPostingSize", 10, "Min posting size for HFresh index (default 10)")
 	annBenchmarkCommand.PersistentFlags().IntVar(&globalConfig.Replicas,
-		"replicas", 8, "Number of replicas for SPFresh index (default 8)")
+		"replicas", 8, "Number of replicas for HFresh index (default 8)")
 	annBenchmarkCommand.PersistentFlags().Float64Var(&globalConfig.RngFactor,
-		"rngFactor", 10.0, "RNG factor for SPFresh index (default 10.0)")
+		"rngFactor", 10.0, "RNG factor for HFresh index (default 10.0)")
 }
 
 func benchmarkANN(cfg Config, queries Queries, neighbors Neighbors, filters []int) Results {

benchmarker/cmd/metrics.go

@@ -58,13 +58,13 @@ func readMemoryMetrics(cfg *Config) (*Memstats, error) {
 	return &memstats, nil
 }
 
-type SPFreshPendingMetrics struct {
+type HFreshPendingMetrics struct {
 	PendingSplitOperations    int `json:"pending_split_operations"`
 	PendingMergeOperations    int `json:"pending_merge_operations"`
 	PendingReassignOperations int `json:"pending_reassign_operations"`
 }
 
-func readSPFreshMetrics(cfg *Config) (*SPFreshPendingMetrics, error) {
+func readHFreshMetrics(cfg *Config) (*HFreshPendingMetrics, error) {
 	prometheusURL := fmt.Sprintf("http://%s/metrics", strings.Replace(cfg.HttpOrigin, "8080", "2112", -1))
 	response, err := http.Get(prometheusURL)
 	if err != nil {
@@ -88,7 +88,7 @@ func readSPFreshMetrics(cfg *Config) (*SPFreshPendingMetrics, error) {
 		return nil, err
 	}
 
-	var spfreshMetrics SPFreshPendingMetrics
+	var hfreshMetrics HFreshPendingMetrics
 
 	if metric, ok := metrics["vector_index_pending_background_operations"]; ok {
 		for _, m := range metric.Metric {
@@ -105,16 +105,16 @@ func readSPFreshMetrics(cfg *Config) (*SPFreshPendingMetrics, error) {
 			value := int(m.GetGauge().GetValue())
 			switch op {
 			case "split":
-				spfreshMetrics.PendingSplitOperations = value
+				hfreshMetrics.PendingSplitOperations = value
 			case "merge":
-				spfreshMetrics.PendingMergeOperations = value
+				hfreshMetrics.PendingMergeOperations = value
 			case "reassign":
-				spfreshMetrics.PendingReassignOperations = value
+				hfreshMetrics.PendingReassignOperations = value
 			}
 		}
 	}
 
-	return &spfreshMetrics, nil
+	return &hfreshMetrics, nil
 }
 
 func waitTombstonesEmpty(cfg *Config) error {