2020-10-22 15:08:00 +00:00
|
|
|
#include <DataTypes/DataTypeString.h>
|
2020-04-05 12:18:51 +00:00
|
|
|
#include <Databases/DatabaseReplicated.h>
|
|
|
|
#include <IO/ReadBufferFromFile.h>
|
2020-05-11 12:55:17 +00:00
|
|
|
#include <IO/ReadBufferFromString.h>
|
2020-04-05 12:18:51 +00:00
|
|
|
#include <IO/ReadHelpers.h>
|
|
|
|
#include <IO/WriteHelpers.h>
|
|
|
|
#include <Interpreters/Context.h>
|
2020-05-11 12:55:17 +00:00
|
|
|
#include <Interpreters/executeQuery.h>
|
2020-04-05 12:18:51 +00:00
|
|
|
#include <Parsers/queryToString.h>
|
2020-05-12 13:35:05 +00:00
|
|
|
#include <Common/Exception.h>
|
2020-10-22 15:08:00 +00:00
|
|
|
#include <Common/Stopwatch.h>
|
2020-04-05 12:18:51 +00:00
|
|
|
#include <Common/ZooKeeper/KeeperException.h>
|
|
|
|
#include <Common/ZooKeeper/Types.h>
|
|
|
|
#include <Common/ZooKeeper/ZooKeeper.h>
|
2020-11-27 14:04:03 +00:00
|
|
|
#include <Databases/DatabaseReplicatedWorker.h>
|
2020-11-05 09:52:23 +00:00
|
|
|
#include <Interpreters/DDLTask.h>
|
|
|
|
#include <Interpreters/executeDDLQueryOnCluster.h>
|
2020-11-13 18:35:45 +00:00
|
|
|
#include <Interpreters/Cluster.h>
|
|
|
|
#include <common/getFQDNOrHostName.h>
|
2020-11-05 09:52:23 +00:00
|
|
|
#include <Parsers/ASTAlterQuery.h>
|
2021-03-08 21:34:52 +00:00
|
|
|
#include <Parsers/ASTDropQuery.h>
|
2020-11-27 14:04:03 +00:00
|
|
|
#include <Parsers/ParserCreateQuery.h>
|
|
|
|
#include <Parsers/parseQuery.h>
|
|
|
|
#include <Interpreters/InterpreterCreateQuery.h>
|
2021-06-01 10:24:06 +00:00
|
|
|
#include <Interpreters/evaluateConstantExpression.h>
|
2020-11-27 14:04:03 +00:00
|
|
|
#include <Parsers/formatAST.h>
|
2021-03-16 20:01:20 +00:00
|
|
|
#include <Common/Macros.h>
|
2020-04-05 12:18:51 +00:00
|
|
|
|
|
|
|
namespace DB
|
|
|
|
{
|
|
|
|
namespace ErrorCodes
|
|
|
|
{
|
|
|
|
extern const int NO_ZOOKEEPER;
|
2020-06-22 14:19:26 +00:00
|
|
|
extern const int LOGICAL_ERROR;
|
2020-06-24 12:45:42 +00:00
|
|
|
extern const int BAD_ARGUMENTS;
|
2020-11-13 18:35:45 +00:00
|
|
|
extern const int REPLICA_IS_ALREADY_EXIST;
|
2020-11-19 10:34:45 +00:00
|
|
|
extern const int DATABASE_REPLICATION_FAILED;
|
2020-11-20 16:06:27 +00:00
|
|
|
extern const int UNKNOWN_DATABASE;
|
2021-02-12 16:22:01 +00:00
|
|
|
extern const int UNKNOWN_TABLE;
|
2020-11-29 11:45:32 +00:00
|
|
|
extern const int NOT_IMPLEMENTED;
|
2021-02-04 19:41:44 +00:00
|
|
|
extern const int INCORRECT_QUERY;
|
2021-02-08 09:14:17 +00:00
|
|
|
extern const int ALL_CONNECTION_TRIES_FAILED;
|
2020-04-05 12:18:51 +00:00
|
|
|
}
|
|
|
|
|
2021-02-08 09:14:17 +00:00
|
|
|
static constexpr const char * DROPPED_MARK = "DROPPED";
|
2021-02-11 22:23:40 +00:00
|
|
|
static constexpr const char * BROKEN_TABLES_SUFFIX = "_broken_tables";
|
2021-02-10 20:30:40 +00:00
|
|
|
|
2021-02-08 09:14:17 +00:00
|
|
|
|
2020-11-13 18:35:45 +00:00
|
|
|
zkutil::ZooKeeperPtr DatabaseReplicated::getZooKeeper() const
|
2020-04-05 12:18:51 +00:00
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
return getContext()->getZooKeeper();
|
2020-04-05 12:18:51 +00:00
|
|
|
}
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
static inline String getHostID(ContextPtr global_context, const UUID & db_uuid)
|
2020-04-05 12:18:51 +00:00
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
return Cluster::Address::toString(getFQDNOrHostName(), global_context->getTCPPort()) + ':' + toString(db_uuid);
|
2020-04-05 12:18:51 +00:00
|
|
|
}
|
|
|
|
|
2020-11-13 18:35:45 +00:00
|
|
|
|
2020-11-05 09:52:23 +00:00
|
|
|
DatabaseReplicated::~DatabaseReplicated() = default;
|
2020-04-05 12:18:51 +00:00
|
|
|
|
|
|
|
DatabaseReplicated::DatabaseReplicated(
|
|
|
|
const String & name_,
|
|
|
|
const String & metadata_path_,
|
2020-10-20 16:14:54 +00:00
|
|
|
UUID uuid,
|
2020-04-05 12:18:51 +00:00
|
|
|
const String & zookeeper_path_,
|
2020-10-27 09:19:45 +00:00
|
|
|
const String & shard_name_,
|
2020-04-05 12:18:51 +00:00
|
|
|
const String & replica_name_,
|
2021-02-19 23:41:58 +00:00
|
|
|
DatabaseReplicatedSettings db_settings_,
|
2021-04-10 23:33:54 +00:00
|
|
|
ContextPtr context_)
|
2020-10-20 16:14:54 +00:00
|
|
|
: DatabaseAtomic(name_, metadata_path_, uuid, "DatabaseReplicated (" + name_ + ")", context_)
|
2020-04-05 12:18:51 +00:00
|
|
|
, zookeeper_path(zookeeper_path_)
|
2020-10-27 09:19:45 +00:00
|
|
|
, shard_name(shard_name_)
|
2020-04-05 12:18:51 +00:00
|
|
|
, replica_name(replica_name_)
|
2021-02-19 23:41:58 +00:00
|
|
|
, db_settings(std::move(db_settings_))
|
2020-04-05 12:18:51 +00:00
|
|
|
{
|
2020-10-27 09:19:45 +00:00
|
|
|
if (zookeeper_path.empty() || shard_name.empty() || replica_name.empty())
|
2020-11-13 18:35:45 +00:00
|
|
|
throw Exception("ZooKeeper path, shard and replica names must be non-empty", ErrorCodes::BAD_ARGUMENTS);
|
|
|
|
if (shard_name.find('/') != std::string::npos || replica_name.find('/') != std::string::npos)
|
|
|
|
throw Exception("Shard and replica names should not contain '/'", ErrorCodes::BAD_ARGUMENTS);
|
2021-02-08 09:14:17 +00:00
|
|
|
if (shard_name.find('|') != std::string::npos || replica_name.find('|') != std::string::npos)
|
|
|
|
throw Exception("Shard and replica names should not contain '|'", ErrorCodes::BAD_ARGUMENTS);
|
2020-06-24 12:45:42 +00:00
|
|
|
|
|
|
|
if (zookeeper_path.back() == '/')
|
2020-04-05 12:18:51 +00:00
|
|
|
zookeeper_path.resize(zookeeper_path.size() - 1);
|
2020-11-13 18:35:45 +00:00
|
|
|
|
2020-10-22 15:08:00 +00:00
|
|
|
/// If zookeeper chroot prefix is used, path should start with '/', because chroot concatenates without it.
|
2020-06-24 12:45:42 +00:00
|
|
|
if (zookeeper_path.front() != '/')
|
2020-04-05 12:18:51 +00:00
|
|
|
zookeeper_path = "/" + zookeeper_path;
|
2020-11-13 18:35:45 +00:00
|
|
|
}
|
2020-05-24 17:13:53 +00:00
|
|
|
|
2021-02-08 09:14:17 +00:00
|
|
|
String DatabaseReplicated::getFullReplicaName() const
|
|
|
|
{
|
|
|
|
return shard_name + '|' + replica_name;
|
|
|
|
}
|
|
|
|
|
|
|
|
std::pair<String, String> DatabaseReplicated::parseFullReplicaName(const String & name)
|
|
|
|
{
|
|
|
|
String shard;
|
|
|
|
String replica;
|
|
|
|
auto pos = name.find('|');
|
|
|
|
if (pos == std::string::npos || name.find('|', pos + 1) != std::string::npos)
|
|
|
|
throw Exception(ErrorCodes::LOGICAL_ERROR, "Incorrect replica identifier: {}", name);
|
|
|
|
shard = name.substr(0, pos);
|
|
|
|
replica = name.substr(pos + 1);
|
|
|
|
return {shard, replica};
|
|
|
|
}
|
|
|
|
|
|
|
|
ClusterPtr DatabaseReplicated::getCluster() const
|
|
|
|
{
|
2021-03-09 17:05:24 +00:00
|
|
|
std::lock_guard lock{mutex};
|
2021-03-29 13:44:41 +00:00
|
|
|
if (cluster)
|
|
|
|
return cluster;
|
|
|
|
|
|
|
|
cluster = getClusterImpl();
|
2021-03-09 17:05:24 +00:00
|
|
|
return cluster;
|
|
|
|
}
|
|
|
|
|
|
|
|
void DatabaseReplicated::setCluster(ClusterPtr && new_cluster)
|
|
|
|
{
|
|
|
|
std::lock_guard lock{mutex};
|
|
|
|
cluster = std::move(new_cluster);
|
|
|
|
}
|
|
|
|
|
|
|
|
ClusterPtr DatabaseReplicated::getClusterImpl() const
|
|
|
|
{
|
2021-02-08 09:14:17 +00:00
|
|
|
Strings hosts;
|
|
|
|
Strings host_ids;
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
auto zookeeper = getContext()->getZooKeeper();
|
2021-02-08 09:14:17 +00:00
|
|
|
constexpr int max_retries = 10;
|
|
|
|
int iteration = 0;
|
|
|
|
bool success = false;
|
|
|
|
while (++iteration <= max_retries)
|
|
|
|
{
|
|
|
|
host_ids.resize(0);
|
|
|
|
Coordination::Stat stat;
|
|
|
|
hosts = zookeeper->getChildren(zookeeper_path + "/replicas", &stat);
|
|
|
|
if (hosts.empty())
|
|
|
|
throw Exception(ErrorCodes::LOGICAL_ERROR, "No hosts found");
|
2021-03-15 08:22:42 +00:00
|
|
|
Int32 cversion = stat.cversion;
|
2021-02-08 19:36:17 +00:00
|
|
|
std::sort(hosts.begin(), hosts.end());
|
2021-02-08 09:14:17 +00:00
|
|
|
|
|
|
|
std::vector<zkutil::ZooKeeper::FutureGet> futures;
|
|
|
|
futures.reserve(hosts.size());
|
|
|
|
host_ids.reserve(hosts.size());
|
|
|
|
for (const auto & host : hosts)
|
|
|
|
futures.emplace_back(zookeeper->asyncTryGet(zookeeper_path + "/replicas/" + host));
|
|
|
|
|
|
|
|
success = true;
|
|
|
|
for (auto & future : futures)
|
|
|
|
{
|
|
|
|
auto res = future.get();
|
|
|
|
if (res.error != Coordination::Error::ZOK)
|
|
|
|
success = false;
|
|
|
|
host_ids.emplace_back(res.data);
|
|
|
|
}
|
|
|
|
|
|
|
|
zookeeper->get(zookeeper_path + "/replicas", &stat);
|
2021-03-15 08:22:42 +00:00
|
|
|
if (cversion != stat.cversion)
|
|
|
|
success = false;
|
|
|
|
if (success)
|
2021-02-08 09:14:17 +00:00
|
|
|
break;
|
|
|
|
}
|
|
|
|
if (!success)
|
2021-02-19 23:41:58 +00:00
|
|
|
throw Exception(ErrorCodes::ALL_CONNECTION_TRIES_FAILED, "Cannot get consistent cluster snapshot,"
|
|
|
|
"because replicas are created or removed concurrently");
|
2021-02-08 09:14:17 +00:00
|
|
|
|
|
|
|
assert(!hosts.empty());
|
|
|
|
assert(hosts.size() == host_ids.size());
|
|
|
|
String current_shard = parseFullReplicaName(hosts.front()).first;
|
|
|
|
std::vector<Strings> shards;
|
|
|
|
shards.emplace_back();
|
|
|
|
for (size_t i = 0; i < hosts.size(); ++i)
|
|
|
|
{
|
|
|
|
const auto & id = host_ids[i];
|
|
|
|
if (id == DROPPED_MARK)
|
|
|
|
continue;
|
|
|
|
auto [shard, replica] = parseFullReplicaName(hosts[i]);
|
2021-03-09 21:41:04 +00:00
|
|
|
auto pos = id.rfind(':');
|
|
|
|
String host_port = id.substr(0, pos);
|
2021-02-08 09:14:17 +00:00
|
|
|
if (shard != current_shard)
|
|
|
|
{
|
|
|
|
current_shard = shard;
|
|
|
|
if (!shards.back().empty())
|
|
|
|
shards.emplace_back();
|
|
|
|
}
|
2021-03-09 21:41:04 +00:00
|
|
|
shards.back().emplace_back(unescapeForFileName(host_port));
|
2021-02-08 09:14:17 +00:00
|
|
|
}
|
|
|
|
|
2021-03-09 21:41:04 +00:00
|
|
|
String username = db_settings.cluster_username;
|
|
|
|
String password = db_settings.cluster_password;
|
2021-04-10 23:33:54 +00:00
|
|
|
UInt16 default_port = getContext()->getTCPPort();
|
2021-03-09 21:41:04 +00:00
|
|
|
bool secure = db_settings.cluster_secure_connection;
|
2021-02-08 09:14:17 +00:00
|
|
|
|
2021-07-23 11:16:35 +00:00
|
|
|
bool treat_local_as_remote = false;
|
|
|
|
bool treat_local_port_as_remote = getContext()->getApplicationType() == Context::ApplicationType::LOCAL;
|
|
|
|
return std::make_shared<Cluster>(
|
|
|
|
getContext()->getSettingsRef(),
|
|
|
|
shards,
|
|
|
|
username,
|
|
|
|
password,
|
|
|
|
default_port,
|
|
|
|
treat_local_as_remote,
|
|
|
|
treat_local_port_as_remote,
|
|
|
|
secure);
|
2021-02-08 09:14:17 +00:00
|
|
|
}
|
|
|
|
|
2021-02-19 23:41:58 +00:00
|
|
|
void DatabaseReplicated::tryConnectToZooKeeperAndInitDatabase(bool force_attach)
|
2021-02-15 00:04:46 +00:00
|
|
|
{
|
|
|
|
try
|
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
if (!getContext()->hasZooKeeper())
|
2021-02-15 00:04:46 +00:00
|
|
|
{
|
|
|
|
throw Exception("Can't create replicated database without ZooKeeper", ErrorCodes::NO_ZOOKEEPER);
|
|
|
|
}
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
auto current_zookeeper = getContext()->getZooKeeper();
|
2021-02-15 00:04:46 +00:00
|
|
|
|
|
|
|
if (!current_zookeeper->exists(zookeeper_path))
|
|
|
|
{
|
|
|
|
/// Create new database, multiple nodes can execute it concurrently
|
|
|
|
createDatabaseNodesInZooKeeper(current_zookeeper);
|
|
|
|
}
|
|
|
|
|
2021-05-09 11:59:49 +00:00
|
|
|
replica_path = fs::path(zookeeper_path) / "replicas" / getFullReplicaName();
|
2021-02-15 00:04:46 +00:00
|
|
|
|
|
|
|
String replica_host_id;
|
|
|
|
if (current_zookeeper->tryGet(replica_path, replica_host_id))
|
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
String host_id = getHostID(getContext(), db_uuid);
|
2021-02-15 00:04:46 +00:00
|
|
|
if (replica_host_id != host_id)
|
|
|
|
throw Exception(ErrorCodes::REPLICA_IS_ALREADY_EXIST,
|
|
|
|
"Replica {} of shard {} of replicated database at {} already exists. Replica host ID: '{}', current host ID: '{}'",
|
|
|
|
replica_name, shard_name, zookeeper_path, replica_host_id, host_id);
|
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
|
|
|
/// Throws if replica with the same name already exists
|
|
|
|
createReplicaNodesInZooKeeper(current_zookeeper);
|
|
|
|
}
|
|
|
|
|
|
|
|
is_readonly = false;
|
|
|
|
}
|
2021-02-15 20:00:59 +00:00
|
|
|
catch (...)
|
2021-02-15 00:04:46 +00:00
|
|
|
{
|
|
|
|
if (!force_attach)
|
|
|
|
throw;
|
2021-02-15 20:00:59 +00:00
|
|
|
|
|
|
|
/// It's server startup, ignore error.
|
|
|
|
/// Worker thread will try to setup ZooKeeper connection
|
2021-02-15 00:04:46 +00:00
|
|
|
tryLogCurrentException(log);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2020-11-13 18:35:45 +00:00
|
|
|
bool DatabaseReplicated::createDatabaseNodesInZooKeeper(const zkutil::ZooKeeperPtr & current_zookeeper)
|
|
|
|
{
|
|
|
|
current_zookeeper->createAncestors(zookeeper_path);
|
2020-06-20 15:39:58 +00:00
|
|
|
|
2020-11-13 18:35:45 +00:00
|
|
|
Coordination::Requests ops;
|
|
|
|
ops.emplace_back(zkutil::makeCreateRequest(zookeeper_path, "", zkutil::CreateMode::Persistent));
|
|
|
|
ops.emplace_back(zkutil::makeCreateRequest(zookeeper_path + "/log", "", zkutil::CreateMode::Persistent));
|
2020-11-19 10:34:45 +00:00
|
|
|
ops.emplace_back(zkutil::makeCreateRequest(zookeeper_path + "/replicas", "", zkutil::CreateMode::Persistent));
|
2020-11-20 16:06:27 +00:00
|
|
|
ops.emplace_back(zkutil::makeCreateRequest(zookeeper_path + "/counter", "", zkutil::CreateMode::Persistent));
|
2021-02-19 23:41:58 +00:00
|
|
|
/// We create and remove counter/cnt- node to increment sequential number of counter/ node and make log entry numbers start from 1.
|
|
|
|
/// New replicas are created with log pointer equal to 0 and log pointer is a number of the last executed entry.
|
|
|
|
/// It means that we cannot have log entry with number 0.
|
2020-11-27 14:04:03 +00:00
|
|
|
ops.emplace_back(zkutil::makeCreateRequest(zookeeper_path + "/counter/cnt-", "", zkutil::CreateMode::Persistent));
|
|
|
|
ops.emplace_back(zkutil::makeRemoveRequest(zookeeper_path + "/counter/cnt-", -1));
|
2020-11-19 10:34:45 +00:00
|
|
|
ops.emplace_back(zkutil::makeCreateRequest(zookeeper_path + "/metadata", "", zkutil::CreateMode::Persistent));
|
2020-11-27 14:04:03 +00:00
|
|
|
ops.emplace_back(zkutil::makeCreateRequest(zookeeper_path + "/max_log_ptr", "1", zkutil::CreateMode::Persistent));
|
2021-02-04 19:41:44 +00:00
|
|
|
ops.emplace_back(zkutil::makeCreateRequest(zookeeper_path + "/logs_to_keep", "1000", zkutil::CreateMode::Persistent));
|
2020-11-13 18:35:45 +00:00
|
|
|
|
|
|
|
Coordination::Responses responses;
|
|
|
|
auto res = current_zookeeper->tryMulti(ops, responses);
|
|
|
|
if (res == Coordination::Error::ZOK)
|
2021-02-15 20:00:59 +00:00
|
|
|
return true; /// Created new database (it's the first replica)
|
2020-11-13 18:35:45 +00:00
|
|
|
if (res == Coordination::Error::ZNODEEXISTS)
|
2021-02-15 20:00:59 +00:00
|
|
|
return false; /// Database exists, we will add new replica
|
2020-11-13 18:35:45 +00:00
|
|
|
|
2021-02-15 20:00:59 +00:00
|
|
|
/// Other codes are unexpected, will throw
|
2020-11-13 18:35:45 +00:00
|
|
|
zkutil::KeeperMultiException::check(res, ops, responses);
|
|
|
|
assert(false);
|
2020-11-27 14:04:03 +00:00
|
|
|
__builtin_unreachable();
|
2020-05-11 12:55:17 +00:00
|
|
|
}
|
|
|
|
|
2021-05-31 13:31:03 +00:00
|
|
|
void DatabaseReplicated::createEmptyLogEntry(const ZooKeeperPtr & current_zookeeper)
|
2020-10-22 15:08:00 +00:00
|
|
|
{
|
2020-11-19 10:34:45 +00:00
|
|
|
/// On replica creation add empty entry to log. Can be used to trigger some actions on other replicas (e.g. update cluster info).
|
2021-02-19 23:41:58 +00:00
|
|
|
DDLLogEntry entry{};
|
2021-05-31 13:31:03 +00:00
|
|
|
DatabaseReplicatedDDLWorker::enqueueQueryImpl(current_zookeeper, entry, this, true);
|
2021-03-09 17:05:24 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
void DatabaseReplicated::createReplicaNodesInZooKeeper(const zkutil::ZooKeeperPtr & current_zookeeper)
|
|
|
|
{
|
|
|
|
/// Write host name to replica_path, it will protect from multiple replicas with the same name
|
2021-04-10 23:33:54 +00:00
|
|
|
auto host_id = getHostID(getContext(), db_uuid);
|
2021-03-09 17:05:24 +00:00
|
|
|
|
2020-11-13 18:35:45 +00:00
|
|
|
Coordination::Requests ops;
|
|
|
|
ops.emplace_back(zkutil::makeCreateRequest(replica_path, host_id, zkutil::CreateMode::Persistent));
|
2020-11-27 14:04:03 +00:00
|
|
|
ops.emplace_back(zkutil::makeCreateRequest(replica_path + "/log_ptr", "0", zkutil::CreateMode::Persistent));
|
2020-11-13 18:35:45 +00:00
|
|
|
current_zookeeper->multi(ops);
|
2021-05-31 13:31:03 +00:00
|
|
|
createEmptyLogEntry(current_zookeeper);
|
2020-06-20 15:39:58 +00:00
|
|
|
}
|
|
|
|
|
2021-05-31 14:49:02 +00:00
|
|
|
void DatabaseReplicated::loadStoredObjects(ContextMutablePtr local_context, bool has_force_restore_data_flag, bool force_attach)
|
2020-10-27 09:19:45 +00:00
|
|
|
{
|
2021-02-19 23:41:58 +00:00
|
|
|
tryConnectToZooKeeperAndInitDatabase(force_attach);
|
2021-02-15 00:04:46 +00:00
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
DatabaseAtomic::loadStoredObjects(local_context, has_force_restore_data_flag, force_attach);
|
2020-11-13 18:35:45 +00:00
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
ddl_worker = std::make_unique<DatabaseReplicatedDDLWorker>(this, getContext());
|
2020-12-01 17:20:42 +00:00
|
|
|
ddl_worker->startup();
|
2020-10-27 09:19:45 +00:00
|
|
|
}
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
void DatabaseReplicated::checkQueryValid(const ASTPtr & query, ContextPtr query_context) const
|
2020-10-22 15:08:00 +00:00
|
|
|
{
|
2021-02-15 20:00:59 +00:00
|
|
|
/// Replicas will set correct name of current database in query context (database name can be different on replicas)
|
2021-03-16 20:01:20 +00:00
|
|
|
if (auto * ddl_query = dynamic_cast<ASTQueryWithTableAndOutput *>(query.get()))
|
2021-03-08 21:34:52 +00:00
|
|
|
{
|
|
|
|
if (ddl_query->database != getDatabaseName())
|
|
|
|
throw Exception(ErrorCodes::UNKNOWN_DATABASE, "Database was renamed");
|
2021-02-08 09:46:30 +00:00
|
|
|
ddl_query->database.clear();
|
2021-03-16 20:01:20 +00:00
|
|
|
|
|
|
|
if (auto * create = query->as<ASTCreateQuery>())
|
|
|
|
{
|
|
|
|
bool replicated_table = create->storage && create->storage->engine && startsWith(create->storage->engine->name, "Replicated");
|
|
|
|
if (!replicated_table || !create->storage->engine->arguments)
|
|
|
|
return;
|
|
|
|
|
2021-06-01 10:24:06 +00:00
|
|
|
ASTs & args_ref = create->storage->engine->arguments->children;
|
|
|
|
ASTs args = args_ref;
|
2021-03-16 20:01:20 +00:00
|
|
|
if (args.size() < 2)
|
|
|
|
return;
|
|
|
|
|
2021-06-01 10:24:06 +00:00
|
|
|
/// It can be a constant expression. Try to evaluate it, ignore exception if we cannot.
|
2021-06-01 12:15:13 +00:00
|
|
|
bool has_expression_argument = args_ref[0]->as<ASTFunction>() || args_ref[1]->as<ASTFunction>();
|
2021-06-01 10:24:06 +00:00
|
|
|
if (has_expression_argument)
|
|
|
|
{
|
|
|
|
try
|
|
|
|
{
|
|
|
|
args[0] = evaluateConstantExpressionAsLiteral(args_ref[0]->clone(), query_context);
|
|
|
|
args[1] = evaluateConstantExpressionAsLiteral(args_ref[1]->clone(), query_context);
|
|
|
|
}
|
|
|
|
catch (...)
|
|
|
|
{
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-03-16 20:01:20 +00:00
|
|
|
ASTLiteral * arg1 = args[0]->as<ASTLiteral>();
|
|
|
|
ASTLiteral * arg2 = args[1]->as<ASTLiteral>();
|
|
|
|
if (!arg1 || !arg2 || arg1->value.getType() != Field::Types::String || arg2->value.getType() != Field::Types::String)
|
|
|
|
return;
|
|
|
|
|
|
|
|
String maybe_path = arg1->value.get<String>();
|
|
|
|
String maybe_replica = arg2->value.get<String>();
|
|
|
|
|
|
|
|
/// Looks like it's ReplicatedMergeTree with explicit zookeeper_path and replica_name arguments.
|
|
|
|
/// Let's ensure that some macros are used.
|
|
|
|
/// NOTE: we cannot check here that substituted values will be actually different on shards and replicas.
|
|
|
|
|
|
|
|
Macros::MacroExpansionInfo info;
|
|
|
|
info.table_id = {getDatabaseName(), create->table, create->uuid};
|
2021-04-10 23:33:54 +00:00
|
|
|
query_context->getMacros()->expand(maybe_path, info);
|
2021-03-16 20:01:20 +00:00
|
|
|
bool maybe_shard_macros = info.expanded_other;
|
|
|
|
info.expanded_other = false;
|
2021-04-10 23:33:54 +00:00
|
|
|
query_context->getMacros()->expand(maybe_replica, info);
|
2021-03-16 20:01:20 +00:00
|
|
|
bool maybe_replica_macros = info.expanded_other;
|
2021-04-10 23:33:54 +00:00
|
|
|
bool enable_functional_tests_helper = getContext()->getConfigRef().has("_functional_tests_helper_database_replicated_replace_args_macros");
|
2021-03-16 20:01:20 +00:00
|
|
|
|
2021-03-18 12:49:31 +00:00
|
|
|
if (!enable_functional_tests_helper)
|
|
|
|
LOG_WARNING(log, "It's not recommended to explicitly specify zookeeper_path and replica_name in ReplicatedMergeTree arguments");
|
|
|
|
|
|
|
|
if (maybe_shard_macros && maybe_replica_macros)
|
|
|
|
return;
|
|
|
|
|
2021-06-01 10:24:06 +00:00
|
|
|
if (enable_functional_tests_helper && !has_expression_argument)
|
2021-03-16 20:01:20 +00:00
|
|
|
{
|
|
|
|
if (maybe_path.empty() || maybe_path.back() != '/')
|
|
|
|
maybe_path += '/';
|
2021-06-01 10:24:06 +00:00
|
|
|
args_ref[0]->as<ASTLiteral>()->value = maybe_path + "auto_{shard}";
|
|
|
|
args_ref[1]->as<ASTLiteral>()->value = maybe_replica + "auto_{replica}";
|
2021-03-16 20:01:20 +00:00
|
|
|
return;
|
|
|
|
}
|
|
|
|
|
|
|
|
throw Exception(ErrorCodes::INCORRECT_QUERY,
|
|
|
|
"Explicit zookeeper_path and replica_name are specified in ReplicatedMergeTree arguments. "
|
|
|
|
"If you really want to specify it explicitly, then you should use some macros "
|
|
|
|
"to distinguish different shards and replicas");
|
|
|
|
}
|
2021-03-08 21:34:52 +00:00
|
|
|
}
|
2021-02-08 09:46:30 +00:00
|
|
|
|
2020-11-05 09:52:23 +00:00
|
|
|
if (const auto * query_alter = query->as<ASTAlterQuery>())
|
2020-06-27 13:39:41 +00:00
|
|
|
{
|
2021-01-18 14:09:39 +00:00
|
|
|
for (const auto & command : query_alter->command_list->children)
|
2020-11-05 09:52:23 +00:00
|
|
|
{
|
2021-01-18 14:09:39 +00:00
|
|
|
if (!isSupportedAlterType(command->as<ASTAlterCommand&>().type))
|
2020-11-05 09:52:23 +00:00
|
|
|
throw Exception("Unsupported type of ALTER query", ErrorCodes::NOT_IMPLEMENTED);
|
|
|
|
}
|
2020-06-27 13:39:41 +00:00
|
|
|
}
|
2020-05-27 18:33:37 +00:00
|
|
|
|
2021-03-08 21:34:52 +00:00
|
|
|
if (auto * query_drop = query->as<ASTDropQuery>())
|
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
if (query_drop->kind == ASTDropQuery::Kind::Detach && query_context->getSettingsRef().database_replicated_always_detach_permanently)
|
2021-03-08 21:34:52 +00:00
|
|
|
query_drop->permanently = true;
|
|
|
|
if (query_drop->kind == ASTDropQuery::Kind::Detach && !query_drop->permanently)
|
|
|
|
throw Exception(ErrorCodes::INCORRECT_QUERY, "DETACH TABLE is not allowed for Replicated databases. "
|
|
|
|
"Use DETACH TABLE PERMANENTLY or SYSTEM RESTART REPLICA or set "
|
|
|
|
"database_replicated_always_detach_permanently to 1");
|
|
|
|
}
|
2021-03-16 20:01:20 +00:00
|
|
|
}
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
BlockIO DatabaseReplicated::tryEnqueueReplicatedDDL(const ASTPtr & query, ContextPtr query_context)
|
2021-03-16 20:01:20 +00:00
|
|
|
{
|
|
|
|
if (is_readonly)
|
|
|
|
throw Exception(ErrorCodes::NO_ZOOKEEPER, "Database is in readonly mode, because it cannot connect to ZooKeeper");
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
if (query_context->getClientInfo().query_kind != ClientInfo::QueryKind::INITIAL_QUERY)
|
2021-03-16 20:01:20 +00:00
|
|
|
throw Exception(ErrorCodes::INCORRECT_QUERY, "It's not initial query. ON CLUSTER is not allowed for Replicated database.");
|
2021-03-08 21:34:52 +00:00
|
|
|
|
2021-03-16 20:01:20 +00:00
|
|
|
checkQueryValid(query, query_context);
|
2020-11-05 09:52:23 +00:00
|
|
|
LOG_DEBUG(log, "Proposing query: {}", queryToString(query));
|
2020-05-13 17:00:47 +00:00
|
|
|
|
2020-11-05 09:52:23 +00:00
|
|
|
DDLLogEntry entry;
|
|
|
|
entry.query = queryToString(query);
|
|
|
|
entry.initiator = ddl_worker->getCommonHostID();
|
2021-03-08 22:57:53 +00:00
|
|
|
entry.setSettingsIfRequired(query_context);
|
2021-02-04 19:41:44 +00:00
|
|
|
String node_path = ddl_worker->tryEnqueueAndExecuteEntry(entry, query_context);
|
2020-07-04 16:32:23 +00:00
|
|
|
|
2021-02-08 19:36:17 +00:00
|
|
|
Strings hosts_to_wait = getZooKeeper()->getChildren(zookeeper_path + "/replicas");
|
2021-03-08 20:35:09 +00:00
|
|
|
return getDistributedDDLStatus(node_path, entry, query_context, hosts_to_wait);
|
2020-07-04 16:32:23 +00:00
|
|
|
}
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
static UUID getTableUUIDIfReplicated(const String & metadata, ContextPtr context)
|
2021-02-15 00:04:46 +00:00
|
|
|
{
|
|
|
|
bool looks_like_replicated = metadata.find("ReplicatedMergeTree") != std::string::npos;
|
|
|
|
if (!looks_like_replicated)
|
|
|
|
return UUIDHelpers::Nil;
|
|
|
|
|
|
|
|
ParserCreateQuery parser;
|
2021-04-10 23:33:54 +00:00
|
|
|
auto size = context->getSettingsRef().max_query_size;
|
|
|
|
auto depth = context->getSettingsRef().max_parser_depth;
|
2021-02-15 00:04:46 +00:00
|
|
|
ASTPtr query = parseQuery(parser, metadata, size, depth);
|
|
|
|
const ASTCreateQuery & create = query->as<const ASTCreateQuery &>();
|
|
|
|
if (!create.storage || !create.storage->engine)
|
|
|
|
return UUIDHelpers::Nil;
|
|
|
|
if (!startsWith(create.storage->engine->name, "Replicated") || !endsWith(create.storage->engine->name, "MergeTree"))
|
|
|
|
return UUIDHelpers::Nil;
|
|
|
|
assert(create.uuid != UUIDHelpers::Nil);
|
|
|
|
return create.uuid;
|
|
|
|
}
|
2020-11-05 09:52:23 +00:00
|
|
|
|
2021-02-09 15:14:20 +00:00
|
|
|
void DatabaseReplicated::recoverLostReplica(const ZooKeeperPtr & current_zookeeper, UInt32 our_log_ptr, UInt32 max_log_ptr)
|
2020-10-22 15:08:00 +00:00
|
|
|
{
|
2021-02-15 20:00:59 +00:00
|
|
|
/// Let's compare local (possibly outdated) metadata with (most actual) metadata stored in ZooKeeper
|
|
|
|
/// and try to update the set of local tables.
|
|
|
|
/// We could drop all local tables and create the new ones just like it's new replica.
|
|
|
|
/// But it will cause all ReplicatedMergeTree tables to fetch all data parts again and data in other tables will be lost.
|
|
|
|
|
2021-02-09 15:14:20 +00:00
|
|
|
bool new_replica = our_log_ptr == 0;
|
|
|
|
if (new_replica)
|
|
|
|
LOG_INFO(log, "Will create new replica from log pointer {}", max_log_ptr);
|
|
|
|
else
|
|
|
|
LOG_WARNING(log, "Will recover replica with staled log pointer {} from log pointer {}", our_log_ptr, max_log_ptr);
|
|
|
|
|
|
|
|
if (new_replica && !empty())
|
|
|
|
throw Exception(ErrorCodes::LOGICAL_ERROR, "It's new replica, but database is not empty");
|
2020-05-24 17:13:53 +00:00
|
|
|
|
2021-02-09 15:14:20 +00:00
|
|
|
auto table_name_to_metadata = tryGetConsistentMetadataSnapshot(current_zookeeper, max_log_ptr);
|
2020-06-24 12:45:42 +00:00
|
|
|
|
2021-02-15 00:04:46 +00:00
|
|
|
/// For ReplicatedMergeTree tables we can compare only UUIDs to ensure that it's the same table.
|
|
|
|
/// Metadata can be different, it's handled on table replication level.
|
2021-02-15 20:00:59 +00:00
|
|
|
/// We need to handle renamed tables only.
|
2021-02-15 00:04:46 +00:00
|
|
|
/// TODO maybe we should also update MergeTree SETTINGS if required?
|
|
|
|
std::unordered_map<UUID, String> zk_replicated_id_to_name;
|
|
|
|
for (const auto & zk_table : table_name_to_metadata)
|
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
UUID zk_replicated_id = getTableUUIDIfReplicated(zk_table.second, getContext());
|
2021-02-15 00:04:46 +00:00
|
|
|
if (zk_replicated_id != UUIDHelpers::Nil)
|
|
|
|
zk_replicated_id_to_name.emplace(zk_replicated_id, zk_table.first);
|
|
|
|
}
|
|
|
|
|
2021-02-15 20:00:59 +00:00
|
|
|
/// We will drop or move tables which exist only in local metadata
|
2021-02-10 20:30:40 +00:00
|
|
|
Strings tables_to_detach;
|
2021-02-15 00:04:46 +00:00
|
|
|
std::vector<std::pair<String, String>> replicated_tables_to_rename;
|
2021-02-10 20:30:40 +00:00
|
|
|
size_t total_tables = 0;
|
2021-02-15 00:04:46 +00:00
|
|
|
std::vector<UUID> replicated_ids;
|
2021-04-10 23:33:54 +00:00
|
|
|
for (auto existing_tables_it = getTablesIterator(getContext(), {}); existing_tables_it->isValid();
|
|
|
|
existing_tables_it->next(), ++total_tables)
|
2021-02-10 20:30:40 +00:00
|
|
|
{
|
|
|
|
String name = existing_tables_it->name();
|
2021-02-15 00:04:46 +00:00
|
|
|
UUID local_replicated_id = UUIDHelpers::Nil;
|
|
|
|
if (existing_tables_it->table()->supportsReplication())
|
2021-02-10 20:30:40 +00:00
|
|
|
{
|
2021-02-15 20:00:59 +00:00
|
|
|
/// Check if replicated tables have the same UUID
|
2021-02-15 00:04:46 +00:00
|
|
|
local_replicated_id = existing_tables_it->table()->getStorageID().uuid;
|
|
|
|
auto it = zk_replicated_id_to_name.find(local_replicated_id);
|
|
|
|
if (it != zk_replicated_id_to_name.end())
|
2021-02-10 20:30:40 +00:00
|
|
|
{
|
2021-02-15 00:04:46 +00:00
|
|
|
if (name != it->second)
|
2021-02-15 20:00:59 +00:00
|
|
|
{
|
|
|
|
/// Need just update table name
|
2021-02-15 00:04:46 +00:00
|
|
|
replicated_tables_to_rename.emplace_back(name, it->second);
|
2021-02-15 20:00:59 +00:00
|
|
|
}
|
2021-02-15 00:04:46 +00:00
|
|
|
continue;
|
2021-02-10 20:30:40 +00:00
|
|
|
}
|
2021-02-15 00:04:46 +00:00
|
|
|
}
|
2021-02-10 20:30:40 +00:00
|
|
|
|
2021-02-15 00:04:46 +00:00
|
|
|
auto in_zk = table_name_to_metadata.find(name);
|
|
|
|
if (in_zk == table_name_to_metadata.end() || in_zk->second != readMetadataFile(name))
|
|
|
|
{
|
2021-02-15 20:00:59 +00:00
|
|
|
/// Local table does not exits in ZooKeeper or has different metadata
|
|
|
|
tables_to_detach.emplace_back(std::move(name));
|
2021-02-10 20:30:40 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-02-11 22:23:40 +00:00
|
|
|
String db_name = getDatabaseName();
|
|
|
|
String to_db_name = getDatabaseName() + BROKEN_TABLES_SUFFIX;
|
2021-02-19 23:41:58 +00:00
|
|
|
if (total_tables * db_settings.max_broken_tables_ratio < tables_to_detach.size())
|
2021-02-11 22:23:40 +00:00
|
|
|
throw Exception(ErrorCodes::DATABASE_REPLICATION_FAILED, "Too many tables to recreate: {} of {}", tables_to_detach.size(), total_tables);
|
2021-02-10 20:30:40 +00:00
|
|
|
else if (!tables_to_detach.empty())
|
2021-02-11 22:23:40 +00:00
|
|
|
{
|
|
|
|
LOG_WARNING(log, "Will recreate {} broken tables to recover replica", tables_to_detach.size());
|
|
|
|
/// It's too dangerous to automatically drop tables, so we will move them to special database.
|
|
|
|
/// We use Ordinary engine for destination database, because it's the only way to discard table UUID
|
|
|
|
/// and make possible creation of new table with the same UUID.
|
|
|
|
String query = fmt::format("CREATE DATABASE IF NOT EXISTS {} ENGINE=Ordinary", backQuoteIfNeed(to_db_name));
|
2021-04-10 23:33:54 +00:00
|
|
|
auto query_context = Context::createCopy(getContext());
|
2021-02-11 22:23:40 +00:00
|
|
|
executeQuery(query, query_context, true);
|
|
|
|
}
|
2021-02-10 20:30:40 +00:00
|
|
|
|
2021-02-11 22:23:40 +00:00
|
|
|
size_t moved_tables = 0;
|
|
|
|
std::vector<UUID> dropped_tables;
|
2021-04-26 10:20:57 +00:00
|
|
|
size_t dropped_dictionaries = 0;
|
|
|
|
|
2021-02-10 20:30:40 +00:00
|
|
|
for (const auto & table_name : tables_to_detach)
|
|
|
|
{
|
2021-02-11 22:23:40 +00:00
|
|
|
DDLGuardPtr table_guard = DatabaseCatalog::instance().getDDLGuard(db_name, table_name);
|
|
|
|
if (getDatabaseName() != db_name)
|
|
|
|
throw Exception(ErrorCodes::UNKNOWN_DATABASE, "Database was renamed, will retry");
|
2021-02-10 20:30:40 +00:00
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
auto table = tryGetTable(table_name, getContext());
|
2021-04-21 13:45:13 +00:00
|
|
|
|
|
|
|
if (!table->storesDataOnDisk())
|
2021-02-11 22:23:40 +00:00
|
|
|
{
|
|
|
|
LOG_DEBUG(log, "Will DROP TABLE {}, because it does not store data on disk and can be safely dropped", backQuoteIfNeed(table_name));
|
|
|
|
dropped_tables.push_back(tryGetTableUUID(table_name));
|
2021-04-26 10:20:57 +00:00
|
|
|
dropped_dictionaries += table->isDictionary();
|
|
|
|
|
2021-05-13 07:03:00 +00:00
|
|
|
table->flushAndShutdown();
|
2021-04-10 23:33:54 +00:00
|
|
|
DatabaseAtomic::dropTable(getContext(), table_name, true);
|
2021-02-10 20:30:40 +00:00
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
2021-02-15 20:00:59 +00:00
|
|
|
/// Table probably stores some data. Let's move it to another database.
|
|
|
|
String to_name = fmt::format("{}_{}_{}", table_name, max_log_ptr, thread_local_rng() % 1000);
|
2021-02-11 22:23:40 +00:00
|
|
|
LOG_DEBUG(log, "Will RENAME TABLE {} TO {}.{}", backQuoteIfNeed(table_name), backQuoteIfNeed(to_db_name), backQuoteIfNeed(to_name));
|
2021-02-15 20:00:59 +00:00
|
|
|
assert(db_name < to_db_name);
|
|
|
|
DDLGuardPtr to_table_guard = DatabaseCatalog::instance().getDDLGuard(to_db_name, to_name);
|
2021-02-11 22:23:40 +00:00
|
|
|
auto to_db_ptr = DatabaseCatalog::instance().getDatabase(to_db_name);
|
2021-04-10 23:33:54 +00:00
|
|
|
DatabaseAtomic::renameTable(getContext(), table_name, *to_db_ptr, to_name, false, false);
|
2021-02-11 22:23:40 +00:00
|
|
|
++moved_tables;
|
2021-02-10 20:30:40 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2021-02-11 22:23:40 +00:00
|
|
|
if (!tables_to_detach.empty())
|
2021-04-26 10:20:57 +00:00
|
|
|
LOG_WARNING(log, "Cleaned {} outdated objects: dropped {} dictionaries and {} tables, moved {} tables",
|
|
|
|
tables_to_detach.size(), dropped_dictionaries, dropped_tables.size() - dropped_dictionaries, moved_tables);
|
2021-02-11 22:23:40 +00:00
|
|
|
|
2021-02-15 00:04:46 +00:00
|
|
|
/// Now database is cleared from outdated tables, let's rename ReplicatedMergeTree tables to actual names
|
|
|
|
for (const auto & old_to_new : replicated_tables_to_rename)
|
|
|
|
{
|
|
|
|
const String & from = old_to_new.first;
|
|
|
|
const String & to = old_to_new.second;
|
|
|
|
|
|
|
|
LOG_DEBUG(log, "Will RENAME TABLE {} TO {}", backQuoteIfNeed(from), backQuoteIfNeed(to));
|
|
|
|
/// TODO Maybe we should do it in two steps: rename all tables to temporary names and then rename them to actual names?
|
|
|
|
DDLGuardPtr table_guard = DatabaseCatalog::instance().getDDLGuard(db_name, std::min(from, to));
|
|
|
|
DDLGuardPtr to_table_guard = DatabaseCatalog::instance().getDDLGuard(db_name, std::max(from, to));
|
2021-04-10 23:33:54 +00:00
|
|
|
DatabaseAtomic::renameTable(getContext(), from, *this, to, false, false);
|
2021-02-15 00:04:46 +00:00
|
|
|
}
|
|
|
|
|
2021-02-11 22:23:40 +00:00
|
|
|
for (const auto & id : dropped_tables)
|
|
|
|
DatabaseCatalog::instance().waitTableFinallyDropped(id);
|
|
|
|
|
2021-02-09 15:14:20 +00:00
|
|
|
for (const auto & name_and_meta : table_name_to_metadata)
|
2020-10-22 15:08:00 +00:00
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
if (isTableExist(name_and_meta.first, getContext()))
|
2021-02-10 20:30:40 +00:00
|
|
|
{
|
|
|
|
assert(name_and_meta.second == readMetadataFile(name_and_meta.first));
|
|
|
|
continue;
|
|
|
|
}
|
|
|
|
|
2021-02-09 15:14:20 +00:00
|
|
|
auto query_ast = parseQueryFromMetadataInZooKeeper(name_and_meta.first, name_and_meta.second);
|
2020-05-24 17:13:53 +00:00
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
auto query_context = Context::createCopy(getContext());
|
|
|
|
query_context->makeQueryContext();
|
|
|
|
query_context->getClientInfo().query_kind = ClientInfo::QueryKind::SECONDARY_QUERY;
|
|
|
|
query_context->setCurrentDatabase(database_name);
|
|
|
|
query_context->setCurrentQueryId(""); // generate random query_id
|
2020-07-04 16:32:23 +00:00
|
|
|
|
2020-11-27 14:04:03 +00:00
|
|
|
LOG_INFO(log, "Executing {}", serializeAST(*query_ast));
|
|
|
|
InterpreterCreateQuery(query_ast, query_context).execute();
|
2020-11-19 10:34:45 +00:00
|
|
|
}
|
2020-07-04 16:32:23 +00:00
|
|
|
|
2021-02-09 15:14:20 +00:00
|
|
|
current_zookeeper->set(replica_path + "/log_ptr", toString(max_log_ptr));
|
|
|
|
}
|
|
|
|
|
|
|
|
std::map<String, String> DatabaseReplicated::tryGetConsistentMetadataSnapshot(const ZooKeeperPtr & zookeeper, UInt32 & max_log_ptr)
|
|
|
|
{
|
|
|
|
std::map<String, String> table_name_to_metadata;
|
|
|
|
constexpr int max_retries = 10;
|
|
|
|
int iteration = 0;
|
|
|
|
while (++iteration <= max_retries)
|
|
|
|
{
|
|
|
|
table_name_to_metadata.clear();
|
|
|
|
LOG_DEBUG(log, "Trying to get consistent metadata snapshot for log pointer {}", max_log_ptr);
|
|
|
|
Strings table_names = zookeeper->getChildren(zookeeper_path + "/metadata");
|
|
|
|
|
|
|
|
std::vector<zkutil::ZooKeeper::FutureGet> futures;
|
|
|
|
futures.reserve(table_names.size());
|
|
|
|
for (const auto & table : table_names)
|
|
|
|
futures.emplace_back(zookeeper->asyncTryGet(zookeeper_path + "/metadata/" + table));
|
|
|
|
|
|
|
|
for (size_t i = 0; i < table_names.size(); ++i)
|
|
|
|
{
|
|
|
|
auto res = futures[i].get();
|
|
|
|
if (res.error != Coordination::Error::ZOK)
|
|
|
|
break;
|
2021-02-10 20:30:40 +00:00
|
|
|
table_name_to_metadata.emplace(unescapeForFileName(table_names[i]), res.data);
|
2021-02-09 15:14:20 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
UInt32 new_max_log_ptr = parse<UInt32>(zookeeper->get(zookeeper_path + "/max_log_ptr"));
|
|
|
|
if (new_max_log_ptr == max_log_ptr && table_names.size() == table_name_to_metadata.size())
|
|
|
|
break;
|
|
|
|
|
|
|
|
if (max_log_ptr < new_max_log_ptr)
|
|
|
|
{
|
|
|
|
LOG_DEBUG(log, "Log pointer moved from {} to {}, will retry", max_log_ptr, new_max_log_ptr);
|
|
|
|
max_log_ptr = new_max_log_ptr;
|
|
|
|
}
|
|
|
|
else
|
|
|
|
{
|
|
|
|
assert(max_log_ptr == new_max_log_ptr);
|
|
|
|
assert(table_names.size() != table_name_to_metadata.size());
|
|
|
|
LOG_DEBUG(log, "Cannot get metadata of some tables due to ZooKeeper error, will retry");
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if (max_retries < iteration)
|
|
|
|
throw Exception(ErrorCodes::DATABASE_REPLICATION_FAILED, "Cannot get consistent metadata snapshot");
|
|
|
|
|
|
|
|
LOG_DEBUG(log, "Got consistent metadata snapshot for log pointer {}", max_log_ptr);
|
|
|
|
|
|
|
|
return table_name_to_metadata;
|
2020-06-20 15:39:58 +00:00
|
|
|
}
|
|
|
|
|
2020-11-27 14:04:03 +00:00
|
|
|
ASTPtr DatabaseReplicated::parseQueryFromMetadataInZooKeeper(const String & node_name, const String & query)
|
|
|
|
{
|
|
|
|
ParserCreateQuery parser;
|
|
|
|
String description = "in ZooKeeper " + zookeeper_path + "/metadata/" + node_name;
|
2021-04-10 23:33:54 +00:00
|
|
|
auto ast = parseQuery(parser, query, description, 0, getContext()->getSettingsRef().max_parser_depth);
|
2020-11-27 14:04:03 +00:00
|
|
|
|
|
|
|
auto & create = ast->as<ASTCreateQuery &>();
|
2021-03-08 17:26:38 +00:00
|
|
|
if (create.uuid == UUIDHelpers::Nil || create.table != TABLE_WITH_UUID_NAME_PLACEHOLDER || !create.database.empty())
|
2020-11-27 14:04:03 +00:00
|
|
|
throw Exception(ErrorCodes::LOGICAL_ERROR, "Got unexpected query from {}: {}", node_name, query);
|
|
|
|
|
2021-03-08 17:26:38 +00:00
|
|
|
bool is_materialized_view_with_inner_table = create.is_materialized_view && create.to_table_id.empty();
|
|
|
|
|
2020-11-27 14:04:03 +00:00
|
|
|
create.database = getDatabaseName();
|
|
|
|
create.table = unescapeForFileName(node_name);
|
2021-03-08 17:26:38 +00:00
|
|
|
create.attach = is_materialized_view_with_inner_table;
|
2020-11-27 14:04:03 +00:00
|
|
|
|
|
|
|
return ast;
|
|
|
|
}
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
void DatabaseReplicated::drop(ContextPtr context_)
|
2020-06-20 15:39:58 +00:00
|
|
|
{
|
2020-11-13 18:35:45 +00:00
|
|
|
auto current_zookeeper = getZooKeeper();
|
2021-05-31 13:31:03 +00:00
|
|
|
current_zookeeper->set(replica_path, DROPPED_MARK, -1);
|
|
|
|
createEmptyLogEntry(current_zookeeper);
|
2021-03-09 17:05:24 +00:00
|
|
|
|
2020-06-20 15:39:58 +00:00
|
|
|
DatabaseAtomic::drop(context_);
|
2021-03-09 17:05:24 +00:00
|
|
|
|
2021-02-04 19:41:44 +00:00
|
|
|
current_zookeeper->tryRemoveRecursive(replica_path);
|
2021-02-08 09:14:17 +00:00
|
|
|
/// TODO it may leave garbage in ZooKeeper if the last node lost connection here
|
|
|
|
if (current_zookeeper->tryRemove(zookeeper_path + "/replicas") == Coordination::Error::ZOK)
|
|
|
|
{
|
|
|
|
/// It was the last replica, remove all metadata
|
|
|
|
current_zookeeper->tryRemoveRecursive(zookeeper_path);
|
|
|
|
}
|
2020-05-24 17:13:53 +00:00
|
|
|
}
|
|
|
|
|
2021-02-04 19:41:44 +00:00
|
|
|
void DatabaseReplicated::stopReplication()
|
2020-11-13 18:35:45 +00:00
|
|
|
{
|
|
|
|
if (ddl_worker)
|
|
|
|
ddl_worker->shutdown();
|
2021-02-04 19:41:44 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
void DatabaseReplicated::shutdown()
|
|
|
|
{
|
|
|
|
stopReplication();
|
|
|
|
ddl_worker = nullptr;
|
2020-11-13 18:35:45 +00:00
|
|
|
DatabaseAtomic::shutdown();
|
|
|
|
}
|
|
|
|
|
2020-11-29 11:45:32 +00:00
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
void DatabaseReplicated::dropTable(ContextPtr local_context, const String & table_name, bool no_delay)
|
2020-11-29 11:45:32 +00:00
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
auto txn = local_context->getZooKeeperMetadataTransaction();
|
2021-03-08 17:26:38 +00:00
|
|
|
assert(!ddl_worker->isCurrentlyActive() || txn || startsWith(table_name, ".inner_id."));
|
2021-02-19 23:41:58 +00:00
|
|
|
if (txn && txn->isInitialQuery())
|
2020-11-29 11:45:32 +00:00
|
|
|
{
|
|
|
|
String metadata_zk_path = zookeeper_path + "/metadata/" + escapeForFileName(table_name);
|
2021-02-19 23:41:58 +00:00
|
|
|
txn->addOp(zkutil::makeRemoveRequest(metadata_zk_path, -1));
|
2020-11-29 11:45:32 +00:00
|
|
|
}
|
2021-04-10 23:33:54 +00:00
|
|
|
DatabaseAtomic::dropTable(local_context, table_name, no_delay);
|
2020-11-29 11:45:32 +00:00
|
|
|
}
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
void DatabaseReplicated::renameTable(ContextPtr local_context, const String & table_name, IDatabase & to_database,
|
2020-11-29 11:45:32 +00:00
|
|
|
const String & to_table_name, bool exchange, bool dictionary)
|
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
auto txn = local_context->getZooKeeperMetadataTransaction();
|
2020-11-29 11:45:32 +00:00
|
|
|
assert(txn);
|
|
|
|
|
2021-02-19 23:41:58 +00:00
|
|
|
if (txn->isInitialQuery())
|
2020-11-29 11:45:32 +00:00
|
|
|
{
|
2021-02-08 09:14:17 +00:00
|
|
|
if (this != &to_database)
|
|
|
|
throw Exception(ErrorCodes::NOT_IMPLEMENTED, "Moving tables between databases is not supported for Replicated engine");
|
2021-02-08 19:36:17 +00:00
|
|
|
if (table_name == to_table_name)
|
|
|
|
throw Exception(ErrorCodes::INCORRECT_QUERY, "Cannot rename table to itself");
|
2021-04-10 23:33:54 +00:00
|
|
|
if (!isTableExist(table_name, local_context))
|
2021-02-04 19:41:44 +00:00
|
|
|
throw Exception(ErrorCodes::UNKNOWN_TABLE, "Table {} does not exist", table_name);
|
2021-04-10 23:33:54 +00:00
|
|
|
if (exchange && !to_database.isTableExist(to_table_name, local_context))
|
2021-02-04 19:41:44 +00:00
|
|
|
throw Exception(ErrorCodes::UNKNOWN_TABLE, "Table {} does not exist", to_table_name);
|
|
|
|
|
2021-02-10 20:30:40 +00:00
|
|
|
String statement = readMetadataFile(table_name);
|
2021-02-19 23:41:58 +00:00
|
|
|
String metadata_zk_path = zookeeper_path + "/metadata/" + escapeForFileName(table_name);
|
|
|
|
String metadata_zk_path_to = zookeeper_path + "/metadata/" + escapeForFileName(to_table_name);
|
|
|
|
txn->addOp(zkutil::makeRemoveRequest(metadata_zk_path, -1));
|
2020-11-29 11:45:32 +00:00
|
|
|
if (exchange)
|
|
|
|
{
|
2021-02-11 15:14:38 +00:00
|
|
|
String statement_to = readMetadataFile(to_table_name);
|
2021-02-19 23:41:58 +00:00
|
|
|
txn->addOp(zkutil::makeRemoveRequest(metadata_zk_path_to, -1));
|
|
|
|
txn->addOp(zkutil::makeCreateRequest(metadata_zk_path, statement_to, zkutil::CreateMode::Persistent));
|
2020-11-29 11:45:32 +00:00
|
|
|
}
|
2021-02-19 23:41:58 +00:00
|
|
|
txn->addOp(zkutil::makeCreateRequest(metadata_zk_path_to, statement, zkutil::CreateMode::Persistent));
|
2020-11-29 11:45:32 +00:00
|
|
|
}
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
DatabaseAtomic::renameTable(local_context, table_name, to_database, to_table_name, exchange, dictionary);
|
2020-11-29 11:45:32 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
void DatabaseReplicated::commitCreateTable(const ASTCreateQuery & query, const StoragePtr & table,
|
|
|
|
const String & table_metadata_tmp_path, const String & table_metadata_path,
|
2021-04-10 23:33:54 +00:00
|
|
|
ContextPtr query_context)
|
2020-11-29 11:45:32 +00:00
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
auto txn = query_context->getZooKeeperMetadataTransaction();
|
2020-11-29 11:45:32 +00:00
|
|
|
assert(!ddl_worker->isCurrentlyActive() || txn);
|
2021-02-19 23:41:58 +00:00
|
|
|
if (txn && txn->isInitialQuery())
|
2020-11-29 11:45:32 +00:00
|
|
|
{
|
2021-02-19 23:41:58 +00:00
|
|
|
String metadata_zk_path = zookeeper_path + "/metadata/" + escapeForFileName(query.table);
|
2020-11-29 11:45:32 +00:00
|
|
|
String statement = getObjectDefinitionFromCreateQuery(query.clone());
|
|
|
|
/// zk::multi(...) will throw if `metadata_zk_path` exists
|
2021-02-19 23:41:58 +00:00
|
|
|
txn->addOp(zkutil::makeCreateRequest(metadata_zk_path, statement, zkutil::CreateMode::Persistent));
|
2020-11-29 11:45:32 +00:00
|
|
|
}
|
|
|
|
DatabaseAtomic::commitCreateTable(query, table, table_metadata_tmp_path, table_metadata_path, query_context);
|
|
|
|
}
|
|
|
|
|
|
|
|
void DatabaseReplicated::commitAlterTable(const StorageID & table_id,
|
|
|
|
const String & table_metadata_tmp_path, const String & table_metadata_path,
|
2021-04-10 23:33:54 +00:00
|
|
|
const String & statement, ContextPtr query_context)
|
2020-11-29 11:45:32 +00:00
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
auto txn = query_context->getZooKeeperMetadataTransaction();
|
2021-02-19 23:41:58 +00:00
|
|
|
if (txn && txn->isInitialQuery())
|
2020-11-29 11:45:32 +00:00
|
|
|
{
|
2021-02-19 23:41:58 +00:00
|
|
|
String metadata_zk_path = zookeeper_path + "/metadata/" + escapeForFileName(table_id.table_name);
|
|
|
|
txn->addOp(zkutil::makeSetRequest(metadata_zk_path, statement, -1));
|
2020-11-29 11:45:32 +00:00
|
|
|
}
|
|
|
|
DatabaseAtomic::commitAlterTable(table_id, table_metadata_tmp_path, table_metadata_path, statement, query_context);
|
|
|
|
}
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
void DatabaseReplicated::detachTablePermanently(ContextPtr local_context, const String & table_name)
|
2021-02-10 20:30:40 +00:00
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
auto txn = local_context->getZooKeeperMetadataTransaction();
|
2021-02-10 20:30:40 +00:00
|
|
|
assert(!ddl_worker->isCurrentlyActive() || txn);
|
2021-02-19 23:41:58 +00:00
|
|
|
if (txn && txn->isInitialQuery())
|
2021-02-10 20:30:40 +00:00
|
|
|
{
|
2021-03-09 00:48:37 +00:00
|
|
|
/// We have to remove metadata from zookeeper, because we do not distinguish permanently detached tables
|
|
|
|
/// from attached tables when recovering replica.
|
2021-02-10 20:30:40 +00:00
|
|
|
String metadata_zk_path = zookeeper_path + "/metadata/" + escapeForFileName(table_name);
|
2021-02-19 23:41:58 +00:00
|
|
|
txn->addOp(zkutil::makeRemoveRequest(metadata_zk_path, -1));
|
2021-02-10 20:30:40 +00:00
|
|
|
}
|
2021-04-10 23:33:54 +00:00
|
|
|
DatabaseAtomic::detachTablePermanently(local_context, table_name);
|
2021-02-10 20:30:40 +00:00
|
|
|
}
|
|
|
|
|
2021-04-10 23:33:54 +00:00
|
|
|
void DatabaseReplicated::removeDetachedPermanentlyFlag(ContextPtr local_context, const String & table_name, const String & table_metadata_path, bool attach) const
|
2021-03-09 00:48:37 +00:00
|
|
|
{
|
2021-04-10 23:33:54 +00:00
|
|
|
auto txn = local_context->getZooKeeperMetadataTransaction();
|
2021-03-09 00:48:37 +00:00
|
|
|
assert(!ddl_worker->isCurrentlyActive() || txn);
|
2021-03-09 10:24:08 +00:00
|
|
|
if (txn && txn->isInitialQuery() && attach)
|
2021-03-09 00:48:37 +00:00
|
|
|
{
|
|
|
|
String metadata_zk_path = zookeeper_path + "/metadata/" + escapeForFileName(table_name);
|
|
|
|
String statement = readMetadataFile(table_name);
|
|
|
|
txn->addOp(zkutil::makeCreateRequest(metadata_zk_path, statement, zkutil::CreateMode::Persistent));
|
|
|
|
}
|
2021-04-10 23:33:54 +00:00
|
|
|
DatabaseAtomic::removeDetachedPermanentlyFlag(local_context, table_name, table_metadata_path, attach);
|
2021-03-09 00:48:37 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
|
2021-02-10 20:30:40 +00:00
|
|
|
String DatabaseReplicated::readMetadataFile(const String & table_name) const
|
|
|
|
{
|
|
|
|
String statement;
|
|
|
|
ReadBufferFromFile in(getObjectMetadataPath(table_name), 4096);
|
|
|
|
readStringUntilEOF(statement, in);
|
|
|
|
return statement;
|
|
|
|
}
|
|
|
|
|
2020-04-05 12:18:51 +00:00
|
|
|
}
|