Add SVS support (#23)

Author: mpozniak95

engine/clients/redis/search.py

@@ -37,7 +37,10 @@ def init_client(cls, host, distance, connection_params: dict, search_params: dic
             # 'EF_RUNTIME' is irrelevant for 'ADHOC_BF' policy
             if cls.hybrid_policy != "ADHOC_BF":
                 cls.knn_conditions = "EF_RUNTIME $EF"
-
+        elif cls.algorithm == "SVS":
+            cls.knn_conditions = "WS_SEARCH $WS_SEARCH"
+        elif cls.algorithm == "SVS_TIERED":
+            cls.knn_conditions = "WS_SEARCH $WS_SEARCH"
         cls.data_type = "FLOAT32"
         if "search_params" in cls.search_params:
             cls.data_type = (
@@ -95,6 +98,10 @@ def search_one(cls, vector, meta_conditions, top) -> List[Tuple[int, float]]:
             # 'EF_RUNTIME' is irrelevant for 'ADHOC_BF' policy
             if cls.hybrid_policy != "ADHOC_BF":
                 params_dict["EF"] = cls.search_params["search_params"]["ef"]
+        if cls.algorithm == "SVS":
+            params_dict["WS_SEARCH"] = cls.search_params["search_params"]["WS_SEARCH"]
+        if cls.algorithm == "SVS_TIERED":
+            params_dict["WS_SEARCH"] = cls.search_params["search_params"]["WS_SEARCH"]
         results = cls._ft.search(q, query_params=params_dict)
 
         return [(int(result.id), float(result.vector_score)) for result in results.docs]

engine/clients/redis/upload.py

@@ -95,21 +95,21 @@ def upload_batch(
 
     @classmethod
     def post_upload(cls, _distance):
-        if cls.algorithm != "HNSW" and cls.algorithm != "FLAT":
+        if cls.algorithm != "HNSW" and cls.algorithm != "FLAT" and cls.algorithm != "SVS" and cls.algorithm != "SVS_TIERED":
             print(f"TODO: FIXME!! Avoiding calling ft.info for {cls.algorithm}...")
             return {}
         index_info = cls.client.ft().info()
         # redisearch / memorystore for redis
-        if "percent_index" in index_info:
-            percent_index = float(index_info["percent_index"])
+        if "percent_indexed" in index_info:
+            percent_index = float(index_info["percent_indexed"])
             while percent_index < 1.0:
                 print(
                     "waiting for index to be fully processed. current percent index: {}".format(
                         percent_index * 100.0
                     )
                 )
                 time.sleep(1)
-                percent_index = float(cls.client.ft().info()["percent_index"])
+                percent_index = float(cls.client.ft().info()["percent_indexed"])
         # memorydb
         if "current_lag" in index_info:
             current_lag = float(index_info["current_lag"])
@@ -136,7 +136,7 @@ def get_memory_usage(cls):
             used_memory.append(used_memory_shard)
         index_info = {}
         device_info = {}
-        if cls.algorithm != "HNSW" and cls.algorithm != "FLAT":
+        if cls.algorithm != "HNSW" and cls.algorithm != "FLAT" and cls.algorithm != "SVS" and cls.algorithm != "SVS_TIERED":
             print(f"TODO: FIXME!! Avoiding calling ft.info for {cls.algorithm}...")
         else:
             index_info = cls.client_decode.ft().info()

experiments/configurations/create-svs.py

@@ -0,0 +1,51 @@
+import json
+
+threads = [16]
+ws_constructs = [100]
+ws_search = [32, 40, 48, 64]
+#ws_search = [48]
+graph_degree = [32]
+#quantization = ["0", "4x4", "4x8", "8", "4"]
+quantization = ["8"]
+topKs = [10]
+data_types = ["FLOAT32"]
+
+for algo in ["svs_tiered"]:
+    for data_type in data_types:
+        for ws_construct in ws_constructs:
+            for graph_d in graph_degree:
+                for quant in quantization:
+                    configs = []
+                    for thread in threads:
+                        config = {
+                            "name": f"svs-test-algo-{algo}-graph-{graph_d}-ws-con-{ws_construct}-quant-{quant}-threads-{thread}-dt-{data_type}",
+                            "engine": "redis",
+                            "connection_params": {},
+                            "collection_params": {
+                                "algorithm": algo,
+                                "data_type": data_type,
+                                f"{algo}_config": {"NUM_THREADS": thread, "GRAPH_DEGREE": graph_d, "WS_CONSTRUCTION": ws_construct, "QUANTIZATION": quant},
+                            },
+                            "search_params": [],
+                            "upload_params": {
+                                "parallel": 128,
+                                "data_type": data_type,
+                                "algorithm": algo,
+                            },
+                        }
+                        for client in [1, 8, 16, 32, 64, 128]:
+                            for ws_s in ws_search:
+                                for top in topKs:
+                                    test_config = {
+                                        "algorithm": algo,
+                                        "parallel": client,
+                                        "top": top,
+                                        "search_params": {"WS_SEARCH": ws_s, "data_type": data_type},
+                                    }
+                                    config["search_params"].append(test_config)
+                        configs.append(config)
+
+                    fname = f"svs-test-algo-{algo}-graph-{graph_d}-ws-con-{ws_construct}-quant-{quant}-threads-{thread}-dt-{data_type}.json"
+                    with open(fname, "w") as json_fd:
+                        json.dump(configs, json_fd, indent=2)
+                        print(f"Created {len(configs)} configs for {fname}.")