Added chinese support (infiniflow#1108)

Added chinese support

### Type of change

- [x] New Feature (non-breaking change which adds functionality)
- [x] Refactoring

Author: yuzhichang

python/benchmark/legacy_benchmark/test_remote_benchmark_basic.py

@@ -12,7 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import functools
-import multiprocessing
 import threading
 import time
 import traceback
@@ -37,37 +36,22 @@ def wrapped_func(*args, **kwargs):
     return wrapped_func
 
 
-@trace_unhandled_exceptions
-def go():
-    print(1)
-    raise Exception()
-    print(2)
+def worker_thread(thread_id, num_iterations, some_function, ip='127.0.0.1', port=9090):
+    infinity_obj = infinity.connect(NetworkAddress(ip, port))
+    try:
+        for j in range(num_iterations):
+            some_function(infinity_obj, port, thread_id, j)
+    except Exception as e:
+        print(f"Exception: {e}")
+    finally:
+        infinity_obj.disconnect()
 
 
-def test():
-    p = multiprocessing.Pool(1)
-
-    p.apply_async(go)
-    p.close()
-    p.join()
-
-
-def worker_thread(process_id, thread_id, num_iterations, some_function, ip='0.0.0.0', port=9090):
-    for j in range(num_iterations):
-        infinity_obj = infinity.connect(NetworkAddress(ip, port))
-        try:
-            some_function(infinity_obj, port, process_id, thread_id, j)
-        except Exception as e:
-            print(f"Exception: {e}")
-        finally:
-            infinity_obj.disconnect()
-
-
-def worker_internal_connection(process_id, num_threads, num_iterations, some_function, ip=None, port=None):
+def worker_internal_connection(num_threads, num_iterations, some_function, ip=None, port=None):
     threads = []
     for j in range(num_threads):
         thread = threading.Thread(target=worker_thread, args=(
-            process_id, j, num_iterations, some_function, ip, port))
+            j, num_iterations, some_function, ip, port))
         threads.append(thread)
         thread.start()
 
@@ -76,38 +60,29 @@ def worker_internal_connection(process_id, num_threads, num_iterations, some_fun
         thread.join()
 
 
-def measure_time_internal(num_processes, num_threads, num_times, some_function, ip=None, port=None):
-    # Calculate how many iterations each process should do
-    num_iterations = num_times // num_processes // num_threads
+def measure_time_internal(num_threads, num_times, some_function, ip=None, port=None):
+    # Calculate how many iterations each thread should do
+    num_iterations = num_times // num_threads
 
     start_time = time.perf_counter()
-    processes = []
-    for i in range(num_processes):
-        process = multiprocessing.Process(target=worker_internal_connection,
-                                          args=(i, num_threads, num_iterations, some_function, ip, port))
-        processes.append(process)
-        process.start()
-
-    # Wait for all threads to finish
-    for process in processes:
-        process.join()
-
+    worker_internal_connection(
+        num_threads, num_iterations, some_function, ip, port)
     end_time = time.perf_counter()
 
     elapsed_time = end_time - start_time
     return elapsed_time
 
 
-def execute(some_functions: list, protocols: list, num_processes, num_threads, num_times) -> pd.DataFrame:
+def execute(some_functions: list, protocols: list, num_threads, num_times) -> pd.DataFrame:
     results = pd.DataFrame(
-        columns=['rpc name', 'function', 'qps', 'elapsed_time', 'average_latency', 'num_processes', 'num_threads',
+        columns=['rpc name', 'function', 'qps', 'elapsed_time', 'average_latency', 'num_threads',
                  'num_times'])
     print(f"\n")
 
     for (protocol, ip, port) in protocols:
         for some_function in some_functions:
             elapsed_time = measure_time_internal(
-                num_processes, num_threads, num_times, some_function, ip, port)
+                num_threads, num_times, some_function, ip, port)
             qps = num_times / elapsed_time  # queries per second
             avg_latency = (elapsed_time / num_times) * 1000  # in ms
 
@@ -116,7 +91,6 @@ def execute(some_functions: list, protocols: list, num_processes, num_threads, n
                                          qps,
                                          elapsed_time,
                                          avg_latency,
-                                         num_processes,
                                          num_threads,
                                          num_times]
 
@@ -129,117 +103,118 @@ class TestBenchmark:
 
     def test_measure_time(self):
         @trace_unhandled_exceptions
-        def create_database(infinity_obj, port, process_id, thread_id, num_iteration):
+        def create_database(infinity_obj, port, thread_id, num_iteration):
             res = infinity_obj.create_database(
-                f"my_database_{port}_{process_id}_{thread_id}_{num_iteration}")
+                f"my_database_{port}_{thread_id}_{num_iteration}")
             if res.error_code != ErrorCode.OK:
                 raise Exception(f"create_database failed: {res.error_msg}")
 
         @trace_unhandled_exceptions
-        def get_database(infinity_obj, port, process_id, thread_id, num_iteration):
+        def get_database(infinity_obj, port, thread_id, num_iteration):
             db_obj = infinity_obj.get_database(f"default_db")
             if db_obj is None:
                 raise Exception(f"get_database failed")
 
         @trace_unhandled_exceptions
-        def list_databases(infinity_obj, port, process_id, thread_id, num_iteration):
+        def list_databases(infinity_obj, port, thread_id, num_iteration):
             res = infinity_obj.list_databases()
             if res.error_code != ErrorCode.OK:
                 raise Exception(f"list_databases failed: {res.error_msg}")
 
         @trace_unhandled_exceptions
-        def drop_database(infinity_obj, port, process_id, thread_id, num_iteration):
+        def drop_database(infinity_obj, port, thread_id, num_iteration):
             res = infinity_obj.drop_database(
-                f"my_database_{port}_{process_id}_{thread_id}_{num_iteration}")
+                f"my_database_{port}_{thread_id}_{num_iteration}")
             if res.error_code != ErrorCode.OK:
                 raise Exception(f"drop_database failed: {res.error_msg}")
 
         @trace_unhandled_exceptions
-        def create_table(infinity_obj, port, process_id, thread_id, num_iteration):
+        def create_table(infinity_obj, port, thread_id, num_iteration):
             res = infinity_obj.get_database(f"default_db").create_table(
-                f"table_{port}_{process_id}_{thread_id}_{num_iteration}",
+                f"table_{port}_{thread_id}_{num_iteration}",
                 {"c1": {"type": "int", "constraints": ["primary key"]}, "c2": {"type": "float"}})
             if res.error_code != ErrorCode.OK:
                 raise Exception(f"create_table failed: {res.error_msg}")
 
         @trace_unhandled_exceptions
-        def insert_table(infinity_obj, port, process_id, thread_id, num_iteration):
+        def insert_table(infinity_obj, port, thread_id, num_iteration):
             res = (infinity_obj
                    .get_database(f"default_db")
-                   .get_table(f"table_{port}_{process_id}_{thread_id}_{num_iteration}")
+                   .get_table(f"table_{port}_{thread_id}_{num_iteration}")
                    .insert([{"c1": 1, "c2": 1.1}, {"c1": 2, "c2": 2.2}]))
             if res.error_code != ErrorCode.OK:
                 raise Exception(f"insert_table failed: {res.error_msg}")
 
         @trace_unhandled_exceptions
-        def list_tables(infinity_obj, port, process_id, thread_id, num_iteration):
+        def list_tables(infinity_obj, port, thread_id, num_iteration):
             (infinity_obj
              .get_database(f"default_db")
              .list_tables())
 
         @trace_unhandled_exceptions
-        def select_table(infinity_obj, port, process_id, thread_id, num_iteration):
+        def select_table(infinity_obj, port, thread_id, num_iteration):
             res = (infinity_obj
                    .get_database(f"default_db")
-                   .get_table(f"table_{port}_{process_id}_{thread_id}_{num_iteration}")
+                   .get_table(f"table_{port}_{thread_id}_{num_iteration}")
                    .query_builder()
                    .output(["*"])
                    .filter("c1 > 1").to_df())
             if res is None:
                 raise Exception(f"select_table failed: {res}")
 
         @trace_unhandled_exceptions
-        def drop_table(infinity_obj, port, process_id, thread_id, num_iteration):
+        def drop_table(infinity_obj, port, thread_id, num_iteration):
             res = (infinity_obj
                    .get_database(f"default_db")
-                   .drop_table(f"table_{port}_{process_id}_{thread_id}_{num_iteration}"))
+                   .drop_table(f"table_{port}_{thread_id}_{num_iteration}"))
             if res.error_code != ErrorCode.OK:
                 raise Exception(f"drop_table failed: {res.error_msg}")
 
         @trace_unhandled_exceptions
-        def create_index(infinity_obj, port, process_id, thread_id, num_iteration):
+        def create_index(infinity_obj, port, thread_id, num_iteration):
             res = (infinity_obj
                    .get_database(f"default_db")
-                   .get_table(f"table_{port}_{process_id}_{thread_id}_{num_iteration}")
+                   .get_table(f"table_{port}_{thread_id}_{num_iteration}")
                    .create_index("my_index", ["c1"], "IVF_FLAT", None))
             if res.error_code != ErrorCode.OK:
                 raise Exception(f"create_index failed: {res.error_msg}")
 
         @trace_unhandled_exceptions
-        def drop_index(infinity_obj, port, process_id, thread_id, num_iteration):
+        def drop_index(infinity_obj, port, thread_id, num_iteration):
             res = (infinity_obj
                    .get_database(f"default_db")
-                   .get_table(f"table_{port}_{process_id}_{thread_id}_{num_iteration}")
+                   .get_table(f"table_{port}_{thread_id}_{num_iteration}")
                    .drop_index("my_index"))
             if res.error_code != ErrorCode.OK:
                 raise Exception(f"drop_index failed: {res.error_msg}")
 
         ############################################
         # Using the tune
 
-        ip: str = '0.0.0.0'
-        thrift = ("Thrift", ip, 9090)
+        ip: str = '127.0.0.1'
+        thrift = ("Thrift", ip, 23817)
         thread_pool_thrift = ("Thread Pool Thrift", ip, 23817)
-        async_thrift = ("AsyncThrift", ip, 9070)
-        num_processes = 16
-        num_threads = 16
-        num_times = 16 * 16 * 10
+        async_thrift = ("AsyncThrift", ip, 23817)
+        num_threads = 1
+        num_times = 10
         protocols = [thread_pool_thrift]
 
-        database_functions = [create_database,
-                              get_database, list_databases, drop_database]
+        database_functions = [create_database]
 
         db_df = execute(database_functions, protocols,
-                        num_processes, num_threads, num_times)
+                        num_threads, num_times)
 
-        table_functions = [create_table, insert_table,
-                           select_table, list_tables, drop_table]
+        table_functions = []
         tbl_df = execute(table_functions, protocols,
-                         num_processes, num_threads, num_times)
+                         num_threads, num_times)
 
         # index_functions = []
-        # idx_df = execute(index_functions, protocols, num_processes, num_threads, num_times)
+        # idx_df = execute(index_functions, protocols, num_threads, num_times)
 
         df = pd.concat([db_df, tbl_df])
         print(df)
         df.to_excel(f"{datetime.now()}_benchmark.xlsx")
+
+
+if __name__ == "__main__":
+    TestBenchmark().test_measure_time()

python/hello_infinity.py

@@ -43,7 +43,7 @@ def main():
     print(res)
 
 
-def test():
+def test_english():
     try:
         infinity_obj = infinity.connect(REMOTE_HOST)
         db = infinity_obj.get_database("default_db")
@@ -105,6 +105,64 @@ def test():
     # print(qb_result)
 
 
+def test_chinese():
+    """
+    Checkout https://github.com/infiniflow/resource.git under /var/infinity. The jieba dict is
+    /var/infinity/resource/jieba/dict/jieba.dict.utf8
+    """
+    try:
+        infinity_obj = infinity.connect(REMOTE_HOST)
+        infinity_obj.create_database("default_db", ConflictType.Ignore)
+        db = infinity_obj.get_database("default_db")
+        # Drop my_table if it already exists
+        db.drop_table("my_table", ConflictType.Ignore)
+        # Create a table named "my_table"
+        table = db.create_table(
+            "my_table", {"num": "integer", "body": "varchar", "vec": "vector, 4, float"})
+        table.insert(
+            [{"num": 1, "body": "据Wccftech报道，苹果正在开发一种定制芯片，旨在为人工智能（AI）服务器提供动力。暂时还不清楚这款芯片的具体规格，以及具体的实现目标。传闻苹果已选择台积电（TSMC）的3nm制程节点来制造这款芯片，预计2025年下半年量产。如果按照量产时间和台积电的半导体工艺进度，那么对应的很可能是N3E工艺。", "vec": [1.0, 1.2, 0.8, 0.9]}])
+        table.insert(
+            [{"num": 2, "body": "两个月前有报道称，苹果已正式放弃了努力超过十年、投下海量资金的“泰坦计划（Project Titan）”电动车项目。苹果随后解散了大约2000人的开发团队，各人会被分配到其他地方，其中一个很重要的去处就是人工智能部门。有传言称，苹果已经将注意力转向生成式AI，希望能够为业务找到新的增长动力。", "vec": [4.0, 4.2, 4.3, 4.5]}])
+
+        # `create_index()` is required before match() or fusion()
+        res = table.create_index("my_index",
+                                 [index.IndexInfo("body",
+                                                  index.IndexType.FullText,
+                                                  [index.InitParameter("ANALYZER", "chinese")]),
+                                  ], ConflictType.Error)
+        # assert res.success
+
+        res = table.output(["num", "body"]).knn(
+            "vec", [3.0, 2.8, 2.7, 3.1], "float", "ip", 2).to_pl()
+
+        pds_df = pds.DataFrame(res)
+        json_data = pds_df.to_json()
+        print("------json-------")
+        print(json_data)
+
+        table_obj = db.get_table("my_table")
+        qb_result = table_obj.output(["num", "body"]).knn(
+            "vec", [3.0, 2.8, 2.7, 3.1], "float", "ip", 3).to_pl()
+        print("------tabular -------")
+        print("------knn-------")
+        print(qb_result)
+
+        qb_result1 = table_obj.match(
+            "body", "芯片", "topn=1").output(["num", "body"]).to_pl()
+        print("------match-------")
+        print(qb_result1)
+
+        qb_result2 = table_obj.output(["num", "body"]).knn(
+            "vec", [3.0, 2.8, 2.7, 3.1], "float", "ip", 3).match(
+            "body", "芯片", "topn=1").fusion('rrf').to_pl()
+        print("------knn+match-------")
+        print(qb_result2)
+
+    except Exception as e:
+        print(str(e))
+
+
 if __name__ == '__main__':
     main()
-    test()
+    test_english()
+    test_chinese()

src/common/analyzer/analyzer_pool.cpp

@@ -43,7 +43,14 @@ UniquePtr<Analyzer> AnalyzerPool::Get(const std::string_view &name) {
         case Str2Int(CHINESE.data()): {
             Analyzer *prototype = cache_[CHINESE].get();
             if (prototype == nullptr) {
-                String path = InfinityContext::instance().config()->resource_dict_path();
+                String path;
+                Config *config = InfinityContext::instance().config();
+                if (config == nullptr) {
+                    // InfinityContext has not been initialized.
+                    path = "/var/infinity/resource";
+                } else {
+                    path = config->resource_dict_path();
+                }
                 UniquePtr<ChineseAnalyzer> analyzer = MakeUnique<ChineseAnalyzer>(std::move(path));
                 if (!analyzer->Load()) {
                     return nullptr;

src/common/analyzer/chinese_analyzer.cpp

@@ -65,23 +65,23 @@ bool ChineseAnalyzer::Load() {
     fs::path stopwords_path(root / STOP_WORD_PATH);
 
     if (!fs::exists(dict_path)) {
-        LOG_INFO(fmt::format("Invalid jieba config {} dict for jieba_analyzer does not exist", dict_path.string()));
+        // LOG_INFO(fmt::format("Invalid jieba config {} dict for jieba_analyzer does not exist", dict_path.string()));
         return false;
     }
     if (!fs::exists(hmm_path)) {
-        LOG_INFO(fmt::format("Invalid jieba config {} hmm for jieba_analyzer does not exist", hmm_path.string()));
+        // LOG_INFO(fmt::format("Invalid jieba config {} hmm for jieba_analyzer does not exist", hmm_path.string()));
         return false;
     }
     if (!fs::exists(userdict_path)) {
-        LOG_INFO(fmt::format("Invalid jieba config {} user_dict for jieba_analyzer does not exist", userdict_path.string()));
+        // LOG_INFO(fmt::format("Invalid jieba config {} user_dict for jieba_analyzer does not exist", userdict_path.string()));
         return false;
     }
     if (!fs::exists(idf_path)) {
-        LOG_INFO(fmt::format("Invalid jieba config {} idf for jieba_analyzer does not exist", idf_path.string()));
+        // LOG_INFO(fmt::format("Invalid jieba config {} idf for jieba_analyzer does not exist", idf_path.string()));
         return false;
     }
     if (!fs::exists(stopwords_path)) {
-        LOG_INFO(fmt::format("Invalid jieba config {} stopword for jieba_analyzer does not exist", stopwords_path.string()));
+        // LOG_INFO(fmt::format("Invalid jieba config {} stopword for jieba_analyzer does not exist", stopwords_path.string()));
         return false;
     }