Fix stateful mergetree (#83)

Author: auxten

programs/local/LocalServer.cpp

@@ -694,7 +694,14 @@ void LocalServer::processConfig()
     // global once flag
     /// We load temporary database first, because projections need it.
     static std::once_flag db_catalog_once;
-    std::call_once(db_catalog_once, [&] { DatabaseCatalog::instance().initializeAndLoadTemporaryDatabase(); });
+    if (config().has("path"))
+    {
+        DatabaseCatalog::instance().initializeAndLoadTemporaryDatabase();
+    }
+    else
+    {
+        std::call_once(db_catalog_once, [&] { DatabaseCatalog::instance().initializeAndLoadTemporaryDatabase(); });
+    }
 
     /** Init dummy default DB
       * NOTE: We force using isolated default database to avoid conflicts with default database from server environment

src/Core/ServerSettings.h

@@ -71,16 +71,16 @@ namespace DB
     M(UInt64, concurrent_threads_soft_limit_num, 0, "Sets how many concurrent thread can be allocated before applying CPU pressure. Zero means Unlimited.", 0) \
     M(UInt64, concurrent_threads_soft_limit_ratio_to_cores, 0, "Same as concurrent_threads_soft_limit_num, but with ratio to cores.", 0) \
     \
-    M(UInt64, background_pool_size, 16, "The maximum number of threads what will be used for merging or mutating data parts for *MergeTree-engine tables in a background.", 0) \
+    M(UInt64, background_pool_size, 0, "The maximum number of threads what will be used for merging or mutating data parts for *MergeTree-engine tables in a background.", 0) \
     M(Float, background_merges_mutations_concurrency_ratio, 2, "The number of part mutation tasks that can be executed concurrently by each thread in background pool.", 0) \
     M(String, background_merges_mutations_scheduling_policy, "round_robin", "The policy on how to perform a scheduling for background merges and mutations. Possible values are: `round_robin` and `shortest_task_first`. ", 0) \
-    M(UInt64, background_move_pool_size, 8, "The maximum number of threads that will be used for moving data parts to another disk or volume for *MergeTree-engine tables in a background.", 0) \
-    M(UInt64, background_fetches_pool_size, 8, "The maximum number of threads that will be used for fetching data parts from another replica for *MergeTree-engine tables in a background.", 0) \
-    M(UInt64, background_common_pool_size, 8, "The maximum number of threads that will be used for performing a variety of operations (mostly garbage collection) for *MergeTree-engine tables in a background.", 0) \
-    M(UInt64, background_buffer_flush_schedule_pool_size, 16, "The maximum number of threads that will be used for performing flush operations for Buffer-engine tables in a background.", 0) \
+    M(UInt64, background_move_pool_size, 0, "The maximum number of threads that will be used for moving data parts to another disk or volume for *MergeTree-engine tables in a background.", 0) \
+    M(UInt64, background_fetches_pool_size, 0, "The maximum number of threads that will be used for fetching data parts from another replica for *MergeTree-engine tables in a background.", 0) \
+    M(UInt64, background_common_pool_size, 0, "The maximum number of threads that will be used for performing a variety of operations (mostly garbage collection) for *MergeTree-engine tables in a background.", 0) \
+    M(UInt64, background_buffer_flush_schedule_pool_size, 0, "The maximum number of threads that will be used for performing flush operations for Buffer-engine tables in a background.", 0) \
     M(UInt64, background_schedule_pool_size, 0, "The maximum number of threads that will be used for constantly executing some lightweight periodic operations.", 0) \
-    M(UInt64, background_message_broker_schedule_pool_size, 16, "The maximum number of threads that will be used for executing background operations for message streaming.", 0) \
-    M(UInt64, background_distributed_schedule_pool_size, 16, "The maximum number of threads that will be used for executing distributed sends.", 0) \
+    M(UInt64, background_message_broker_schedule_pool_size, 0, "The maximum number of threads that will be used for executing background operations for message streaming.", 0) \
+    M(UInt64, background_distributed_schedule_pool_size, 0, "The maximum number of threads that will be used for executing distributed sends.", 0) \
     M(Bool, display_secrets_in_show_and_select, false, "Allow showing secrets in SHOW and SELECT queries via a format setting and a grant", 0)
 
 

src/Storages/MergeTree/MergeTreeBackgroundExecutor.cpp

@@ -42,8 +42,8 @@ MergeTreeBackgroundExecutor<Queue>::MergeTreeBackgroundExecutor(
     , pool(std::make_unique<ThreadPool>(
           CurrentMetrics::MergeTreeBackgroundExecutorThreads, CurrentMetrics::MergeTreeBackgroundExecutorThreadsActive))
 {
-    if (max_tasks_count == 0)
-        throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Task count for MergeTreeBackgroundExecutor must not be zero");
+    // if (max_tasks_count == 0)
+    //     throw Exception(ErrorCodes::INVALID_CONFIG_PARAMETER, "Task count for MergeTreeBackgroundExecutor must not be zero");
 
     pending.setCapacity(max_tasks_count);
     active.set_capacity(max_tasks_count);

src/Storages/MergeTree/MergeTreeSettings.h

@@ -43,8 +43,8 @@ struct Settings;
     M(UInt64, max_replicated_merges_in_queue, 1000, "How many tasks of merging and mutating parts are allowed simultaneously in ReplicatedMergeTree queue.", 0) \
     M(UInt64, max_replicated_mutations_in_queue, 8, "How many tasks of mutating parts are allowed simultaneously in ReplicatedMergeTree queue.", 0) \
     M(UInt64, max_replicated_merges_with_ttl_in_queue, 1, "How many tasks of merging parts with TTL are allowed simultaneously in ReplicatedMergeTree queue.", 0) \
-    M(UInt64, number_of_free_entries_in_pool_to_lower_max_size_of_merge, 8, "When there is less than specified number of free entries in pool (or replicated queue), start to lower maximum size of merge to process (or to put in queue). This is to allow small merges to process - not filling the pool with long running merges.", 0) \
-    M(UInt64, number_of_free_entries_in_pool_to_execute_mutation, 20, "When there is less than specified number of free entries in pool, do not execute part mutations. This is to leave free threads for regular merges and avoid \"Too many parts\"", 0) \
+    M(UInt64, number_of_free_entries_in_pool_to_lower_max_size_of_merge, 0, "When there is less than specified number of free entries in pool (or replicated queue), start to lower maximum size of merge to process (or to put in queue). This is to allow small merges to process - not filling the pool with long running merges.", 0) \
+    M(UInt64, number_of_free_entries_in_pool_to_execute_mutation, 0, "When there is less than specified number of free entries in pool, do not execute part mutations. This is to leave free threads for regular merges and avoid \"Too many parts\"", 0) \
     M(UInt64, max_number_of_mutations_for_replica, 0, "Limit the number of part mutations per replica to the specified amount. Zero means no limit on the number of mutations per replica (the execution can still be constrained by other settings).", 0) \
     M(UInt64, max_number_of_merges_with_ttl_in_pool, 2, "When there is more than specified number of merges with TTL entries in pool, do not assign new merge with TTL. This is to leave free threads for regular merges and avoid \"Too many parts\"", 0) \
     M(Seconds, old_parts_lifetime, 8 * 60, "How many seconds to keep obsolete parts.", 0) \

tests/test_stateful.py

@@ -11,6 +11,7 @@
 current_process = psutil.Process()
 check_thread_count = False
 
+
 class TestStateful(unittest.TestCase):
     def setUp(self) -> None:
         shutil.rmtree(test_state_dir, ignore_errors=True)
@@ -41,7 +42,7 @@ def test_path(self):
         ret = sess.query("SELECT * FROM db_xxx.view_xxx", "CSV")
         self.assertEqual(str(ret), "1\n2\n")
 
-        del sess # name sess dir will not be deleted
+        del sess  # name sess dir will not be deleted
 
         sess = session.Session(test_state_dir)
         ret = sess.query("SELECT chdb_xxx()", "CSV")
@@ -64,6 +65,22 @@ def test_path(self):
         ret = sess2.query("SELECT chdb_xxx()", "CSV")
         self.assertEqual(str(ret), "")
 
+    def test_mergetree(self):
+        sess = session.Session()
+        sess.query(
+            "CREATE DATABASE IF NOT EXISTS db_xxx_merge ENGINE = Atomic;", "CSV"
+        )
+        sess.query(
+            "CREATE TABLE IF NOT EXISTS db_xxx_merge.log_table_xxx (x String, y Int) ENGINE = MergeTree ORDER BY x;"
+        )
+        # insert 1000000 random rows
+        sess.query(
+            "INSERT INTO db_xxx_merge.log_table_xxx (x, y) SELECT toString(rand()), rand() FROM numbers(1000000);"
+        )
+        sess.query("Optimize TABLE db_xxx_merge.log_table_xxx;")
+        ret = sess.query("SELECT count(*) FROM db_xxx_merge.log_table_xxx;")
+        self.assertEqual(str(ret), "1000000\n")
+
     def test_tmp(self):
         sess = session.Session()
         sess.query("CREATE FUNCTION chdb_xxx AS () -> '0.12.0'", "CSV")
@@ -83,6 +100,7 @@ def test_zfree_thread_count(self):
         if check_thread_count:
             self.assertEqual(thread_count, 1)
 
+
 if __name__ == "__main__":
     shutil.rmtree(test_state_dir, ignore_errors=True)
     check_thread_count = True