ClickHouse/src/Storages/ExternalDataSourceConfiguration.cpp

286 lines
13 KiB
C++
Raw Normal View History

2021-09-01 17:59:11 +00:00
#include "ExternalDataSourceConfiguration.h"
#include <Interpreters/Context.h>
#include <Interpreters/evaluateConstantExpression.h>
#include <Parsers/ASTIdentifier.h>
#include <Parsers/ASTFunction.h>
#include <Parsers/ASTLiteral.h>
#include <Poco/Util/AbstractConfiguration.h>
2021-09-01 23:17:15 +00:00
#include <IO/WriteBufferFromString.h>
2021-09-01 17:59:11 +00:00
namespace DB
{
namespace ErrorCodes
{
extern const int BAD_ARGUMENTS;
}
2021-09-01 23:17:15 +00:00
String ExternalDataSourceConfiguration::toString() const
{
WriteBufferFromOwnString configuration_info;
2021-09-07 11:17:25 +00:00
configuration_info << "username: " << username << "\t";
if (addresses.empty())
{
configuration_info << "host: " << host << "\t";
configuration_info << "port: " << port << "\t";
}
else
{
for (const auto & [replica_host, replica_port] : addresses)
{
configuration_info << "host: " << replica_host << "\t";
configuration_info << "port: " << replica_port << "\t";
}
}
2021-09-01 23:17:15 +00:00
return configuration_info.str();
}
2021-09-01 17:59:11 +00:00
std::tuple<ExternalDataSourceConfiguration, EngineArgs, bool>
2021-09-07 11:17:25 +00:00
getExternalDataSourceConfiguration(const ASTs & args, ContextPtr context, bool is_database_engine)
2021-09-01 17:59:11 +00:00
{
2021-09-07 11:17:25 +00:00
if (args.empty())
throw Exception(ErrorCodes::BAD_ARGUMENTS, "External data source must have arguments");
2021-09-01 17:59:11 +00:00
ExternalDataSourceConfiguration configuration;
EngineArgs non_common_args;
if (const auto * collection = typeid_cast<const ASTIdentifier *>(args[0].get()))
{
const auto & config = context->getConfigRef();
const auto & config_prefix = fmt::format("named_collections.{}", collection->name());
if (!config.has(config_prefix))
throw Exception(ErrorCodes::BAD_ARGUMENTS, "There is no collection named `{}` in config", collection->name());
configuration.host = config.getString(config_prefix + ".host", "");
configuration.port = config.getInt(config_prefix + ".port", 0);
configuration.username = config.getString(config_prefix + ".user", "");
configuration.password = config.getString(config_prefix + ".password", "");
configuration.database = config.getString(config_prefix + ".database", "");
configuration.table = config.getString(config_prefix + ".table", "");
2021-09-01 23:17:15 +00:00
configuration.schema = config.getString(config_prefix + ".schema", "");
2021-09-01 17:59:11 +00:00
if ((args.size() == 1) && (configuration.host.empty() || configuration.port == 0
|| configuration.username.empty() || configuration.password.empty()
2021-09-01 23:17:15 +00:00
|| configuration.database.empty() || (configuration.table.empty() && !is_database_engine)))
2021-09-01 17:59:11 +00:00
{
throw Exception(ErrorCodes::BAD_ARGUMENTS,
"Named collection of connection parameters is missing some of the parameters and no key-value arguments are added");
}
for (size_t i = 1; i < args.size(); ++i)
{
if (const auto * ast_function = typeid_cast<const ASTFunction *>(args[i].get()))
{
const auto * args_expr = assert_cast<const ASTExpressionList *>(ast_function->arguments.get());
auto function_args = args_expr->children;
if (function_args.size() != 2)
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Expected key-value defined argument");
auto arg_name = function_args[0]->as<ASTIdentifier>()->name();
auto arg_value = evaluateConstantExpressionOrIdentifierAsLiteral(function_args[1], context)->as<ASTLiteral>()->value;
if (arg_name == "host")
configuration.host = arg_value.safeGet<String>();
else if (arg_name == "port")
configuration.port = arg_value.safeGet<UInt64>();
else if (arg_name == "user")
configuration.username = arg_value.safeGet<String>();
else if (arg_name == "password")
configuration.password = arg_value.safeGet<String>();
else if (arg_name == "database")
configuration.database = arg_value.safeGet<String>();
else if (arg_name == "table")
configuration.table = arg_value.safeGet<String>();
2021-09-01 23:17:15 +00:00
else if (arg_name == "schema")
configuration.schema = arg_value.safeGet<String>();
2021-09-01 17:59:11 +00:00
else
non_common_args.emplace_back(std::make_pair(arg_name, arg_value));
}
else
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Expected key-value defined argument");
}
}
return std::make_tuple(configuration, non_common_args, true);
}
return std::make_tuple(configuration, non_common_args, false);
}
2021-09-01 23:17:15 +00:00
2021-09-03 11:16:32 +00:00
ExternalDataSourceConfiguration getExternalDataSourceConfiguration(
2021-09-01 23:17:15 +00:00
const Poco::Util::AbstractConfiguration & dict_config, const String & dict_config_prefix, ContextPtr context)
{
ExternalDataSourceConfiguration configuration;
auto collection_name = dict_config.getString(dict_config_prefix + ".name", "");
if (!collection_name.empty())
{
const auto & config = context->getConfigRef();
const auto & config_prefix = fmt::format("named_collections.{}", collection_name);
if (!config.has(config_prefix))
throw Exception(ErrorCodes::BAD_ARGUMENTS, "There is no collection named `{}` in config", collection_name);
configuration.host = dict_config.getString(dict_config_prefix + ".host", config.getString(config_prefix + ".host", ""));
configuration.port = dict_config.getInt(dict_config_prefix + ".port", config.getUInt(config_prefix + ".port", 0));
configuration.username = dict_config.getString(dict_config_prefix + ".user", config.getString(config_prefix + ".user", ""));
configuration.password = dict_config.getString(dict_config_prefix + ".password", config.getString(config_prefix + ".password", ""));
configuration.database = dict_config.getString(dict_config_prefix + ".db", config.getString(config_prefix + ".database", ""));
configuration.table = dict_config.getString(dict_config_prefix + ".table", config.getString(config_prefix + ".table", ""));
configuration.schema = dict_config.getString(dict_config_prefix + ".schema", config.getString(config_prefix + ".schema", ""));
if (configuration.host.empty() || configuration.port == 0 || configuration.username.empty() || configuration.password.empty()
|| configuration.database.empty() || configuration.table.empty())
{
throw Exception(ErrorCodes::BAD_ARGUMENTS,
"Named collection of connection parameters is missing some of the parameters and dictionary parameters are added");
}
}
else
{
configuration.host = dict_config.getString(dict_config_prefix + ".host", "");
configuration.port = dict_config.getUInt(dict_config_prefix + ".port", 0);
configuration.username = dict_config.getString(dict_config_prefix + ".user", "");
configuration.password = dict_config.getString(dict_config_prefix + ".password", "");
configuration.database = dict_config.getString(dict_config_prefix + ".db", "");
configuration.table = dict_config.getString(fmt::format("{}.table", dict_config_prefix), "");
configuration.schema = dict_config.getString(fmt::format("{}.schema", dict_config_prefix), "");
}
return configuration;
}
2021-09-03 11:16:32 +00:00
ExternalDataSourcesByPriority getExternalDataSourceConfigurationByPriority(
2021-09-01 23:17:15 +00:00
const Poco::Util::AbstractConfiguration & dict_config, const String & dict_config_prefix, ContextPtr context)
{
2021-09-03 11:16:32 +00:00
auto common_configuration = getExternalDataSourceConfiguration(dict_config, dict_config_prefix, context);
2021-09-02 13:01:26 +00:00
ExternalDataSourcesByPriority configuration
{
.database = common_configuration.database,
.table = common_configuration.table,
2021-09-10 11:11:52 +00:00
.schema = common_configuration.schema,
.replicas_configurations = {}
2021-09-02 13:01:26 +00:00
};
2021-09-01 23:17:15 +00:00
if (dict_config.has(dict_config_prefix + ".replica"))
{
Poco::Util::AbstractConfiguration::Keys config_keys;
dict_config.keys(dict_config_prefix, config_keys);
for (const auto & config_key : config_keys)
{
if (config_key.starts_with("replica"))
{
ExternalDataSourceConfiguration replica_configuration(common_configuration);
String replica_name = dict_config_prefix + "." + config_key;
size_t priority = dict_config.getInt(replica_name + ".priority", 0);
replica_configuration.host = dict_config.getString(replica_name + ".host", common_configuration.host);
replica_configuration.port = dict_config.getUInt(replica_name + ".port", common_configuration.port);
replica_configuration.username = dict_config.getString(replica_name + ".user", common_configuration.username);
replica_configuration.password = dict_config.getString(replica_name + ".password", common_configuration.password);
if (replica_configuration.host.empty() || replica_configuration.port == 0
|| replica_configuration.username.empty() || replica_configuration.password.empty())
{
throw Exception(ErrorCodes::BAD_ARGUMENTS,
"Named collection of connection parameters is missing some of the parameters and no other dictionary parameters are added");
}
2021-09-02 13:01:26 +00:00
configuration.replicas_configurations[priority].emplace_back(replica_configuration);
2021-09-01 23:17:15 +00:00
}
}
}
else
{
2021-09-02 13:01:26 +00:00
configuration.replicas_configurations[0].emplace_back(common_configuration);
2021-09-01 23:17:15 +00:00
}
2021-09-02 13:01:26 +00:00
return configuration;
2021-09-01 23:17:15 +00:00
}
2021-09-07 11:17:25 +00:00
std::tuple<URLBasedDataSourceConfiguration, EngineArgs, bool>
getURLBasedDataSourceConfiguration(const ASTs & args, ContextPtr context)
{
if (args.empty())
throw Exception(ErrorCodes::BAD_ARGUMENTS, "External data source must have arguments");
URLBasedDataSourceConfiguration configuration;
EngineArgs non_common_args;
if (const auto * collection = typeid_cast<const ASTIdentifier *>(args[0].get()))
{
const auto & config = context->getConfigRef();
auto config_prefix = fmt::format("named_collections.{}", collection->name());
if (!config.has(config_prefix))
throw Exception(ErrorCodes::BAD_ARGUMENTS, "There is no collection named `{}` in config", collection->name());
Poco::Util::AbstractConfiguration::Keys keys;
config.keys(config_prefix, keys);
for (const auto & key : keys)
{
if (key == "url")
2021-09-09 09:18:08 +00:00
{
2021-09-07 11:17:25 +00:00
configuration.url = config.getString(config_prefix + ".url", "");
2021-09-09 09:18:08 +00:00
}
else if (key == "headers")
2021-09-07 11:17:25 +00:00
{
Poco::Util::AbstractConfiguration::Keys header_keys;
2021-09-15 18:11:49 +00:00
config.keys(config_prefix + ".headers", header_keys);
2021-09-07 11:17:25 +00:00
for (const auto & header : header_keys)
{
const auto header_prefix = config_prefix + ".headers." + header;
configuration.headers.emplace_back(std::make_pair(config.getString(header_prefix + ".name"), config.getString(header_prefix + ".value")));
}
}
else
non_common_args.emplace_back(std::make_pair(key, config.getString(config_prefix + '.' + key)));
}
for (size_t i = 1; i < args.size(); ++i)
{
if (const auto * ast_function = typeid_cast<const ASTFunction *>(args[i].get()))
{
const auto * args_expr = assert_cast<const ASTExpressionList *>(ast_function->arguments.get());
auto function_args = args_expr->children;
if (function_args.size() != 2)
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Expected key-value defined argument");
auto arg_name = function_args[0]->as<ASTIdentifier>()->name();
auto arg_value = evaluateConstantExpressionOrIdentifierAsLiteral(function_args[1], context)->as<ASTLiteral>()->value;
if (arg_name == "url")
configuration.url = arg_value.safeGet<String>();
else if (arg_name == "format")
configuration.format = arg_value.safeGet<String>();
else if (arg_name == "compression_method")
configuration.compression_method = arg_value.safeGet<String>();
else if (arg_name == "structure")
configuration.structure = arg_value.safeGet<String>();
else
non_common_args.emplace_back(std::make_pair(arg_name, arg_value));
}
else
{
throw Exception(ErrorCodes::BAD_ARGUMENTS, "Expected key-value defined argument");
}
}
if (configuration.url.empty() || configuration.format.empty())
throw Exception(ErrorCodes::BAD_ARGUMENTS,
2021-09-09 09:18:08 +00:00
"Storage requires {}", configuration.url.empty() ? "url" : "format");
2021-09-07 11:17:25 +00:00
return std::make_tuple(configuration, non_common_args, true);
}
return std::make_tuple(configuration, non_common_args, false);
}
2021-09-01 17:59:11 +00:00
}