ClickHouse/programs/copier/ClusterCopier.h

240 lines
9.1 KiB
C++
Raw Normal View History

#pragma once
2020-02-11 18:34:48 +00:00
2020-02-20 09:01:06 +00:00
#include "Aliases.h"
2020-02-11 18:34:48 +00:00
#include "Internals.h"
2020-02-13 10:52:46 +00:00
#include "TaskCluster.h"
2020-02-20 09:01:06 +00:00
#include "TaskTableAndShard.h"
2020-02-13 10:52:46 +00:00
#include "ShardPartition.h"
#include "ShardPartitionPiece.h"
2020-02-20 09:01:06 +00:00
#include "ZooKeeperStaff.h"
namespace DB
{
2021-05-31 14:49:02 +00:00
class ClusterCopier : WithMutableContext
{
public:
2020-03-17 18:07:54 +00:00
ClusterCopier(const String & task_path_,
const String & host_id_,
const String & proxy_database_name_,
ContextMutablePtr context_,
2021-04-22 18:04:32 +00:00
Poco::Logger * log_)
2021-05-31 14:49:02 +00:00
: WithMutableContext(context_),
2020-03-17 18:07:54 +00:00
task_zookeeper_path(task_path_),
host_id(host_id_),
working_database_name(proxy_database_name_),
2021-04-29 22:56:41 +00:00
log(log_) {}
2020-02-11 18:34:48 +00:00
2020-02-18 13:39:22 +00:00
void init();
2020-02-11 18:34:48 +00:00
template <typename T>
2020-02-18 13:39:22 +00:00
decltype(auto) retry(T && func, UInt64 max_tries = 100);
2020-02-11 18:34:48 +00:00
2020-02-18 13:39:22 +00:00
void discoverShardPartitions(const ConnectionTimeouts & timeouts, const TaskShardPtr & task_shard);
2020-02-11 18:34:48 +00:00
/// Compute set of partitions, assume set of partitions aren't changed during the processing
2020-02-18 13:39:22 +00:00
void discoverTablePartitions(const ConnectionTimeouts & timeouts, TaskTable & task_table, UInt64 num_threads = 0);
2020-02-11 18:34:48 +00:00
void uploadTaskDescription(const std::string & task_path, const std::string & task_file, bool force);
2020-02-11 18:34:48 +00:00
2020-02-18 13:39:22 +00:00
void reloadTaskDescription();
2020-02-11 18:34:48 +00:00
2020-02-18 13:39:22 +00:00
void updateConfigIfNeeded();
2020-02-11 18:34:48 +00:00
2020-02-18 13:39:22 +00:00
void process(const ConnectionTimeouts & timeouts);
2020-02-11 18:34:48 +00:00
/// Disables DROP PARTITION commands that used to clear data after errors
2020-02-20 10:01:02 +00:00
void setSafeMode(bool is_safe_mode_ = true)
{
is_safe_mode = is_safe_mode_;
}
2020-02-11 18:34:48 +00:00
2020-02-20 10:01:02 +00:00
void setCopyFaultProbability(double copy_fault_probability_)
{
copy_fault_probability = copy_fault_probability_;
}
2020-02-11 18:34:48 +00:00
2020-03-16 21:05:38 +00:00
void setMoveFaultProbability(double move_fault_probability_)
{
move_fault_probability = move_fault_probability_;
}
2020-04-21 17:37:40 +00:00
void setExperimentalUseSampleOffset(bool value)
{
experimental_use_sample_offset = value;
}
void setMaxTableTries(UInt64 tries)
{
max_table_tries = tries;
}
void setMaxShardPartitionTries(UInt64 tries)
{
max_shard_partition_tries = tries;
}
void setMaxShardPartitionPieceTriesForAlter(UInt64 tries)
{
max_shard_partition_piece_tries_for_alter = tries;
}
void setRetryDelayMs(std::chrono::milliseconds ms)
{
retry_delay_ms = ms;
}
2020-02-20 10:01:02 +00:00
protected:
2020-02-11 18:34:48 +00:00
2020-02-20 10:01:02 +00:00
String getWorkersPath() const
{
return task_cluster->task_zookeeper_path + "/task_active_workers";
}
2020-02-11 18:34:48 +00:00
2020-02-20 10:01:02 +00:00
String getWorkersPathVersion() const
{
return getWorkersPath() + "_version";
}
2020-02-11 18:34:48 +00:00
2020-02-20 10:01:02 +00:00
String getCurrentWorkerNodePath() const
{
return getWorkersPath() + "/" + host_id;
}
2020-02-11 18:34:48 +00:00
zkutil::EphemeralNodeHolder::Ptr createTaskWorkerNodeAndWaitIfNeed(
const zkutil::ZooKeeperPtr & zookeeper,
const String & description,
2020-02-18 13:39:22 +00:00
bool unprioritized);
2020-02-11 18:34:48 +00:00
2020-02-21 16:00:50 +00:00
/*
* Checks that partition piece or some other entity is clean.
* The only requirement is that you have to pass is_dirty_flag_path and is_dirty_cleaned_path to the function.
* And is_dirty_flag_path is a parent of is_dirty_cleaned_path.
* */
2020-03-27 22:44:13 +00:00
static bool checkPartitionPieceIsClean(
2020-02-21 16:00:50 +00:00
const zkutil::ZooKeeperPtr & zookeeper,
const CleanStateClock & clean_state_clock,
2020-03-27 22:44:13 +00:00
const String & task_status_path);
2020-02-21 16:00:50 +00:00
bool checkAllPiecesInPartitionAreDone(const TaskTable & task_table, const String & partition_name, const TasksShard & shards_with_partition);
2020-02-11 18:34:48 +00:00
/** Checks that the whole partition of a table was copied. We should do it carefully due to dirty lock.
* State of some task could change during the processing.
* We have to ensure that all shards have the finished state and there is no dirty flag.
* Moreover, we have to check status twice and check zxid, because state can change during the checking.
*/
2020-02-18 13:26:08 +00:00
/* The same as function above
* Assume that we don't know on which shards do we have partition certain piece.
* We'll check them all (I mean shards that contain the whole partition)
* And shards that don't have certain piece MUST mark that piece is_done true.
* */
bool checkPartitionPieceIsDone(const TaskTable & task_table, const String & partition_name,
2020-02-18 13:39:22 +00:00
size_t piece_number, const TasksShard & shards_with_partition);
2020-02-18 13:26:08 +00:00
2020-03-13 14:19:20 +00:00
/*Alter successful insertion to helping tables it will move all pieces to destination table*/
2020-03-18 13:25:49 +00:00
TaskStatus tryMoveAllPiecesToDestinationTable(const TaskTable & task_table, const String & partition_name);
2020-03-13 14:19:20 +00:00
2020-02-11 18:34:48 +00:00
/// Removes MATERIALIZED and ALIAS columns from create table query
2021-04-22 22:32:16 +00:00
static ASTPtr removeAliasMaterializedAndTTLColumnsFromCreateQuery(const ASTPtr & query_ast, bool allow_to_copy_alias_and_materialized_columns);
2020-02-11 18:34:48 +00:00
bool tryDropPartitionPiece(ShardPartition & task_partition, size_t current_piece_number,
2020-02-20 18:58:00 +00:00
const zkutil::ZooKeeperPtr & zookeeper, const CleanStateClock & clean_state_clock);
2020-02-11 18:34:48 +00:00
2020-02-18 13:39:22 +00:00
bool tryProcessTable(const ConnectionTimeouts & timeouts, TaskTable & task_table);
2020-02-11 18:34:48 +00:00
TaskStatus tryCreateDestinationTable(const ConnectionTimeouts & timeouts, TaskTable & task_table);
2020-02-11 18:34:48 +00:00
/// Job for copying partition from particular shard.
2020-03-18 13:25:49 +00:00
TaskStatus tryProcessPartitionTask(const ConnectionTimeouts & timeouts,
ShardPartition & task_partition,
bool is_unprioritized_task);
TaskStatus iterateThroughAllPiecesInPartition(const ConnectionTimeouts & timeouts,
ShardPartition & task_partition,
bool is_unprioritized_task);
TaskStatus processPartitionPieceTaskImpl(const ConnectionTimeouts & timeouts,
ShardPartition & task_partition,
size_t current_piece_number,
2020-03-18 13:25:49 +00:00
bool is_unprioritized_task);
2020-02-11 18:34:48 +00:00
2020-02-18 13:39:22 +00:00
void dropAndCreateLocalTable(const ASTPtr & create_ast);
2020-02-11 18:34:48 +00:00
2020-02-18 13:39:22 +00:00
void dropLocalTableIfExists(const DatabaseAndTableName & table_name) const;
2020-02-11 18:34:48 +00:00
2020-03-13 16:25:07 +00:00
void dropHelpingTables(const TaskTable & task_table);
void dropHelpingTablesByPieceNumber(const TaskTable & task_table, size_t current_piece_number);
2020-03-18 18:35:58 +00:00
/// Is used for usage less disk space.
/// After all pieces were successfully moved to original destination
/// table we can get rid of partition pieces (partitions in helping tables).
void dropParticularPartitionPieceFromAllHelpingTables(const TaskTable & task_table, const String & partition_name);
String getRemoteCreateTable(const DatabaseAndTableName & table, Connection & connection, const Settings & settings);
2020-02-11 18:34:48 +00:00
2020-02-18 13:39:22 +00:00
ASTPtr getCreateTableForPullShard(const ConnectionTimeouts & timeouts, TaskShard & task_shard);
2020-02-11 18:34:48 +00:00
2020-02-18 13:26:08 +00:00
/// If it is implicitly asked to create split Distributed table for certain piece on current shard, we will do it.
2020-02-21 16:00:50 +00:00
void createShardInternalTables(const ConnectionTimeouts & timeouts, TaskShard & task_shard, bool create_split = true);
2020-02-11 18:34:48 +00:00
2020-02-18 13:39:22 +00:00
std::set<String> getShardPartitions(const ConnectionTimeouts & timeouts, TaskShard & task_shard);
2020-02-11 18:34:48 +00:00
2020-02-18 13:39:22 +00:00
bool checkShardHasPartition(const ConnectionTimeouts & timeouts, TaskShard & task_shard, const String & partition_quoted_name);
2020-02-11 18:34:48 +00:00
bool checkPresentPartitionPiecesOnCurrentShard(const ConnectionTimeouts & timeouts,
2020-02-18 13:39:22 +00:00
TaskShard & task_shard, const String & partition_quoted_name, size_t current_piece_number);
2020-02-11 18:34:48 +00:00
2020-03-10 20:04:08 +00:00
/*
* This class is used in executeQueryOnCluster function
* You can execute query on each shard (no sense it is executed on each replica of a shard or not)
* or you can execute query on each replica on each shard.
* First mode is useful for INSERTS queries.
* */
enum ClusterExecutionMode
{
ON_EACH_SHARD,
ON_EACH_NODE
};
2020-02-11 18:34:48 +00:00
/** Executes simple query (without output streams, for example DDL queries) on each shard of the cluster
* Returns number of shards for which at least one replica executed query successfully
*/
UInt64 executeQueryOnCluster(
const ClusterPtr & cluster,
const String & query,
2020-05-30 17:53:55 +00:00
const Settings & current_settings,
2021-04-22 18:04:32 +00:00
ClusterExecutionMode execution_mode = ClusterExecutionMode::ON_EACH_SHARD) const;
private:
2020-02-11 18:34:48 +00:00
String task_zookeeper_path;
String task_description_path;
String host_id;
String working_database_name;
/// Auto update config stuff
2020-02-20 17:26:20 +00:00
UInt64 task_description_current_version = 1;
std::atomic<UInt64> task_description_version{1};
2020-02-11 18:34:48 +00:00
Coordination::WatchCallback task_description_watch_callback;
/// ZooKeeper session used to set the callback
zkutil::ZooKeeperPtr task_description_watch_zookeeper;
2020-02-11 18:34:48 +00:00
ConfigurationPtr task_cluster_initial_config;
ConfigurationPtr task_cluster_current_config;
2020-02-11 18:34:48 +00:00
std::unique_ptr<TaskCluster> task_cluster;
bool is_safe_mode = false;
2020-02-11 18:34:48 +00:00
double copy_fault_probability = 0.0;
2020-03-16 21:05:38 +00:00
double move_fault_probability = 0.0;
2020-04-21 17:37:40 +00:00
bool experimental_use_sample_offset{false};
2020-02-11 18:34:48 +00:00
Poco::Logger * log;
UInt64 max_table_tries = 3;
UInt64 max_shard_partition_tries = 3;
UInt64 max_shard_partition_piece_tries_for_alter = 10;
std::chrono::milliseconds retry_delay_ms{1000};
2020-02-11 18:34:48 +00:00
};
}