put the background threads, kill operation and system table behind server experimental setting

Author: arthurpassos

src/Core/ServerSettings.cpp

@@ -1144,6 +1144,7 @@ The policy on how to perform a scheduling of CPU slots specified by `concurrent_
     DECLARE(UInt64, object_storage_list_objects_cache_max_entries, 1000, "Maximum size of ObjectStorage list objects cache in entries. Zero means disabled.", 0)                      \
     DECLARE(UInt64, object_storage_list_objects_cache_ttl, 3600, "Time to live of records in ObjectStorage list objects cache in seconds. Zero means unlimited", 0)                   \
     DECLARE(UInt64, input_format_parquet_metadata_cache_max_size, 500000000, "Maximum size of parquet file metadata cache", 0) \
+    DECLARE(Bool, enable_experimental_export_merge_tree_partition_feature, false, "Enable export replicated merge tree partition feature. It is experimental and not yet ready for production use.", 0) \
 // clang-format on
 
 /// If you add a setting which can be updated at runtime, please update 'changeable_settings' map in dumpToSystemServerSettingsColumns below

src/Interpreters/InterpreterKillQueryQuery.cpp

@@ -19,6 +19,7 @@
 #include <DataTypes/DataTypeString.h>
 #include <Columns/ColumnsNumber.h>
 #include <DataTypes/DataTypesNumber.h>
+#include <Core/ServerSettings.h>
 #include <Processors/Sources/SourceFromSingleChunk.h>
 #include <Processors/ISource.h>
 #include <Processors/Executors/PullingPipelineExecutor.h>
@@ -37,11 +38,17 @@ namespace Setting
     extern const SettingsUInt64 max_parser_depth;
 }
 
+namespace ServerSetting
+{
+    extern const ServerSettingsBool enable_experimental_export_merge_tree_partition_feature;
+}
+
 namespace ErrorCodes
 {
     extern const int LOGICAL_ERROR;
     extern const int ACCESS_DENIED;
     extern const int NOT_IMPLEMENTED;
+    extern const int SUPPORT_IS_DISABLED;
 }
 
 
@@ -252,6 +259,12 @@ BlockIO InterpreterKillQueryQuery::execute()
     }
     case ASTKillQueryQuery::Type::ExportPartition:
     {
+        if (!getContext()->getServerSettings()[ServerSetting::enable_experimental_export_merge_tree_partition_feature])
+        {
+            throw Exception(ErrorCodes::SUPPORT_IS_DISABLED,
+                "Exporting merge tree partition is experimental. Set the server setting `enable_experimental_export_merge_tree_partition_feature` to enable it");
+        }
+
         Block exports_block = getSelectResult(
             "source_database, source_table, transaction_id, destination_database, destination_table, partition_id",
             "system.replicated_partition_exports");

src/Storages/MergeTree/ReplicatedMergeTreeRestartingThread.cpp

@@ -11,6 +11,7 @@
 #include <Core/BackgroundSchedulePool.h>
 #include <Core/ServerUUID.h>
 #include <boost/algorithm/string/replace.hpp>
+#include <Core/ServerSettings.h>
 
 
 namespace CurrentMetrics
@@ -27,6 +28,11 @@ namespace MergeTreeSetting
     extern const MergeTreeSettingsSeconds zookeeper_session_expiration_check_period;
 }
 
+namespace ServerSetting
+{
+    extern const ServerSettingsBool enable_experimental_export_merge_tree_partition_feature;
+}
+
 namespace ErrorCodes
 {
     extern const int REPLICA_IS_ALREADY_ACTIVE;
@@ -171,12 +177,18 @@ bool ReplicatedMergeTreeRestartingThread::runImpl()
     storage.mutations_updating_task->activateAndSchedule();
     storage.mutations_finalizing_task->activateAndSchedule();
     storage.merge_selecting_task->activateAndSchedule();
+
+    if (storage.getContext()->getServerSettings()[ServerSetting::enable_experimental_export_merge_tree_partition_feature])
+    {
+        storage.export_merge_tree_partition_updating_task->activateAndSchedule();
+        storage.export_merge_tree_partition_select_task->activateAndSchedule();
+        storage.export_merge_tree_partition_status_handling_task->activateAndSchedule();
+    }
+
     storage.cleanup_thread.start();
     storage.async_block_ids_cache.start();
     storage.part_check_thread.start();
-    storage.export_merge_tree_partition_updating_task->activateAndSchedule();
-    storage.export_merge_tree_partition_select_task->activateAndSchedule();
-    storage.export_merge_tree_partition_status_handling_task->activateAndSchedule();
+
 
     LOG_DEBUG(log, "Table started successfully");
     return true;

src/Storages/StorageReplicatedMergeTree.cpp

@@ -304,6 +304,11 @@ namespace ErrorCodes
     extern const int INVALID_SETTING_VALUE;
 }
 
+namespace ServerSetting
+{
+    extern const ServerSettingsBool enable_experimental_export_merge_tree_partition_feature;
+}
+
 namespace ActionLocks
 {
     extern const StorageActionBlockType PartsMerge;
@@ -480,26 +485,30 @@ StorageReplicatedMergeTree::StorageReplicatedMergeTree(
     /// Will be activated by restarting thread.
     mutations_finalizing_task->deactivate();
 
-    export_merge_tree_partition_manifest_updater = std::make_shared<ExportPartitionManifestUpdatingTask>(*this);
+    if (getContext()->getServerSettings()[ServerSetting::enable_experimental_export_merge_tree_partition_feature])
+    {
+        export_merge_tree_partition_manifest_updater = std::make_shared<ExportPartitionManifestUpdatingTask>(*this);
+
+        export_merge_tree_partition_task_scheduler = std::make_shared<ExportPartitionTaskScheduler>(*this);
 
-    export_merge_tree_partition_task_scheduler = std::make_shared<ExportPartitionTaskScheduler>(*this);
+        export_merge_tree_partition_updating_task = getContext()->getSchedulePool().createTask(
+            getStorageID().getFullTableName() + " (StorageReplicatedMergeTree::export_merge_tree_partition_updating_task)", [this] { exportMergeTreePartitionUpdatingTask(); });
 
-    export_merge_tree_partition_updating_task = getContext()->getSchedulePool().createTask(
-        getStorageID().getFullTableName() + " (StorageReplicatedMergeTree::export_merge_tree_partition_updating_task)", [this] { exportMergeTreePartitionUpdatingTask(); });
+        export_merge_tree_partition_updating_task->deactivate();
 
-    export_merge_tree_partition_updating_task->deactivate();
+        export_merge_tree_partition_status_handling_task = getContext()->getSchedulePool().createTask(
+            getStorageID().getFullTableName() + " (StorageReplicatedMergeTree::export_merge_tree_partition_status_handling_task)", [this] { exportMergeTreePartitionStatusHandlingTask(); });
 
-    export_merge_tree_partition_status_handling_task = getContext()->getSchedulePool().createTask(
-        getStorageID().getFullTableName() + " (StorageReplicatedMergeTree::export_merge_tree_partition_status_handling_task)", [this] { exportMergeTreePartitionStatusHandlingTask(); });
+        export_merge_tree_partition_status_handling_task->deactivate();
 
-    export_merge_tree_partition_status_handling_task->deactivate();
+        export_merge_tree_partition_watch_callback = std::make_shared<Coordination::WatchCallback>(export_merge_tree_partition_updating_task->getWatchCallback());
 
-    export_merge_tree_partition_watch_callback = std::make_shared<Coordination::WatchCallback>(export_merge_tree_partition_updating_task->getWatchCallback());
-        
-    export_merge_tree_partition_select_task = getContext()->getSchedulePool().createTask(
-        getStorageID().getFullTableName() + " (StorageReplicatedMergeTree::export_merge_tree_partition_select_task)", [this] { selectPartsToExport(); });
+        export_merge_tree_partition_select_task = getContext()->getSchedulePool().createTask(
+            getStorageID().getFullTableName() + " (StorageReplicatedMergeTree::export_merge_tree_partition_select_task)", [this] { selectPartsToExport(); });
+
+        export_merge_tree_partition_select_task->deactivate();
+    }
 
-    export_merge_tree_partition_select_task->deactivate();
 
     bool has_zookeeper = getContext()->hasZooKeeper() || getContext()->hasAuxiliaryZooKeeper(zookeeper_info.zookeeper_name);
     if (has_zookeeper)
@@ -5923,9 +5932,13 @@ void StorageReplicatedMergeTree::partialShutdown()
     queue_updating_task->deactivate();
     mutations_updating_task->deactivate();
     mutations_finalizing_task->deactivate();
-    export_merge_tree_partition_updating_task->deactivate();
-    export_merge_tree_partition_select_task->deactivate();
-    export_merge_tree_partition_status_handling_task->deactivate();
+
+    if (getContext()->getServerSettings()[ServerSetting::enable_experimental_export_merge_tree_partition_feature])
+    {
+        export_merge_tree_partition_updating_task->deactivate();
+        export_merge_tree_partition_select_task->deactivate();
+        export_merge_tree_partition_status_handling_task->deactivate();
+    }
 
     cleanup_thread.stop();
     async_block_ids_cache.stop();
@@ -8073,10 +8086,10 @@ void StorageReplicatedMergeTree::fetchPartition(
 
 void StorageReplicatedMergeTree::exportPartitionToTable(const PartitionCommand & command, ContextPtr query_context)
 {
-    if (!query_context->getSettingsRef()[Setting::allow_experimental_export_merge_tree_part])
+    if (!query_context->getServerSettings()[ServerSetting::enable_experimental_export_merge_tree_partition_feature])
     {
         throw Exception(ErrorCodes::SUPPORT_IS_DISABLED,
-            "Exporting merge tree part is experimental. Set `allow_experimental_export_merge_tree_part` to enable it");
+            "Exporting merge tree partition is experimental. Set the server setting `enable_experimental_export_merge_tree_partition_feature` to enable it");
     }
 
     const auto dest_database = query_context->resolveDatabase(command.to_database);

src/Storages/System/attachSystemTables.cpp

@@ -5,7 +5,7 @@
 #include <Databases/IDatabase.h>
 #include <Storages/System/attachSystemTables.h>
 #include <Storages/System/attachSystemTablesImpl.h>
-
+#include <Core/ServerSettings.h>
 #include <Storages/System/StorageSystemAggregateFunctionCombinators.h>
 #include <Storages/System/StorageSystemAsynchronousMetrics.h>
 #include <Storages/System/StorageSystemAsyncLoader.h>
@@ -130,6 +130,11 @@
 namespace DB
 {
 
+namespace ServerSetting
+{
+    extern const ServerSettingsBool enable_experimental_export_merge_tree_partition_feature;
+}
+
 void attachSystemTablesServer(ContextPtr context, IDatabase & system_database, bool has_zookeeper)
 {
     attachNoDescription<StorageSystemOne>(context, system_database, "one", "This table contains a single row with a single dummy UInt8 column containing the value 0. Used when the table is not specified explicitly, for example in queries like `SELECT 1`.");
@@ -212,7 +217,10 @@ void attachSystemTablesServer(ContextPtr context, IDatabase & system_database, b
     attach<StorageSystemMerges>(context, system_database, "merges", "Contains a list of merges currently executing merges of MergeTree tables and their progress. Each merge operation is represented by a single row.");
     attach<StorageSystemMoves>(context, system_database, "moves", "Contains information about in-progress data part moves of MergeTree tables. Each data part movement is represented by a single row.");
     attach<StorageSystemExports>(context, system_database, "exports", "Contains a list of exports currently executing exports of MergeTree tables and their progress. Each export operation is represented by a single row.");
-    attach<StorageSystemReplicatedPartitionExports>(context, system_database, "replicated_partition_exports", "Contains a list of partition exports of ReplicatedMergeTree tables and their progress. Each export operation is represented by a single row.");
+    if (context->getServerSettings()[ServerSetting::enable_experimental_export_merge_tree_partition_feature])
+    {
+        attach<StorageSystemReplicatedPartitionExports>(context, system_database, "replicated_partition_exports", "Contains a list of partition exports of ReplicatedMergeTree tables and their progress. Each export operation is represented by a single row.");
+    }
     attach<StorageSystemMutations>(context, system_database, "mutations", "Contains a list of mutations and their progress. Each mutation command is represented by a single row.");
     attachNoDescription<StorageSystemReplicas>(context, system_database, "replicas", "Contains information and status of all table replicas on current server. Each replica is represented by a single row.");
     attach<StorageSystemReplicationQueue>(context, system_database, "replication_queue", "Contains information about tasks from replication queues stored in ClickHouse Keeper, or ZooKeeper, for each table replica.");

tests/config/config.d/enable_experimental_export_merge_tree_partition.xml

@@ -0,0 +1,3 @@
+<clickhouse>
+    <enable_export_merge_tree_partition_feature>0</enable_export_merge_tree_partition_feature>
+</clickhouse>

tests/config/install.sh

@@ -85,6 +85,7 @@ ln -sf $SRC_PATH/config.d/blob_storage_log.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/config.d/custom_settings_prefixes.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/config.d/database_catalog_drop_table_concurrency.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/config.d/enable_access_control_improvements.xml $DEST_SERVER_PATH/config.d/
+ln -sf $SRC_PATH/config.d/enable_experimental_export_merge_tree_partition.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/config.d/macros.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/config.d/secure_ports.xml $DEST_SERVER_PATH/config.d/
 ln -sf $SRC_PATH/config.d/clusters.xml $DEST_SERVER_PATH/config.d/

tests/integration/test_export_replicated_mt_partition_to_object_storage/configs/allow_experimental_export_partition.xml

@@ -0,0 +1,3 @@
+<clickhouse>
+    <enable_experimental_export_merge_tree_partition_feature>1</enable_experimental_export_merge_tree_partition_feature>
+</clickhouse>

tests/integration/test_export_replicated_mt_partition_to_object_storage/configs/disable_experimental_export_partition.xml

@@ -0,0 +1,3 @@
+<clickhouse>
+    <enable_experimental_export_merge_tree_partition_feature>0</enable_experimental_export_merge_tree_partition_feature>
+</clickhouse>

tests/integration/test_export_replicated_mt_partition_to_object_storage/test.py

@@ -15,15 +15,15 @@ def cluster():
         cluster = ClickHouseCluster(__file__)
         cluster.add_instance(
             "replica1", 
-            main_configs=["configs/named_collections.xml"],
+            main_configs=["configs/named_collections.xml", "configs/allow_experimental_export_partition.xml"],
             user_configs=["configs/users.d/profile.xml"],
             with_minio=True,
             stay_alive=True,
             with_zookeeper=True,
         )
         cluster.add_instance(
             "replica2", 
-            main_configs=["configs/named_collections.xml"],
+            main_configs=["configs/named_collections.xml", "configs/allow_experimental_export_partition.xml"],
             user_configs=["configs/users.d/profile.xml"],
             with_minio=True,
             stay_alive=True,
@@ -36,6 +36,14 @@ def cluster():
             user_configs=[],
             with_minio=True,
         )
+        cluster.add_instance(
+            "replica_with_export_disabled", 
+            main_configs=["configs/named_collections.xml", "configs/disable_experimental_export_partition.xml"],
+            user_configs=["configs/users.d/profile.xml"],
+            with_minio=True,
+            stay_alive=True,
+            with_zookeeper=True,
+        )
         logging.info("Starting cluster...")
         cluster.start()
         yield cluster
@@ -538,6 +546,21 @@ def test_export_partition_file_already_exists_policy(cluster):
     ) == "FAILED\n", "Export should be marked as FAILED"
 
 
+def test_export_partition_feature_is_disabled(cluster):
+    replica_with_export_disabled = cluster.instances["replica_with_export_disabled"]
+
+    mt_table = "export_partition_feature_is_disabled_mt_table"
+    s3_table = "export_partition_feature_is_disabled_s3_table"
+
+    create_tables_and_insert_data(replica_with_export_disabled, mt_table, s3_table, "replica1")
+
+    error = replica_with_export_disabled.query_and_get_error(f"ALTER TABLE {mt_table} EXPORT PARTITION ID '2020' TO TABLE {s3_table} SETTINGS allow_experimental_export_merge_tree_part=1;")
+    assert "experimental" in error, "Expected error about disabled feature"
+
+    # make sure kill operation also throws
+    error = replica_with_export_disabled.query_and_get_error(f"KILL EXPORT PARTITION WHERE partition_id = '2020' and source_table = '{mt_table}' and destination_table = '{s3_table}'")
+    assert "experimental" in error, "Expected error about disabled feature"
+
 # def test_source_mutations_during_export_snapshot(cluster):
 #     node = cluster.instances["replica1"]